X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=acfaea63c272aa74bc676dcac90765d027c8ddd2;hb=3f74a195c346809d3075b1351705eb9ad543afd5;hp=6fa76550db1031ea901ff55d71998d1aeb9fd640;hpb=4c173a5448990cd4da398be1bf3479bef17b1048;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index 6fa7655..acfaea6 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -15,17 +15,18 @@ Forewarnings: """ from datetime import datetime, timezone +from enum import Enum from inspect import isclass from logging import getLogger from struct import pack, unpack __all__ = ( "handle_packet", + "inline_response", "make_object", "make_response", "parse_message", "proto_by_name", - "set_config", "GPS303Pkt", "UNKNOWN", "LOGIN", @@ -56,9 +57,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 @@ -88,23 +96,29 @@ class GPS303Pkt: return pack("BB", self.length, self.PROTO) + self.payload @classmethod - def response(cls, *args): - if len(args) == 0: - return None - assert len(args) == 1 and isinstance(args[0], bytes) - payload = args[0] - length = len(payload) + 1 - if length > 6: - length -= 6 + 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 + 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): @@ -113,20 +127,16 @@ class LOGIN(GPS303Pkt): self.ver = unpack("B", payload[-1:])[0] return self - @classmethod - def response(cls): - return super().response(b"") - -class SUPERVISION(GPS303Pkt): # Server sends supervision number status +class SUPERVISION(GPS303Pkt): PROTO = 0x05 + DIR = Dir.OUT - @classmethod - def response(cls, supnum=0): + def response(self, 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 self.make_packet(pack("B", status)) class HEARTBEAT(GPS303Pkt): @@ -148,17 +158,20 @@ class _GPS_POSITIONING(GPS303Pkt): self.gps_nb_sat = payload[6] & 0x0F lat, lon, speed, flags = unpack("!IIBH", payload[7:18]) self.gps_is_valid = bool(flags & 0b0001000000000000) # bit 3 - flip_lon = bool(flags & 0b0000100000000000) # bit 4 - flip_lat = not bool(flags & 0b0000010000000000) # bit 5 - self.heading = flags & 0b0000001111111111 # bits 6 - last + flip_lon = bool(flags & 0b0000100000000000) # bit 4 + flip_lat = not bool(flags & 0b0000010000000000) # bit 5 + self.heading = flags & 0b0000001111111111 # bits 6 - last self.latitude = lat / (30000 * 60) * (-1 if flip_lat else 1) self.longitude = lon / (30000 * 60) * (-2 if flip_lon else 1) self.speed = speed 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): @@ -171,6 +184,7 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): class STATUS(GPS303Pkt): PROTO = 0x13 + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -191,25 +205,31 @@ class STATUS(GPS303Pkt): return self def response(self, upload_interval=25): # Set interval in minutes - return super().response(pack("B", upload_interval)) + return self.make_packet(pack("B", upload_interval)) class HIBERNATION(GPS303Pkt): PROTO = 0x14 + DIR = Dir.EXT + + def response(self): # Server can send to send devicee to sleep + return self.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"") + return self.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)) + return self.make_packet(pack("B", number)) class _WIFI_POSITIONING(GPS303Pkt): @@ -244,76 +264,182 @@ 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 def response(self, status=1): # Server sent, 0-off, 1-on - return super().response(pack("B", status)) + return self.make_packet(pack("B", status)) + + +class GPS_LBS_SWITCH_TIMES(GPS303Pkt): + PROTO = 0x34 + DIR = Dir.OUT + + def response(self): + # 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 self.make_packet(b"") # TODO + + +class _SET_PHONE(GPS303Pkt): + DIR = Dir.OUT + + def response(self, phone): + return self.make_packet(phone.encode()) + +class REMOTE_MONITOR_PHONE(_SET_PHONE): + PROTO = 0x40 -class MOM_PHONE(GPS303Pkt): + +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 def response(self): - return super().response(b"") + return self.make_packet(b"") + + +class GPS_OFF_PERIOD(GPS303Pkt): + PROTO = 0x46 + DIR = Dir.OUT + + def response(self, onoff=0, fm="0000", to="2359"): + return self.make_packet( + pack("B", onoff) + bytes.fromhex(fm) + bytes.fromhex(to) + ) + + +class DND_PERIOD(GPS303Pkt): + PROTO = 0x47 + DIR = Dir.OUT + + def response( + self, onoff=0, week=3, fm1="0000", to1="2359", fm2="0000", to2="2359" + ): + return self.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 + + def response(self, flag=0): + # 1 - restart + # 2 - shutdown + return self.make_packet(pack("B", flag)) + + +class DEVICE(GPS303Pkt): + PROTO = 0x49 + DIR = Dir.OUT + + def response(self, flag=0): + # 0 - Stop looking for equipment + # 1 - Start looking for equipment + return self.make_packet(pack("B", flag)) + + +class ALARM_CLOCK(GPS303Pkt): + PROTO = 0x50 + DIR = Dir.OUT + + def response(self, alarms=((0, "0000"), (0, "0000"), (0, "0000"))): + return b"".join( + pack("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 def response( self, - uploadIntervalSeconds=0x0300, - binarySwitch=0b00110001, + 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 self.make_packet(payload) class SYNCHRONOUS_WHITELIST(GPS303Pkt): @@ -326,6 +452,7 @@ class RESTORE_PASSWORD(GPS303Pkt): class WIFI_POSITIONING(_WIFI_POSITIONING): PROTO = 0x69 + DIR = Dir.EXT def response(self, lat=None, lon=None): if lat is None or lon is None: @@ -334,11 +461,29 @@ class WIFI_POSITIONING(_WIFI_POSITIONING): payload = "{:+#010.8g},{:+#010.8g}".format(lat, lon).encode( "ascii" ) - return super().response(payload) + return self.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") + + def response(self): + return self.make_packet(b"") class BATTERY_CHARGE(GPS303Pkt): @@ -359,6 +504,7 @@ class VIBRATION_RECEIVED(GPS303Pkt): class POSITION_UPLOAD_INTERVAL(GPS303Pkt): PROTO = 0x98 + DIR = Dir.EXT @classmethod def from_packet(cls, length, payload): @@ -366,8 +512,8 @@ class POSITION_UPLOAD_INTERVAL(GPS303Pkt): self.interval = unpack("!H", payload[:2]) return self - def response(self): - return super().response(pack("!H", self.interval)) + def response(self, interval=10): + return self.make_packet(pack("!H", interval)) class SOS_ALARM(GPS303Pkt): @@ -398,6 +544,14 @@ 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].inline_response(packet) + else: + return None + + def make_object(length, proto, payload): if proto in CLASSES: return CLASSES[proto].from_packet(length, payload) @@ -412,8 +566,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 ): @@ -436,7 +589,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