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
13 from .protomodule import ProtoModule
15 log = getLogger("loctrkd/qry")
18 pmods: List[ProtoModule] = []
22 conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
26 cast(ProtoModule, import_module("." + modnm, __package__))
27 for modnm in conf.get("common", "protocols").split(",")
29 db = connect(conf.get("storage", "dbfn"))
33 selector = " where proto = :proto"
38 if len(args) > 1 and "-o" in dopts:
46 """select tstamp, imei, peeraddr, is_incoming, proto, packet
52 for tstamp, imei, peeraddr, is_incoming, proto, packet in c:
53 msg: Any = f"Unparseable({packet.hex()})"
55 if pmod.proto_handled(proto):
56 msg = pmod.parse_message(packet, is_incoming)
58 datetime.fromtimestamp(tstamp)
59 .astimezone(tz=timezone.utc)
65 if fn and hasattr(msg, attr):
66 with open(fn, "wb") as fl: # TODO support multiple files
67 fl.write(getattr(msg, attr))
70 if __name__.endswith("__main__"):
71 opts, args = getopt(argv[1:], "o:c:d")
72 main(common.init(log, opts=opts), opts, args)