X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=b9b463961cc3c2df90003474cccdb05c1bada786;hb=f864ae02fafc77fa7ddda90dc11e81e5270bcb5c;hp=93216c98ca39ce8ce570aff7cb7033ee6b556620;hpb=84861997657f7a8daab41aa13790981bd77749f8;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 93216c9..b9b4639 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -23,6 +23,12 @@ from typing import ( from types import SimpleNamespace from .protomodule import ProtoClass +from .common import ( + CoordReport, + HintReport, + StatusReport, + Report, +) __all__ = ( "Stream", @@ -33,7 +39,6 @@ __all__ = ( "proto_handled", "parse_message", "probe_buffer", - "proto_name", "DecodeError", "Respond", ) @@ -135,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) @@ -272,14 +285,21 @@ class BeeSurePkt(ProtoClass): # 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() @@ -313,7 +333,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), @@ -323,11 +343,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 = [ @@ -351,29 +371,27 @@ class _LOC_DATA(BeeSurePkt): self.latitude = p.lat * p.nors self.longitude = p.lon * p.eorw - def rectified(self) -> Dict[str, Any]: # JSON-able dict + def rectified(self) -> Report: if self.gps_valid: - return { - "type": "location", - "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, - } + 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 { - "type": "approximate_location", - "devtime": str(self.devtime), - "battery_percentage": self.battery_percentage, - "mcc": self.mcc, - "mnc": self.mnc, - "base_stations": self.base_stations, - "wifi_aps": self.wifi_aps, - } + 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 AL(_LOC_DATA): @@ -549,12 +567,6 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) -def proto_name(obj: Union[Type[BeeSurePkt], BeeSurePkt]) -> str: - return PROTO_PREFIX + ( - obj.__class__.__name__ if isinstance(obj, BeeSurePkt) else obj.__name__ - ) - - def proto_of_message(packet: bytes) -> str: return PROTO_PREFIX + packet[20:-1].split(b",")[0].decode() @@ -618,7 +630,7 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt: def exposed_protos() -> List[Tuple[str, bool]]: return [ - (proto_name(cls), False) + (cls.proto_name(), False) for cls in CLASSES.values() if hasattr(cls, "rectified") ]