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.
 
 
 

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