X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=0a222d05634930679995e6e4807d939d35104346;hb=18eda7307e92eeee6a9e2fdd2e810f98d25df654;hp=136f86ba3df3e5a47044bfac09536cd691a1f45e;hpb=2a7ebf5ab639f4f3a30d5b737f292ac0958d7d1b;p=loctrkd.git

diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py
index 136f86b..0a222d0 100755
--- a/gps303/gps303proto.py
+++ b/gps303/gps303proto.py
@@ -17,13 +17,25 @@ Forewarnings:
 from datetime import datetime, timezone
 from enum import Enum
 from inspect import isclass
-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",
     "parse_message",
     "proto_by_name",
+    "DecodeError",
     "Respond",
     "GPS303Pkt",
     "UNKNOWN",
@@ -66,13 +78,34 @@ __all__ = (
 )
 
 
-def intx(x):
+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 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")
@@ -83,21 +116,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:],
+        )
+
+    if isinstance(x, str):
+        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):
+def l3int(x: Union[str, List[int]]) -> List[int]:
     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 = [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):
@@ -112,7 +182,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,
@@ -146,10 +229,21 @@ class Respond(Enum):
 class GPS303Pkt(metaclass=MetaPkt):
     RESPOND = Respond.NON  # Do not send anything back by default
     PROTO: int
-    IN_KWARGS = ()
-    OUT_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 __init__(self, *args, **kwargs):
+        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
@@ -157,7 +251,10 @@ class GPS303Pkt(metaclass=MetaPkt):
         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
-            self.decode(self.length, self.payload)
+            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)))
@@ -166,7 +263,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(
@@ -181,30 +278,32 @@ 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):
-        # Necessary to emulate terminal, which is not implemented
-        raise NotImplementedError(
-            self.__class__.__name__ + ".decode() not implemented"
-        )
+    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):
+    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
 
 
@@ -216,18 +315,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))
 
-    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[: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
@@ -242,7 +346,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
@@ -262,9 +366,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)
@@ -281,25 +384,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),)
 
-    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 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):
@@ -312,12 +428,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):
-    def in_decode(self, 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
@@ -340,14 +465,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"))
 
 
@@ -355,7 +501,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])
 
 
@@ -363,31 +509,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):
@@ -418,7 +576,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)
@@ -437,7 +595,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)
@@ -452,7 +610,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)
@@ -464,27 +622,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):
@@ -502,8 +662,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(
@@ -524,6 +684,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
@@ -536,19 +699,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
 
-    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",
@@ -558,7 +728,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):
@@ -582,12 +751,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):
@@ -598,9 +766,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
@@ -619,7 +786,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()
@@ -631,15 +800,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]
@@ -648,13 +817,26 @@ def inline_response(packet):
     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:]
-    if proto in CLASSES:
-        return CLASSES[proto].In(length, payload)
+    if proto not in CLASSES:
+        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, ValueError, IndexError) as e:
+            cause = e
+    if is_incoming:
         retobj = UNKNOWN.In(length, payload)
-        retobj.PROTO = proto  # Override class attr with object attr
-        return retobj
+    else:
+        retobj = UNKNOWN.Out(length, payload)
+    retobj.PROTO = proto  # Override class attr with object attr
+    retobj.cause = cause
+    return retobj