1 """ sqlite event store """
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
8 __all__ = "fetch", "initdb", "stow", "stowloc"
13 """create table if not exists events (
16 peeraddr text not null,
17 is_incoming int not null default TRUE,
21 """create table if not exists reports (
23 devtime text not null,
29 """create table if not exists pmodmap (
30 imei text not null unique,
32 tstamp real not null default (unixepoch())
37 def initdb(dbname: str) -> None:
41 need_populate_pmodmap = False
43 DB.execute("select count(pmod) from pmodmap")
45 DB.execute("select count(tstamp) from pmodmap")
46 except OperationalError:
47 need_populate_pmodmap = True
48 DB.execute("alter table pmodmap rename to old_pmodmap")
49 except OperationalError:
53 if need_populate_pmodmap:
55 """insert into pmodmap(imei, pmod)
56 select imei, pmod from old_pmodmap"""
58 DB.execute("drop table old_pmodmap")
62 def stow(**kwargs: Any) -> None:
65 k: kwargs[k] if k in kwargs else v
67 ("is_incoming", True),
75 assert len(kwargs) <= len(parms)
77 """insert or ignore into events
78 (tstamp, imei, peeraddr, proto, packet, is_incoming)
80 (:when, :imei, :peeraddr, :proto, :packet, :is_incoming)
87 def stowloc(**kwargs: Dict[str, Any]) -> None:
90 k: kwargs.pop(k) if k in kwargs else v
93 ("devtime", str(datetime.now())),
99 parms["remainder"] = dumps(kwargs)
101 """insert or ignore into reports
102 (imei, devtime, accuracy, latitude, longitude, remainder)
104 (:imei, :devtime, :accuracy, :latitude, :longitude, :remainder)
111 def stowpmod(imei: str, pmod: str) -> None:
112 assert DB is not None
114 """insert or replace into pmodmap
115 (imei, pmod) values (:imei, :pmod)
117 {"imei": imei, "pmod": pmod},
122 def fetch(imei: str, backlog: int) -> List[Dict[str, Any]]:
123 assert DB is not None
126 """select imei, devtime, accuracy, latitude, longitude, remainder
127 from reports where imei = ?
128 order by devtime desc limit ?""",
134 remainder = loads(dic.pop("remainder"))
135 dic.update(remainder)
138 return list(reversed(result))
141 def fetchpmod(imei: str) -> Optional[Any]:
142 assert DB is not None
146 """select pmod from pmodmap where imei = ?
147 and tstamp > unixepoch() - 3600.0""",
150 result = cur.fetchone()