beesure: handle "TK" (Voice message)

[loctrkd.git] / loctrkd / beesure.py

diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py
index 369abc2db39bdd3be3ebaa15347f94dddee333d2..51a1ec9bcdf2f63bd5e54c6fcdf048f5fb297680 100755 (executable)
--- a/loctrkd/beesure.py
+++ b/loctrkd/beesure.py
@@ -26,6 +26,7 @@ __all__ = (
     "Stream",
     "class_by_prefix",
     "enframe",
+    "exposed_protos",
     "inline_response",
     "proto_handled",
     "parse_message",
@@ -170,6 +171,21 @@ def l3str(x: Union[str, List[str]]) -> List[str]:
     return lx
 
 
+def pblist(x: Union[str, List[Tuple[str, str]]]) -> List[Tuple[str, str]]:
+    if isinstance(x, str):
+
+        def splitpair(s: str) -> Tuple[str, str]:
+            a, b = s.split(":")
+            return a, b
+
+        lx = [splitpair(el) for el in x.split(",")]
+    else:
+        lx = x
+    if len(lx) > 5:
+        raise ValueError(str(lx) + " has too many elements (max 5)")
+    return lx
+
+
 class MetaPkt(type):
     """
     For each class corresponding to a message, automatically create
@@ -247,11 +263,15 @@ class BeeSurePkt(metaclass=MetaPkt):
         Construct the object _either_ from (length, payload),
         _or_ from the values of individual fields
         """
+        self.payload: Union[List[str], bytes]
         assert not args or (len(args) == 4 and not kwargs)
         if args:  # guaranteed to be two arguments at this point
             self.vendor, self.imei, self.datalength, self.payload = args
             try:
-                self.decode(*self.payload)
+                if isinstance(self.payload, list):
+                    self.decode(*self.payload)
+                else:
+                    self.decode(self.payload)
             except error as e:
                 raise DecodeError(e, obj=self)
         else:
@@ -277,7 +297,7 @@ class BeeSurePkt(metaclass=MetaPkt):
             ),
         )
 
-    def decode(self, *args: str) -> None:
+    def decode(self, *args: Any) -> None:
         ...
 
     def in_decode(self, *args: str) -> None:
@@ -325,12 +345,12 @@ class LK(BeeSurePkt):
 
     def in_decode(self, *args: str) -> None:
         numargs = len(args)
+        if numargs > 0:
+            self.step = args[0]
         if numargs > 1:
-            self.step = args[1]
+            self.tumbling_number = args[1]
         if numargs > 2:
-            self.tumbling_number = args[2]
-        if numargs > 3:
-            self.battery_percentage = args[3]
+            self.battery_percentage = args[2]
 
     def in_encode(self) -> str:
         return "LK"
@@ -350,7 +370,6 @@ class _LOC_DATA(BeeSurePkt):
         _id = lambda x: x
         for (obj, attr, func), val in zip(
             (
-                (p, "verb", _id),
                 (p, "date", _id),
                 (p, "time", _id),
                 (self, "gps_valid", lambda x: x == "A"),
@@ -372,10 +391,10 @@ class _LOC_DATA(BeeSurePkt):
                 (self, "mcc", int),
                 (self, "mnc", int),
             ),
-            args[:21],
+            args[:20],
         ):
             setattr(obj, attr, func(val))  # type: ignore
-        rest_args = args[21:]
+        rest_args = args[20:]
         # (area_id, cell_id, strength)*
         self.base_stations = [
             tuple(int(el) for el in rest_args[i * 3 : 3 + i * 3])
@@ -438,6 +457,29 @@ class FLOWER(BeeSurePkt):
         return str(self.number)
 
 
+class _PHB(BeeSurePkt):
+    OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = (
+        ("entries", pblist, []),
+    )
+
+    def out_encode(self) -> str:
+        self.entries: List[Tuple[str, str]]
+        return ",".join(
+            [
+                ",".join((num, name.encode("utf_16_be").hex()))
+                for name, num in self.entries
+            ]
+        )
+
+
+class PHB(_PHB):
+    pass
+
+
+class PHB2(_PHB):
+    pass
+
+
 class POWEROFF(BeeSurePkt):
     pass
 
@@ -474,6 +516,24 @@ class SOS3(_SET_PHONE):
     pass
 
 
+class TK(BeeSurePkt):
+    RESPOND = Respond.INL
+
+    def in_decode(self, *args: Any) -> None:
+        assert len(args) == 1 and isinstance(args[0], bytes)
+        self.amr_data = (
+            args[0]
+            .replace(b"}*", b"*")
+            .replace(b"},", b",")
+            .replace(b"}[", b"[")
+            .replace(b"}]", b"]")
+            .replace(b"}}", b"}")
+        )
+
+    def out_encode(self) -> str:
+        return "1"  # 0 - receive failure, 1 - receive success
+
+
 # Build dicts protocol number -> class and class name -> protocol number
 CLASSES = {}
 if True:  # just to indent the code, sorry!
@@ -544,8 +604,15 @@ def probe_buffer(buffer: bytes) -> bool:
 def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt:
     """From a packet (without framing bytes) derive the XXX.In object"""
     toskip, vendor, imei, datalength = _framestart(packet)
-    payload = packet[20:-1].decode().split(",")
-    proto = payload[0] if len(payload) > 0 else ""
+    try:
+        splits = packet[20:-1].decode().split(",")
+        proto = splits[0] if len(splits) > 0 else ""
+        payload: Union[List[str], bytes] = splits[1:]
+    except UnicodeDecodeError:
+        bsplits = packet[20:-1].split(b",", 1)
+        if len(bsplits) == 2:
+            proto = bsplits[0].decode("ascii")
+            payload = bsplits[1]
     if proto not in CLASSES:
         cause: Union[DecodeError, ValueError, IndexError] = ValueError(
             f"Proto {proto} is unknown"
@@ -565,3 +632,10 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt:
     retobj.proto = proto  # Override class attr with object attr
     retobj.cause = cause
     return retobj
+
+
+def exposed_protos() -> List[Tuple[str, bool]]:
+    return [
+        (proto_name(UD), True),
+        (proto_name(UD2), False),
+    ]