loctrkd/evstore.py

   1 """ sqlite event store """
   2
   3 from datetime import datetime
   4 from json import dumps, loads
   5 from sqlite3 import connect, OperationalError, Row
   6 from typing import Any, Dict, List, Optional, Tuple
   7
   8 __all__ = "fetch", "initdb", "stow", "stowloc"
   9
  10 DB = None
  11
  12 SCHEMA = (
  13     """create table if not exists events (
  14     tstamp real not null,
  15     imei text,
  16     peeraddr text not null,
  17     is_incoming int not null default TRUE,
  18     proto text not null,
  19     packet blob
  20 )""",
  21     """create table if not exists reports (
  22     imei text,
  23     devtime text not null,
  24     accuracy real,
  25     latitude real,
  26     longitude real,
  27     remainder text
  28 )""",
  29     """create table if not exists pmodmap (
  30     imei text not null unique,
  31     pmod text not null
  32 )""",
  33 )
  34
  35
  36 def initdb(dbname: str) -> None:
  37     global DB
  38     DB = connect(dbname)
  39     DB.row_factory = Row
  40     for stmt in SCHEMA:
  41         DB.execute(stmt)
  42
  43
  44 def stow(**kwargs: Any) -> None:
  45     assert DB is not None
  46     parms = {
  47         k: kwargs[k] if k in kwargs else v
  48         for k, v in (
  49             ("is_incoming", True),
  50             ("peeraddr", None),
  51             ("when", 0.0),
  52             ("imei", None),
  53             ("proto", "UNKNOWN"),
  54             ("packet", b""),
  55         )
  56     }
  57     assert len(kwargs) <= len(parms)
  58     DB.execute(
  59         """insert or ignore into events
  60                 (tstamp, imei, peeraddr, proto, packet, is_incoming)
  61                 values
  62                 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
  63         """,
  64         parms,
  65     )
  66     DB.commit()
  67
  68
  69 def stowloc(**kwargs: Dict[str, Any]) -> None:
  70     assert DB is not None
  71     parms = {
  72         k: kwargs.pop(k) if k in kwargs else v
  73         for k, v in (
  74             ("imei", None),
  75             ("devtime", str(datetime.now())),
  76             ("accuracy", None),
  77             ("latitude", None),
  78             ("longitude", None),
  79         )
  80     }
  81     parms["remainder"] = dumps(kwargs)
  82     DB.execute(
  83         """insert or ignore into reports
  84                 (imei, devtime, accuracy, latitude, longitude, remainder)
  85                 values
  86                 (:imei, :devtime, :accuracy, :latitude, :longitude, :remainder)
  87         """,
  88         parms,
  89     )
  90     DB.commit()
  91
  92
  93 def stowpmod(imei: str, pmod: str) -> None:
  94     assert DB is not None
  95     DB.execute(
  96         """insert or replace into pmodmap
  97                 (imei, pmod) values (:imei, :pmod)
  98         """,
  99         {"imei": imei, "pmod": pmod},
 100     )
 101     DB.commit()
 102
 103
 104 def fetch(imei: str, backlog: int) -> List[Dict[str, Any]]:
 105     assert DB is not None
 106     cur = DB.cursor()
 107     cur.execute(
 108         """select imei, devtime, accuracy, latitude, longitude, remainder
 109                     from reports where imei = ?
 110                     order by devtime desc limit ?""",
 111         (imei, backlog),
 112     )
 113     result = []
 114     for row in cur:
 115         dic = dict(row)
 116         remainder = loads(dic.pop("remainder"))
 117         dic.update(remainder)
 118         result.append(dic)
 119     cur.close()
 120     return list(reversed(result))
 121
 122
 123 def fetchpmod(imei: str) -> Optional[Any]:
 124     assert DB is not None
 125     ret = None
 126     cur = DB.cursor()
 127     cur.execute("select pmod from pmodmap where imei = ?", (imei,))
 128     result = cur.fetchone()
 129     if result:
 130         ret = result[0]
 131     cur.close()
 132     return ret