improve diagnistic message about left data

[loctrkd.git] / gps303 / gps303proto.py

diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py
index 2d777d94fa5bfbec0b4c4b15192a0a56f9427954..e3776c3a529276fd333d619ea1946a8264f8337b 100755 (executable)
--- a/gps303/gps303proto.py
+++ b/gps303/gps303proto.py
@@ -18,6 +18,7 @@ from datetime import datetime, timezone
 from enum import Enum
 from inspect import isclass
 from struct import error, pack, unpack
 from enum import Enum
 from inspect import isclass
 from struct import error, pack, unpack

+from time import time

 from typing import (
     Any,
     Callable,
 from typing import (
     Any,
     Callable,


@@ -31,9 +32,11 @@ from typing import (
 )
 
 __all__ = (
 )
 
 __all__ = (

+    "Stream",

     "class_by_prefix",
     "inline_response",
     "parse_message",
     "class_by_prefix",
     "inline_response",
     "parse_message",

+    "probe_buffer",

     "proto_by_name",
     "DecodeError",
     "Respond",
     "proto_by_name",
     "DecodeError",
     "Respond",


@@ -77,6 +80,75 @@ __all__ = (
     "UNKNOWN_B3",
 )
 
     "UNKNOWN_B3",
 )
 

+### 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:
 
 class DecodeError(Exception):
     def __init__(self, e: Exception, **kwargs: Any) -> None:


@@ -85,6 +157,10 @@ class DecodeError(Exception):
             setattr(self, k, v)
 
 
             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)
 def intx(x: Union[str, int]) -> int:
     if isinstance(x, str):
         x = int(x, 0)


@@ -299,7 +375,7 @@ class GPS303Pkt(metaclass=MetaPkt):
     @property
     def packed(self) -> bytes:
         payload = self.encode()
     @property
     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
 
 
         return pack("BB", length, self.PROTO) + payload
 
 


@@ -311,10 +387,16 @@ class LOGIN(GPS303Pkt):
     PROTO = 0x01
     RESPOND = Respond.INL
     # Default response for ACK, can also respond with STOP_UPLOAD
     PROTO = 0x01
     RESPOND = Respond.INL
     # Default response for ACK, can also respond with STOP_UPLOAD

+    IN_KWARGS = (("imei", str, "0000000000000000"), ("ver", int, 0))

 
     def in_decode(self, length: int, payload: bytes) -> None:
 
     def in_decode(self, length: int, payload: bytes) -> None:

-        self.imei = payload[:16].hex()
-        self.ver = payload[16]
+        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):
 
 
 class SUPERVISION(GPS303Pkt):


@@ -374,6 +456,13 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING):
 class STATUS(GPS303Pkt):
     PROTO = 0x13
     RESPOND = Respond.EXT
 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),)
 
     def in_decode(self, length: int, payload: bytes) -> None:
     OUT_KWARGS = (("upload_interval", int, 25),)
 
     def in_decode(self, length: int, payload: bytes) -> None:


@@ -385,6 +474,11 @@ class STATUS(GPS303Pkt):
         else:
             self.signal = None
 
         else:
             self.signal = None
 

+    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)
 
     def out_encode(self) -> bytes:  # Set interval in minutes
         return pack("B", self.upload_interval)
 


@@ -392,6 +486,9 @@ class STATUS(GPS303Pkt):
 class HIBERNATION(GPS303Pkt):  # Server can send to send devicee to sleep
     PROTO = 0x14
 
 class HIBERNATION(GPS303Pkt):  # Server can send to send devicee to sleep
     PROTO = 0x14
 

+    def in_encode(self) -> bytes:
+        return b""
+

 
 class RESET(GPS303Pkt):
     # Device sends when it got reset SMS
 
 class RESET(GPS303Pkt):
     # Device sends when it got reset SMS


@@ -408,6 +505,15 @@ class WHITELIST_TOTAL(GPS303Pkt):  # Server sends to initiage sync (0x58)
 
 
 class _WIFI_POSITIONING(GPS303Pkt):
 
 
 class _WIFI_POSITIONING(GPS303Pkt):

+    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":
     def in_decode(self, length: int, payload: bytes) -> None:
         self.dtime = payload[:6]
         if self.dtime == b"\0\0\0\0\0\0":


@@ -432,6 +538,28 @@ class _WIFI_POSITIONING(GPS303Pkt):
             )
             self.gsm_cells.append((locac, cellid, -sigstr))
 
             )
             self.gsm_cells.append((locac, cellid, -sigstr))
 

+    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
 
 class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING):
     PROTO = 0x17


@@ -628,6 +756,9 @@ class SETUP(GPS303Pkt):
             + [b";".join([el.encode() for el in self.phonenumbers])]
         )
 
             + [b";".join([el.encode() for el in self.phonenumbers])]
         )
 

+    def in_encode(self) -> bytes:
+        return b""
+

 
 class SYNCHRONOUS_WHITELIST(GPS303Pkt):
     PROTO = 0x58
 
 class SYNCHRONOUS_WHITELIST(GPS303Pkt):
     PROTO = 0x58


@@ -749,6 +880,18 @@ def proto_of_message(packet: bytes) -> int:
     return packet[1]
 
 
     return packet[1]
 
 

+def imei_from_packet(packet: bytes) -> Optional[str]:
+    if proto_of_message(packet) == LOGIN.PROTO:
+        msg = parse_message(packet)
+        if isinstance(msg, LOGIN):
+            return msg.imei
+    return None
+
+
+def is_goodbye_packet(packet: bytes) -> bool:
+    return proto_of_message(packet) == HIBERNATION.PROTO
+
+

 def inline_response(packet: bytes) -> Optional[bytes]:
     proto = proto_of_message(packet)
     if proto in CLASSES:
 def inline_response(packet: bytes) -> Optional[bytes]:
     proto = proto_of_message(packet)
     if proto in CLASSES:


@@ -758,6 +901,15 @@ def inline_response(packet: bytes) -> Optional[bytes]:
     return None
 
 
     return None
 
 

+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: bytes, is_incoming: bool = True) -> GPS303Pkt:
     """From a packet (without framing bytes) derive the XXX.In object"""
     length, proto = unpack("BB", packet[:2])
 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])