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.
 
 
 

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