evstore: do not use unixepoch() - for older sqlite

[loctrkd.git] / loctrkd / common.py

diff --git a/loctrkd/common.py b/loctrkd/common.py
index 30a17194ffcb5f645bc01d43cf265261a718e339..026f5f3100cc6729c946bdc56581152219daf05f 100644 (file)
--- 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 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] = []


@@ -71,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


@@ -80,9 +96,19 @@ def exposed_protos() -> List[Tuple[str, bool]]:
     return [item for pmod in pmods for item in pmod.exposed_protos()]
 
 
     return [item for pmod in pmods for item in pmod.exposed_protos()]
 
 

-class Report(SimpleNamespace):
+class Report:

     TYPE: str
 
     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
     @property
     def json(self) -> str:
         self.type = self.TYPE


@@ -96,24 +122,22 @@ class CoordReport(Report):
         self,
         *,
         devtime: str,
         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,
         latitude: float,

-        longitude: float
+        longitude: float,

     ) -> None:
     ) -> 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):
 
 
 class HintReport(Report):


@@ -123,24 +147,22 @@ class HintReport(Report):
         self,
         *,
         devtime: str,
         self,
         *,
         devtime: str,

-        battery_percentage: int,
+        battery_percentage: Optional[int],

         mcc: int,
         mnc: int,
         gsm_cells: List[Tuple[int, int, 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:
     ) -> 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:
 
 
 class StatusReport(Report):
     TYPE = "status"
 
     def __init__(self, *, battery_percentage: int) -> None:

-        super().__init__(battery_percentage=battery_percentage)
+        self.battery_percentage = battery_percentage