X-Git-Url: http://www.average.org/gitweb/?p=loctrkd.git;a=blobdiff_plain;f=loctrkd%2Fcommon.py;h=162fe0db07abbbd5a14e0cfbc429d93d193eb9d5;hp=81f8bb99c914644e55ba3274a4e756cb178fae49;hb=2cf0fd9d215dda17eae4261ab7967367f6aa0028;hpb=f24ba18e34cbfe34bf648459f0e80bd874334fe4 diff --git a/loctrkd/common.py b/loctrkd/common.py index 81f8bb9..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 @@ -106,11 +122,11 @@ 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, ) -> None: @@ -131,7 +147,7 @@ class HintReport(Report): self, *, devtime: str, - battery_percentage: int, + battery_percentage: Optional[int], mcc: int, mnc: int, gsm_cells: List[Tuple[int, int, int]],