1 """ Watch for locevt and print them """
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
11 from .zmsg import Bcast
13 log = getLogger("loctrkd/watch")
18 def proto_handled(proto: str) -> bool:
22 def parse_message(packet: bytes, is_incoming: bool = True) -> Any:
26 pmods: List[ProtoModule] = []
29 def runserver(conf: ConfigParser) -> None:
32 cast(ProtoModule, import_module("." + modnm, __package__))
33 for modnm in conf.get("collector", "protocols").split(",")
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"")
43 zmsg = Bcast(zsub.recv())
44 print("I" if zmsg.is_incoming else "O", zmsg.proto, zmsg.imei)
46 if pmod.proto_handled(zmsg.proto):
47 msg = pmod.parse_message(zmsg.packet, zmsg.is_incoming)
49 except KeyboardInterrupt:
53 if __name__.endswith("__main__"):
54 runserver(common.init(log))