Context Navigation

source: trunk/lib/nanownlib/storage.py @ 21

Last change on this file since 21 was 16, checked in by tim, 9 years ago
.
File size: 11.3 KB

Rev	Line
[4]	1	#-- mode: Python;--
	2
	3	import sys
	4	import os
	5	import uuid
[10]	6	import random
[4]	7	import threading
	8	import sqlite3
[16]	9	try:
	10	import numpy
	11	except:
	12	sys.stderr.write('ERROR: Could not import numpy module. Ensure it is installed.\n')
	13	sys.stderr.write(' Under Debian, the package name is "python3-numpy"\n.')
	14	sys.exit(1)
[4]	15
[10]	16	# Don't trust numpy's seeding
	17	numpy.random.seed(random.SystemRandom().randint(0,2**32-1))
[8]	18
[4]	19	def _newid():
	20	return uuid.uuid4().hex
	21
	22
	23	class db(threading.local):
	24	conn = None
	25	cursor = None
[7]	26	_population_sizes = None
[8]	27	_population_cache = None
[10]	28	_offset_cache = None
	29	_cur_offsets = None
[7]	30
[4]	31	def __init__(self, path):
	32	exists = os.path.exists(path)
	33	self.conn = sqlite3.connect(path)
	34	self.conn.execute("PRAGMA foreign_keys = ON;")
	35	self.conn.row_factory = sqlite3.Row
[7]	36	self._population_sizes = {}
[8]	37	self._population_cache = {}
[10]	38	self._offset_cache = {}
	39	self._cur_offsets = {}
[4]	40
	41	if not exists:
	42	self.conn.execute(
	43	"""CREATE TABLE meta (id BLOB PRIMARY KEY,
	44	tcpts_mean REAL,
	45	tcpts_stddev REAL,
[16]	46	tcpts_slopes TEXT,
	47	unusual_case TEXT,
	48	greater INTEGER)
[4]	49	""")
	50
	51	self.conn.execute(
	52	"""CREATE TABLE probes (id BLOB PRIMARY KEY,
	53	sample INTEGER,
	54	test_case TEXT,
	55	type TEXT,
	56	tc_order INTEGER,
	57	time_of_day INTEGER,
	58	local_port INTEGER,
	59	reported INTEGER,
	60	userspace_rtt INTEGER,
	61	UNIQUE (sample, test_case))
	62	""")
	63
	64	self.conn.execute(
	65	"""CREATE TABLE packets (id BLOB PRIMARY KEY,
	66	probe_id REFERENCES probes(id) ON DELETE CASCADE,
	67	sent INTEGER,
	68	observed INTEGER,
	69	tsval INTEGER,
	70	payload_len INTEGER,
	71	tcpseq INTEGER,
	72	tcpack INTEGER)
	73	""")
	74
	75	self.conn.execute(
	76	"""CREATE TABLE analysis (id BLOB PRIMARY KEY,
	77	probe_id UNIQUE REFERENCES probes(id) ON DELETE CASCADE,
	78	suspect TEXT,
	79	packet_rtt INTEGER,
	80	tsval_rtt INTEGER)
	81	""")
	82
	83	self.conn.execute(
	84	"""CREATE TABLE trim_analysis (id BLOB PRIMARY KEY,
	85	probe_id REFERENCES probes(id) ON DELETE CASCADE,
	86	suspect TEXT,
	87	packet_rtt INTEGER,
	88	tsval_rtt INTEGER,
	89	sent_trimmed INTEGER,
	90	rcvd_trimmed INTEGER)
	91	""")
	92
	93	self.conn.execute(
	94	"""CREATE TABLE classifier_results (id BLOB PRIMARY KEY,
[10]	95	classifier TEXT,
	96	trial_type TEXT,
	97	num_observations INTEGER,
	98	num_trials INTEGER,
[4]	99	params TEXT,
	100	false_positives REAL,
	101	false_negatives REAL)
	102	""")
	103
	104	def __del__(self):
	105	if self.conn:
	106	self.conn.commit()
	107	self.conn.close()
	108
[8]	109
[7]	110	def populationSize(self, probe_type):
	111	if probe_type in self._population_sizes:
	112	return self._population_sizes[probe_type]
	113
	114	try:
	115	cursor = self.conn.cursor()
	116	cursor.execute("SELECT max(c) FROM (SELECT count(sample) c FROM probes WHERE type=? GROUP BY test_case)", (probe_type,))
	117	self._population_sizes[probe_type] = cursor.fetchone()[0]
	118	return self._population_sizes[probe_type]
	119	except Exception as e:
	120	print(e)
	121	return 0
[8]	122
	123
[11]	124	def subseries(self, probe_type, unusual_case, size=None, offset=None):
	125	cache_key = (probe_type,unusual_case)
[10]	126	if cache_key not in self._population_cache:
[8]	127	query="""
[11]	128	SELECT packet_rtt AS unusual_packet,
	129	(SELECT avg(packet_rtt) FROM probes,analysis
	130	WHERE analysis.probe_id=probes.id AND probes.test_case!=:unusual_case AND probes.type=:probe_type AND sample=u.sample) AS other_packet,
	131
	132	tsval_rtt AS unusual_tsval,
	133	(SELECT avg(tsval_rtt) FROM probes,analysis
	134	WHERE analysis.probe_id=probes.id AND probes.test_case!=:unusual_case AND probes.type=:probe_type AND sample=u.sample) AS other_tsval,
	135
	136	reported AS unusual_reported,
	137	(SELECT avg(reported) FROM probes,analysis
	138	WHERE analysis.probe_id=probes.id AND probes.test_case!=:unusual_case AND probes.type=:probe_type AND sample=u.sample) AS other_reported
	139
	140	FROM (SELECT probes.sample,packet_rtt,tsval_rtt,reported FROM probes,analysis
[8]	141	WHERE analysis.probe_id=probes.id AND probes.test_case =:unusual_case AND probes.type=:probe_type) u
[11]	142	"""
[8]	143
	144	params = {"probe_type":probe_type, "unusual_case":unusual_case}
	145	cursor = self.conn.cursor()
	146	cursor.execute(query, params)
[10]	147	p = [dict(row) for row in cursor.fetchall()]
	148	self._population_cache[cache_key] = p
	149	self._offset_cache[cache_key] = tuple(numpy.random.random_integers(0,len(p)-1, len(p)/5))
	150	self._cur_offsets[cache_key] = 0
[8]	151
[10]	152	population = self._population_cache[cache_key]
[8]	153
	154	if size == None or size > len(population):
	155	size = len(population)
	156	if offset == None or offset >= len(population) or offset < 0:
[10]	157	offset = self._offset_cache[cache_key][self._cur_offsets[cache_key]]
	158	self._cur_offsets[cache_key] = (offset + 1) % len(self._offset_cache[cache_key])
	159
[9]	160	try:
[10]	161	offset = int(offset)
	162	size = int(size)
[9]	163	except Exception as e:
	164	print(e, offset, size)
[10]	165	return None
	166
	167	ret_val = population[offset:offset+size]
[8]	168	if len(ret_val) < size:
	169	ret_val += population[0:size-len(ret_val)]
[7]	170
[8]	171	return ret_val
[10]	172
	173
	174	def resetOffsets(self):
	175	for k in self._cur_offsets.keys():
	176	self._cur_offsets[k] = 0
[8]	177
[10]	178
[8]	179	def clearCache(self):
	180	self._population_cache = {}
[10]	181	self._offset_cache = {}
	182	self._cur_offsets = {}
[8]	183
	184
[4]	185	def _insert(self, table, row):
	186	rid = _newid()
	187	keys = row.keys()
	188	columns = ','.join(keys)
	189	placeholders = ':'+', :'.join(keys)
	190	query = "INSERT INTO %s (id,%s) VALUES ('%s',%s)" % (table, columns, rid, placeholders)
	191	#print(row)
	192	self.conn.execute(query, row)
	193	return rid
	194
	195	def addMeta(self, meta):
	196	ret_val = self._insert('meta', meta)
	197	self.conn.commit()
	198	return ret_val
	199
	200	def addProbes(self, p):
	201	return [self._insert('probes', row) for row in p]
	202
	203	def addPackets(self, pkts, window_size):
	204	query = ("INSERT INTO packets (id,probe_id,sent,observed,tsval,payload_len,tcpseq,tcpack)"
[16]	205	" VALUES(hex(randomblob(16)),"
[4]	206	"(SELECT id FROM probes WHERE local_port=:local_port AND :observed>time_of_day"
	207	" AND :observed<time_of_day+userspace_rtt+%d"
	208	" ORDER BY time_of_day ASC LIMIT 1),"
	209	":sent,:observed,:tsval,:payload_len,:tcpseq,:tcpack)") % window_size
	210	self.conn.execute("PRAGMA foreign_keys = OFF;")
	211	self.conn.execute("CREATE INDEX IF NOT EXISTS probes_port ON probes (local_port)")
	212	cursor = self.conn.cursor()
	213	#print(query, list(pkts)[0:3])
	214	cursor.executemany(query, pkts)
	215	self.conn.commit()
	216	self.conn.execute("PRAGMA foreign_keys = ON;")
	217
	218	def addAnalyses(self, analyses):
	219	return [self._insert('analysis', row) for row in analyses]
	220
	221	def addTrimAnalyses(self, analyses):
	222	return [self._insert('trim_analysis', row) for row in analyses]
	223
[11]	224	def addClassifierResult(self, results):
[4]	225	ret_val = self._insert('classifier_results', results)
	226	self.conn.commit()
	227	return ret_val
[10]	228
[11]	229	def fetchClassifierResult(self, classifier, trial_type, num_observations, params=None):
[10]	230	query = """
	231	SELECT * FROM classifier_results
[11]	232	WHERE classifier=:classifier
	233	AND trial_type=:trial_type
	234	AND num_observations=:num_observations"""
	235	if params != None:
	236	query += """
	237	AND params=:params"""
	238	query += """
	239	ORDER BY false_positives+false_negatives
	240	LIMIT 1
[10]	241	"""
[11]	242
	243	qparams = {'classifier':classifier, 'trial_type':trial_type,
	244	'num_observations':num_observations,'params':params}
[10]	245	cursor = self.conn.cursor()
[11]	246	cursor.execute(query, qparams)
[10]	247	ret_val = cursor.fetchone()
	248	if ret_val != None:
	249	ret_val = dict(ret_val)
	250	return ret_val
[11]	251
[10]	252	def deleteClassifierResults(self, classifier, trial_type, num_observations=None):
	253	params = {"classifier":classifier,"trial_type":trial_type,"num_observations":num_observations}
	254	query = """
	255	DELETE FROM classifier_results
	256	WHERE classifier=:classifier AND trial_type=:trial_type
	257	"""
	258	if num_observations != None:
	259	query += " AND num_observations=:num_observations"
	260
	261	self.conn.execute(query, params)
	262	self.conn.commit()
[16]	263
	264	def setUnusualCase(self, unusual_case, greater):
	265	query = """SELECT * FROM meta LIMIT 1"""
	266	cursor = self.conn.cursor()
	267	cursor.execute(query)
	268	row = cursor.fetchone()
	269	if row == None:
	270	params = {"id":_newid()}
	271	else:
	272	params = dict(row)
	273
	274	params["unusual_case"]=unusual_case
	275	params["greater"]=greater
[10]	276
[16]	277	keys = params.keys()
	278	columns = ','.join(keys)
	279	placeholders = ':'+', :'.join(keys)
	280
	281	query = """INSERT OR REPLACE INTO meta (%s) VALUES (%s)""" % (columns, placeholders)
	282	cursor.execute(query, params)
	283
	284
	285	def getUnusualCase(self):
	286	query = """SELECT unusual_case,greater FROM meta LIMIT 1"""
	287	cursor = self.conn.cursor()
	288	cursor.execute(query)
	289	row = cursor.fetchone()
	290	if row == None or row[0] == None or row[1] == None:
	291	return None
	292	else:
	293	return tuple(row)

Note: See TracBrowser for help on using the repository browser.

Download in other formats: