1 """ Print out contens of the event database """
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
10 from typing import Any, cast, List, Tuple
14 log = getLogger("loctrkd/qry")
19 def proto_handled(proto: str) -> bool:
23 def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
27 pmods: List[ProtoModule] = []
31 conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
35 cast(ProtoModule, import_module("." + modnm, __package__))
36 for modnm in conf.get("collector", "protocols").split(",")
38 db = connect(conf.get("storage", "dbfn"))
42 selector = " where proto = :proto"
48 """select tstamp, imei, peeraddr, is_incoming, proto, packet
54 for tstamp, imei, peeraddr, is_incoming, proto, packet in c:
55 msg: Any = f"Unparseable({packet.hex()})"
57 if pmod.proto_handled(proto):
58 msg = pmod.parse_message(packet, is_incoming)
60 datetime.fromtimestamp(tstamp)
61 .astimezone(tz=timezone.utc)
69 if __name__.endswith("__main__"):
70 opts, args = getopt(argv[1:], "c:d")
71 main(common.init(log, opts=opts), opts, args)