X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=efb02d249f6c118185ce2a66b23675c9acf85efd;hb=c3bc6d5bbdc0d0bf10e338c6e3bad1a519d5afa0;hp=59a700e4ebb2fdb11a06f60476ef95478bd58884;hpb=66860d17a8d88c7e1084486cfe9cb9379d8ffe31;p=loctrkd.git

diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py
index 59a700e..efb02d2 100755
--- a/gps303/gps303proto.py
+++ b/gps303/gps303proto.py
@@ -17,15 +17,29 @@ 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 time import time
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    TYPE_CHECKING,
+    Union,
+)
 
 __all__ = (
+    "Stream",
     "class_by_prefix",
     "inline_response",
-    "make_object",
     "parse_message",
+    "probe_buffer",
     "proto_by_name",
+    "proto_name",
+    "DecodeError",
     "Respond",
     "GPS303Pkt",
     "UNKNOWN",
@@ -40,7 +54,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,18 +77,110 @@ __all__ = (
     "CHARGER_DISCONNECTED",
     "VIBRATION_RECEIVED",
     "POSITION_UPLOAD_INTERVAL",
+    "SOS_ALARM",
+    "UNKNOWN_B3",
 )
 
-log = getLogger("gps303")
+PROTO_PREFIX = "ZX"
+
+### Deframer ###
+
+MAXBUFFER: int = 4096
+
+
+class Stream:
+    def __init__(self) -> None:
+        self.buffer = b""
+
+    @staticmethod
+    def enframe(buffer: bytes) -> bytes:
+        return b"xx" + buffer + b"\r\n"
+
+    def recv(self, segment: bytes) -> List[Union[bytes, str]]:
+        """
+        Process next segment of the stream. Return successfully deframed
+        packets as `bytes` and error messages as `str`.
+        """
+        when = time()
+        self.buffer += segment
+        if len(self.buffer) > MAXBUFFER:
+            # We are receiving junk. Let's drop it or we run out of memory.
+            self.buffer = b""
+            return [f"More than {MAXBUFFER} unparseable data, dropping"]
+        msgs: List[Union[bytes, str]] = []
+        while True:
+            framestart = self.buffer.find(b"xx")
+            if framestart == -1:  # No frames, return whatever we have
+                break
+            if framestart > 0:  # Should not happen, report
+                msgs.append(
+                    f'Undecodable data ({framestart}) "{self.buffer[:framestart][:64].hex()}"'
+                )
+                self.buffer = self.buffer[framestart:]
+            # At this point, buffer starts with a packet
+            if len(self.buffer) < 6:  # no len and proto - cannot proceed
+                break
+            exp_end = self.buffer[2] + 3  # Expect '\r\n' here
+            frameend = 0
+            # Length field can legitimeely be much less than the
+            # length of the packet (e.g. WiFi positioning), but
+            # it _should not_ be greater. Still sometimes it is.
+            # Luckily, not by too much: by maybe two or three bytes?
+            # Do this embarrassing hack to avoid accidental match
+            # of some binary data in the packet against '\r\n'.
+            while True:
+                frameend = self.buffer.find(b"\r\n", frameend + 1)
+                if frameend == -1 or frameend >= (
+                    exp_end - 3
+                ):  # Found realistic match or none
+                    break
+            if frameend == -1:  # Incomplete frame, return what we have
+                break
+            packet = self.buffer[2:frameend]
+            self.buffer = self.buffer[frameend + 2 :]
+            if len(packet) < 2:  # frameend comes too early
+                msgs.append(f"Packet too short: {packet.hex()}")
+            else:
+                msgs.append(packet)
+        return msgs
+
+    def close(self) -> bytes:
+        ret = self.buffer
+        self.buffer = b""
+        return ret
+
+
+### Parser/Constructor ###
+
+
+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 maybe(typ: type) -> Callable[[Any], Any]:
+    return lambda x: None if x is None else typ(x)
+
+
+def intx(x: Union[str, int]) -> int:
+    if isinstance(x, str):
+        x = int(x, 0)
+    return x
 
 
-def intx(x):
+def boolx(x: Union[str, bool]) -> bool:
     if isinstance(x, str):
-        x = int(x, 0)
+        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):
+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")
@@ -74,21 +191,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 l3int(x):
+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:],
+        )
+
     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):
@@ -103,19 +257,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
 
@@ -129,22 +304,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(
@@ -159,24 +353,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
+        length = getattr(self, "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
@@ -186,20 +390,23 @@ class LOGIN(GPS303Pkt):
     PROTO = 0x01
     RESPOND = Respond.INL
     # Default response for ACK, can also respond with STOP_UPLOAD
+    IN_KWARGS = (("imei", str, "0000000000000000"), ("ver", int, 0))
 
-    @classmethod
-    def from_packet(cls, length, payload):
-        self = super().from_packet(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[:8].ljust(8, b"\0").hex()
+        self.ver = payload[8]
+
+    def in_encode(self) -> bytes:
+        return bytes.fromhex(self.imei).ljust(8, b"\0")[:8] + pack(
+            "B", self.ver
+        )
 
 
 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
@@ -214,15 +421,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
@@ -232,12 +438,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)
@@ -254,33 +459,38 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING):
 class STATUS(GPS303Pkt):
     PROTO = 0x13
     RESPOND = Respond.EXT
+    IN_KWARGS = (
+        ("batt", int, 100),
+        ("ver", int, 0),
+        ("timezone", int, 0),
+        ("intvl", int, 0),
+        ("signal", maybe(int), None),
+    )
     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 in_encode(self) -> bytes:
+        return pack("BBBB", self.batt, self.ver, self.timezone, self.intvl) + (
+            b"" if self.signal is None else pack("B", self.signal)
+        )
+
+    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
+
+    def in_encode(self) -> bytes:
+        return b""
 
 
 class RESET(GPS303Pkt):
@@ -293,14 +503,21 @@ 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)
+    IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = (
+        # IN_KWARGS = (
+        ("dtime", bytes, b"\0\0\0\0\0\0"),
+        ("wifi_aps", list, []),
+        ("mcc", int, 0),
+        ("mnc", int, 0),
+        ("gsm_cells", list, []),
+    )
+
+    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
@@ -323,14 +540,35 @@ class _WIFI_POSITIONING(GPS303Pkt):
                 "!HHB", gsm_slice[4 + i * 5 : 9 + i * 5]
             )
             self.gsm_cells.append((locac, cellid, -sigstr))
-        return self
+
+    def in_encode(self) -> bytes:
+        self.length = len(self.wifi_aps)
+        return b"".join(
+            [
+                self.dtime,
+                b"".join(
+                    [
+                        bytes.fromhex(mac.replace(":", "")).ljust(6, b"\0")[:6]
+                        + pack("B", -sigstr)
+                        for mac, sigstr in self.wifi_aps
+                    ]
+                ),
+                pack("!BHB", len(self.gsm_cells), self.mcc, self.mnc),
+                b"".join(
+                    [
+                        pack("!HHB", locac, cellid, -sigstr)
+                        for locac, cellid, sigstr in self.gsm_cells
+                    ]
+                ),
+            ]
+        )
 
 
 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"))
 
 
@@ -338,7 +576,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])
 
 
@@ -346,31 +584,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):
@@ -401,7 +651,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)
@@ -420,7 +670,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)
@@ -433,9 +683,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)
@@ -447,29 +697,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
 
-    @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):
@@ -487,8 +737,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(
@@ -509,6 +759,9 @@ class SETUP(GPS303Pkt):
             + [b";".join([el.encode() for el in self.phonenumbers])]
         )
 
+    def in_encode(self) -> bytes:
+        return b""
+
 
 class SYNCHRONOUS_WHITELIST(GPS303Pkt):
     PROTO = 0x58
@@ -521,21 +774,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",
@@ -545,7 +803,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):
@@ -569,20 +826,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 = {}
@@ -599,7 +861,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()
@@ -611,16 +875,40 @@ def class_by_prefix(prefix):
     return CLASSES[proto]
 
 
-def proto_by_name(name):
+def proto_name(obj: Union[MetaPkt, GPS303Pkt]) -> str:
+    return (
+        PROTO_PREFIX
+        + ":"
+        + (
+            obj.__class__.__name__
+            if isinstance(obj, GPS303Pkt)
+            else obj.__name__
+        )
+    ).ljust(16, "\0")[:16]
+
+
+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) -> str:
+    return proto_name(CLASSES.get(packet[1], UNKNOWN))
 
 
-def inline_response(packet):
-    proto = proto_of_message(packet)
+def imei_from_packet(packet: bytes) -> Optional[str]:
+    if packet[1] == LOGIN.PROTO:
+        msg = parse_message(packet)
+        if isinstance(msg, LOGIN):
+            return msg.imei
+    return None
+
+
+def is_goodbye_packet(packet: bytes) -> bool:
+    return packet[1] == HIBERNATION.PROTO
+
+
+def inline_response(packet: bytes) -> Optional[bytes]:
+    proto = packet[1]
     if proto in CLASSES:
         cls = CLASSES[proto]
         if cls.RESPOND is Respond.INL:
@@ -628,29 +916,35 @@ 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 probe_buffer(buffer: bytes) -> bool:
+    framestart = buffer.find(b"xx")
+    if framestart < 0:
+        return False
+    if len(buffer) - framestart < 6:
+        return False
+    return True
 
 
-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, IndexError] = 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, ValueError, IndexError) 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