X-Git-Url: http://www.average.org/gitweb/?p=loctrkd.git;a=blobdiff_plain;f=loctrkd%2Fcommon.py;h=162fe0db07abbbd5a14e0cfbc429d93d193eb9d5;hp=30a17194ffcb5f645bc01d43cf265261a718e339;hb=2cf0fd9d215dda17eae4261ab7967367f6aa0028;hpb=b965fecb08f4149f6f91770e9d6bb6d79f53f11e diff --git a/loctrkd/common.py b/loctrkd/common.py index 30a1719..162fe0d 100644 --- a/loctrkd/common.py +++ b/loctrkd/common.py @@ -11,7 +11,7 @@ from sys import argv, stderr, stdout from typing import Any, cast, Dict, List, Optional, Tuple, Union from types import SimpleNamespace -from .protomodule import ProtoModule +from .protomodule import ProtoClass, ProtoModule CONF = "/etc/loctrkd.conf" pmods: List[ProtoModule] = [] @@ -71,6 +71,22 @@ def pmod_for_proto(proto: str) -> Optional[ProtoModule]: return None +def pmod_by_name(pmodname: str) -> Optional[ProtoModule]: + for pmod in pmods: + if pmod.__name__.split(".")[-1] == pmodname: + return pmod + return None + + +def make_response( + pmodname: str, cmd: str, imei: str, **kwargs: Any +) -> Optional[ProtoClass.Out]: + pmod = pmod_by_name(pmodname) + if pmod is None: + return None + return pmod.make_response(cmd, imei, **kwargs) + + def parse_message(proto: str, packet: bytes, is_incoming: bool = True) -> Any: pmod = pmod_for_proto(proto) return pmod.parse_message(packet, is_incoming) if pmod else None @@ -80,9 +96,19 @@ def exposed_protos() -> List[Tuple[str, bool]]: return [item for pmod in pmods for item in pmod.exposed_protos()] -class Report(SimpleNamespace): +class Report: TYPE: str + def __repr__(self) -> str: + return ( + self.__class__.__name__ + + "(" + + ", ".join( + [f"{k}={v.__repr__()}" for k, v in self.__dict__.items()] + ) + + ")" + ) + @property def json(self) -> str: self.type = self.TYPE @@ -96,24 +122,22 @@ class CoordReport(Report): self, *, devtime: str, - battery_percentage: int, - accuracy: float, - altitude: float, - speed: float, - direction: float, + battery_percentage: Optional[int], + accuracy: Optional[float], + altitude: Optional[float], + speed: Optional[float], + direction: Optional[float], latitude: float, - longitude: float + longitude: float, ) -> None: - super().__init__( - devtime=devtime, - battery_percentage=battery_percentage, - accuracy=accuracy, - altitude=altitude, - speed=speed, - direction=direction, - latitude=latitude, - longitude=longitude, - ) + self.devtime = devtime + self.battery_percentage = battery_percentage + self.accuracy = accuracy + self.altitude = altitude + self.speed = speed + self.direction = direction + self.latitude = latitude + self.longitude = longitude class HintReport(Report): @@ -123,24 +147,22 @@ class HintReport(Report): self, *, devtime: str, - battery_percentage: int, + battery_percentage: Optional[int], mcc: int, mnc: int, gsm_cells: List[Tuple[int, int, int]], - wifi_aps: List[Tuple[str, str, int]] + wifi_aps: List[Tuple[str, str, int]], ) -> None: - super().__init__( - devtime=devtime, - battery_percentage=battery_percentage, - mcc=mcc, - mnc=mnc, - gsm_cells=gsm_cells, - wifi_aps=wifi_aps, - ) + self.devtime = devtime + self.battery_percentage = battery_percentage + self.mcc = mcc + self.mnc = mnc + self.gsm_cells = gsm_cells + self.wifi_aps = wifi_aps class StatusReport(Report): TYPE = "status" def __init__(self, *, battery_percentage: int) -> None: - super().__init__(battery_percentage=battery_percentage) + self.battery_percentage = battery_percentage