loctrkd/qry.py

   1 """ Print out contens of the event database """
   2
   3 from configparser import ConfigParser
   4 from datetime import datetime, timezone
   5 from getopt import getopt
   6 from importlib import import_module
   7 from logging import getLogger
   8 from sqlite3 import connect
   9 from sys import argv
  10 from typing import Any, cast, List, Tuple
  11
  12 from . import common
  13
  14 log = getLogger("loctrkd/qry")
  15
  16
  17 class ProtoModule:
  18     @staticmethod
  19     def proto_handled(proto: str) -> bool:
  20         ...
  21
  22     @staticmethod
  23     def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
  24         ...
  25
  26
  27 pmods: List[ProtoModule] = []
  28
  29
  30 def main(
  31     conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
  32 ) -> None:
  33     global pmods
  34     pmods = [
  35         cast(ProtoModule, import_module("." + modnm, __package__))
  36         for modnm in conf.get("collector", "protocols").split(",")
  37     ]
  38     db = connect(conf.get("storage", "dbfn"))
  39     c = db.cursor()
  40     if len(args) > 0:
  41         proto = args[0]
  42         selector = " where proto = :proto"
  43     else:
  44         proto = ""
  45         selector = ""
  46
  47     c.execute(
  48         """select tstamp, imei, peeraddr, is_incoming, proto, packet
  49            from events"""
  50         + selector,
  51         {"proto": proto},
  52     )
  53
  54     for tstamp, imei, peeraddr, is_incoming, proto, packet in c:
  55         msg: Any = f"Unparseable({packet.hex()})"
  56         for pmod in pmods:
  57             if pmod.proto_handled(proto):
  58                 msg = pmod.parse_message(packet, is_incoming)
  59         print(
  60             datetime.fromtimestamp(tstamp)
  61             .astimezone(tz=timezone.utc)
  62             .isoformat(),
  63             imei,
  64             peeraddr,
  65             msg,
  66         )
  67
  68
  69 if __name__.endswith("__main__"):
  70     opts, args = getopt(argv[1:], "c:d")
  71     main(common.init(log, opts=opts), opts, args)