Bump Python version requirement

Git doesn't save empty dirs, so put a placeholder there that is
ignored when the test data is copied to its final location during
testing.

README.md

@@ -1,7 +1,7 @@
 # nilmdb: Non-Intrusive Load Monitor Database
 by Jim Paris <jim@jtan.com>
 
-NilmDB requires Python 3.7 or newer.
+NilmDB requires Python 3.8 or newer.
 
 ## Prerequisites:
 

extras/fix-oversize-files.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
 
 import os
 import sys

nilmdb/client/client.py

@@ -145,8 +145,8 @@ class Client():
 
         Example:
           with client.stream_insert_context('/path', start, end) as ctx:
-            ctx.insert('1234567890.0 1 2 3 4\\n')
+            ctx.insert('1234567890000000 1 2 3 4\\n')
-            ctx.insert('1234567891.0 1 2 3 4\\n')
+            ctx.insert('1234567891000000 1 2 3 4\\n')
 
         For more details, see help for nilmdb.client.client.StreamInserter
 

nilmdb/client/httpclient.py

@@ -109,7 +109,7 @@ class HTTPClient():
                                           stream=False, headers=headers)
         if isjson:
             return json.loads(response.content)
-        return response.content
+        return response.text
 
     def get(self, url, params=None):
         """Simple GET (parameters in URL)"""

nilmdb/client/numpyclient.py

@@ -59,7 +59,7 @@ class NumpyClient(nilmdb.client.client.Client):
         dtype = self._get_dtype(path, layout)
 
         def to_numpy(data):
-            a = numpy.fromstring(data, dtype)
+            a = numpy.frombuffer(data, dtype)
             if structured:
                 return a
             return numpy.c_[a['timestamp'], a['data']]

nilmdb/cmdline/cmdline.py

@@ -66,7 +66,7 @@ class Complete():
         layouts = []
         for i in range(1, 10):
             layouts.extend([(t + "_" + str(i)) for t in types])
-        return (l for l in layouts if l.startswith(prefix))
+        return (lay for lay in layouts if lay.startswith(prefix))
 
     def meta_key(self, prefix, parsed_args, **kwargs):
         return (kv.split('=')[0] for kv

nilmdb/fsck/fsck.py

@@ -1,7 +1,5 @@
 # -*- coding: utf-8 -*-
 
-raise Exception("todo: fix path bytes issues")
-
 """Check database consistency, with some ability to fix problems.
 This should be able to fix cases where a database gets corrupted due
 to unexpected system shutdown, and detect other cases that may cause
@@ -13,7 +11,6 @@ import nilmdb.client.numpyclient
 from nilmdb.utils.interval import IntervalError
 from nilmdb.server.interval import Interval, IntervalSet
 from nilmdb.utils.printf import printf, fprintf, sprintf
-from nilmdb.utils.time import timestamp_to_string
 
 from collections import defaultdict
 import sqlite3
@@ -21,70 +18,79 @@ import os
 import sys
 import progressbar
 import re
-import time
 import shutil
 import pickle
 import numpy
 
+
 class FsckError(Exception):
-    def __init__(self, msg = "", *args):
+    def __init__(self, msg="", *args):
         if args:
             msg = sprintf(msg, *args)
         Exception.__init__(self, msg)
+
+
 class FixableFsckError(FsckError):
-    def __init__(self, msg = "", *args):
+    def __init__(self, msg=""):
-        if args:
+        FsckError.__init__(self, f'{msg}\nThis may be fixable with "--fix".')
-            msg = sprintf(msg, *args)
+
-        FsckError.__init__(self, "%s\nThis may be fixable with \"--fix\".", msg)
+
 class RetryFsck(FsckError):
     pass
 
+
 def log(format, *args):
     printf(format, *args)
 
+
 def err(format, *args):
     fprintf(sys.stderr, format, *args)
 
+
 # Decorator that retries a function if it returns a specific value
-def retry_if_raised(exc, message = None, max_retries = 100):
+def retry_if_raised(exc, message=None, max_retries=100):
     def f1(func):
         def f2(*args, **kwargs):
             for n in range(max_retries):
                 try:
                     return func(*args, **kwargs)
-                except exc as e:
+                except exc:
                     if message:
                         log("%s\n\n", message)
-            raise Exception("Max number of retries (%d) exceeded; giving up")
+            raise Exception("Max number of retries (%d) exceeded; giving up" %
+                            max_retries)
         return f2
     return f1
 
+
 class Progress(object):
     def __init__(self, maxval):
         if maxval == 0:
             maxval = 1
         self.bar = progressbar.ProgressBar(
-            maxval = maxval,
+            maxval=maxval,
-            widgets = [ progressbar.Percentage(), ' ',
+            widgets=[progressbar.Percentage(), ' ',
-                        progressbar.Bar(), ' ',
+                     progressbar.Bar(), ' ',
-                        progressbar.ETA() ])
+                     progressbar.ETA()])
-        if self.bar.term_width == 0:
+        self.bar.term_width = self.bar.term_width or 75
-            self.bar.term_width = 75
+
     def __enter__(self):
         self.bar.start()
         self.last_update = 0
         return self
+
     def __exit__(self, exc_type, exc_value, traceback):
         if exc_type is None:
             self.bar.finish()
         else:
             printf("\n")
+
     def update(self, val):
         self.bar.update(val)
 
-class Fsck(object):
 
-    def __init__(self, path, fix = False):
+class Fsck(object):
+    def __init__(self, path, fix=False):
         self.basepath = path
         self.sqlpath = os.path.join(path, "data.sql")
         self.bulkpath = os.path.join(path, "data")
@@ -94,7 +100,7 @@ class Fsck(object):
     ### Main checks
 
     @retry_if_raised(RetryFsck, "Something was fixed: restarting fsck")
-    def check(self, skip_data = False):
+    def check(self, skip_data=False):
         self.bulk = None
         self.sql = None
         try:
@@ -109,7 +115,9 @@ class Fsck(object):
         finally:
             if self.bulk:
                 self.bulk.close()
-            if self.sql:
+            if self.sql:  # pragma: no cover
+                # (coverage doesn't handle finally clauses correctly;
+                # both branches here are tested)
                 self.sql.commit()
                 self.sql.close()
         log("ok\n")
@@ -164,7 +172,7 @@ class Fsck(object):
                                  "ORDER BY start_time")
             for r in result:
                 if r[0] not in self.stream_path:
-                    raise FsckError("interval ID %d not in streams", k)
+                    raise FsckError("interval ID %d not in streams", r[0])
                 self.stream_interval[r[0]].append((r[1], r[2], r[3], r[4]))
 
             log("  loading metadata\n")
@@ -172,10 +180,11 @@ class Fsck(object):
             result = cur.execute("SELECT stream_id, key, value FROM metadata")
             for r in result:
                 if r[0] not in self.stream_path:
-                    raise FsckError("metadata ID %d not in streams", k)
+                    raise FsckError("metadata ID %d not in streams", r[0])
                 if r[1] in self.stream_meta[r[0]]:
-                    raise FsckError("duplicate metadata key '%s' for stream %d",
+                    raise FsckError(
-                                    r[1], r[0])
+                        "duplicate metadata key '%s' for stream %d",
+                        r[1], r[0])
                 self.stream_meta[r[0]][r[1]] = r[2]
 
     ### Check streams and basic interval overlap
@@ -202,6 +211,7 @@ class Fsck(object):
 
                 # must exist in bulkdata
                 bulk = self.bulkpath + path
+                bulk = bulk.encode('utf-8')
                 if not os.path.isdir(bulk):
                     raise FsckError("%s: missing bulkdata dir", path)
                 if not nilmdb.server.bulkdata.Table.exists(bulk):
@@ -233,27 +243,30 @@ class Fsck(object):
 
                 # Check that we can open bulkdata
                 try:
-                    tab = None
+                    tab = nilmdb.server.bulkdata.Table(bulk)
-                    try:
+                except Exception as e:  # pragma: no cover --
-                        tab = nilmdb.server.bulkdata.Table(bulk)
+                    # No coverage here because, in the current code,
-                    except Exception as e:
+                    # everything that would cause the bulkdata to fail
-                        raise FsckError("%s: can't open bulkdata: %s",
+                    # has been already checked.
-                                        path, str(e))
+                    raise FsckError("%s: can't open bulkdata: %s",
-                finally:
+                                    path, str(e))
-                    if tab:
+                tab.close()
-                        tab.close()
 
     ### Check that bulkdata is good enough to be opened
 
     @retry_if_raised(RetryFsck)
     def check_bulkdata(self, sid, path, bulk):
-        with open(os.path.join(bulk, "_format"), "rb") as f:
+        with open(os.path.join(bulk, b"_format"), "rb") as f:
             fmt = pickle.load(f)
         if fmt["version"] != 3:
             raise FsckError("%s: bad or unsupported bulkdata version %d",
                             path, fmt["version"])
-        row_per_file = int(fmt["rows_per_file"])
+        rows_per_file = int(fmt["rows_per_file"])
+        if rows_per_file < 1:
+            raise FsckError(f"{path}: bad rows_per_file {rows_per_file}")
         files_per_dir = int(fmt["files_per_dir"])
+        if files_per_dir < 1:
+            raise FsckError(f"{path}: bad files_per_dir {files_per_dir}")
         layout = fmt["layout"]
         if layout != self.stream_layout[sid]:
             raise FsckError("%s: layout mismatch %s != %s", path,
@@ -265,16 +278,16 @@ class Fsck(object):
         rkt.close()
 
         # Find all directories
-        regex = re.compile("^[0-9a-f]{4,}$")
+        regex = re.compile(b"^[0-9a-f]{4,}$")
         subdirs = sorted(filter(regex.search, os.listdir(bulk)),
-                         key = lambda x: int(x, 16), reverse = True)
+                         key=lambda x: int(x, 16), reverse=True)
         for subdir in subdirs:
             # Find all files in that dir
             subpath = os.path.join(bulk, subdir)
             files = list(filter(regex.search, os.listdir(subpath)))
             if not files:
                 self.fix_empty_subdir(subpath)
-                raise RetryFsck
+                raise RetryFsck  # pragma: no cover; raised by fix_empty_subdir
             # Verify that their size is a multiple of the row size
             for filename in files:
                 filepath = os.path.join(subpath, filename)
@@ -290,10 +303,11 @@ class Fsck(object):
         # as long as it's only ".removed" files.
         err("\n%s\n", msg)
         for fn in os.listdir(subpath):
-            if not fn.endswith(".removed"):
+            if not fn.endswith(b".removed"):
                 raise FsckError("can't fix automatically: please manually "
-                                "remove the file %s and try again",
+                                "remove the file '%s' and try again",
-                                os.path.join(subpath, fn))
+                                os.path.join(subpath, fn).decode(
+                                    'utf-8', errors='backslashreplace'))
         # Remove the whole thing
         err("Removing empty subpath\n")
         shutil.rmtree(subpath)
@@ -324,9 +338,12 @@ class Fsck(object):
             for sid in self.stream_interval:
                 try:
                     bulk = self.bulkpath + self.stream_path[sid]
+                    bulk = bulk.encode('utf-8')
                     tab = nilmdb.server.bulkdata.Table(bulk)
+
                     def update(x):
                         pbar.update(done + x)
+
                     ints = self.stream_interval[sid]
                     done += self.check_table_intervals(sid, ints, tab, update)
                 finally:
@@ -335,7 +352,7 @@ class Fsck(object):
     def check_table_intervals(self, sid, ints, tab, update):
         # look in the table to make sure we can pick out the interval's
         # endpoints
-        path = self.stream_path[sid]
+        path = self.stream_path[sid]  # noqa: F841 unused
         tab.file_open.cache_remove_all()
         for (i, intv) in enumerate(ints):
             update(i)
@@ -343,11 +360,11 @@ class Fsck(object):
             if spos == epos and spos >= 0 and spos <= tab.nrows:
                 continue
             try:
-                srow = tab[spos]
+                srow = tab[spos]    # noqa: F841 unused
-                erow = tab[epos-1]
+                erow = tab[epos-1]  # noqa: F841 unused
             except Exception as e:
                 self.fix_bad_interval(sid, intv, tab, str(e))
-                raise RetryFsck
+                raise RetryFsck  # pragma: no cover; raised by fix_bad_interval
         return len(ints)
 
     def fix_bad_interval(self, sid, intv, tab, msg):
@@ -376,10 +393,10 @@ class Fsck(object):
                                 "end_time=? AND start_pos=? AND end_pos=?",
                                 (new_etime, new_epos, sid, stime, etime,
                                  spos, epos))
-                    if cur.rowcount != 1:
+                    if cur.rowcount != 1:  # pragma: no cover (shouldn't fail)
                         raise FsckError("failed to fix SQL database")
                 raise RetryFsck
-            err("actually it can't be truncated; times are bad too")
+            err("actually it can't be truncated; times are bad too\n")
 
         # Otherwise, the only hope is to delete the interval entirely.
         err("*** Deleting the entire interval from SQL.\n")
@@ -392,7 +409,7 @@ class Fsck(object):
                         "stream_id=? AND start_time=? AND "
                         "end_time=? AND start_pos=? AND end_pos=?",
                         (sid, stime, etime, spos, epos))
-            if cur.rowcount != 1:
+            if cur.rowcount != 1:  # pragma: no cover (shouldn't fail)
                 raise FsckError("failed to remove interval")
         raise RetryFsck
 
@@ -407,9 +424,12 @@ class Fsck(object):
             for sid in self.stream_interval:
                 try:
                     bulk = self.bulkpath + self.stream_path[sid]
+                    bulk = bulk.encode('utf-8')
                     tab = nilmdb.server.bulkdata.Table(bulk)
+
                     def update(x):
                         pbar.update(done + x)
+
                     ints = self.stream_interval[sid]
                     done += self.check_table_data(sid, ints, tab, update)
                 finally:
@@ -418,7 +438,7 @@ class Fsck(object):
     def check_table_data(self, sid, ints, tab, update):
         # Pull out all of the interval's data and verify that it's
         # monotonic.
-        maxrows = 100000
+        maxrows = getattr(self, 'maxrows_override', 100000)
         path = self.stream_path[sid]
         layout = self.stream_layout[sid]
         dtype = nilmdb.client.numpyclient.layout_to_dtype(layout)
@@ -438,27 +458,45 @@ class Fsck(object):
 
                 # Get raw data, convert to NumPy arary
                 try:
-                    raw = tab.get_data(start, stop, binary = True)
+                    raw = tab.get_data(start, stop, binary=True)
-                    data = numpy.fromstring(raw, dtype)
+                    data = numpy.frombuffer(raw, dtype)
-                except Exception as e:
+                except Exception as e:  # pragma: no cover
-                    raise FsckError("%s: failed to grab rows %d through %d: %s",
+                    # No coverage because it's hard to trigger this -- earlier
-                                    path, start, stop, repr(e))
+                    # checks check the ranges, so this would probably be a real
+                    # disk error, malloc failure, etc.
+                    raise FsckError(
+                        "%s: failed to grab rows %d through %d: %s",
+                        path, start, stop, repr(e))
+
+                ts = data['timestamp']
+
+                # Verify that all timestamps are in range.
+                match = (ts < stime) | (ts >= etime)
+                if match.any():
+                    row = start + numpy.argmax(match)
+                    raise FsckError("%s: data timestamp %d at row %d "
+                                    "outside interval range [%d,%d)",
+                                    path, data['timestamp'][row], row,
+                                    stime, etime)
 
                 # Verify that timestamps are monotonic
-                if (numpy.diff(data['timestamp']) <= 0).any():
+                match = numpy.diff(ts) <= 0
-                    raise FsckError("%s: non-monotonic timestamp(s) in rows "
+                if match.any():
-                                    "%d through %d", path, start, stop)
+                    row = start + numpy.argmax(match)
-                first_ts = data['timestamp'][0]
+                    raise FsckError("%s: non-monotonic timestamp (%d -> %d) "
+                                    "at row %d", path, ts[row], ts[row+1], row)
+                first_ts = ts[0]
                 if last_ts is not None and first_ts <= last_ts:
                     raise FsckError("%s: first interval timestamp %d is not "
                                     "greater than the previous last interval "
                                     "timestamp %d, at row %d",
                                     path, first_ts, last_ts, start)
-                last_ts = data['timestamp'][-1]
+                last_ts = ts[-1]
 
-                # These are probably fixable, by removing the offending
+                # The previous errors are fixable, by removing the
-                # intervals.  But I'm not going to bother implementing
+                # offending intervals, or changing the data
-                # that yet.
+                # timestamps.  But these are probably unlikely errors,
+                # so it's not worth implementing that yet.
 
                 # Done
                 done += count

nilmdb/scripts/nilmdb_fsck.py

@@ -1,27 +1,27 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
 
 import nilmdb.fsck
 import argparse
-import os
+
-import sys
 
 def main():
     """Main entry point for the 'nilmdb-fsck' command line script"""
 
     parser = argparse.ArgumentParser(
-        description = 'Check database consistency',
+        description='Check database consistency',
-        formatter_class = argparse.ArgumentDefaultsHelpFormatter)
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
     parser.add_argument("-v", "--version", action="version",
-                        version = nilmdb.__version__)
+                        version=nilmdb.__version__)
     parser.add_argument("-f", "--fix", action="store_true",
-                        default=False, help = 'Fix errors when possible '
+                        default=False, help='Fix errors when possible '
                         '(which may involve removing data)')
     parser.add_argument("-n", "--no-data", action="store_true",
-                        default=False, help = 'Skip the slow full-data check')
+                        default=False, help='Skip the slow full-data check')
-    parser.add_argument('database', help = 'Database directory')
+    parser.add_argument('database', help='Database directory')
     args = parser.parse_args()
 
-    nilmdb.fsck.Fsck(args.database, args.fix).check(skip_data = args.no_data)
+    nilmdb.fsck.Fsck(args.database, args.fix).check(skip_data=args.no_data)
+
 
 if __name__ == "__main__":
     main()

nilmdb/scripts/nilmdb_server.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
 
 import os
 import sys
@@ -78,9 +78,12 @@ def main():
                 stats = yappi.get_func_stats()
                 stats.sort("ttot")
                 stats.print_all()
-                from IPython import embed
+                try:
-                embed(header="Use the `yappi` or `stats` object to explore "
+                    from IPython import embed
-                      "further, quit to exit")
+                    embed(header="Use the `yappi` or `stats` object to "
+                          "explore further, `quit` to exit")
+                except ModuleNotFoundError:
+                    print("\nInstall ipython to explore further")
         else:
             server.start(blocking=True)
     except nilmdb.server.serverutil.CherryPyExit:

nilmdb/scripts/nilmtool.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
 
 import nilmdb.cmdline
 

nilmdb/server/serverutil.py

@@ -6,6 +6,7 @@ import sys
 import json
 import decorator
 import functools
+import threading
 
 import cherrypy
 
@@ -178,6 +179,19 @@ def cherrypy_patch_exit():
             os._exit = real_exit
     bus.exit = functools.partial(patched_exit, bus.exit)
 
+    # A behavior change in Python 3.8 means that some thread exceptions,
+    # derived from SystemExit, now print tracebacks where they didn't
+    # used to: https://bugs.python.org/issue1230540
+    # Install a thread exception hook that ignores CherryPyExit;
+    # to make this match the behavior where we didn't set
+    # threading.excepthook, we also need to ignore SystemExit.
+    def hook(args):
+        if args.exc_type == CherryPyExit or args.exc_type == SystemExit:
+            return
+        sys.excepthook(args.exc_type, args.exc_value,
+                       args.exc_traceback)  # pragma: no cover
+    threading.excepthook = hook
+
 
 # Start/stop CherryPy standalone server
 def cherrypy_start(blocking=False, event=False):

nilmdb/utils/timestamper.py

@@ -26,7 +26,9 @@ class Timestamper():
                 return b""
             if line[0:1] == b'#':
                 continue
-            break
+            # For some reason, coverage on python 3.8 reports that
+            # we never hit this break, even though we definitely do.
+            break  # pragma: no cover
         try:
             return next(self.ts_iter) + line
         except StopIteration:

requirements.txt

@@ -1,16 +1,41 @@
-argcomplete>=1.10.0
+argcomplete==1.12.0
-CherryPy>=18.1.2
+CherryPy==18.6.0
-coverage>=4.5.4
+coverage==5.2.1
-cython>=0.29.13
+Cython==0.29.21
-decorator>=4.4.0
+decorator==4.4.2
-fallocate>=1.6.4
+fallocate==1.6.4
-flake8>=3.7.8
+flake8==3.8.3
-nose>=1.3.7
+nose==1.3.7
-numpy>=1.17.0
+numpy==1.19.1
-progressbar>=2.5
+progressbar==2.5
-psutil>=5.6.3
+psutil==5.7.2
-python-datetime-tz>=0.5.4
+python-datetime-tz==0.5.4
-python-dateutil>=2.8.0
+python-dateutil==2.8.1
-requests>=2.22.0
+requests==2.24.0
-tz>=0.2.2
+tz==0.2.2
-WebTest>=2.0.33
+yappi==1.2.5
+
+## The following requirements were added by pip freeze:
+beautifulsoup4==4.9.1
+certifi==2020.6.20
+chardet==3.0.4
+cheroot==8.4.2
+idna==2.10
+jaraco.classes==3.1.0
+jaraco.collections==3.0.0
+jaraco.functools==3.0.1
+jaraco.text==3.2.0
+mccabe==0.6.1
+more-itertools==8.4.0
+portend==2.6
+pycodestyle==2.6.0
+pyflakes==2.2.0
+pytz==2020.1
+six==1.15.0
+soupsieve==2.0.1
+tempora==4.0.0
+urllib3==1.25.10
+waitress==1.4.4
+WebOb==1.8.6
+WebTest==2.0.35
+zc.lockfile==2.0

setup.cfg

@@ -47,10 +47,13 @@ tag_prefix=nilmdb-
 parentdir_prefix=nilmdb-
 
 [flake8]
-exclude=_version.py,fsck.py,nilmdb_fsck.py
+exclude=_version.py
 extend-ignore=E731
-per-file-ignores=__init__.py:F401,E402 serializer.py:E722 mustclose.py:E722
+per-file-ignores=__init__.py:F401,E402 \
+        serializer.py:E722 \
+        mustclose.py:E722 \
+        fsck.py:E266
 
 [pylint]
-ignore=_version.py,fsck.py,nilmdb_fsck.py
+ignore=_version.py
 disable=C0103,C0111,R0913,R0914

setup.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
 
 # To release a new version, tag it:
 #   git tag -a nilmdb-1.1 -m "Version 1.1"

tests/fsck-data/test1/data/git-empty-dir-placeholder

@@ -0,0 +1 @@
+hi

tests/fsck-data/test1a/data/git-empty-dir-placeholder

@@ -0,0 +1 @@
+hi

tests/fsck-data/test1c/data/git-empty-dir-placeholder

@@ -0,0 +1 @@
+hi

tests/fsck-data/test2h/data/git-empty-dir-placeholder

@@ -0,0 +1 @@
+hi

tests/fsck-data/test2p/data/a/b/0001/git-empty-dir-placeholder

@@ -0,0 +1 @@
+hi

tests/fsck-data/test2p1/data/a/b/0001/hello

@@ -0,0 +1 @@
+world

tests/fsck-data/test2p2/data/a/b/0001/hello.removed

@@ -0,0 +1 @@
+world