X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=83ac19cd4d68a67ba06091c8b3e0c9274f027110;hb=c772f9219b7a90be63174ef8e66c2f1249200c85;hp=4789b033a3936873150d12b9cad4e1502869be23;hpb=b613110bb16c95d9b641882c2ad6869e3ced1a0c;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index 4789b03..83ac19c 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -23,7 +23,9 @@ __all__ = ( "handle_packet", "make_object", "make_response", - "set_config", + "parse_message", + "proto_by_name", + "GPS303Pkt", "UNKNOWN", "LOGIN", "SUPERVISION", @@ -53,9 +55,8 @@ __all__ = ( log = getLogger("gps303") -class _GT06pkt: +class GPS303Pkt: PROTO: int - CONFIG = None def __init__(self, *args, **kwargs): assert len(args) == 0 @@ -78,10 +79,14 @@ 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 response(self, *args): + def to_packet(self): + 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) @@ -89,45 +94,47 @@ class _GT06pkt: length = len(payload) + 1 if length > 6: length -= 6 - return b"xx" + pack("BB", length, self.proto) + payload + b"\r\n" + return pack("BB", length, cls.PROTO) + payload -class UNKNOWN(_GT06pkt): - pass +class UNKNOWN(GPS303Pkt): + PROTO = 256 # > 255 is impossible in real packets -class LOGIN(_GT06pkt): +class LOGIN(GPS303Pkt): PROTO = 0x01 @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): + @classmethod + def response(cls): return super().response(b"") -class SUPERVISION(_GT06pkt): # Server sends supervision number status +class SUPERVISION(GPS303Pkt): # Server sends supervision number status PROTO = 0x05 - def response(self, supnum=0): + @classmethod + def response(cls, supnum=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"") -class HEARTBEAT(_GT06pkt): +class HEARTBEAT(GPS303Pkt): PROTO = 0x08 -class _GPS_POSITIONING(_GT06pkt): +class _GPS_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 @@ -139,9 +146,9 @@ class _GPS_POSITIONING(_GT06pkt): 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 @@ -160,12 +167,12 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): PROTO = 0x11 -class STATUS(_GT06pkt): +class STATUS(GPS303Pkt): PROTO = 0x13 @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, @@ -185,28 +192,28 @@ class STATUS(_GT06pkt): return super().response(pack("B", upload_interval)) -class HIBERNATION(_GT06pkt): +class HIBERNATION(GPS303Pkt): PROTO = 0x14 -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"") -class WHITELIST_TOTAL(_GT06pkt): # Server sends to initiage sync (0x58) +class WHITELIST_TOTAL(GPS303Pkt): # Server sends to initiage sync (0x58) PROTO = 0x16 def response(self, number=3): # Number of whitelist entries return super().response(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 @@ -239,7 +246,7 @@ class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING): return super().response(self.dtime) -class TIME(_GT06pkt): +class TIME(GPS303Pkt): PROTO = 0x30 def response(self): @@ -247,29 +254,29 @@ class TIME(_GT06pkt): return super().response(payload) -class PROHIBIT_LBS(_GT06pkt): +class PROHIBIT_LBS(GPS303Pkt): PROTO = 0x33 def response(self, status=1): # Server sent, 0-off, 1-on return super().response(pack("B", status)) -class MOM_PHONE(_GT06pkt): +class MOM_PHONE(GPS303Pkt): 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 def response(self): return super().response(b"") -class STOP_ALARM(_GT06pkt): +class STOP_ALARM(GPS303Pkt): PROTO = 0x56 -class SETUP(_GT06pkt): +class SETUP(GPS303Pkt): PROTO = 0x57 def response( @@ -307,11 +314,11 @@ class SETUP(_GT06pkt): return super().response(payload) -class SYNCHRONOUS_WHITELIST(_GT06pkt): +class SYNCHRONOUS_WHITELIST(GPS303Pkt): PROTO = 0x58 -class RESTORE_PASSWORD(_GT06pkt): +class RESTORE_PASSWORD(GPS303Pkt): PROTO = 0x67 @@ -328,32 +335,32 @@ class WIFI_POSITIONING(_WIFI_POSITIONING): return super().response(payload) -class MANUAL_POSITIONING(_GT06pkt): +class MANUAL_POSITIONING(GPS303Pkt): PROTO = 0x80 -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 @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 @@ -361,61 +368,70 @@ class POSITION_UPLOAD_INTERVAL(_GT06pkt): return super().response(pack("!H", self.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 proto_by_name(name): + return PROTOS.get(name, -1) + + +def proto_of_message(packet): + return unpack("B", packet[1:2])[0] def make_object(length, proto, payload): if proto in CLASSES: - return CLASSES[proto].from_packet(length, proto, payload) + return CLASSES[proto].from_packet(length, payload) else: - return UNKNOWN.from_packet(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 handle_packet(packet, addr, when): - if len(packet) < 6: - return UNKNOWN.from_packet(0, 0, packet) - 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) +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 make_response(msg, **kwargs): - return msg.response(**kwargs) +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 - _GT06pkt.CONFIG = config