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 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 parse_message(proto: str, packet: bytes, is_incoming: bool = True) -> Any:
75 pmod = pmod_for_proto(proto)
76 return pmod.parse_message(packet, is_incoming) if pmod else None
79 def exposed_protos() -> List[Tuple[str, bool]]:
80 return [item for pmod in pmods for item in pmod.exposed_protos()]
86 def __repr__(self) -> str:
88 self.__class__.__name__
91 [f"{k}={v.__repr__()}" for k, v in self.__dict__.items()]
97 def json(self) -> str:
99 return dumps(self.__dict__)
102 class CoordReport(Report):
109 battery_percentage: Optional[int],
110 accuracy: Optional[float],
111 altitude: Optional[float],
112 speed: Optional[float],
113 direction: Optional[float],
117 self.devtime = devtime
118 self.battery_percentage = battery_percentage
119 self.accuracy = accuracy
120 self.altitude = altitude
122 self.direction = direction
123 self.latitude = latitude
124 self.longitude = longitude
127 class HintReport(Report):
128 TYPE = "approximate_location"
134 battery_percentage: Optional[int],
137 gsm_cells: List[Tuple[int, int, int]],
138 wifi_aps: List[Tuple[str, str, int]],
140 self.devtime = devtime
141 self.battery_percentage = battery_percentage
144 self.gsm_cells = gsm_cells
145 self.wifi_aps = wifi_aps
148 class StatusReport(Report):
151 def __init__(self, *, battery_percentage: int) -> None:
152 self.battery_percentage = battery_percentage