X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fbeesure.py;h=baf14dbee9600eb25252f9d44ea38c80fac0890a;hb=be1de0bb68c5c4acf2edd6f04c5ea8e40d9f31fb;hp=369abc2db39bdd3be3ebaa15347f94dddee333d2;hpb=b0bfb1a7b499ca18bf707858b0650e04acec9881;p=loctrkd.git diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 369abc2..baf14db 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -22,10 +22,13 @@ from typing import ( ) from types import SimpleNamespace +from .protomodule import ProtoClass + __all__ = ( "Stream", "class_by_prefix", "enframe", + "exposed_protos", "inline_response", "proto_handled", "parse_message", @@ -33,7 +36,6 @@ __all__ = ( "proto_name", "DecodeError", "Respond", - "LK", ) PROTO_PREFIX = "BS:" @@ -170,54 +172,19 @@ def l3str(x: Union[str, List[str]]) -> List[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 pblist(x: Union[str, List[Tuple[str, str]]]) -> List[Tuple[str, str]]: + if isinstance(x, str): - def __setattr__(self, name: str, value: Any) -> None: - pass + def splitpair(s: str) -> Tuple[str, str]: + a, b = s.split(":") + return a, b - 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 + 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 Respond(Enum): @@ -226,7 +193,7 @@ class Respond(Enum): 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], ...] = () @@ -247,11 +214,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: @@ -277,7 +248,7 @@ class BeeSurePkt(metaclass=MetaPkt): ), ) - def decode(self, *args: str) -> None: + def decode(self, *args: Any) -> None: ... def in_decode(self, *args: str) -> None: @@ -320,37 +291,12 @@ class UNKNOWN(BeeSurePkt): pass -class LK(BeeSurePkt): - RESPOND = Respond.INL - - def in_decode(self, *args: str) -> None: - numargs = len(args) - if numargs > 1: - self.step = args[1] - if numargs > 2: - self.tumbling_number = args[2] - if numargs > 3: - self.battery_percentage = args[3] - - 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() _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"), @@ -372,10 +318,10 @@ class _LOC_DATA(BeeSurePkt): (self, "mcc", int), (self, "mnc", int), ), - args[:21], + args[:20], ): setattr(obj, attr, func(val)) # type: ignore - rest_args = args[21:] + 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]) @@ -405,37 +351,99 @@ class _LOC_DATA(BeeSurePkt): self.latitude = p.lat * p.nors self.longitude = p.lon * p.eorw + def rectified(self) -> SimpleNamespace: # JSON-able dict + if self.gps_valid: + return SimpleNamespace( + 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 SimpleNamespace( + 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): + 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): @@ -474,6 +482,40 @@ 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 + + +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! @@ -507,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__ ) @@ -544,8 +586,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" @@ -565,3 +614,11 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> BeeSurePkt: retobj.proto = proto # Override class attr with object attr retobj.cause = cause return retobj + + +def exposed_protos() -> List[Tuple[str, bool]]: + return [ + (proto_name(cls), False) + for cls in CLASSES.values() + if hasattr(cls, "rectified") + ]