X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=27733a3fa48f246212d7d86e33a0106348ec651c;hb=ba20de1b9666bd56818d5f951b5a9d2fc98af133;hp=1280fe15ebb6106b8890f6139077e7c4879d8485;hpb=2c60838ac44f02802480eeca626040231014caaa;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index 1280fe1..27733a3 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -15,17 +15,19 @@ 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", - "set_config", "GPS303Pkt", "UNKNOWN", "LOGIN", @@ -56,9 +58,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 - CONFIG = None + DIR = Dir.INLINE # Most packets anticipate simple acknowledgement def __init__(self, *args, **kwargs): assert len(args) == 0 @@ -87,23 +96,30 @@ class GPS303Pkt: def to_packet(self): return pack("BB", self.length, self.PROTO) + self.payload - def response(self, *args): - if len(args) == 0: + @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 + + @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 pack("BB", length, self.PROTO) + payload class UNKNOWN(GPS303Pkt): PROTO = 256 # > 255 is impossible in real packets + DIR = Dir.IN class LOGIN(GPS303Pkt): PROTO = 0x01 + # Default response for ACK, can also respond with STOP_UPLOAD @classmethod def from_packet(cls, length, payload): @@ -112,18 +128,17 @@ class LOGIN(GPS303Pkt): self.ver = unpack("B", payload[-1:])[0] return self - def response(self): - return super().response(b"") - -class SUPERVISION(GPS303Pkt): # 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(GPS303Pkt): @@ -154,8 +169,11 @@ class _GPS_POSITIONING(GPS303Pkt): 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): @@ -168,6 +186,7 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): class STATUS(GPS303Pkt): PROTO = 0x13 + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -187,26 +206,36 @@ class STATUS(GPS303Pkt): 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(GPS303Pkt): PROTO = 0x14 + DIR = Dir.EXT + + @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 + DIR = Dir.EXT - 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(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(GPS303Pkt): @@ -241,76 +270,192 @@ class _WIFI_POSITIONING(GPS303Pkt): class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING): PROTO = 0x17 - 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(GPS303Pkt): PROTO = 0x30 - 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(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)) - def response(self, status=1): # Server sent, 0-off, 1-on - return super().response(pack("B", status)) +class GPS_LBS_SWITCH_TIMES(GPS303Pkt): + PROTO = 0x34 + DIR = Dir.OUT -class MOM_PHONE(GPS303Pkt): + @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 + + +class DAD_PHONE(_SET_PHONE): + PROTO = 0x42 + + +class MOM_PHONE(_SET_PHONE): PROTO = 0x43 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 + + @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] + 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(GPS303Pkt): @@ -323,19 +468,40 @@ class RESTORE_PASSWORD(GPS303Pkt): 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(GPS303Pkt): PROTO = 0x80 + DIR = Dir.EXT + + @classmethod + def from_packet(cls, length, payload): + self = super().from_packet(length, payload) + self.flag = payload[0] + 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") + + @classmethod + def response(cls): + return cls.make_packet(b"") class BATTERY_CHARGE(GPS303Pkt): @@ -356,6 +522,7 @@ class VIBRATION_RECEIVED(GPS303Pkt): class POSITION_UPLOAD_INTERVAL(GPS303Pkt): PROTO = 0x98 + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -363,8 +530,9 @@ class POSITION_UPLOAD_INTERVAL(GPS303Pkt): 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(GPS303Pkt): @@ -387,12 +555,29 @@ 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) def proto_of_message(packet): - return unpack("B", packet[1:2]) + return unpack("B", packet[1:2])[0] + + +def inline_response(packet): + proto = proto_of_message(packet) + if proto in CLASSES: + return CLASSES[proto].inline_response(packet) + else: + return None def make_object(length, proto, payload): @@ -409,8 +594,7 @@ def parse_message(packet): 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) + proto not in (WIFI_POSITIONING.PROTO, WIFI_OFFLINE_POSITIONING.PROTO) and length > 1 and len(payload) + adjust != length ): @@ -433,7 +617,3 @@ def handle_packet(packet): # DEPRECATED def make_response(msg, **kwargs): # DEPRECATED inframe = msg.response(**kwargs) return None if inframe is None else b"xx" + inframe + b"\r\n" - - -def set_config(config): # Note that we are setting _class_ attribute - GPS303Pkt.CONFIG = config