loctrkd/watch.py

   1 """ Watch for locevt and print them """
   2
   3 from configparser import ConfigParser
   4 from datetime import datetime, timezone
   5 from importlib import import_module
   6 from logging import getLogger
   7 from typing import Any, cast, List
   8 import zmq
   9
  10 from . import common
  11 from .zmsg import Bcast
  12
  13 log = getLogger("loctrkd/watch")
  14
  15
  16 class ProtoModule:
  17     @staticmethod
  18     def proto_handled(proto: str) -> bool:
  19         ...
  20
  21     @staticmethod
  22     def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
  23         ...
  24
  25
  26 pmods: List[ProtoModule] = []
  27
  28
  29 def runserver(conf: ConfigParser) -> None:
  30     global pmods
  31     pmods = [
  32         cast(ProtoModule, import_module("." + modnm, __package__))
  33         for modnm in conf.get("collector", "protocols").split(",")
  34     ]
  35     # Is this https://github.com/zeromq/pyzmq/issues/1627 still not fixed?!
  36     zctx = zmq.Context()  # type: ignore
  37     zsub = zctx.socket(zmq.SUB)  # type: ignore
  38     zsub.connect(conf.get("collector", "publishurl"))
  39     zsub.setsockopt(zmq.SUBSCRIBE, b"")
  40
  41     try:
  42         while True:
  43             zmsg = Bcast(zsub.recv())
  44             print("I" if zmsg.is_incoming else "O", zmsg.proto, zmsg.imei)
  45             for pmod in pmods:
  46                 if pmod.proto_handled(zmsg.proto):
  47                     msg = pmod.parse_message(zmsg.packet, zmsg.is_incoming)
  48                     print(msg)
  49     except KeyboardInterrupt:
  50         pass
  51
  52
  53 if __name__.endswith("__main__"):
  54     runserver(common.init(log))