X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=9346b55979eaf49540172a2b61810d1812084031;hb=15537c8be40f3ba25c5a51af75d9bed53a8a215d;hp=9b0255c2401a7d65f817afcb7363d9190122f073;hpb=63a086cf3956b93f760b1a0344afd757e0d0392f;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 9b0255c..9346b55 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -20,14 +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", @@ -67,11 +70,6 @@ class Stream: self.imei: Optional[str] = None self.datalen: int = 0 - @staticmethod - def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes: - assert imei is not None and len(imei) == 10 - return f"[LT*{imei:10s}*{len(buffer):04X}*".encode() + buffer + b"]" - def recv(self, segment: bytes) -> List[Union[bytes, str]]: """ Process next segment of the stream. Return successfully deframed @@ -96,12 +94,13 @@ class Stream: ) self.buffer = self.buffer[toskip:] # From this point, buffer starts with a packet header - if self.imei is not None and self.imei != imei: + if self.imei is None: + self.imei = imei + if self.imei != imei: msgs.append( f"Packet's imei {imei} mismatches" - f" previous value {self.imei}" + f" previous value {self.imei}, old value kept" ) - self.imei = imei self.datalen = datalen if len(self.buffer) < self.datalen + 21: # Incomplete packet break @@ -125,6 +124,13 @@ class Stream: return ret +def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes: + assert imei is not None and len(imei) == 10 + off, vid, _, dlen = _framestart(buffer) + assert off == 0 + return f"[{vid:2s}*{imei:10s}*{dlen:04X}*".encode() + buffer[20:] + + ### Parser/Constructor ### @@ -155,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 @@ -213,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], ...] = () @@ -285,19 +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: - return self.encode().encode() # first is object's, second str's + 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: @@ -314,65 +338,145 @@ class LK(BeeSurePkt): class CONFIG(BeeSurePkt): - PROTO = "CONFIG" - RESPOND = Respond.INL + pass + +class ICCID(BeeSurePkt): + pass -class UD(BeeSurePkt): - PROTO = "UD" +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, "verb", _id), + (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[:21], + ): + setattr(obj, attr, func(val)) # type: ignore + rest_args = args[21:] + # (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 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 @@ -381,23 +485,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: @@ -406,12 +514,8 @@ 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.split(b",")[0].decode() + return PROTO_PREFIX + packet[20:-1].split(b",")[0].decode() def imei_from_packet(packet: bytes) -> Optional[str]: @@ -459,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), + ]