X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=51a1ec9bcdf2f63bd5e54c6fcdf048f5fb297680;hb=550c4b98348628f43bfb679ddac8b4fdacb6bbec;hp=2a81b6eb290927dd49d26dfed0d9af2e938a8576;hpb=c602ee4c0f76feb819010a17e426de55b3ad89bd;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 2a81b6e..51a1ec9 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", @@ -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 @@ -228,7 +244,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], ...] = () @@ -248,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: @@ -278,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: @@ -302,6 +321,14 @@ class BeeSurePkt(metaclass=MetaPkt): # Overridden in subclasses, otherwise command verb only 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: data = self.encode() @@ -310,32 +337,31 @@ class BeeSurePkt(metaclass=MetaPkt): class UNKNOWN(BeeSurePkt): - PROTO = "UNKNOWN" + pass class LK(BeeSurePkt): - PROTO = "LK" RESPOND = Respond.INL 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" class CONFIG(BeeSurePkt): - PROTO = "CONFIG" + pass class ICCID(BeeSurePkt): - PROTO = "ICCID" + pass class _LOC_DATA(BeeSurePkt): @@ -344,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"), @@ -366,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]) @@ -401,34 +426,30 @@ 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): - PROTO = "CR" + pass class FLOWER(BeeSurePkt): - PROTO = "FLOWER" OUT_KWARGS = (("number", int, 1),) def out_encode(self) -> str: @@ -436,16 +457,38 @@ 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): - PROTO = "POWEROFF" + pass class RESET(BeeSurePkt): - PROTO = "RESET" + pass class SOS(BeeSurePkt): - PROTO = "SOS" OUT_KWARGS = (("phonenumbers", l3str, ["", "", ""]),) def out_encode(self) -> str: @@ -462,20 +505,37 @@ class _SET_PHONE(BeeSurePkt): class SOS1(_SET_PHONE): - PROTO = "SOS1" + pass class SOS2(_SET_PHONE): - PROTO = "SOS2" + pass class SOS3(_SET_PHONE): - PROTO = "SOS3" + 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 = {} -PROTOS = {} if True: # just to indent the code, sorry! for cls in [ cls @@ -484,24 +544,18 @@ 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]]]: +) -> Union[Type[BeeSurePkt], List[str]]: if prefix.startswith(PROTO_PREFIX): pname = prefix[len(PROTO_PREFIX) :].upper() else: raise KeyError(pname) - lst = [ - (name, proto) - for name, proto in PROTOS.items() - if name.upper().startswith(pname) - ] - for _, proto in lst: + 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 @@ -550,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" @@ -568,6 +629,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), + ]