X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=51a1ec9bcdf2f63bd5e54c6fcdf048f5fb297680;hb=550c4b98348628f43bfb679ddac8b4fdacb6bbec;hp=a0d802bb1df306e888e87218028241210ffd5695;hpb=5ef83cb7db7464a5a625b0b7c86c4e25ebbb0de1;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index a0d802b..51a1ec9 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -20,15 +20,17 @@ from typing import ( TYPE_CHECKING, Union, ) +from types import SimpleNamespace __all__ = ( "Stream", "class_by_prefix", "enframe", + "exposed_protos", "inline_response", + "proto_handled", "parse_message", "probe_buffer", - "proto_by_name", "proto_name", "DecodeError", "Respond", @@ -159,6 +161,31 @@ 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 + + +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 @@ -217,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], ...] = () @@ -237,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: @@ -267,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: @@ -289,95 +319,223 @@ 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: 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" - RESPOND = Respond.INL + pass -class UD(BeeSurePkt): - PROTO = "UD" +class ICCID(BeeSurePkt): + pass + +class _LOC_DATA(BeeSurePkt): def in_decode(self, *args: str) -> None: - ( - _, - self.date, - self.time, - self.gps_valid, - self.lat, - self.nors, - self.lon, - self.eorw, - self.speed, - self.direction, - self.altitude, - self.num_of_sats, - self.gsm_strength_percentage, - self.battery_percentage, - self.pedometer, - self.tubmling_times, - self.device_status, - ) = args[:17] - rest_args = args[17:] - self.base_stations_number = int(rest_args[0]) - self.base_stations = rest_args[1 : 4 + 3 * self.base_stations_number] - rest_args = rest_args[3 + 3 * self.base_stations_number + 1 :] - self.wifi_ap_number = int(rest_args[0]) - self.wifi_ap = rest_args[1 : self.wifi_ap_number] - # rest_args = rest_args[self_wifi_ap_number+1:] - self.positioning_accuracy = rest_args[-1] - - -class UD2(BeeSurePkt): - PROTO = "UD2" + p = SimpleNamespace() + _id = lambda x: x + for (obj, attr, func), val in zip( + ( + (p, "date", _id), + (p, "time", _id), + (self, "gps_valid", lambda x: x == "A"), + (p, "lat", float), + (p, "nors", lambda x: 1 if x == "N" else -1), + (p, "lon", float), + (p, "eorw", lambda x: 1 if x == "E" else -1), + (self, "speed", float), + (self, "direction", float), + (self, "altitude", float), + (self, "num_of_sats", int), + (self, "gsm_strength_percentage", int), + (self, "battery_percentage", int), + (self, "pedometer", int), + (self, "tubmling_times", int), + (self, "device_status", lambda x: int(x, 16)), + (self, "base_stations_number", int), + (self, "connect_base_station_number", int), + (self, "mcc", int), + (self, "mnc", int), + ), + args[:20], + ): + 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) + ] + rest_args = rest_args[3 * self.base_stations_number :] + self.wifi_aps_number = int(rest_args[0]) + # (SSID, MAC, strength)* + self.wifi_aps = [ + ( + rest_args[1 + i * 3], + rest_args[2 + i * 3], + int(rest_args[3 + i * 3]), + ) + for i in range(self.wifi_aps_number) + ] + rest_args = rest_args[1 + 3 * self.wifi_aps_number :] + self.positioning_accuracy = float(rest_args[0]) + self.devtime = ( + datetime.strptime( + p.date + p.time, + "%d%m%y%H%M%S", + ) + # .replace(tzinfo=timezone.utc) + # .astimezone(tz=timezone.utc) + ) + self.latitude = p.lat * p.nors + self.longitude = p.lon * p.eorw + + +class UD(_LOC_DATA): + pass + + +class UD2(_LOC_DATA): + pass class TKQ(BeeSurePkt): - PROTO = "TKQ" RESPOND = Respond.INL class TKQ2(BeeSurePkt): - PROTO = "TKQ2" RESPOND = Respond.INL -class AL(BeeSurePkt): - PROTO = "AL" +class AL(_LOC_DATA): + 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 _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 + + +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 + + +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 @@ -386,23 +544,27 @@ 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: + return proto.startswith(PROTO_PREFIX) def proto_name(obj: Union[MetaPkt, BeeSurePkt]) -> str: @@ -411,10 +573,6 @@ def proto_name(obj: Union[MetaPkt, BeeSurePkt]) -> str: ) -def proto_by_name(name: str) -> str: - return PROTO_PREFIX + PROTOS.get(name, "UNKNOWN") - - def proto_of_message(packet: bytes) -> str: return PROTO_PREFIX + packet[20:-1].split(b",")[0].decode() @@ -446,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" @@ -464,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), + ]