X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=da5796f988eeb281f4fa2e85c64e55e12c7e44f1;hb=dbcd32b54fe0832a9020ab3cc32f886ad9761093;hp=acfaea63c272aa74bc676dcac90765d027c8ddd2;hpb=3f74a195c346809d3075b1351705eb9ad543afd5;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index acfaea6..da5796f 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -17,16 +17,16 @@ 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, Union __all__ = ( - "handle_packet", + "class_by_prefix", "inline_response", - "make_object", - "make_response", "parse_message", "proto_by_name", + "DecodeError", + "Respond", "GPS303Pkt", "UNKNOWN", "LOGIN", @@ -40,7 +40,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,28 +63,134 @@ __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) -class Dir(Enum): - IN = 0 # Incoming, no response needed - INLINE = 2 # Birirectional, use `inline_response()` - EXT = 3 # Birirectional, use external responder - OUT = 4 # Outgoing, should not appear on input +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): + """ + 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. + """ + + def __new__( + cls, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any] + ) -> "MetaPkt": + newcls = super().__new__(cls, name, bases, attrs) + setattr( + newcls, + "In", + super().__new__( + cls, + name + ".In", + (newcls,) + bases, + { + "KWARGS": newcls.IN_KWARGS, + "decode": newcls.in_decode, + "encode": newcls.in_encode, + }, + ), + ) + setattr( + newcls, + "Out", + super().__new__( + cls, + name + ".Out", + (newcls,) + bases, + { + "KWARGS": newcls.OUT_KWARGS, + "decode": newcls.out_decode, + "encode": newcls.out_encode, + }, + ), + ) + return newcls -class GPS303Pkt: - PROTO: int - DIR = Dir.INLINE # Most packets anticipate simple acknowledgement - def __init__(self, *args, **kwargs): - assert len(args) == 0 - for k, v in kwargs.items(): - setattr(self, k, v) +class Respond(Enum): + NON = 0 # Incoming, no response needed + INL = 1 # Birirectional, use `inline_response()` + EXT = 2 # Birirectional, use external responder + + +class GPS303Pkt(metaclass=MetaPkt): + RESPOND = Respond.NON # Do not send anything back by default + PROTO: int + 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: 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( @@ -88,71 +205,76 @@ class GPS303Pkt: ), ) - @classmethod - def from_packet(cls, length, payload): - return cls(payload=payload, length=length) - - def to_packet(self): - return pack("BB", self.length, self.PROTO) + self.payload - - @classmethod - def make_packet(cls, payload): - assert isinstance(payload, bytes) - length = len(payload) + 1 # plus proto byte - # if length > 6: - # length -= 6 - return pack("BB", length, cls.PROTO) + payload - - @classmethod - def inline_response(cls, packet): - if cls.DIR is Dir.INLINE: - return cls.make_packet(b"") - else: - return None + 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) -> bytes: + # Overridden in subclasses, otherwise make empty payload + return b"" + + @property + def packed(self) -> bytes: + payload = self.encode() + length = len(payload) + 1 + return pack("BB", length, self.PROTO) + payload class UNKNOWN(GPS303Pkt): PROTO = 256 # > 255 is impossible in real packets - DIR = Dir.IN class LOGIN(GPS303Pkt): PROTO = 0x01 + 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 - DIR = Dir.OUT + OUT_KWARGS = (("status", int, 1),) - def response(self, status=0): + 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 - return self.make_packet(pack("B", status)) + return pack("B", self.status) class HEARTBEAT(GPS303Pkt): PROTO = 0x08 + RESPOND = Respond.INL class _GPS_POSITIONING(GPS303Pkt): - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(length, payload) + RESPOND = Respond.INL + + 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 @@ -162,16 +284,14 @@ 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 - @classmethod - def inline_response(cls, packet): + def out_encode(self) -> bytes: tup = datetime.utcnow().timetuple() ttup = (tup[0] % 100,) + tup[1:6] - return cls.make_packet(pack("BBBBBB", *ttup)) + return pack("BBBBBB", *ttup) class GPS_POSITIONING(_GPS_POSITIONING): @@ -184,58 +304,42 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): class STATUS(GPS303Pkt): PROTO = 0x13 - DIR = Dir.EXT - - @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 - ) + RESPOND = Respond.EXT + OUT_KWARGS = (("upload_interval", int, 25),) + + 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 response(self, upload_interval=25): # Set interval in minutes - return self.make_packet(pack("B", upload_interval)) + def out_encode(self) -> bytes: # Set interval in minutes + return pack("B", self.upload_interval) -class HIBERNATION(GPS303Pkt): +class HIBERNATION(GPS303Pkt): # Server can send to send devicee to sleep PROTO = 0x14 - DIR = Dir.EXT - - def response(self): # Server can send to send devicee to sleep - return self.make_packet(b"") -class RESET(GPS303Pkt): # Device sends when it got reset SMS +class RESET(GPS303Pkt): + # Device sends when it got reset SMS + # Server can send to initiate factory reset PROTO = 0x15 - DIR = Dir.EXT - - def response(self): # Server can send to initiate factory reset - return self.make_packet(b"") class WHITELIST_TOTAL(GPS303Pkt): # Server sends to initiage sync (0x58) PROTO = 0x16 - DIR = Dir.OUT + OUT_KWARGS = (("number", int, 3),) - def response(self, number=3): # Number of whitelist entries - return self.make_packet(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 @@ -258,59 +362,54 @@ 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 - @classmethod - def inline_response(cls, packet): - return cls.make_packet( - bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S")) - ) + def out_encode(self) -> bytes: + return bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S")) class TIME(GPS303Pkt): PROTO = 0x30 + RESPOND = Respond.INL - @classmethod - def inline_response(cls, packet): - return cls.make_packet( - pack("!HBBBBB", *datetime.utcnow().timetuple()[:6]) - ) + def out_encode(self) -> bytes: + return pack("!HBBBBB", *datetime.utcnow().timetuple()[:6]) class PROHIBIT_LBS(GPS303Pkt): PROTO = 0x33 - DIR = Dir.OUT + OUT_KWARGS = (("status", int, 1),) - def response(self, status=1): # Server sent, 0-off, 1-on - return self.make_packet(pack("B", status)) + def out_encode(self) -> bytes: # Server sent, 0-off, 1-on + return pack("B", self.status) class GPS_LBS_SWITCH_TIMES(GPS303Pkt): PROTO = 0x34 - DIR = Dir.OUT - def response(self): - # 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 - return self.make_packet(b"") # TODO + # 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) -> bytes: + return b"" # TODO class _SET_PHONE(GPS303Pkt): - DIR = Dir.OUT + OUT_KWARGS = (("phone", str, ""),) - def response(self, phone): - return self.make_packet(phone.encode()) + def out_encode(self) -> bytes: + self.phone: str + return self.phone.encode("") class REMOTE_MONITOR_PHONE(_SET_PHONE): @@ -331,64 +430,72 @@ class MOM_PHONE(_SET_PHONE): class STOP_UPLOAD(GPS303Pkt): # Server response to LOGIN to thwart the device PROTO = 0x44 - DIR = Dir.OUT - - def response(self): - return self.make_packet(b"") class GPS_OFF_PERIOD(GPS303Pkt): PROTO = 0x46 - DIR = Dir.OUT - - def response(self, onoff=0, fm="0000", to="2359"): - return self.make_packet( - pack("B", onoff) + bytes.fromhex(fm) + bytes.fromhex(to) + OUT_KWARGS = ( + ("onoff", int, 0), + ("fm", hhmm, "0000"), + ("to", hhmm, "2359"), + ) + + def out_encode(self) -> bytes: + return ( + pack("B", self.onoff) + + bytes.fromhex(self.fm) + + bytes.fromhex(self.to) ) class DND_PERIOD(GPS303Pkt): PROTO = 0x47 - DIR = Dir.OUT - - def response( - self, onoff=0, week=3, fm1="0000", to1="2359", fm2="0000", to2="2359" - ): - return self.make_packet( - pack("B", onoff) - + pack("B", week) - + bytes.fromhex(fm1) - + bytes.fromhex(to1) - + bytes.fromhex(fm2) - + bytes.fromhex(to2) + OUT_KWARGS = ( + ("onoff", int, 0), + ("week", int, 3), + ("fm1", hhmm, "0000"), + ("to1", hhmm, "2359"), + ("fm2", hhmm, "0000"), + ("to2", hhmm, "2359"), + ) + + def out_encode(self) -> bytes: + return ( + pack("B", self.onoff) + + pack("B", self.week) + + bytes.fromhex(self.fm1) + + bytes.fromhex(self.to1) + + bytes.fromhex(self.fm2) + + bytes.fromhex(self.to2) ) class RESTART_SHUTDOWN(GPS303Pkt): PROTO = 0x48 - DIR = Dir.OUT + OUT_KWARGS = (("flag", int, 0),) - def response(self, flag=0): + def out_encode(self) -> bytes: # 1 - restart # 2 - shutdown - return self.make_packet(pack("B", flag)) + return pack("B", self.flag) class DEVICE(GPS303Pkt): PROTO = 0x49 - DIR = Dir.OUT + OUT_KWARGS = (("flag", int, 0),) - def response(self, flag=0): - # 0 - Stop looking for equipment - # 1 - Start looking for equipment - return self.make_packet(pack("B", flag)) + # 0 - Stop looking for equipment + # 1 - Start looking for equipment + def out_encode(self) -> bytes: + return pack("B", self.flag) class ALARM_CLOCK(GPS303Pkt): PROTO = 0x50 - DIR = Dir.OUT - def response(self, alarms=((0, "0000"), (0, "0000"), (0, "0000"))): + def out_encode(self) -> bytes: + # TODO implement parsing kwargs + alarms = ((0, "0000"), (0, "0000"), (0, "0000")) return b"".join( pack("B", day) + bytes.fromhex(tm) for day, tm in alarms ) @@ -397,49 +504,46 @@ class ALARM_CLOCK(GPS303Pkt): 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] class SETUP(GPS303Pkt): PROTO = 0x57 - DIR = Dir.EXT - - def response( - self, - uploadintervalseconds=0x0300, - binaryswitch=0b00110001, - alarms=[0, 0, 0], - dndtimeswitch=0, - dndtimes=[0, 0, 0], - gpstimeswitch=0, - gpstimestart=0, - gpstimestop=0, - phonenumbers=["", "", ""], - ): - def pack3b(x): + RESPOND = Respond.EXT + OUT_KWARGS = ( + ("uploadintervalseconds", intx, 0x0300), + ("binaryswitch", intx, 0b00110001), + ("alarms", l3int, [0, 0, 0]), + ("dndtimeswitch", int, 0), + ("dndtimes", l3int, [0, 0, 0]), + ("gpstimeswitch", int, 0), + ("gpstimestart", int, 0), + ("gpstimestop", int, 0), + ("phonenumbers", l3str, ["", "", ""]), + ) + + def out_encode(self) -> bytes: + def pack3b(x: int) -> bytes: return pack("!I", x)[1:] - payload = b"".join( + return b"".join( [ - pack("!H", uploadintervalseconds), - pack("B", binaryswitch), + pack("!H", self.uploadintervalseconds), + pack("B", self.binaryswitch), ] - + [pack3b(el) for el in alarms] + + [pack3b(el) for el in self.alarms] + [ - pack("B", dndtimeswitch), + pack("B", self.dndtimeswitch), ] - + [pack3b(el) for el in dndtimes] + + [pack3b(el) for el in self.dndtimes] + [ - pack("B", gpstimeswitch), - pack("!H", gpstimestart), - pack("!H", gpstimestop), + pack("B", self.gpstimeswitch), + pack("!H", self.gpstimestart), + pack("!H", self.gpstimestop), ] - + [b";".join([el.encode() for el in phonenumbers])] + + [b";".join([el.encode() for el in self.phonenumbers])] ) - return self.make_packet(payload) class SYNCHRONOUS_WHITELIST(GPS303Pkt): @@ -452,26 +556,27 @@ class RESTORE_PASSWORD(GPS303Pkt): class WIFI_POSITIONING(_WIFI_POSITIONING): PROTO = 0x69 - DIR = Dir.EXT + RESPOND = Respond.EXT + OUT_KWARGS = (("latitude", float, None), ("longitude", float, None)) - def response(self, lat=None, lon=None): - if lat is None or lon is None: - payload = b"" - else: - payload = "{:+#010.8g},{:+#010.8g}".format(lat, lon).encode( - "ascii" - ) - return self.make_packet(payload) + 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: int, payload: bytes) -> None: + lat, lon = payload.decode().split(",") + self.latitude = float(lat) + self.longitude = float(lon) class MANUAL_POSITIONING(GPS303Pkt): PROTO = 0x80 - DIR = Dir.EXT - @classmethod - def from_packet(cls, length, payload): - self = super().from_packet(length, payload) - self.flag = payload[0] + 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", @@ -482,9 +587,6 @@ class MANUAL_POSITIONING(GPS303Pkt): 7: "GPS spacing < 50 m", }.get(self.flag, "Unknown") - def response(self): - return self.make_packet(b"") - class BATTERY_CHARGE(GPS303Pkt): PROTO = 0x81 @@ -504,22 +606,28 @@ class VIBRATION_RECEIVED(GPS303Pkt): class POSITION_UPLOAD_INTERVAL(GPS303Pkt): PROTO = 0x98 - DIR = Dir.EXT + 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 response(self, interval=10): - return self.make_packet(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 = {} @@ -536,56 +644,55 @@ if True: # just to indent the code, sorry! PROTOS[cls.__name__] = cls.PROTO -def proto_by_name(name): - return PROTOS.get(name, -1) +def class_by_prefix(prefix: str) -> Union[type, List[Tuple[str, int]]]: + lst = [ + (name, proto) + for name, proto in PROTOS.items() + if name.upper().startswith(prefix.upper()) + ] + if len(lst) != 1: + return lst + _, proto = lst[0] + return CLASSES[proto] -def proto_of_message(packet): - return unpack("B", packet[1:2])[0] +def proto_by_name(name: str) -> int: + return PROTOS.get(name, -1) -def inline_response(packet): - proto = proto_of_message(packet) - if proto in CLASSES: - return CLASSES[proto].inline_response(packet) - else: - return None +def proto_of_message(packet: bytes) -> int: + return packet[1] -def make_object(length, proto, payload): +def inline_response(packet: bytes) -> Optional[bytes]: + proto = proto_of_message(packet) 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 + cls = CLASSES[proto] + if cls.RESPOND is Respond.INL: + return cls.Out().packed + return None -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) - - -def handle_packet(packet): # DEPRECATED - if len(packet) < 6 or packet[:2] != b"xx" or packet[-2:] != b"\r\n": - return UNKNOWN.from_packet(len(packet), packet) - return parse_message(packet[2:-2]) - - -def make_response(msg, **kwargs): # DEPRECATED - inframe = msg.response(**kwargs) - return None if inframe is None else b"xx" + inframe + b"\r\n" + 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