X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=93216c98ca39ce8ce570aff7cb7033ee6b556620;hb=84861997657f7a8daab41aa13790981bd77749f8;hp=51a1ec9bcdf2f63bd5e54c6fcdf048f5fb297680;hpb=550c4b98348628f43bfb679ddac8b4fdacb6bbec;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 51a1ec9..93216c9 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -22,6 +22,8 @@ from typing import ( ) from types import SimpleNamespace +from .protomodule import ProtoClass + __all__ = ( "Stream", "class_by_prefix", @@ -34,7 +36,6 @@ __all__ = ( "proto_name", "DecodeError", "Respond", - "LK", ) PROTO_PREFIX = "BS:" @@ -186,63 +187,13 @@ 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): 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], ...] = () @@ -340,30 +291,6 @@ 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 ICCID(BeeSurePkt): - pass - - class _LOC_DATA(BeeSurePkt): def in_decode(self, *args: str) -> None: p = SimpleNamespace() @@ -424,37 +351,76 @@ 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 + 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, + } + 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, + } -class UD(_LOC_DATA): + +class AL(_LOC_DATA): + RESPOND = Respond.INL + + +class CONFIG(BeeSurePkt): pass -class UD2(_LOC_DATA): +class CR(BeeSurePkt): pass -class TKQ(BeeSurePkt): - RESPOND = Respond.INL +class FLOWER(BeeSurePkt): + OUT_KWARGS = (("number", int, 1),) + def out_encode(self) -> str: + self.number: int + return str(self.number) -class TKQ2(BeeSurePkt): - RESPOND = Respond.INL + +class ICCID(BeeSurePkt): + pass -class AL(_LOC_DATA): +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] -class CR(BeeSurePkt): - pass + def in_encode(self) -> str: + return "LK" -class FLOWER(BeeSurePkt): - OUT_KWARGS = (("number", int, 1),) +class MESSAGE(BeeSurePkt): + OUT_KWARGS = (("message", str, ""),) def out_encode(self) -> str: - self.number: int - return str(self.number) + return str(self.message.encode("utf_16_be").hex()) class _PHB(BeeSurePkt): @@ -534,6 +500,22 @@ 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 + + # Build dicts protocol number -> class and class name -> protocol number CLASSES = {} if True: # just to indent the code, sorry! @@ -567,7 +549,7 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) -def proto_name(obj: Union[MetaPkt, BeeSurePkt]) -> str: +def proto_name(obj: Union[Type[BeeSurePkt], BeeSurePkt]) -> str: return PROTO_PREFIX + ( obj.__class__.__name__ if isinstance(obj, BeeSurePkt) else obj.__name__ ) @@ -636,6 +618,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), + (proto_name(cls), False) + for cls in CLASSES.values() + if hasattr(cls, "rectified") ]