X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=b3afae149e619875f7ec089a97690de182ba0459;hb=a4a6606a30e8ba743c269083f1922222b0e1e81a;hp=b9b463961cc3c2df90003474cccdb05c1bada786;hpb=456fcc5a8964c84385d34a6687e83ae05ab2ddc3;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index b9b4639..b3afae1 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -43,6 +43,7 @@ __all__ = ( "Respond", ) +PMODNAME = __name__.split(".")[-1] PROTO_PREFIX = "BS:" ### Deframer ### @@ -116,7 +117,7 @@ class Stream: else: msgs.append( f"Packet does not end with ']'" - f" at {self.datalen+20}: {self.buffer=!r}" + f" at {self.datalen+20}: {self.buffer[:64]=!r}" ) self.buffer = self.buffer[self.datalen + 21 :] self.datalen = 0 @@ -207,6 +208,7 @@ class Respond(Enum): class BeeSurePkt(ProtoClass): + BINARY = False RESPOND = Respond.NON # Do not send anything back by default IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () @@ -311,6 +313,14 @@ class UNKNOWN(BeeSurePkt): pass +class _SET_PHONE(BeeSurePkt): + OUT_KWARGS = (("phonenumber", str, ""),) + + def out_encode(self) -> str: + self.phonenumber: str + return self.phonenumber + + class _LOC_DATA(BeeSurePkt): def in_decode(self, *args: str) -> None: p = SimpleNamespace() @@ -372,7 +382,10 @@ class _LOC_DATA(BeeSurePkt): self.longitude = p.lon * p.eorw def rectified(self) -> Report: - if self.gps_valid: + # self.gps_valid is supposed to mean it, but it does not. Perfectly + # good looking coordinates, with ten satellites, still get 'V'. + # I suspect that in reality, 'A' means "hint data is absent". + if self.gps_valid or self.num_of_sats > 3: return CoordReport( devtime=str(self.devtime), battery_percentage=self.battery_percentage, @@ -398,6 +411,14 @@ class AL(_LOC_DATA): RESPOND = Respond.INL +class CALL(_SET_PHONE): + pass + + +class CENTER(_SET_PHONE): + pass + + class CONFIG(BeeSurePkt): pass @@ -406,6 +427,10 @@ class CR(BeeSurePkt): pass +class FIND(BeeSurePkt): + pass + + class FLOWER(BeeSurePkt): OUT_KWARGS = (("number", int, 1),) @@ -434,6 +459,13 @@ class LK(BeeSurePkt): return "LK" +class LZ(BeeSurePkt): + OUT_KWARGS = (("language", int, 1), ("timezone", int, 0)) + + def out_encode(self) -> str: + return f"{self.language},{self.timezone}" + + class MESSAGE(BeeSurePkt): OUT_KWARGS = (("message", str, ""),) @@ -441,6 +473,10 @@ class MESSAGE(BeeSurePkt): return str(self.message.encode("utf_16_be").hex()) +class MONITOR(BeeSurePkt): + pass + + class _PHB(BeeSurePkt): OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ( ("entries", pblist, []), @@ -480,14 +516,6 @@ class SOS(BeeSurePkt): 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 @@ -501,6 +529,7 @@ class SOS3(_SET_PHONE): class TK(BeeSurePkt): + BINARY = True RESPOND = Respond.INL def in_decode(self, *args: Any) -> None: @@ -534,6 +563,13 @@ class UD2(_LOC_DATA): pass +class UPLOAD(BeeSurePkt): + OUT_KWARGS = (("interval", int, 600),) + + def out_encode(self) -> str: + return str(self.interval) + + # Build dicts protocol number -> class and class name -> protocol number CLASSES = {} if True: # just to indent the code, sorry! @@ -567,8 +603,15 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) +def _local_proto(packet: bytes) -> str: + try: + return packet[20:-1].split(b",")[0].decode() + except UnicodeDecodeError: + return "UNKNOWN" + + def proto_of_message(packet: bytes) -> str: - return PROTO_PREFIX + packet[20:-1].split(b",")[0].decode() + return PROTO_PREFIX + _local_proto(packet) def imei_from_packet(packet: bytes) -> Optional[str]: @@ -583,7 +626,7 @@ def is_goodbye_packet(packet: bytes) -> bool: def inline_response(packet: bytes) -> Optional[bytes]: - proto = packet[20:-1].split(b",")[0].decode() + proto = _local_proto(packet) if proto in CLASSES: cls = CLASSES[proto] if cls.RESPOND is Respond.INL: @@ -598,27 +641,30 @@ 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) + bsplits = packet[20:-1].split(b",", 1) try: - splits = packet[20:-1].decode().split(",") - proto = splits[0] if len(splits) > 0 else "" - payload: Union[List[str], bytes] = splits[1:] + proto = bsplits[0].decode("ascii") 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" - ) + proto = str(bsplits[0]) + if len(bsplits) == 2: + rest = bsplits[1] else: + rest = b"" + if proto in CLASSES: + cls = CLASSES[proto].In if is_incoming else CLASSES[proto].Out + payload = ( + # Some people encode their SSIDs in non-utf8 + rest + if cls.BINARY + else rest.decode("Windows-1252").split(",") + ) try: - if is_incoming: - return CLASSES[proto].In(vendor, imei, datalength, payload) - else: - return CLASSES[proto].Out(vendor, imei, datalength, payload) + return cls(vendor, imei, datalength, payload) except (DecodeError, ValueError, IndexError) as e: - cause = e + cause: Union[DecodeError, ValueError, IndexError] = e + else: + payload = rest + cause = ValueError(f"Proto {proto} is unknown") if is_incoming: retobj = UNKNOWN.In(vendor, imei, datalength, payload) else: @@ -634,3 +680,13 @@ def exposed_protos() -> List[Tuple[str, bool]]: for cls in CLASSES.values() if hasattr(cls, "rectified") ] + + +def make_response(cmd: str, imei: str, **kwargs: Any) -> Optional[BeeSurePkt]: + if cmd == "poweroff": + return POWEROFF.Out() + elif cmd == "refresh": + return MONITOR.Out() + elif cmd == "message": + return MESSAGE.Out(message=kwargs.get("txt", "Hello")) + return None