X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=9346b55979eaf49540172a2b61810d1812084031;hb=15537c8be40f3ba25c5a51af75d9bed53a8a215d;hp=b83b68b755ba8e4543d22a0aa1b7e8c79331501c;hpb=b2775cb13e254b42e4b44d2ee8fe3c8b9be4d5dd;p=loctrkd.git

diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py
index b83b68b..9346b55 100755
--- 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",
@@ -160,6 +161,16 @@ def boolx(x: Union[str, bool]) -> bool:
     return x
 
 
+def l3str(x: Union[str, List[str]]) -> List[str]:
+    if isinstance(x, str):
+        lx = x.split(",")
+    else:
+        lx = x
+    if len(lx) != 3 or not all(isinstance(el, str) for el in x):
+        raise ValueError(str(lx) + " is not a list of three strings")
+    return lx
+
+
 class MetaPkt(type):
     """
     For each class corresponding to a message, automatically create
@@ -218,7 +229,6 @@ class Respond(Enum):
 
 class BeeSurePkt(metaclass=MetaPkt):
     RESPOND = Respond.NON  # Do not send anything back by default
-    PROTO: str
     IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
     OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
     KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
@@ -290,20 +300,28 @@ class BeeSurePkt(metaclass=MetaPkt):
 
     def out_encode(self) -> str:
         # Overridden in subclasses, otherwise command verb only
-        return self.PROTO
+        return ""
+
+    @property
+    def PROTO(self) -> str:
+        try:
+            proto, _ = self.__class__.__name__.split(".")
+        except ValueError:
+            proto = self.__class__.__name__
+        return proto
 
     @property
     def packed(self) -> bytes:
-        buffer = self.encode().encode()
-        return f"[LT*0000000000*{len(buffer):04X}*".encode() + buffer + b"]"
+        data = self.encode()
+        payload = self.PROTO + "," + data if data else self.PROTO
+        return f"[LT*0000000000*{len(payload):04X}*{payload}]".encode()
 
 
 class UNKNOWN(BeeSurePkt):
-    PROTO = "UNKNOWN"
+    pass
 
 
 class LK(BeeSurePkt):
-    PROTO = "LK"
     RESPOND = Respond.INL
 
     def in_decode(self, *args: str) -> None:
@@ -320,11 +338,11 @@ class LK(BeeSurePkt):
 
 
 class CONFIG(BeeSurePkt):
-    PROTO = "CONFIG"
+    pass
 
 
 class ICCID(BeeSurePkt):
-    PROTO = "ICCID"
+    pass
 
 
 class _LOC_DATA(BeeSurePkt):
@@ -390,31 +408,75 @@ class _LOC_DATA(BeeSurePkt):
 
 
 class UD(_LOC_DATA):
-    PROTO = "UD"
+    pass
 
 
 class UD2(_LOC_DATA):
-    PROTO = "UD2"
+    pass
 
 
 class TKQ(BeeSurePkt):
-    PROTO = "TKQ"
     RESPOND = Respond.INL
 
 
 class TKQ2(BeeSurePkt):
-    PROTO = "TKQ2"
     RESPOND = Respond.INL
 
 
 class AL(_LOC_DATA):
-    PROTO = "AL"
     RESPOND = Respond.INL
 
 
+class CR(BeeSurePkt):
+    pass
+
+
+class FLOWER(BeeSurePkt):
+    OUT_KWARGS = (("number", int, 1),)
+
+    def out_encode(self) -> str:
+        self.number: int
+        return str(self.number)
+
+
+class POWEROFF(BeeSurePkt):
+    pass
+
+
+class RESET(BeeSurePkt):
+    pass
+
+
+class SOS(BeeSurePkt):
+    OUT_KWARGS = (("phonenumbers", l3str, ["", "", ""]),)
+
+    def out_encode(self) -> str:
+        self.phonenumbers: List[str]
+        return ",".join(self.phonenumbers)
+
+
+class _SET_PHONE(BeeSurePkt):
+    OUT_KWARGS = (("phonenumber", str, ""),)
+
+    def out_encode(self) -> str:
+        self.phonenumber: str
+        return self.phonenumber
+
+
+class SOS1(_SET_PHONE):
+    pass
+
+
+class SOS2(_SET_PHONE):
+    pass
+
+
+class SOS3(_SET_PHONE):
+    pass
+
+
 # Build dicts protocol number -> class and class name -> protocol number
 CLASSES = {}
-PROTOS = {}
 if True:  # just to indent the code, sorry!
     for cls in [
         cls
@@ -423,23 +485,23 @@ if True:  # just to indent the code, sorry!
         and issubclass(cls, BeeSurePkt)
         and not name.startswith("_")
     ]:
-        if hasattr(cls, "PROTO"):
-            CLASSES[cls.PROTO] = cls
-            PROTOS[cls.__name__] = cls.PROTO
+        CLASSES[cls.__name__] = cls
 
 
 def class_by_prefix(
     prefix: str,
-) -> Union[Type[BeeSurePkt], List[Tuple[str, str]]]:
-    lst = [
-        (name, proto)
-        for name, proto in PROTOS.items()
-        if name.upper().startswith(prefix.upper())
-    ]
-    if len(lst) != 1:
-        return lst
-    _, proto = lst[0]
-    return CLASSES[proto]
+) -> Union[Type[BeeSurePkt], List[str]]:
+    if prefix.startswith(PROTO_PREFIX):
+        pname = prefix[len(PROTO_PREFIX) :].upper()
+    else:
+        raise KeyError(pname)
+    lst = [name for name in CLASSES.keys() if name.upper().startswith(pname)]
+    for proto in lst:
+        if len(lst) == 1:  # unique prefix match
+            return CLASSES[proto]
+        if proto == pname:  # exact match
+            return CLASSES[proto]
+    return lst
 
 
 def proto_handled(proto: str) -> bool:
@@ -501,6 +563,13 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt:
         retobj = UNKNOWN.In(vendor, imei, datalength, payload)
     else:
         retobj = UNKNOWN.Out(vendor, imei, datalength, payload)
-    retobj.PROTO = proto  # Override class attr with object attr
+    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),
+    ]