X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=2a81b6eb290927dd49d26dfed0d9af2e938a8576;hb=e2e977603507f9121e5c60c5b69d98fb425fa833;hp=8fa83f51902b0e472286ceff089ba49aeb389964;hpb=ba4cb894d37f24ac333b316cf9487dfc913eaf74;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 8fa83f5..2a81b6e 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -20,6 +20,7 @@ from typing import ( TYPE_CHECKING, Union, ) +from types import SimpleNamespace __all__ = ( "Stream", @@ -29,7 +30,6 @@ __all__ = ( "proto_handled", "parse_message", "probe_buffer", - "proto_by_name", "proto_name", "DecodeError", "Respond", @@ -160,6 +160,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 @@ -290,12 +300,13 @@ class BeeSurePkt(metaclass=MetaPkt): def out_encode(self) -> str: # Overridden in subclasses, otherwise command verb only - return self.PROTO + return "" @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): @@ -327,40 +338,73 @@ class ICCID(BeeSurePkt): PROTO = "ICCID" -class UD(BeeSurePkt): +class _LOC_DATA(BeeSurePkt): + def in_decode(self, *args: str) -> None: + 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): PROTO = "UD" - 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): + +class UD2(_LOC_DATA): PROTO = "UD2" @@ -374,11 +418,61 @@ class TKQ2(BeeSurePkt): RESPOND = Respond.INL -class AL(BeeSurePkt): +class AL(_LOC_DATA): PROTO = "AL" RESPOND = Respond.INL +class CR(BeeSurePkt): + PROTO = "CR" + + +class FLOWER(BeeSurePkt): + PROTO = "FLOWER" + OUT_KWARGS = (("number", int, 1),) + + def out_encode(self) -> str: + self.number: int + return str(self.number) + + +class POWEROFF(BeeSurePkt): + PROTO = "POWEROFF" + + +class RESET(BeeSurePkt): + PROTO = "RESET" + + +class SOS(BeeSurePkt): + PROTO = "SOS" + 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): + PROTO = "SOS1" + + +class SOS2(_SET_PHONE): + PROTO = "SOS2" + + +class SOS3(_SET_PHONE): + PROTO = "SOS3" + + # Build dicts protocol number -> class and class name -> protocol number CLASSES = {} PROTOS = {} @@ -398,15 +492,21 @@ if True: # just to indent the code, sorry! def class_by_prefix( prefix: str, ) -> Union[Type[BeeSurePkt], List[Tuple[str, 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(prefix.upper()) + if name.upper().startswith(pname) ] - if len(lst) != 1: - return lst - _, proto = lst[0] - return CLASSES[proto] + 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: @@ -419,10 +519,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()