1 """ Common housekeeping for all daemons """
3 from configparser import ConfigParser
4 from importlib import import_module
5 from getopt import getopt
7 from logging import Formatter, getLogger, Logger, StreamHandler, DEBUG, INFO
8 from logging.handlers import SysLogHandler
9 from pkg_resources import get_distribution, DistributionNotFound
10 from sys import argv, stderr, stdout
11 from typing import Any, cast, Dict, List, Optional, Tuple, Union
12 from types import SimpleNamespace
14 from .protomodule import ProtoClass, ProtoModule
16 CONF = "/etc/loctrkd.conf"
17 pmods: List[ProtoModule] = []
20 version = get_distribution("loctrkd").version
21 except DistributionNotFound:
25 def init_protocols(conf: ConfigParser) -> None:
28 cast(ProtoModule, import_module("." + modnm, __package__))
29 for modnm in conf.get("common", "protocols").split(",")
34 log: Logger, opts: Optional[List[Tuple[str, str]]] = None
37 opts, _ = getopt(argv[1:], "c:d")
40 conf.read(dopts["-c"] if "-c" in dopts else CONF)
41 log.setLevel(DEBUG if "-d" in dopts else INFO)
43 fhdl = StreamHandler(stderr)
45 Formatter("%(asctime)s - %(levelname)s - %(message)s")
48 log.debug("%s starting with options: %s", version, dopts)
50 lhdl = SysLogHandler(address="/dev/log")
52 Formatter("%(name)s[%(process)d]: %(levelname)s - %(message)s")
55 log.info("%s starting with options: %s", version, dopts)
60 def probe_pmod(segment: bytes) -> Optional[ProtoModule]:
62 if pmod.probe_buffer(segment):
67 def pmod_for_proto(proto: str) -> Optional[ProtoModule]:
69 if pmod.proto_handled(proto):
74 def pmod_by_name(pmodname: str) -> Optional[ProtoModule]:
76 if pmod.PMODNAME == pmodname:
82 pmodname: str, cmd: str, imei: str, **kwargs: Any
83 ) -> Optional[ProtoClass.Out]:
84 pmod = pmod_by_name(pmodname)
87 return pmod.make_response(cmd, imei, **kwargs)
90 def parse_message(proto: str, packet: bytes, is_incoming: bool = True) -> Any:
91 pmod = pmod_for_proto(proto)
92 return pmod.parse_message(packet, is_incoming) if pmod else None
95 def exposed_protos() -> List[Tuple[str, bool]]:
96 return [item for pmod in pmods for item in pmod.exposed_protos()]
102 def __repr__(self) -> str:
104 self.__class__.__name__
107 [f"{k}={v.__repr__()}" for k, v in self.__dict__.items()]
113 def json(self) -> str:
114 self.type = self.TYPE
115 return dumps(self.__dict__)
118 class CoordReport(Report):
125 battery_percentage: Optional[int],
126 accuracy: Optional[float],
127 altitude: Optional[float],
128 speed: Optional[float],
129 direction: Optional[float],
133 self.devtime = devtime
134 self.battery_percentage = battery_percentage
135 self.accuracy = accuracy
136 self.altitude = altitude
138 self.direction = direction
139 self.latitude = latitude
140 self.longitude = longitude
143 class HintReport(Report):
144 TYPE = "approximate_location"
150 battery_percentage: Optional[int],
153 gsm_cells: List[Tuple[int, int, int]],
154 wifi_aps: List[Tuple[str, str, int]],
156 self.devtime = devtime
157 self.battery_percentage = battery_percentage
160 self.gsm_cells = gsm_cells
161 self.wifi_aps = wifi_aps
164 class StatusReport(Report):
167 def __init__(self, *, battery_percentage: int) -> None:
168 self.battery_percentage = battery_percentage