X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=c96a6dc945312ef4b0f9df46239afa06c73f6370;hb=66f552e63fb7568c6cd30e3c3c0b2735acadc418;hp=3959dc82242cb0181ee0204f1a7ea88cc63d86f5;hpb=5fb0153ac3c0d8b0986575aa23f6f33a17512031;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index 3959dc8..c96a6dc 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -17,15 +17,25 @@ Forewarnings: from datetime import datetime, timezone from enum import Enum from inspect import isclass -from logging import getLogger -from struct import pack, unpack +from struct import error, pack, unpack +from typing import ( + Any, + Callable, + Dict, + List, + Optional, + Tuple, + Type, + TYPE_CHECKING, + Union, +) __all__ = ( "class_by_prefix", "inline_response", - "make_object", "parse_message", "proto_by_name", + "DecodeError", "Respond", "GPS303Pkt", "UNKNOWN", @@ -40,7 +50,18 @@ __all__ = ( "WHITELIST_TOTAL", "WIFI_OFFLINE_POSITIONING", "TIME", + "PROHIBIT_LBS", + "GPS_LBS_SWITCH_TIMES", + "REMOTE_MONITOR_PHONE", + "SOS_PHONE", + "DAD_PHONE", "MOM_PHONE", + "STOP_UPLOAD", + "GPS_OFF_PERIOD", + "DND_PERIOD", + "RESTART_SHUTDOWN", + "DEVICE", + "ALARM_CLOCK", "STOP_ALARM", "SETUP", "SYNCHRONOUS_WHITELIST", @@ -52,9 +73,49 @@ __all__ = ( "CHARGER_DISCONNECTED", "VIBRATION_RECEIVED", "POSITION_UPLOAD_INTERVAL", + "SOS_ALARM", + "UNKNOWN_B3", ) -log = getLogger("gps303") + +class DecodeError(Exception): + def __init__(self, e: Exception, **kwargs: Any) -> None: + super().__init__(e) + for k, v in kwargs.items(): + setattr(self, k, v) + + +def intx(x: Union[str, int]) -> int: + if isinstance(x, str): + x = int(x, 0) + 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") + hh = int(x[:2]) + mm = int(x[2:]) + if hh < 0 or hh > 23 or mm < 0 or mm > 59: + raise ValueError(str(x) + " does not contain valid hours and minutes") + return x + + +def l3str(x: Union[str, List[str]]) -> List[str]: + if isinstance(x, str): + lx = x.split(",") + 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 l3int(x: Union[str, List[int]]) -> List[int]: + if isinstance(x, str): + lx = [int(el) for el in x.split(",")] + 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): @@ -69,19 +130,40 @@ 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, name + ".In", (newcls,) + bases, - {"KWARGS": newcls.IN_KWARGS, "encode": newcls.in_encode}, + { + "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, "encode": newcls.out_encode}, + { + "KWARGS": newcls.OUT_KWARGS, + "decode": newcls.out_decode, + "encode": newcls.out_encode, + }, ) return newcls @@ -95,22 +177,41 @@ class Respond(Enum): class GPS303Pkt(metaclass=MetaPkt): RESPOND = Respond.NON # Do not send anything back by default PROTO: int - # Have these kwargs for now, TODO redo - IN_KWARGS = (("length", int, 0), ("payload", bytes, b"")) - OUT_KWARGS = () - - def __init__(self, *args, **kwargs): - assert len(args) == 0 - for kw, typ, dfl in self.KWARGS: - setattr(self, kw, typ(kwargs.pop(kw, dfl))) - if kwargs: - print("KWARGS", self.KWARGS) - print("kwargs", kwargs) - raise TypeError( - self.__class__.__name__ + " stray kwargs " + str(kwargs) - ) + 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"] + + 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 + """ + assert not args or (len(args) == 2 and not kwargs) + if args: # guaranteed to be two arguments at this point + self.length, self.payload = args + try: + self.decode(self.length, self.payload) + except error as e: + raise DecodeError(e, obj=self) + else: + for kw, typ, dfl in self.KWARGS: + setattr(self, kw, typ(kwargs.pop(kw, dfl))) + if kwargs: + raise ValueError( + self.__class__.__name__ + " stray kwargs " + str(kwargs) + ) - def __repr__(self): + def __repr__(self) -> str: return "{}({})".format( self.__class__.__name__, ", ".join( @@ -125,24 +226,34 @@ class GPS303Pkt(metaclass=MetaPkt): ), ) - def in_encode(self): + 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: int, packet: bytes) -> None: + # Overridden in subclasses, otherwise do not decode payload + return + + 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 - @classmethod - def from_packet(cls, length, payload): - return cls.In(payload=payload, length=length) - class UNKNOWN(GPS303Pkt): PROTO = 256 # > 255 is impossible in real packets @@ -153,19 +264,16 @@ class LOGIN(GPS303Pkt): RESPOND = Respond.INL # Default response for ACK, can also respond with STOP_UPLOAD - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(length, payload) + def in_decode(self, length: int, payload: bytes) -> None: self.imei = payload[:-1].hex() self.ver = unpack("B", payload[-1:])[0] - return self 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 @@ -180,15 +288,14 @@ class HEARTBEAT(GPS303Pkt): class _GPS_POSITIONING(GPS303Pkt): RESPOND = Respond.INL - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(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 else: + yr, mo, da, hr, mi, se = unpack("BBBBBB", self.dtime) self.devtime = datetime( - *unpack("BBBBBB", self.dtime), tzinfo=timezone.utc + 2000 + yr, mo, da, hr, mi, se, tzinfo=timezone.utc ) self.gps_data_length = payload[6] >> 4 self.gps_nb_sat = payload[6] & 0x0F @@ -198,12 +305,11 @@ class _GPS_POSITIONING(GPS303Pkt): flip_lat = not bool(flags & 0b0000010000000000) # bit 5 self.heading = flags & 0b0000001111111111 # bits 6 - last self.latitude = lat / (30000 * 60) * (-1 if flip_lat else 1) - self.longitude = lon / (30000 * 60) * (-2 if flip_lon else 1) + 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) @@ -222,31 +328,21 @@ class STATUS(GPS303Pkt): RESPOND = Respond.EXT OUT_KWARGS = (("upload_interval", int, 25),) - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(length, payload) - if len(payload) == 5: - ( - self.batt, - self.ver, - self.timezone, - self.intvl, - self.signal, - ) = unpack("BBBBB", payload) - elif len(payload) == 4: - self.batt, self.ver, self.timezone, self.intvl = unpack( - "BBBB", 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: Optional[int] = payload[4] + else: self.signal = None - return self - def out_encode(self): # Set interval in minutes - return cls.make_packet(pack("B", self.upload_interval)) + def out_encode(self) -> bytes: # Set interval in minutes + return pack("B", self.upload_interval) class HIBERNATION(GPS303Pkt): # Server can send to send devicee to sleep PROTO = 0x14 - RESPOND = Respond.INL class RESET(GPS303Pkt): @@ -259,14 +355,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): - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(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 @@ -289,14 +383,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")) @@ -304,7 +397,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]) @@ -312,7 +405,7 @@ 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) @@ -328,15 +421,16 @@ class GPS_LBS_SWITCH_TIMES(GPS303Pkt): # HHMM - Time of boot # 00/01 - Don't set / Set time of shutdown # HHMM - Time of shutdown - def out_encode(self): + def out_encode(self) -> bytes: return b"" # TODO 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): @@ -361,9 +455,13 @@ class STOP_UPLOAD(GPS303Pkt): # Server response to LOGIN to thwart the device class GPS_OFF_PERIOD(GPS303Pkt): PROTO = 0x46 - OUT_KWARGS = (("onoff", int, 0), ("fm", str, "0000"), ("to", str, "2359")) + OUT_KWARGS = ( + ("onoff", int, 0), + ("fm", hhmm, "0000"), + ("to", hhmm, "2359"), + ) - def out_encode(self): + def out_encode(self) -> bytes: return ( pack("B", self.onoff) + bytes.fromhex(self.fm) @@ -376,13 +474,13 @@ class DND_PERIOD(GPS303Pkt): OUT_KWARGS = ( ("onoff", int, 0), ("week", int, 3), - ("fm1", str, "0000"), - ("to1", str, "2359"), - ("fm2", str, "0000"), - ("to2", str, "2359"), + ("fm1", hhmm, "0000"), + ("to1", hhmm, "2359"), + ("fm2", hhmm, "0000"), + ("to2", hhmm, "2359"), ) - def out_endode(self): + def out_encode(self) -> bytes: return ( pack("B", self.onoff) + pack("B", self.week) @@ -395,9 +493,9 @@ class DND_PERIOD(GPS303Pkt): class RESTART_SHUTDOWN(GPS303Pkt): PROTO = 0x48 - OUT_KWARGS = (("flag", int, 2),) + OUT_KWARGS = (("flag", int, 0),) - def out_encode(self): + def out_encode(self) -> bytes: # 1 - restart # 2 - shutdown return pack("B", self.flag) @@ -409,48 +507,45 @@ 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 - def out_encode(self): + def out_encode(self) -> bytes: # TODO implement parsing kwargs alarms = ((0, "0000"), (0, "0000"), (0, "0000")) return b"".join( - cls("B", day) + bytes.fromhex(tm) for day, tm in alarms + pack("B", day) + bytes.fromhex(tm) for day, tm in alarms ) class STOP_ALARM(GPS303Pkt): PROTO = 0x56 - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(length, payload) + def in_decode(self, length: int, payload: bytes) -> None: self.flag = payload[0] - return self class SETUP(GPS303Pkt): PROTO = 0x57 RESPOND = Respond.EXT - OUT_KWARGS = ( # TODO handle properly - ("uploadintervalseconds", int, 0x0300), - ("binaryswitch", int, 0b00110001), - ("alarms", lambda x: x, [0, 0, 0]), + OUT_KWARGS = ( + ("uploadintervalseconds", intx, 0x0300), + ("binaryswitch", intx, 0b00110001), + ("alarms", l3int, [0, 0, 0]), ("dndtimeswitch", int, 0), - ("dndtimes", lambda x: x, [0, 0, 0]), + ("dndtimes", l3int, [0, 0, 0]), ("gpstimeswitch", int, 0), ("gpstimestart", int, 0), ("gpstimestop", int, 0), - ("phonenumbers", lambda x: x, ["", "", ""]), + ("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( @@ -483,21 +578,26 @@ class RESTORE_PASSWORD(GPS303Pkt): class WIFI_POSITIONING(_WIFI_POSITIONING): PROTO = 0x69 RESPOND = Respond.EXT - OUT_KWARGS = (("lat", float, None), ("lon", float, None)) + OUT_KWARGS = (("latitude", float, None), ("longitude", float, None)) - def out_encode(self): - if self.lat is None or self.lon is None: + def out_encode(self) -> bytes: + if self.latitude is None or self.longitude is None: return b"" - return "{:+#010.8g},{:+#010.8g}".format(self.lat, self.lon).encode() + return "{:+#010.8g},{:+#010.8g}".format( + self.latitude, self.longitude + ).encode() + + def out_decode(self, length: int, payload: bytes) -> None: + lat, lon = payload.decode().split(",") + self.latitude = float(lat) + self.longitude = float(lon) class MANUAL_POSITIONING(GPS303Pkt): PROTO = 0x80 - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(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", @@ -507,7 +607,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): @@ -531,20 +630,25 @@ class POSITION_UPLOAD_INTERVAL(GPS303Pkt): RESPOND = Respond.EXT OUT_KWARGS = (("interval", int, 10),) - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(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): PROTO = 0x99 +class UNKNOWN_B3(GPS303Pkt): + PROTO = 0xB3 + IN_KWARGS = (("asciidata", str, ""),) + + def in_decode(self, length: int, payload: bytes) -> None: + self.asciidata = payload.decode() + + # Build dicts protocol number -> class and class name -> protocol number CLASSES = {} PROTOS = {} @@ -561,7 +665,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() @@ -573,15 +679,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] @@ -590,29 +696,26 @@ def inline_response(packet): return None -def make_object(length, proto, payload): - if proto in CLASSES: - return CLASSES[proto].from_packet(length, payload) - else: - retobj = UNKNOWN.from_packet(length, payload) - retobj.PROTO = proto # Override class attr with object attr - return retobj - - -def parse_message(packet): +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:] - adjust = 2 if proto == STATUS.PROTO else 4 # Weird special case - if ( - proto not in (WIFI_POSITIONING.PROTO, WIFI_OFFLINE_POSITIONING.PROTO) - and length > 1 - and len(payload) + adjust != length - ): - log.warning( - "With proto %d length is %d but payload length is %d+%d", - proto, - length, - len(payload), - adjust, + if proto not in CLASSES: + cause: Union[DecodeError, ValueError] = ValueError( + f"Proto {proto} is unknown" ) - return make_object(length, proto, payload) + else: + try: + if is_incoming: + return CLASSES[proto].In(length, payload) + else: + return CLASSES[proto].Out(length, payload) + except DecodeError as e: + cause = e + if is_incoming: + retobj = UNKNOWN.In(length, payload) + else: + retobj = UNKNOWN.Out(length, payload) + retobj.PROTO = proto # Override class attr with object attr + retobj.cause = cause + return retobj