Change reporting of pmod to events

[loctrkd.git] / loctrkd / common.py

diff --git a/loctrkd/common.py b/loctrkd/common.py
index f112fbd53a51176e279b77c01cc05e6f32529bd2..026f5f3100cc6729c946bdc56581152219daf05f 100644 (file)
--- a/loctrkd/common.py
+++ b/loctrkd/common.py
@@ -3,13 +3,15 @@
 from configparser import ConfigParser
 from importlib import import_module
 from getopt import getopt
 from configparser import ConfigParser
 from importlib import import_module
 from getopt import getopt

+from json import dumps

 from logging import Formatter, getLogger, Logger, StreamHandler, DEBUG, INFO
 from logging.handlers import SysLogHandler
 from pkg_resources import get_distribution, DistributionNotFound
 from sys import argv, stderr, stdout
 from typing import Any, cast, Dict, List, Optional, Tuple, Union
 from logging import Formatter, getLogger, Logger, StreamHandler, DEBUG, INFO
 from logging.handlers import SysLogHandler
 from pkg_resources import get_distribution, DistributionNotFound
 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] = []
 
 CONF = "/etc/loctrkd.conf"
 pmods: List[ProtoModule] = []


@@ -69,6 +71,22 @@ def pmod_for_proto(proto: str) -> Optional[ProtoModule]:
     return None
 
 
     return None
 
 

+def pmod_by_name(pmodname: str) -> Optional[ProtoModule]:
+    for pmod in pmods:
+        if pmod.PMODNAME == 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
 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


@@ -76,3 +94,75 @@ def parse_message(proto: str, packet: bytes, is_incoming: bool = True) -> Any:
 
 def exposed_protos() -> List[Tuple[str, bool]]:
     return [item for pmod in pmods for item in pmod.exposed_protos()]
 
 def exposed_protos() -> List[Tuple[str, bool]]:
     return [item for pmod in pmods for item in pmod.exposed_protos()]

+
+
+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
+        return dumps(self.__dict__)
+
+
+class CoordReport(Report):
+    TYPE = "location"
+
+    def __init__(
+        self,
+        *,
+        devtime: str,
+        battery_percentage: Optional[int],
+        accuracy: Optional[float],
+        altitude: Optional[float],
+        speed: Optional[float],
+        direction: Optional[float],
+        latitude: float,
+        longitude: float,
+    ) -> None:
+        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):
+    TYPE = "approximate_location"
+
+    def __init__(
+        self,
+        *,
+        devtime: str,
+        battery_percentage: Optional[int],
+        mcc: int,
+        mnc: int,
+        gsm_cells: List[Tuple[int, int, int]],
+        wifi_aps: List[Tuple[str, str, int]],
+    ) -> None:
+        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:
+        self.battery_percentage = battery_percentage