X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=2d777d94fa5bfbec0b4c4b15192a0a56f9427954;hb=f075dbd6eea0e7326d68fd15a103d668baf67c10;hp=b70cacf1eb328571171272657c92c97d0f04b2f1;hpb=7271332d13eadd3e201882288042207fceae50bd;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index b70cacf..2d777d9 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -18,7 +18,17 @@ from datetime import datetime, timezone from enum import Enum from inspect import isclass from struct import error, pack, unpack -from typing import Any, Callable, Tuple +from typing import ( + Any, + Callable, + Dict, + List, + Optional, + Tuple, + Type, + TYPE_CHECKING, + Union, +) __all__ = ( "class_by_prefix", @@ -69,18 +79,29 @@ __all__ = ( class DecodeError(Exception): - def __init__(self, e, **kwargs): + def __init__(self, e: Exception, **kwargs: Any) -> None: super().__init__(e) for k, v in kwargs.items(): setattr(self, k, v) -def intx(x): + +def intx(x: Union[str, int]) -> int: if isinstance(x, str): x = int(x, 0) return x -def hhmm(x): +def boolx(x: Union[str, bool]) -> bool: + if isinstance(x, str): + if x.upper() in ("ON", "TRUE", "1"): + return True + if x.upper() in ("OFF", "FALSE", "0"): + return False + raise ValueError(str(x) + " could not be parsed as a Boolean") + return x + + +def hhmm(x: str) -> str: """Check for the string that represents hours and minutes""" if not isinstance(x, str) or len(x) != 4: raise ValueError(str(x) + " is not a four-character string") @@ -91,21 +112,58 @@ def hhmm(x): return x -def l3str(x): +def hhmmhhmm(x: str) -> str: + """Check for the string that represents hours and minutes twice""" + if not isinstance(x, str) or len(x) != 8: + raise ValueError(str(x) + " is not an eight-character string") + return hhmm(x[:4]) + hhmm(x[4:]) + + +def l3str(x: Union[str, List[str]]) -> List[str]: if isinstance(x, str): - x = x.split(",") - if len(x) != 3 or not all(isinstance(el, str) for el in x): - raise ValueError(str(x) + " is not a list of three strings") - return x + 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 + +def l3alarms(x: Union[str, List[Tuple[int, str]]]) -> List[Tuple[int, str]]: + def alrmspec(sub: str) -> Tuple[int, str]: + if len(sub) != 7: + raise ValueError(sub + " does not represent day and time") + return ( + { + "MON": 1, + "TUE": 2, + "WED": 3, + "THU": 4, + "FRI": 5, + "SAT": 6, + "SUN": 7, + }[sub[:3].upper()], + sub[3:], + ) -def l3int(x): if isinstance(x, str): - x = x.split(",") - x = [int(el) for el in x] - if len(x) != 3 or not all(isinstance(el, int) for el in x): - raise ValueError(str(x) + " is not a list of three integers") - return x + lx = [alrmspec(sub) for sub in x.split(",")] + else: + lx = x + lx.extend([(0, "0000") for _ in range(3 - len(lx))]) + if len(lx) != 3 or any(d < 0 or d > 7 for d, tm in lx): + raise ValueError(str(lx) + " is a wrong alarms specification") + return [(d, hhmm(tm)) for d, tm in lx] + + +def l3int(x: Union[str, List[int]]) -> List[int]: + if isinstance(x, str): + lx = [int(el) for el in x.split(",")] + else: + lx = x + if len(lx) != 3 or not all(isinstance(el, int) for el in lx): + raise ValueError(str(lx) + " is not a list of three integers") + return lx class MetaPkt(type): @@ -120,7 +178,20 @@ class MetaPkt(type): respectively. """ - def __new__(cls, name, bases, attrs): + 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, @@ -154,10 +225,21 @@ class Respond(Enum): class GPS303Pkt(metaclass=MetaPkt): RESPOND = Respond.NON # Do not send anything back by default PROTO: int - IN_KWARGS: Tuple[Tuple[str, Callable, Any], ...] = () - OUT_KWARGS: Tuple[Tuple[str, Callable, Any], ...] = () + 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], ...] = () + In: Type["GPS303Pkt"] + Out: Type["GPS303Pkt"] - def __init__(self, *args, **kwargs): + if TYPE_CHECKING: + + def __getattr__(self, name: str) -> Any: + pass + + def __setattr__(self, name: str, value: Any) -> None: + pass + + def __init__(self, *args: Any, **kwargs: Any): """ Construct the object _either_ from (length, payload), _or_ from the values of individual fields @@ -177,7 +259,7 @@ class GPS303Pkt(metaclass=MetaPkt): self.__class__.__name__ + " stray kwargs " + str(kwargs) ) - def __repr__(self): + def __repr__(self) -> str: return "{}({})".format( self.__class__.__name__, ", ".join( @@ -192,26 +274,30 @@ class GPS303Pkt(metaclass=MetaPkt): ), ) - def in_decode(self, length, packet): + decode: Callable[["GPS303Pkt", int, bytes], None] + + def in_decode(self, length: int, packet: bytes) -> None: # Overridden in subclasses, otherwise do not decode payload return - def out_decode(self, length, packet): + def out_decode(self, length: int, packet: bytes) -> None: # Overridden in subclasses, otherwise do not decode payload return - def in_encode(self): + encode: Callable[["GPS303Pkt"], bytes] + + def in_encode(self) -> bytes: # Necessary to emulate terminal, which is not implemented raise NotImplementedError( self.__class__.__name__ + ".encode() not implemented" ) - def out_encode(self): + def out_encode(self) -> bytes: # Overridden in subclasses, otherwise make empty payload return b"" @property - def packed(self): + def packed(self) -> bytes: payload = self.encode() length = len(payload) + 1 return pack("BB", length, self.PROTO) + payload @@ -226,17 +312,16 @@ class LOGIN(GPS303Pkt): RESPOND = Respond.INL # Default response for ACK, can also respond with STOP_UPLOAD - def in_decode(self, length, payload): - self.imei = payload[:-1].hex() - self.ver = unpack("B", payload[-1:])[0] - return self + def in_decode(self, length: int, payload: bytes) -> None: + self.imei = payload[:16].hex() + self.ver = payload[16] class SUPERVISION(GPS303Pkt): PROTO = 0x05 OUT_KWARGS = (("status", int, 1),) - def out_encode(self): + def out_encode(self) -> bytes: # 1: The device automatically answers Pickup effect # 2: Automatically Answering Two-way Calls # 3: Ring manually answer the two-way call @@ -251,7 +336,7 @@ class HEARTBEAT(GPS303Pkt): class _GPS_POSITIONING(GPS303Pkt): RESPOND = Respond.INL - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.dtime = payload[:6] if self.dtime == b"\0\0\0\0\0\0": self.devtime = None @@ -271,9 +356,8 @@ class _GPS_POSITIONING(GPS303Pkt): self.longitude = lon / (30000 * 60) * (-1 if flip_lon else 1) self.speed = speed self.flags = flags - return self - def out_encode(self): + def out_encode(self) -> bytes: tup = datetime.utcnow().timetuple() ttup = (tup[0] % 100,) + tup[1:6] return pack("BBBBBB", *ttup) @@ -292,17 +376,16 @@ class STATUS(GPS303Pkt): RESPOND = Respond.EXT OUT_KWARGS = (("upload_interval", int, 25),) - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.batt, self.ver, self.timezone, self.intvl = unpack( "BBBB", payload[:4] ) if len(payload) > 4: - self.signal = payload[4] + self.signal: Optional[int] = payload[4] else: self.signal = None - return self - def out_encode(self): # Set interval in minutes + def out_encode(self) -> bytes: # Set interval in minutes return pack("B", self.upload_interval) @@ -320,12 +403,12 @@ class WHITELIST_TOTAL(GPS303Pkt): # Server sends to initiage sync (0x58) PROTO = 0x16 OUT_KWARGS = (("number", int, 3),) - def out_encode(self): # Number of whitelist entries - return pack("B", number) + def out_encode(self) -> bytes: # Number of whitelist entries + return pack("B", self.number) class _WIFI_POSITIONING(GPS303Pkt): - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.dtime = payload[:6] if self.dtime == b"\0\0\0\0\0\0": self.devtime = None @@ -348,14 +431,13 @@ class _WIFI_POSITIONING(GPS303Pkt): "!HHB", gsm_slice[4 + i * 5 : 9 + i * 5] ) self.gsm_cells.append((locac, cellid, -sigstr)) - return self class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING): PROTO = 0x17 RESPOND = Respond.INL - def out_encode(self): + def out_encode(self) -> bytes: return bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S")) @@ -363,7 +445,7 @@ class TIME(GPS303Pkt): PROTO = 0x30 RESPOND = Respond.INL - def out_encode(self): + def out_encode(self) -> bytes: return pack("!HBBBBB", *datetime.utcnow().timetuple()[:6]) @@ -371,31 +453,43 @@ class PROHIBIT_LBS(GPS303Pkt): PROTO = 0x33 OUT_KWARGS = (("status", int, 1),) - def out_encode(self): # Server sent, 0-off, 1-on + def out_encode(self) -> bytes: # Server sent, 0-off, 1-on return pack("B", self.status) class GPS_LBS_SWITCH_TIMES(GPS303Pkt): PROTO = 0x34 - # Data is in packed decimal - # 00/01 - GPS on/off - # 00/01 - Don't set / Set upload period - # HHMMHHMM - Upload period - # 00/01 - LBS on/off - # 00/01 - Don't set / Set time of boot - # HHMM - Time of boot - # 00/01 - Don't set / Set time of shutdown - # HHMM - Time of shutdown - def out_encode(self): - return b"" # TODO + OUT_KWARGS = ( + ("gps_off", boolx, False), # Clarify the meaning of 0/1 + ("gps_interval_set", boolx, False), + ("gps_interval", hhmmhhmm, "00000000"), + ("lbs_off", boolx, False), # Clarify the meaning of 0/1 + ("boot_time_set", boolx, False), + ("boot_time", hhmm, "0000"), + ("shut_time_set", boolx, False), + ("shut_time", hhmm, "0000"), + ) + + def out_encode(self) -> bytes: + return ( + pack("B", self.gps_off) + + pack("B", self.gps_interval_set) + + bytes.fromhex(self.gps_interval) + + pack("B", self.lbs_off) + + pack("B", self.boot_time_set) + + bytes.fromhex(self.boot_time) + + pack("B", self.shut_time_set) + + bytes.fromhex(self.shut_time) + ) class _SET_PHONE(GPS303Pkt): OUT_KWARGS = (("phone", str, ""),) - def out_encode(self): - return self.phone.encode() + def out_encode(self) -> bytes: + self.phone: str + return self.phone.encode("") class REMOTE_MONITOR_PHONE(_SET_PHONE): @@ -426,7 +520,7 @@ class GPS_OFF_PERIOD(GPS303Pkt): ("to", hhmm, "2359"), ) - def out_encode(self): + def out_encode(self) -> bytes: return ( pack("B", self.onoff) + bytes.fromhex(self.fm) @@ -445,7 +539,7 @@ class DND_PERIOD(GPS303Pkt): ("to2", hhmm, "2359"), ) - def out_endode(self): + def out_encode(self) -> bytes: return ( pack("B", self.onoff) + pack("B", self.week) @@ -460,7 +554,7 @@ class RESTART_SHUTDOWN(GPS303Pkt): PROTO = 0x48 OUT_KWARGS = (("flag", int, 0),) - def out_encode(self): + def out_encode(self) -> bytes: # 1 - restart # 2 - shutdown return pack("B", self.flag) @@ -472,27 +566,29 @@ class DEVICE(GPS303Pkt): # 0 - Stop looking for equipment # 1 - Start looking for equipment - def out_encode(self): + def out_encode(self) -> bytes: return pack("B", self.flag) class ALARM_CLOCK(GPS303Pkt): PROTO = 0x50 + OUT_KWARGS: Tuple[ + Tuple[str, Callable[[Any], Any], List[Tuple[int, str]]], ... + ] = ( + ("alarms", l3alarms, []), + ) - def out_encode(self): - # TODO implement parsing kwargs - alarms = ((0, "0000"), (0, "0000"), (0, "0000")) + def out_encode(self) -> bytes: return b"".join( - cls("B", day) + bytes.fromhex(tm) for day, tm in alarms + pack("B", day) + bytes.fromhex(tm) for day, tm in self.alarms ) class STOP_ALARM(GPS303Pkt): PROTO = 0x56 - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.flag = payload[0] - return self class SETUP(GPS303Pkt): @@ -510,8 +606,8 @@ class SETUP(GPS303Pkt): ("phonenumbers", l3str, ["", "", ""]), ) - def out_encode(self): - def pack3b(x): + def out_encode(self) -> bytes: + def pack3b(x: int) -> bytes: return pack("!I", x)[1:] return b"".join( @@ -546,14 +642,14 @@ class WIFI_POSITIONING(_WIFI_POSITIONING): RESPOND = Respond.EXT OUT_KWARGS = (("latitude", float, None), ("longitude", float, None)) - def out_encode(self): + def out_encode(self) -> bytes: if self.latitude is None or self.longitude is None: return b"" return "{:+#010.8g},{:+#010.8g}".format( self.latitude, self.longitude ).encode() - def out_decode(self, length, payload): + def out_decode(self, length: int, payload: bytes) -> None: lat, lon = payload.decode().split(",") self.latitude = float(lat) self.longitude = float(lon) @@ -562,8 +658,8 @@ class WIFI_POSITIONING(_WIFI_POSITIONING): class MANUAL_POSITIONING(GPS303Pkt): PROTO = 0x80 - def in_decode(self, length, payload): - self.flag = payload[0] if len(payload) > 0 else None + def in_decode(self, length: int, payload: bytes) -> None: + self.flag = payload[0] if len(payload) > 0 else -1 self.reason = { 1: "Incorrect time", 2: "LBS less", @@ -573,7 +669,6 @@ class MANUAL_POSITIONING(GPS303Pkt): 6: "LBS prohibited, WiFi absent", 7: "GPS spacing < 50 m", }.get(self.flag, "Unknown") - return self class BATTERY_CHARGE(GPS303Pkt): @@ -597,12 +692,11 @@ class POSITION_UPLOAD_INTERVAL(GPS303Pkt): RESPOND = Respond.EXT OUT_KWARGS = (("interval", int, 10),) - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.interval = unpack("!H", payload[:2]) - return self - def out_encode(self): - return pack("!H", interval) + def out_encode(self) -> bytes: + return pack("!H", self.interval) class SOS_ALARM(GPS303Pkt): @@ -613,9 +707,8 @@ class UNKNOWN_B3(GPS303Pkt): PROTO = 0xB3 IN_KWARGS = (("asciidata", str, ""),) - def in_decode(self, length, payload): + def in_decode(self, length: int, payload: bytes) -> None: self.asciidata = payload.decode() - return self # Build dicts protocol number -> class and class name -> protocol number @@ -634,7 +727,9 @@ if True: # just to indent the code, sorry! PROTOS[cls.__name__] = cls.PROTO -def class_by_prefix(prefix): +def class_by_prefix( + prefix: str, +) -> Union[Type[GPS303Pkt], List[Tuple[str, int]]]: lst = [ (name, proto) for name, proto in PROTOS.items() @@ -646,15 +741,15 @@ def class_by_prefix(prefix): return CLASSES[proto] -def proto_by_name(name): +def proto_by_name(name: str) -> int: return PROTOS.get(name, -1) -def proto_of_message(packet): - return unpack("B", packet[1:2])[0] +def proto_of_message(packet: bytes) -> int: + return packet[1] -def inline_response(packet): +def inline_response(packet: bytes) -> Optional[bytes]: proto = proto_of_message(packet) if proto in CLASSES: cls = CLASSES[proto] @@ -663,19 +758,21 @@ def inline_response(packet): return None -def parse_message(packet, is_incoming=True): +def parse_message(packet: bytes, is_incoming: bool = True) -> GPS303Pkt: """From a packet (without framing bytes) derive the XXX.In object""" length, proto = unpack("BB", packet[:2]) payload = packet[2:] if proto not in CLASSES: - cause = ValueError(f"Proto {proto} is unknown") + cause: Union[DecodeError, ValueError, IndexError] = ValueError( + f"Proto {proto} is unknown" + ) else: try: if is_incoming: return CLASSES[proto].In(length, payload) else: return CLASSES[proto].Out(length, payload) - except DecodeError as e: + except (DecodeError, ValueError, IndexError) as e: cause = e if is_incoming: retobj = UNKNOWN.In(length, payload)