1 """ Command line tool for sending requests to the terminal """
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
10 from typing import Any, cast, List, Tuple, Type, Union
14 from .zmsg import Bcast, Resp
16 log = getLogger("loctrkd")
21 def proto_handled(proto: str) -> bool:
25 def class_by_prefix(prefix: str) -> Any:
29 pmods: List[ProtoModule] = []
33 conf: ConfigParser, opts: List[Tuple[str, str]], args: List[str]
37 cast(ProtoModule, import_module("." + modnm, __package__))
38 for modnm in conf.get("collector", "protocols").split(",")
40 # Is this https://github.com/zeromq/pyzmq/issues/1627 still not fixed?!
41 zctx = zmq.Context() # type: ignore
42 zpush = zctx.socket(zmq.PUSH) # type: ignore
43 zpush.connect(conf.get("collector", "listenurl"))
47 "Too few args, need IMEI and command min: " + str(args)
54 if pmod.proto_handled(cmd):
58 raise NotImplementedError(f"No protocol can handle {cmd}")
59 cls = pmod.class_by_prefix(cmd)
60 if isinstance(cls, list):
61 raise ValueError("Prefix does not select a single class: " + str(cls))
62 kwargs = dict([arg.split("=") for arg in args])
66 resp = Resp(imei=imei, when=time(), packet=cls.Out(**kwargs).packed)
67 log.debug("Response: %s", resp)
68 zpush.send(resp.packed)
71 if __name__.endswith("__main__"):
72 opts, args = getopt(argv[1:], "c:d")
73 main(common.init(log, opts=opts), opts, args)