X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=dcf16ba227fd7ca585baa9371595e19307917fe9;hb=718dd83f618a0ee2674450162aba85cfed447512;hp=51a1ec9bcdf2f63bd5e54c6fcdf048f5fb297680;hpb=550c4b98348628f43bfb679ddac8b4fdacb6bbec;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 51a1ec9..dcf16ba 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -22,6 +22,14 @@ from typing import ( ) from types import SimpleNamespace +from .protomodule import ProtoClass +from .common import ( + CoordReport, + HintReport, + StatusReport, + Report, +) + __all__ = ( "Stream", "class_by_prefix", @@ -31,10 +39,8 @@ __all__ = ( "proto_handled", "parse_message", "probe_buffer", - "proto_name", "DecodeError", "Respond", - "LK", ) PROTO_PREFIX = "BS:" @@ -110,7 +116,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 @@ -134,6 +140,14 @@ def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes: ### Parser/Constructor ### +class classproperty: + def __init__(self, f: Callable[[Any], str]) -> None: + self.f = f + + def __get__(self, obj: Any, owner: Any) -> str: + return self.f(owner) + + class DecodeError(Exception): def __init__(self, e: Exception, **kwargs: Any) -> None: super().__init__(e) @@ -186,63 +200,14 @@ def pblist(x: Union[str, List[Tuple[str, str]]]) -> List[Tuple[str, str]]: return lx -class MetaPkt(type): - """ - For each class corresponding to a message, automatically create - two nested classes `In` and `Out` that also inherit from their - "nest". Class attribute `IN_KWARGS` defined in the "nest" is - copied to the `In` nested class under the name `KWARGS`, and - likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS` - to the nested class `Out`. In addition, method `encode` is - defined in both classes equal to `in_encode()` and `out_encode()` - respectively. - """ - - if TYPE_CHECKING: - - def __getattr__(self, name: str) -> Any: - pass - - def __setattr__(self, name: str, value: Any) -> None: - pass - - def __new__( - cls: Type["MetaPkt"], - name: str, - bases: Tuple[type, ...], - attrs: Dict[str, Any], - ) -> "MetaPkt": - newcls = super().__new__(cls, name, bases, attrs) - newcls.In = super().__new__( - cls, - name + ".In", - (newcls,) + bases, - { - "KWARGS": newcls.IN_KWARGS, - "decode": newcls.in_decode, - "encode": newcls.in_encode, - }, - ) - newcls.Out = super().__new__( - cls, - name + ".Out", - (newcls,) + bases, - { - "KWARGS": newcls.OUT_KWARGS, - "decode": newcls.out_decode, - "encode": newcls.out_encode, - }, - ) - return newcls - - class Respond(Enum): NON = 0 # Incoming, no response needed INL = 1 # Birirectional, use `inline_response()` EXT = 2 # Birirectional, use external responder -class BeeSurePkt(metaclass=MetaPkt): +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], ...] = () @@ -321,14 +286,21 @@ class BeeSurePkt(metaclass=MetaPkt): # Overridden in subclasses, otherwise command verb only return "" - @property - def PROTO(self) -> str: + @classproperty + def PROTO(cls: "BeeSurePkt") -> str: + """Name of the class without possible .In / .Out suffix""" + proto: str try: - proto, _ = self.__class__.__name__.split(".") + proto, _ = cls.__name__.split(".") except ValueError: - proto = self.__class__.__name__ + proto = cls.__name__ return proto + @classmethod + def proto_name(cls) -> str: + """Name of the command as used externally""" + return PROTO_PREFIX + cls.PROTO[:16] + @property def packed(self) -> bytes: data = self.encode() @@ -340,28 +312,12 @@ class UNKNOWN(BeeSurePkt): pass -class LK(BeeSurePkt): - RESPOND = Respond.INL - - def in_decode(self, *args: str) -> None: - numargs = len(args) - if numargs > 0: - self.step = args[0] - if numargs > 1: - self.tumbling_number = args[1] - if numargs > 2: - self.battery_percentage = args[2] - - def in_encode(self) -> str: - return "LK" - - -class CONFIG(BeeSurePkt): - pass - +class _SET_PHONE(BeeSurePkt): + OUT_KWARGS = (("phonenumber", str, ""),) -class ICCID(BeeSurePkt): - pass + def out_encode(self) -> str: + self.phonenumber: str + return self.phonenumber class _LOC_DATA(BeeSurePkt): @@ -386,7 +342,7 @@ class _LOC_DATA(BeeSurePkt): (self, "pedometer", int), (self, "tubmling_times", int), (self, "device_status", lambda x: int(x, 16)), - (self, "base_stations_number", int), + (self, "gsm_cells_number", int), (self, "connect_base_station_number", int), (self, "mcc", int), (self, "mnc", int), @@ -396,11 +352,11 @@ class _LOC_DATA(BeeSurePkt): setattr(obj, attr, func(val)) # type: ignore 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]) - for i in range(self.base_stations_number) + self.gsm_cells: List[Tuple[int, int, int]] = [ + tuple(int(el) for el in rest_args[i * 3 : 3 + i * 3]) # type: ignore + for i in range(self.gsm_cells_number) ] - rest_args = rest_args[3 * self.base_stations_number :] + rest_args = rest_args[3 * self.gsm_cells_number :] self.wifi_aps_number = int(rest_args[0]) # (SSID, MAC, strength)* self.wifi_aps = [ @@ -424,28 +380,53 @@ class _LOC_DATA(BeeSurePkt): self.latitude = p.lat * p.nors self.longitude = p.lon * p.eorw + def rectified(self) -> Report: + # 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, + accuracy=self.positioning_accuracy, + altitude=self.altitude, + speed=self.speed, + direction=self.direction, + latitude=self.latitude, + longitude=self.longitude, + ) + else: + return HintReport( + devtime=str(self.devtime), + battery_percentage=self.battery_percentage, + mcc=self.mcc, + mnc=self.mnc, + gsm_cells=self.gsm_cells, + wifi_aps=self.wifi_aps, + ) + -class UD(_LOC_DATA): - pass +class AL(_LOC_DATA): + RESPOND = Respond.INL -class UD2(_LOC_DATA): +class CALL(_SET_PHONE): pass -class TKQ(BeeSurePkt): - RESPOND = Respond.INL +class CENTER(_SET_PHONE): + pass -class TKQ2(BeeSurePkt): - RESPOND = Respond.INL +class CONFIG(BeeSurePkt): + pass -class AL(_LOC_DATA): - RESPOND = Respond.INL +class CR(BeeSurePkt): + pass -class CR(BeeSurePkt): +class FIND(BeeSurePkt): pass @@ -457,6 +438,44 @@ class FLOWER(BeeSurePkt): return str(self.number) +class ICCID(BeeSurePkt): + pass + + +class LK(BeeSurePkt): + RESPOND = Respond.INL + + def in_decode(self, *args: str) -> None: + numargs = len(args) + if numargs > 0: + self.step = args[0] + if numargs > 1: + self.tumbling_number = args[1] + if numargs > 2: + self.battery_percentage = args[2] + + def in_encode(self) -> str: + 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, ""),) + + def out_encode(self) -> str: + 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, []), @@ -496,14 +515,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 @@ -517,6 +528,7 @@ class SOS3(_SET_PHONE): class TK(BeeSurePkt): + BINARY = True RESPOND = Respond.INL def in_decode(self, *args: Any) -> None: @@ -534,6 +546,29 @@ class TK(BeeSurePkt): return "1" # 0 - receive failure, 1 - receive success +class TKQ(BeeSurePkt): + RESPOND = Respond.INL + + +class TKQ2(BeeSurePkt): + RESPOND = Respond.INL + + +class UD(_LOC_DATA): + pass + + +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,14 +602,15 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) -def proto_name(obj: Union[MetaPkt, BeeSurePkt]) -> str: - return PROTO_PREFIX + ( - obj.__class__.__name__ if isinstance(obj, BeeSurePkt) else obj.__name__ - ) +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]: @@ -589,7 +625,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: @@ -604,27 +640,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: @@ -636,6 +675,7 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt: def exposed_protos() -> List[Tuple[str, bool]]: return [ - (proto_name(UD), True), - (proto_name(UD2), False), + (cls.proto_name(), False) + for cls in CLASSES.values() + if hasattr(cls, "rectified") ]