from typing import Any, cast, List, Tuple
from . import common
+from .protomodule import ProtoModule
log = getLogger("loctrkd/mkgpx")
-class ProtoModule:
- @staticmethod
- def proto_handled(proto: str) -> bool:
- ...
-
- @staticmethod
- def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
- ...
-
-
-pmods: List[ProtoModule] = []
-
-
def main(
conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
) -> None:
- global pmods
- pmods = [
- cast(ProtoModule, import_module("." + modnm, __package__))
- for modnm in conf.get("collector", "protocols").split(",")
- ]
db = connect(conf.get("storage", "dbfn"))
c = db.cursor()
c.execute(
)
for tstamp, is_incoming, proto, packet in c:
- for pmod in pmods:
- if pmod.proto_handled(proto):
- msg = pmod.parse_message(packet, is_incoming=is_incoming)
+ pmod = common.pmod_for_proto(proto)
+ if pmod is not None:
+ msg = pmod.parse_message(packet, is_incoming=is_incoming)
lat, lon = msg.latitude, msg.longitude
isotime = (
datetime.fromtimestamp(tstamp)