You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

555 lines
21 KiB

  1. # -*- coding: utf-8 -*-
  2. """NilmDB
  3. Object that represents a NILM database file.
  4. Manages both the SQL database and the table storage backend.
  5. """
  6. # Need absolute_import so that "import nilmdb" won't pull in
  7. # nilmdb.py, but will pull the parent nilmdb module instead.
  8. from __future__ import absolute_import
  9. import nilmdb
  10. from nilmdb.utils.printf import *
  11. from nilmdb.server.interval import (Interval, DBInterval,
  12. IntervalSet, IntervalError)
  13. from nilmdb.server import bulkdata
  14. from nilmdb.server.errors import NilmDBError, StreamError, OverlapError
  15. import sqlite3
  16. import os
  17. import errno
  18. import bisect
  19. # Note about performance and transactions:
  20. #
  21. # Committing a transaction in the default sync mode (PRAGMA synchronous=FULL)
  22. # takes about 125msec. sqlite3 will commit transactions at 3 times:
  23. # 1: explicit con.commit()
  24. # 2: between a series of DML commands and non-DML commands, e.g.
  25. # after a series of INSERT, SELECT, but before a CREATE TABLE or PRAGMA.
  26. # 3: at the end of an explicit transaction, e.g. "with self.con as con:"
  27. #
  28. # To speed up testing, or if this transaction speed becomes an issue,
  29. # the sync=False option to NilmDB.__init__ will set PRAGMA synchronous=OFF.
  30. # Don't touch old entries -- just add new ones.
  31. _sql_schema_updates = {
  32. 0: """
  33. -- All streams
  34. CREATE TABLE streams(
  35. id INTEGER PRIMARY KEY, -- stream ID
  36. path TEXT UNIQUE NOT NULL, -- path, e.g. '/newton/prep'
  37. layout TEXT NOT NULL -- layout name, e.g. float32_8
  38. );
  39. -- Individual timestamped ranges in those streams.
  40. -- For a given start_time and end_time, this tells us that the
  41. -- data is stored between start_pos and end_pos.
  42. -- Times are stored as μs since Unix epoch
  43. -- Positions are opaque: PyTables rows, file offsets, etc.
  44. --
  45. -- Note: end_pos points to the row _after_ end_time, so end_pos-1
  46. -- is the last valid row.
  47. CREATE TABLE ranges(
  48. stream_id INTEGER NOT NULL,
  49. start_time INTEGER NOT NULL,
  50. end_time INTEGER NOT NULL,
  51. start_pos INTEGER NOT NULL,
  52. end_pos INTEGER NOT NULL
  53. );
  54. CREATE INDEX _ranges_index ON ranges (stream_id, start_time, end_time);
  55. """,
  56. 1: """
  57. -- Generic dictionary-type metadata that can be associated with a stream
  58. CREATE TABLE metadata(
  59. stream_id INTEGER NOT NULL,
  60. key TEXT NOT NULL,
  61. value TEXT
  62. );
  63. """,
  64. }
  65. @nilmdb.utils.must_close()
  66. class NilmDB(object):
  67. verbose = 0
  68. def __init__(self, basepath, sync=True, max_results=None,
  69. bulkdata_args=None):
  70. if bulkdata_args is None:
  71. bulkdata_args = {}
  72. # set up path
  73. self.basepath = os.path.abspath(basepath)
  74. # Create the database path if it doesn't exist
  75. try:
  76. os.makedirs(self.basepath)
  77. except OSError as e:
  78. if e.errno != errno.EEXIST:
  79. raise IOError("can't create tree " + self.basepath)
  80. # Our data goes inside it
  81. self.data = bulkdata.BulkData(self.basepath, **bulkdata_args)
  82. # SQLite database too
  83. sqlfilename = os.path.join(self.basepath, "data.sql")
  84. # We use check_same_thread = False, assuming that the rest
  85. # of the code (e.g. Server) will be smart and not access this
  86. # database from multiple threads simultaneously. Otherwise
  87. # false positives will occur when the database is only opened
  88. # in one thread, and only accessed in another.
  89. self.con = sqlite3.connect(sqlfilename, check_same_thread = False)
  90. self._sql_schema_update()
  91. # See big comment at top about the performance implications of this
  92. if sync:
  93. self.con.execute("PRAGMA synchronous=FULL")
  94. else:
  95. self.con.execute("PRAGMA synchronous=OFF")
  96. # Approximate largest number of elements that we want to send
  97. # in a single reply (for stream_intervals, stream_extract)
  98. if max_results:
  99. self.max_results = max_results
  100. else:
  101. self.max_results = 16384
  102. def get_basepath(self):
  103. return self.basepath
  104. def close(self):
  105. if self.con:
  106. self.con.commit()
  107. self.con.close()
  108. self.data.close()
  109. def _sql_schema_update(self):
  110. cur = self.con.cursor()
  111. version = cur.execute("PRAGMA user_version").fetchone()[0]
  112. oldversion = version
  113. while version in _sql_schema_updates:
  114. cur.executescript(_sql_schema_updates[version])
  115. version = version + 1
  116. if self.verbose: # pragma: no cover
  117. printf("Schema updated to %d\n", version)
  118. if version != oldversion:
  119. with self.con:
  120. cur.execute("PRAGMA user_version = {v:d}".format(v=version))
  121. @nilmdb.utils.lru_cache(size = 16)
  122. def _get_intervals(self, stream_id):
  123. """
  124. Return a mutable IntervalSet corresponding to the given stream ID.
  125. """
  126. iset = IntervalSet()
  127. result = self.con.execute("SELECT start_time, end_time, "
  128. "start_pos, end_pos "
  129. "FROM ranges "
  130. "WHERE stream_id=?", (stream_id,))
  131. try:
  132. for (start_time, end_time, start_pos, end_pos) in result:
  133. iset += DBInterval(start_time, end_time,
  134. start_time, end_time,
  135. start_pos, end_pos)
  136. except IntervalError: # pragma: no cover
  137. raise NilmDBError("unexpected overlap in ranges table!")
  138. return iset
  139. def _sql_interval_insert(self, id, start, end, start_pos, end_pos):
  140. """Helper that adds interval to the SQL database only"""
  141. self.con.execute("INSERT INTO ranges "
  142. "(stream_id,start_time,end_time,start_pos,end_pos) "
  143. "VALUES (?,?,?,?,?)",
  144. (id, start, end, start_pos, end_pos))
  145. def _sql_interval_delete(self, id, start, end, start_pos, end_pos):
  146. """Helper that removes interval from the SQL database only"""
  147. self.con.execute("DELETE FROM ranges WHERE "
  148. "stream_id=? AND start_time=? AND "
  149. "end_time=? AND start_pos=? AND end_pos=?",
  150. (id, start, end, start_pos, end_pos))
  151. def _add_interval(self, stream_id, interval, start_pos, end_pos):
  152. """
  153. Add interval to the internal interval cache, and to the database.
  154. Note: arguments must be ints (not numpy.int64, etc)
  155. """
  156. # Load this stream's intervals
  157. iset = self._get_intervals(stream_id)
  158. # Check for overlap
  159. if iset.intersects(interval): # pragma: no cover (gets caught earlier)
  160. raise NilmDBError("new interval overlaps existing data")
  161. # Check for adjacency. If there's a stream in the database
  162. # that ends exactly when this one starts, and the database
  163. # rows match up, we can make one interval that covers the
  164. # time range [adjacent.start -> interval.end)
  165. # and database rows [ adjacent.start_pos -> end_pos ].
  166. # Only do this if the resulting interval isn't too large.
  167. max_merged_rows = 8000 * 60 * 60 * 1.05 # 1.05 hours at 8 KHz
  168. adjacent = iset.find_end(interval.start)
  169. if (adjacent is not None and
  170. start_pos == adjacent.db_endpos and
  171. (end_pos - adjacent.db_startpos) < max_merged_rows):
  172. # First delete the old one, both from our iset and the
  173. # database
  174. iset -= adjacent
  175. self._sql_interval_delete(stream_id,
  176. adjacent.db_start, adjacent.db_end,
  177. adjacent.db_startpos, adjacent.db_endpos)
  178. # Now update our interval so the fallthrough add is
  179. # correct.
  180. interval.start = adjacent.start
  181. start_pos = adjacent.db_startpos
  182. # Add the new interval to the iset
  183. iset.iadd_nocheck(DBInterval(interval.start, interval.end,
  184. interval.start, interval.end,
  185. start_pos, end_pos))
  186. # Insert into the database
  187. self._sql_interval_insert(stream_id, interval.start, interval.end,
  188. int(start_pos), int(end_pos))
  189. self.con.commit()
  190. def _remove_interval(self, stream_id, original, remove):
  191. """
  192. Remove an interval from the internal cache and the database.
  193. stream_id: id of stream
  194. original: original DBInterval; must be already present in DB
  195. to_remove: DBInterval to remove; must be subset of 'original'
  196. """
  197. # Just return if we have nothing to remove
  198. if remove.start == remove.end: # pragma: no cover
  199. return
  200. # Load this stream's intervals
  201. iset = self._get_intervals(stream_id)
  202. # Remove existing interval from the cached set and the database
  203. iset -= original
  204. self._sql_interval_delete(stream_id,
  205. original.db_start, original.db_end,
  206. original.db_startpos, original.db_endpos)
  207. # Add back the intervals that would be left over if the
  208. # requested interval is removed. There may be two of them, if
  209. # the removed piece was in the middle.
  210. def add(iset, start, end, start_pos, end_pos):
  211. iset += DBInterval(start, end, start, end, start_pos, end_pos)
  212. self._sql_interval_insert(stream_id, start, end, start_pos, end_pos)
  213. if original.start != remove.start:
  214. # Interval before the removed region
  215. add(iset, original.start, remove.start,
  216. original.db_startpos, remove.db_startpos)
  217. if original.end != remove.end:
  218. # Interval after the removed region
  219. add(iset, remove.end, original.end,
  220. remove.db_endpos, original.db_endpos)
  221. # Commit SQL changes
  222. self.con.commit()
  223. return
  224. def stream_list(self, path = None, layout = None):
  225. """Return list of [path, layout] lists of all streams
  226. in the database.
  227. If path is specified, include only streams with a path that
  228. matches the given string.
  229. If layout is specified, include only streams with a layout
  230. that matches the given string.
  231. """
  232. where = "WHERE 1=1"
  233. params = ()
  234. if layout:
  235. where += " AND layout=?"
  236. params += (layout,)
  237. if path:
  238. where += " AND path=?"
  239. params += (path,)
  240. result = self.con.execute("SELECT path, layout "
  241. "FROM streams " + where, params).fetchall()
  242. return sorted(list(x) for x in result)
  243. def stream_intervals(self, path, start = None, end = None):
  244. """
  245. Returns (intervals, restart) tuple.
  246. intervals is a list of [start,end] timestamps of all intervals
  247. that exist for path, between start and end.
  248. restart, if nonzero, means that there were too many results to
  249. return in a single request. The data is complete from the
  250. starting timestamp to the point at which it was truncated,
  251. and a new request with a start time of 'restart' will fetch
  252. the next block of data.
  253. """
  254. stream_id = self._stream_id(path)
  255. intervals = self._get_intervals(stream_id)
  256. requested = Interval(start or 0, end or 1e12)
  257. result = []
  258. for n, i in enumerate(intervals.intersection(requested)):
  259. if n >= self.max_results:
  260. restart = i.start
  261. break
  262. result.append([i.start, i.end])
  263. else:
  264. restart = 0
  265. return (result, restart)
  266. def stream_create(self, path, layout_name):
  267. """Create a new table in the database.
  268. path: path to the data (e.g. '/newton/prep').
  269. Paths must contain at least two elements, e.g.:
  270. /newton/prep
  271. /newton/raw
  272. /newton/upstairs/prep
  273. /newton/upstairs/raw
  274. layout_name: string for nilmdb.layout.get_named(), e.g. 'float32_8'
  275. """
  276. # Create the bulk storage. Raises ValueError on error, which we
  277. # pass along.
  278. self.data.create(path, layout_name)
  279. # Insert into SQL database once the bulk storage is happy
  280. with self.con as con:
  281. con.execute("INSERT INTO streams (path, layout) VALUES (?,?)",
  282. (path, layout_name))
  283. def _stream_id(self, path):
  284. """Return unique stream ID"""
  285. result = self.con.execute("SELECT id FROM streams WHERE path=?",
  286. (path,)).fetchone()
  287. if result is None:
  288. raise StreamError("No stream at path " + path)
  289. return result[0]
  290. def stream_set_metadata(self, path, data):
  291. """Set stream metadata from a dictionary, e.g.
  292. { description = 'Downstairs lighting',
  293. v_scaling = 123.45 }
  294. This replaces all existing metadata.
  295. """
  296. stream_id = self._stream_id(path)
  297. with self.con as con:
  298. con.execute("DELETE FROM metadata WHERE stream_id=?", (stream_id,))
  299. for key in data:
  300. if data[key] != '':
  301. con.execute("INSERT INTO metadata VALUES (?, ?, ?)",
  302. (stream_id, key, data[key]))
  303. def stream_get_metadata(self, path):
  304. """Return stream metadata as a dictionary."""
  305. stream_id = self._stream_id(path)
  306. result = self.con.execute("SELECT metadata.key, metadata.value "
  307. "FROM metadata "
  308. "WHERE metadata.stream_id=?", (stream_id,))
  309. data = {}
  310. for (key, value) in result:
  311. data[key] = value
  312. return data
  313. def stream_update_metadata(self, path, newdata):
  314. """Update stream metadata from a dictionary"""
  315. data = self.stream_get_metadata(path)
  316. data.update(newdata)
  317. self.stream_set_metadata(path, data)
  318. def stream_destroy(self, path):
  319. """Fully remove a table and all of its data from the database.
  320. No way to undo it! Metadata is removed."""
  321. stream_id = self._stream_id(path)
  322. # Delete the cached interval data (if it was cached)
  323. self._get_intervals.cache_remove(self, stream_id)
  324. # Delete the data
  325. self.data.destroy(path)
  326. # Delete metadata, stream, intervals
  327. with self.con as con:
  328. con.execute("DELETE FROM metadata WHERE stream_id=?", (stream_id,))
  329. con.execute("DELETE FROM ranges WHERE stream_id=?", (stream_id,))
  330. con.execute("DELETE FROM streams WHERE id=?", (stream_id,))
  331. def stream_insert(self, path, start, end, data):
  332. """Insert new data into the database.
  333. path: Path at which to add the data
  334. start: Starting timestamp
  335. end: Ending timestamp
  336. data: Rows of data, to be passed to bulkdata table.append
  337. method. E.g. nilmdb.layout.Parser.data
  338. """
  339. # First check for basic overlap using timestamp info given.
  340. stream_id = self._stream_id(path)
  341. iset = self._get_intervals(stream_id)
  342. interval = Interval(start, end)
  343. if iset.intersects(interval):
  344. raise OverlapError("new data overlaps existing data at range: "
  345. + str(iset & interval))
  346. # Insert the data
  347. table = self.data.getnode(path)
  348. row_start = table.nrows
  349. table.append(data)
  350. row_end = table.nrows
  351. # Insert the record into the sql database.
  352. self._add_interval(stream_id, interval, row_start, row_end)
  353. # And that's all
  354. return
  355. def _find_start(self, table, dbinterval):
  356. """
  357. Given a DBInterval, find the row in the database that
  358. corresponds to the start time. Return the first database
  359. position with a timestamp (first element) greater than or
  360. equal to 'start'.
  361. """
  362. # Optimization for the common case where an interval wasn't truncated
  363. if dbinterval.start == dbinterval.db_start:
  364. return dbinterval.db_startpos
  365. return bisect.bisect_left(bulkdata.TimestampOnlyTable(table),
  366. dbinterval.start,
  367. dbinterval.db_startpos,
  368. dbinterval.db_endpos)
  369. def _find_end(self, table, dbinterval):
  370. """
  371. Given a DBInterval, find the row in the database that follows
  372. the end time. Return the first database position after the
  373. row with timestamp (first element) greater than or equal
  374. to 'end'.
  375. """
  376. # Optimization for the common case where an interval wasn't truncated
  377. if dbinterval.end == dbinterval.db_end:
  378. return dbinterval.db_endpos
  379. # Note that we still use bisect_left here, because we don't
  380. # want to include the given timestamp in the results. This is
  381. # so a queries like 1:00 -> 2:00 and 2:00 -> 3:00 return
  382. # non-overlapping data.
  383. return bisect.bisect_left(bulkdata.TimestampOnlyTable(table),
  384. dbinterval.end,
  385. dbinterval.db_startpos,
  386. dbinterval.db_endpos)
  387. def stream_extract(self, path, start = None, end = None, count = False):
  388. """
  389. Returns (data, restart) tuple.
  390. data is a list of raw data from the database, suitable for
  391. passing to e.g. nilmdb.layout.Formatter to translate into
  392. textual form.
  393. restart, if nonzero, means that there were too many results to
  394. return in a single request. The data is complete from the
  395. starting timestamp to the point at which it was truncated,
  396. and a new request with a start time of 'restart' will fetch
  397. the next block of data.
  398. count, if true, means to not return raw data, but just the count
  399. of rows that would have been returned. This is much faster
  400. than actually fetching the data. It is not limited by
  401. max_results.
  402. """
  403. stream_id = self._stream_id(path)
  404. table = self.data.getnode(path)
  405. intervals = self._get_intervals(stream_id)
  406. requested = Interval(start or 0, end or 1e12)
  407. result = []
  408. matched = 0
  409. remaining = self.max_results
  410. restart = 0
  411. for interval in intervals.intersection(requested):
  412. # Reading single rows from the table is too slow, so
  413. # we use two bisections to find both the starting and
  414. # ending row for this particular interval, then
  415. # read the entire range as one slice.
  416. row_start = self._find_start(table, interval)
  417. row_end = self._find_end(table, interval)
  418. if count:
  419. matched += row_end - row_start
  420. continue
  421. # Shorten it if we'll hit the maximum number of results
  422. row_max = row_start + remaining
  423. if row_max < row_end:
  424. row_end = row_max
  425. restart = table[row_max][0]
  426. # Gather these results up
  427. result.extend(table[row_start:row_end])
  428. # Count them
  429. remaining -= row_end - row_start
  430. if restart:
  431. break
  432. if count:
  433. return matched
  434. return (result, restart)
  435. def stream_remove(self, path, start = None, end = None):
  436. """
  437. Remove data from the specified time interval within a stream.
  438. Removes all data in the interval [start, end), and intervals
  439. are truncated or split appropriately. Returns the number of
  440. data points removed.
  441. """
  442. stream_id = self._stream_id(path)
  443. table = self.data.getnode(path)
  444. intervals = self._get_intervals(stream_id)
  445. to_remove = Interval(start or 0, end or 1e12)
  446. removed = 0
  447. if start == end:
  448. return 0
  449. # Can't remove intervals from within the iterator, so we need to
  450. # remember what's currently in the intersection now.
  451. all_candidates = list(intervals.intersection(to_remove, orig = True))
  452. for (dbint, orig) in all_candidates:
  453. # Find row start and end
  454. row_start = self._find_start(table, dbint)
  455. row_end = self._find_end(table, dbint)
  456. # Adjust the DBInterval to match the newly found ends
  457. dbint.db_start = dbint.start
  458. dbint.db_end = dbint.end
  459. dbint.db_startpos = row_start
  460. dbint.db_endpos = row_end
  461. # Remove interval from the database
  462. self._remove_interval(stream_id, orig, dbint)
  463. # Remove data from the underlying table storage
  464. table.remove(row_start, row_end)
  465. # Count how many were removed
  466. removed += row_end - row_start
  467. return removed