X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=dcc335b184fc6fa59249ce2c692f37c54d1587d5;hb=faa8ce9d87530105ec2ad2f4809a9ace581a2ad6;hp=4789b033a3936873150d12b9cad4e1502869be23;hpb=b613110bb16c95d9b641882c2ad6869e3ced1a0c;p=loctrkd.git

diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py
index 4789b03..dcc335b 100755
--- a/gps303/gps303proto.py
+++ b/gps303/gps303proto.py
@@ -15,15 +15,21 @@ Forewarnings:
 """
 
 from datetime import datetime, timezone
+from enum import Enum
 from inspect import isclass
 from logging import getLogger
 from struct import pack, unpack
 
 __all__ = (
+    "class_by_prefix",
     "handle_packet",
+    "inline_response",
     "make_object",
     "make_response",
-    "set_config",
+    "parse_message",
+    "proto_by_name",
+    "Dir",
+    "GPS303Pkt",
     "UNKNOWN",
     "LOGIN",
     "SUPERVISION",
@@ -53,9 +59,16 @@ __all__ = (
 log = getLogger("gps303")
 
 
-class _GT06pkt:
+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
+
+
+class GPS303Pkt:
     PROTO: int
-    CONFIG = None
+    DIR = Dir.IN  # Do not send anything back by default
 
     def __init__(self, *args, **kwargs):
         assert len(args) == 0
@@ -78,56 +91,68 @@ class _GT06pkt:
         )
 
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        return cls(proto=proto, payload=payload, length=length)
+    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
 
-    def response(self, *args):
-        if len(args) == 0:
+    @classmethod
+    def inline_response(cls, packet):
+        if cls.DIR is Dir.INLINE:
+            return cls.make_packet(b"")
+        else:
             return None
-        assert len(args) == 1 and isinstance(args[0], bytes)
-        payload = args[0]
-        length = len(payload) + 1
-        if length > 6:
-            length -= 6
-        return b"xx" + pack("BB", length, self.proto) + payload + b"\r\n"
 
 
-class UNKNOWN(_GT06pkt):
-    pass
+class UNKNOWN(GPS303Pkt):
+    PROTO = 256  # > 255 is impossible in real packets
 
 
-class LOGIN(_GT06pkt):
+class LOGIN(GPS303Pkt):
     PROTO = 0x01
+    DIR = Dir.INLINE
+    # Default response for ACK, can also respond with STOP_UPLOAD
 
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        self = super().from_packet(length, proto, payload)
+    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 response(self):
-        return super().response(b"")
-
 
-class SUPERVISION(_GT06pkt):  # Server sends supervision number status
+class SUPERVISION(GPS303Pkt):
     PROTO = 0x05
+    DIR = Dir.OUT
 
-    def response(self, supnum=0):
+    @classmethod
+    def response(cls, status=0):
         # 1: The device automatically answers Pickup effect
         # 2: Automatically Answering Two-way Calls
         # 3: Ring manually answer the two-way call
-        return super().response(b"")
+        return cls.make_packet(pack("B", status))
 
 
-class HEARTBEAT(_GT06pkt):
+class HEARTBEAT(GPS303Pkt):
     PROTO = 0x08
+    DIR = Dir.INLINE
 
 
-class _GPS_POSITIONING(_GT06pkt):
+class _GPS_POSITIONING(GPS303Pkt):
+    DIR = Dir.INLINE
+
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        self = super().from_packet(length, proto, payload)
+    def from_packet(cls, length, payload):
+        self = super().from_packet(length, payload)
         self.dtime = payload[:6]
         if self.dtime == b"\0\0\0\0\0\0":
             self.devtime = None
@@ -148,8 +173,11 @@ class _GPS_POSITIONING(_GT06pkt):
         self.flags = flags
         return self
 
-    def response(self):
-        return super().response(self.dtime)
+    @classmethod
+    def inline_response(cls, packet):
+        tup = datetime.utcnow().timetuple()
+        ttup = (tup[0] % 100,) + tup[1:6]
+        return cls.make_packet(pack("BBBBBB", *ttup))
 
 
 class GPS_POSITIONING(_GPS_POSITIONING):
@@ -160,12 +188,13 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING):
     PROTO = 0x11
 
 
-class STATUS(_GT06pkt):
+class STATUS(GPS303Pkt):
     PROTO = 0x13
+    DIR = Dir.EXT
 
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        self = super().from_packet(length, proto, payload)
+    def from_packet(cls, length, payload):
+        self = super().from_packet(length, payload)
         if len(payload) == 5:
             (
                 self.batt,
@@ -181,32 +210,41 @@ class STATUS(_GT06pkt):
             self.signal = None
         return self
 
-    def response(self, upload_interval=25):  # Set interval in minutes
-        return super().response(pack("B", upload_interval))
+    @classmethod
+    def response(cls, upload_interval=25):  # Set interval in minutes
+        return cls.make_packet(pack("B", upload_interval))
 
 
-class HIBERNATION(_GT06pkt):
+class HIBERNATION(GPS303Pkt):
     PROTO = 0x14
+    DIR = Dir.INLINE
+
+    @classmethod
+    def response(cls):  # Server can send to send devicee to sleep
+        return cls.make_packet(b"")
 
 
-class RESET(_GT06pkt):  # Device sends when it got reset SMS
+class RESET(GPS303Pkt):  # Device sends when it got reset SMS
     PROTO = 0x15
 
-    def response(self):  # Server can send to initiate factory reset
-        return super().response(b"")
+    @classmethod
+    def response(cls):  # Server can send to initiate factory reset
+        return cls.make_packet(b"")
 
 
-class WHITELIST_TOTAL(_GT06pkt):  # Server sends to initiage sync (0x58)
+class WHITELIST_TOTAL(GPS303Pkt):  # Server sends to initiage sync (0x58)
     PROTO = 0x16
+    DIR = Dir.OUT
 
-    def response(self, number=3):  # Number of whitelist entries
-        return super().response(pack("B", number))
+    @classmethod
+    def response(cls, number=3):  # Number of whitelist entries
+        return cls.make_packet(pack("B", number))
 
 
-class _WIFI_POSITIONING(_GT06pkt):
+class _WIFI_POSITIONING(GPS303Pkt):
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        self = super().from_packet(length, proto, payload)
+    def from_packet(cls, length, payload):
+        self = super().from_packet(length, payload)
         self.dtime = payload[:6]
         if self.dtime == b"\0\0\0\0\0\0":
             self.devtime = None
@@ -234,188 +272,358 @@ class _WIFI_POSITIONING(_GT06pkt):
 
 class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING):
     PROTO = 0x17
+    DIR = Dir.INLINE
 
-    def response(self):
-        return super().response(self.dtime)
+    @classmethod
+    def inline_response(cls, packet):
+        return cls.make_packet(
+            bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S"))
+        )
 
 
-class TIME(_GT06pkt):
+class TIME(GPS303Pkt):
     PROTO = 0x30
+    DIR = Dir.INLINE
 
-    def response(self):
-        payload = pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
-        return super().response(payload)
+    @classmethod
+    def inline_response(cls, packet):
+        return cls.make_packet(
+            pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
+        )
 
 
-class PROHIBIT_LBS(_GT06pkt):
+class PROHIBIT_LBS(GPS303Pkt):
     PROTO = 0x33
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, status=1):  # Server sent, 0-off, 1-on
+        return cls.make_packet(pack("B", status))
+
+
+class GPS_LBS_SWITCH_TIMES(GPS303Pkt):
+    PROTO = 0x34
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls):
+        # 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 cls.make_packet(b"")  # TODO
+
+
+class _SET_PHONE(GPS303Pkt):
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, phone):
+        return cls.make_packet(phone.encode())
+
+
+class REMOTE_MONITOR_PHONE(_SET_PHONE):
+    PROTO = 0x40
+
+
+class SOS_PHONE(_SET_PHONE):
+    PROTO = 0x41
 
-    def response(self, status=1):  # Server sent, 0-off, 1-on
-        return super().response(pack("B", status))
 
+class DAD_PHONE(_SET_PHONE):
+    PROTO = 0x42
 
-class MOM_PHONE(_GT06pkt):
+
+class MOM_PHONE(_SET_PHONE):
     PROTO = 0x43
 
 
-class STOP_UPLOAD(_GT06pkt):  # Server response to LOGIN to thwart the device
+class STOP_UPLOAD(GPS303Pkt):  # Server response to LOGIN to thwart the device
     PROTO = 0x44
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls):
+        return cls.make_packet(b"")
+
+
+class GPS_OFF_PERIOD(GPS303Pkt):
+    PROTO = 0x46
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, onoff=0, fm="0000", to="2359"):
+        return cls.make_packet(
+            pack("B", onoff) + bytes.fromhex(fm) + bytes.fromhex(to)
+        )
 
-    def response(self):
-        return super().response(b"")
 
+class DND_PERIOD(GPS303Pkt):
+    PROTO = 0x47
+    DIR = Dir.OUT
 
-class STOP_ALARM(_GT06pkt):
+    @classmethod
+    def response(
+        cls, onoff=0, week=3, fm1="0000", to1="2359", fm2="0000", to2="2359"
+    ):
+        return cls.make_packet(
+            pack("B", onoff)
+            + pack("B", week)
+            + bytes.fromhex(fm1)
+            + bytes.fromhex(to1)
+            + bytes.fromhex(fm2)
+            + bytes.fromhex(to2)
+        )
+
+
+class RESTART_SHUTDOWN(GPS303Pkt):
+    PROTO = 0x48
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, flag=2):
+        # 1 - restart
+        # 2 - shutdown
+        return cls.make_packet(pack("B", flag))
+
+
+class DEVICE(GPS303Pkt):
+    PROTO = 0x49
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, flag=0):
+        # 0 - Stop looking for equipment
+        # 1 - Start looking for equipment
+        return cls.make_packet(pack("B", flag))
+
+
+class ALARM_CLOCK(GPS303Pkt):
+    PROTO = 0x50
+    DIR = Dir.OUT
+
+    @classmethod
+    def response(cls, alarms=((0, "0000"), (0, "0000"), (0, "0000"))):
+        return b"".join(
+            cls("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)
+        self.flag = payload[0]
+        return self
 
-class SETUP(_GT06pkt):
+
+class SETUP(GPS303Pkt):
     PROTO = 0x57
+    DIR = Dir.EXT
 
+    @classmethod
     def response(
-        self,
-        uploadIntervalSeconds=0x0300,
-        binarySwitch=0b00110001,
+        cls,
+        uploadintervalseconds=0x0300,
+        binaryswitch=0b00110001,
         alarms=[0, 0, 0],
-        dndTimeSwitch=0,
-        dndTimes=[0, 0, 0],
-        gpsTimeSwitch=0,
-        gpsTimeStart=0,
-        gpsTimeStop=0,
-        phoneNumbers=["", "", ""],
+        dndtimeswitch=0,
+        dndtimes=[0, 0, 0],
+        gpstimeswitch=0,
+        gpstimestart=0,
+        gpstimestop=0,
+        phonenumbers=["", "", ""],
     ):
         def pack3b(x):
             return pack("!I", x)[1:]
 
         payload = b"".join(
             [
-                pack("!H", uploadIntervalSeconds),
-                pack("B", binarySwitch),
+                pack("!H", uploadintervalseconds),
+                pack("B", binaryswitch),
             ]
             + [pack3b(el) for el in alarms]
             + [
-                pack("B", dndTimeSwitch),
+                pack("B", dndtimeswitch),
             ]
-            + [pack3b(el) for el in dndTimes]
+            + [pack3b(el) for el in dndtimes]
             + [
-                pack("B", gpsTimeSwitch),
-                pack("!H", gpsTimeStart),
-                pack("!H", gpsTimeStop),
+                pack("B", gpstimeswitch),
+                pack("!H", gpstimestart),
+                pack("!H", gpstimestop),
             ]
-            + [b";".join([el.encode() for el in phoneNumbers])]
+            + [b";".join([el.encode() for el in phonenumbers])]
         )
-        return super().response(payload)
+        return cls.make_packet(payload)
 
 
-class SYNCHRONOUS_WHITELIST(_GT06pkt):
+class SYNCHRONOUS_WHITELIST(GPS303Pkt):
     PROTO = 0x58
 
 
-class RESTORE_PASSWORD(_GT06pkt):
+class RESTORE_PASSWORD(GPS303Pkt):
     PROTO = 0x67
 
 
 class WIFI_POSITIONING(_WIFI_POSITIONING):
     PROTO = 0x69
+    DIR = Dir.EXT
 
-    def response(self, lat=None, lon=None):
+    @classmethod
+    def response(cls, 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 super().response(payload)
+        return cls.make_packet(payload)
 
 
-class MANUAL_POSITIONING(_GT06pkt):
+class MANUAL_POSITIONING(GPS303Pkt):
     PROTO = 0x80
+    DIR = Dir.OUT
+
+    @classmethod
+    def from_packet(cls, length, payload):
+        self = super().from_packet(length, payload)
+        self.flag = payload[0] if len(payload) > 0 else None
+        self.reason = {
+            1: "Incorrect time",
+            2: "LBS less",
+            3: "WiFi less",
+            4: "LBS search > 3 times",
+            5: "Same LBS and WiFi data",
+            6: "LBS prohibited, WiFi absent",
+            7: "GPS spacing < 50 m",
+        }.get(self.flag, "Unknown")
+        return self
+
+    @classmethod
+    def response(cls):
+        return cls.make_packet(b"")
 
 
-class BATTERY_CHARGE(_GT06pkt):
+class BATTERY_CHARGE(GPS303Pkt):
     PROTO = 0x81
 
 
-class CHARGER_CONNECTED(_GT06pkt):
+class CHARGER_CONNECTED(GPS303Pkt):
     PROTO = 0x82
 
 
-class CHARGER_DISCONNECTED(_GT06pkt):
+class CHARGER_DISCONNECTED(GPS303Pkt):
     PROTO = 0x83
 
 
-class VIBRATION_RECEIVED(_GT06pkt):
+class VIBRATION_RECEIVED(GPS303Pkt):
     PROTO = 0x94
 
 
-class POSITION_UPLOAD_INTERVAL(_GT06pkt):
+class POSITION_UPLOAD_INTERVAL(GPS303Pkt):
     PROTO = 0x98
+    DIR = Dir.EXT
 
     @classmethod
-    def from_packet(cls, length, proto, payload):
-        self = super().from_packet(length, proto, payload)
+    def from_packet(cls, length, payload):
+        self = super().from_packet(length, payload)
         self.interval = unpack("!H", payload[:2])
         return self
 
-    def response(self):
-        return super().response(pack("!H", self.interval))
+    @classmethod
+    def response(cls, interval=10):
+        return cls.make_packet(pack("!H", interval))
 
 
-class SOS_ALARM(_GT06pkt):
+class SOS_ALARM(GPS303Pkt):
     PROTO = 0x99
 
 
-# Build a dict protocol number -> class
+# Build dicts protocol number -> class and class name -> protocol number
 CLASSES = {}
+PROTOS = {}
 if True:  # just to indent the code, sorry!
     for cls in [
         cls
         for name, cls in globals().items()
         if isclass(cls)
-        and issubclass(cls, _GT06pkt)
+        and issubclass(cls, GPS303Pkt)
         and not name.startswith("_")
     ]:
         if hasattr(cls, "PROTO"):
             CLASSES[cls.PROTO] = cls
+            PROTOS[cls.__name__] = cls.PROTO
 
 
-def make_object(length, proto, payload):
+def class_by_prefix(prefix):
+    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_by_name(name):
+    return PROTOS.get(name, -1)
+
+
+def proto_of_message(packet):
+    return unpack("B", packet[1:2])[0]
+
+
+def inline_response(packet):
+    proto = proto_of_message(packet)
     if proto in CLASSES:
-        return CLASSES[proto].from_packet(length, proto, payload)
+        return CLASSES[proto].inline_response(packet)
     else:
-        return UNKNOWN.from_packet(length, proto, payload)
+        return None
 
 
-def handle_packet(packet, addr, when):
-    if len(packet) < 6:
-        return UNKNOWN.from_packet(0, 0, packet)
+def make_object(length, proto, payload):
+    if proto in CLASSES:
+        return CLASSES[proto].from_packet(length, payload)
     else:
-        xx, length, proto = unpack("!2sBB", packet[:4])
-        crlf = packet[-2:]
-        payload = packet[4:-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 xx != b"xx" or crlf != b"\r\n":
-            return UNKNOWN.from_packet(length, proto, packet)  # full packet
-        else:
-            return make_object(length, proto, payload)
+        retobj = UNKNOWN.from_packet(length, payload)
+        retobj.PROTO = proto  # Override class attr with object attr
+        return retobj
+
+
+def parse_message(packet):
+    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,
+        )
+    return make_object(length, proto, payload)
 
 
-def make_response(msg, **kwargs):
-    return msg.response(**kwargs)
+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 set_config(config):  # Note that we are setting _class_ attribute
-    _GT06pkt.CONFIG = config
+def make_response(msg, **kwargs):  # DEPRECATED
+    inframe = msg.response(**kwargs)
+    return None if inframe is None else b"xx" + inframe + b"\r\n"