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()]
83 class Report(SimpleNamespace):
87 def json(self) -> str:
89 return dumps(self.__dict__)
92 class CoordReport(Report):
99 battery_percentage: int,
109 battery_percentage=battery_percentage,
119 class HintReport(Report):
120 TYPE = "approximate_location"
126 battery_percentage: int,
129 gsm_cells: List[Tuple[int, int, int]],
130 wifi_aps: List[Tuple[str, str, int]]
134 battery_percentage=battery_percentage,
142 class StatusReport(Report):
145 def __init__(self, *, battery_percentage: int) -> None:
146 super().__init__(battery_percentage=battery_percentage)