X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=dcc335b184fc6fa59249ce2c692f37c54d1587d5;hb=faa8ce9d87530105ec2ad2f4809a9ace581a2ad6;hp=ad8fd7168fe4951d75eda0bf19f87e33bd9229e3;hpb=c91cccfb91f1ea45dc886e2d75796e3af0b7f00f;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index ad8fd71..dcc335b 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -15,17 +15,20 @@ 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", "parse_message", "proto_by_name", + "Dir", "GPS303Pkt", "UNKNOWN", "LOGIN", @@ -56,9 +59,16 @@ __all__ = ( log = getLogger("gps303") +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 - INLINE = True + DIR = Dir.IN # Do not send anything back by default def __init__(self, *args, **kwargs): assert len(args) == 0 @@ -90,14 +100,14 @@ class GPS303Pkt: @classmethod def make_packet(cls, payload): assert isinstance(payload, bytes) - length = len(payload) + 1 - if length > 6: - length -= 6 + 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.INLINE: + if cls.DIR is Dir.INLINE: return cls.make_packet(b"") else: return None @@ -105,11 +115,12 @@ class GPS303Pkt: class UNKNOWN(GPS303Pkt): PROTO = 256 # > 255 is impossible in real packets - INLINE = False class LOGIN(GPS303Pkt): PROTO = 0x01 + DIR = Dir.INLINE + # Default response for ACK, can also respond with STOP_UPLOAD @classmethod def from_packet(cls, length, payload): @@ -119,22 +130,26 @@ class LOGIN(GPS303Pkt): return self -class SUPERVISION(GPS303Pkt): # Server sends supervision number status +class SUPERVISION(GPS303Pkt): PROTO = 0x05 - INLINE = False + 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 self.make_packet(pack("B", supnum)) + return cls.make_packet(pack("B", status)) class HEARTBEAT(GPS303Pkt): PROTO = 0x08 + DIR = Dir.INLINE class _GPS_POSITIONING(GPS303Pkt): + DIR = Dir.INLINE + @classmethod def from_packet(cls, length, payload): self = super().from_packet(length, payload) @@ -160,7 +175,9 @@ class _GPS_POSITIONING(GPS303Pkt): @classmethod def inline_response(cls, packet): - return cls.make_packet(packet[2:8]) + tup = datetime.utcnow().timetuple() + ttup = (tup[0] % 100,) + tup[1:6] + return cls.make_packet(pack("BBBBBB", *ttup)) class GPS_POSITIONING(_GPS_POSITIONING): @@ -173,7 +190,7 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): class STATUS(GPS303Pkt): PROTO = 0x13 - INLINE = False + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -193,28 +210,35 @@ class STATUS(GPS303Pkt): self.signal = None return self - def response(self, upload_interval=25): # Set interval in minutes - return self.make_packet(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(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(GPS303Pkt): # Device sends when it got reset SMS PROTO = 0x15 - INLINE = False - def response(self): # Server can send to initiate factory reset - return self.make_packet(b"") + @classmethod + def response(cls): # Server can send to initiate factory reset + return cls.make_packet(b"") class WHITELIST_TOTAL(GPS303Pkt): # Server sends to initiage sync (0x58) PROTO = 0x16 - INLINE = False + DIR = Dir.OUT - def response(self, number=3): # Number of whitelist entries - return self.make_packet(pack("B", number)) + @classmethod + def response(cls, number=3): # Number of whitelist entries + return cls.make_packet(pack("B", number)) class _WIFI_POSITIONING(GPS303Pkt): @@ -248,103 +272,196 @@ class _WIFI_POSITIONING(GPS303Pkt): class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING): PROTO = 0x17 + DIR = Dir.INLINE @classmethod def inline_response(cls, packet): - return cls.make_packet(packet[2:8]) + return cls.make_packet( + bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S")) + ) class TIME(GPS303Pkt): PROTO = 0x30 + DIR = Dir.INLINE @classmethod def inline_response(cls, packet): - return pack( - "!BBHBBBBB", 7, cls.PROTO, *datetime.utcnow().timetuple()[:6] + return cls.make_packet( + pack("!HBBBBB", *datetime.utcnow().timetuple()[:6]) ) class PROHIBIT_LBS(GPS303Pkt): PROTO = 0x33 - INLINE = False + DIR = Dir.OUT - def response(self, status=1): # Server sent, 0-off, 1-on - return self.make_packet(pack("B", status)) + @classmethod + def response(cls, status=1): # Server sent, 0-off, 1-on + return cls.make_packet(pack("B", status)) -class LBS_SWITCH_TIMES(GPS303Pkt): +class GPS_LBS_SWITCH_TIMES(GPS303Pkt): PROTO = 0x34 - INLINE = False + DIR = Dir.OUT - def response(self): - return self.make_packet(b"") + @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(GPS303Pkt): +class REMOTE_MONITOR_PHONE(_SET_PHONE): PROTO = 0x40 - INLINE = False -class SOS_PHONE(GPS303Pkt): +class SOS_PHONE(_SET_PHONE): PROTO = 0x41 - INLINE = False -class DAD_PHONE(GPS303Pkt): +class DAD_PHONE(_SET_PHONE): PROTO = 0x42 - INLINE = False -class MOM_PHONE(GPS303Pkt): +class MOM_PHONE(_SET_PHONE): PROTO = 0x43 - INLINE = False 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) + ) + + +class DND_PERIOD(GPS303Pkt): + PROTO = 0x47 + DIR = Dir.OUT + + @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(GPS303Pkt): PROTO = 0x57 - INLINE = False + 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 self.make_packet(payload) + return cls.make_packet(payload) class SYNCHRONOUS_WHITELIST(GPS303Pkt): @@ -357,21 +474,41 @@ class RESTORE_PASSWORD(GPS303Pkt): class WIFI_POSITIONING(_WIFI_POSITIONING): PROTO = 0x69 - INLINE = False + 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 self.make_packet(payload) + return cls.make_packet(payload) class MANUAL_POSITIONING(GPS303Pkt): PROTO = 0x80 - INLINE = False + 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(GPS303Pkt): @@ -392,7 +529,7 @@ class VIBRATION_RECEIVED(GPS303Pkt): class POSITION_UPLOAD_INTERVAL(GPS303Pkt): PROTO = 0x98 - INLINE = False + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -400,8 +537,9 @@ class POSITION_UPLOAD_INTERVAL(GPS303Pkt): self.interval = unpack("!H", payload[:2]) return self - def response(self, interval=10): - return self.make_packet(pack("!H", interval)) + @classmethod + def response(cls, interval=10): + return cls.make_packet(pack("!H", interval)) class SOS_ALARM(GPS303Pkt): @@ -424,6 +562,18 @@ if True: # just to indent the code, sorry! PROTOS[cls.__name__] = cls.PROTO +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)