2 Implementation of the protocol used by zx303 GPS+GPRS module
3 Description from https://github.com/tobadia/petGPS/tree/master/resources
6 from datetime import datetime, timezone
7 from inspect import isclass
8 from logging import getLogger
9 from struct import pack, unpack
21 "GPS_OFFLINE_POSITIONING",
26 "WIFI_OFFLINE_POSITIONING",
31 "SYNCHRONOUS_WHITELIST",
37 "CHARGER_DISCONNECTED",
39 "POSITION_UPLOAD_INTERVAL",
42 log = getLogger("gps303")
49 def __init__(self, *args, **kwargs):
51 for k, v in kwargs.items():
55 return "{}({})".format(
56 self.__class__.__name__,
60 'bytes.fromhex("{}")'.format(v.hex())
61 if isinstance(v, bytes)
64 for k, v in self.__dict__.items()
65 if not k.startswith("_")
70 def from_packet(cls, length, proto, payload):
71 return cls(proto=proto, payload=payload, length=length)
73 def response(self, *args):
76 assert len(args) == 1 and isinstance(args[0], bytes)
78 length = len(payload) + 1
81 return b"xx" + pack("BB", length, self.proto) + payload + b"\r\n"
84 class UNKNOWN(_GT06pkt):
88 class LOGIN(_GT06pkt):
92 def from_packet(cls, length, proto, payload):
93 self = super().from_packet(length, proto, payload)
94 self.imei = payload[:-1].hex()
95 self.ver = unpack("B", payload[-1:])[0]
99 return super().response(b"")
102 class SUPERVISION(_GT06pkt): # Server sends supervision number status
105 def response(self, supnum=0):
106 # 1: The device automatically answers Pickup effect
107 # 2: Automatically Answering Two-way Calls
108 # 3: Ring manually answer the two-way call
109 return super().response(b"")
112 class HEARTBEAT(_GT06pkt):
116 class _GPS_POSITIONING(_GT06pkt):
118 def from_packet(cls, length, proto, payload):
119 self = super().from_packet(length, proto, payload)
120 self.dtime = payload[:6]
121 if self.dtime == b"\0\0\0\0\0\0":
124 self.devtime = datetime(
125 *unpack("BBBBBB", self.dtime), tzinfo=timezone.utc
127 self.gps_data_length = payload[6] >> 4
128 self.gps_nb_sat = payload[6] & 0x0F
129 lat, lon, speed, flags = unpack("!IIBH", payload[7:18])
130 self.gps_is_valid = bool(flags & 0b0001000000000000) # bit 3
131 flip_lon = bool(flags & 0b0000100000000000) # bit 4
132 flip_lat = not bool(flags & 0b0000010000000000) # bit 5
133 self.heading = flags & 0b0000001111111111 # bits 6 - last
134 self.latitude = lat / (30000 * 60) * (-1 if flip_lat else 1)
135 self.longitude = lon / (30000 * 60) * (-2 if flip_lon else 1)
141 return super().response(self.dtime)
144 class GPS_POSITIONING(_GPS_POSITIONING):
148 class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING):
152 class STATUS(_GT06pkt):
156 def from_packet(cls, length, proto, payload):
157 self = super().from_packet(length, proto, payload)
158 if len(payload) == 5:
165 ) = unpack("BBBBB", payload)
166 elif len(payload) == 4:
167 self.batt, self.ver, self.timezone, self.intvl = unpack(
173 def response(self, upload_interval=25): # Set interval in minutes
174 return super().response(pack("B", upload_interval))
177 class HIBERNATION(_GT06pkt):
181 class RESET(_GT06pkt): # Device sends when it got reset SMS
184 def response(self): # Server can send to initiate factory reset
185 return super().response(b"")
188 class WHITELIST_TOTAL(_GT06pkt): # Server sends to initiage sync (0x58)
191 def response(self, number=3): # Number of whitelist entries
192 return super().response(pack("B", number))
195 class _WIFI_POSITIONING(_GT06pkt):
197 def from_packet(cls, length, proto, payload):
198 self = super().from_packet(length, proto, payload)
199 self.dtime = payload[:6]
200 if self.dtime == b"\0\0\0\0\0\0":
203 self.devtime = datetime.strptime(
204 self.dtime.hex(), "%y%m%d%H%M%S"
205 ).astimezone(tz=timezone.utc)
207 for i in range(self.length): # length has special meaning here
208 slice = payload[6 + i * 7 : 13 + i * 7]
209 self.wifi_aps.append(
210 (":".join([format(b, "02X") for b in slice[:6]]), -slice[6])
212 gsm_slice = payload[6 + self.length * 7 :]
213 ncells, self.mcc, self.mnc = unpack("!BHB", gsm_slice[:4])
215 for i in range(ncells):
216 slice = gsm_slice[4 + i * 5 : 9 + i * 5]
217 locac, cellid, sigstr = unpack(
218 "!HHB", gsm_slice[4 + i * 5 : 9 + i * 5]
220 self.gsm_cells.append((locac, cellid, -sigstr))
224 class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING):
228 return super().response(self.dtime)
231 class TIME(_GT06pkt):
235 payload = pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
236 return super().response(payload)
239 class PROHIBIT_LBS(_GT06pkt):
242 def response(self, status=1): # Server sent, 0-off, 1-on
243 return super().response(pack("B", status))
246 class MOM_PHONE(_GT06pkt):
250 class STOP_UPLOAD(_GT06pkt): # Server response to LOGIN to thwart the device
254 return super().response(b"")
257 class STOP_ALARM(_GT06pkt):
261 class SETUP(_GT06pkt):
266 uploadIntervalSeconds=0x0300,
267 binarySwitch=0b00110001,
274 phoneNumbers=["", "", ""],
277 return pack("!I", x)[1:]
281 pack("!H", uploadIntervalSeconds),
282 pack("B", binarySwitch),
284 + [pack3b(el) for el in alarms]
286 pack("B", dndTimeSwitch),
288 + [pack3b(el) for el in dndTimes]
290 pack("B", gpsTimeSwitch),
291 pack("!H", gpsTimeStart),
292 pack("!H", gpsTimeStop),
294 + [b";".join([el.encode() for el in phoneNumbers])]
296 return super().response(payload)
299 class SYNCHRONOUS_WHITELIST(_GT06pkt):
303 class RESTORE_PASSWORD(_GT06pkt):
307 class WIFI_POSITIONING(_WIFI_POSITIONING):
310 def response(self, lat=None, lon=None):
311 if lat is None or lon is None:
314 payload = "{:+#010.8g},{:+#010.8g}".format(lat, lon).encode(
317 return super().response(payload)
320 class MANUAL_POSITIONING(_GT06pkt):
324 class BATTERY_CHARGE(_GT06pkt):
328 class CHARGER_CONNECTED(_GT06pkt):
332 class CHARGER_DISCONNECTED(_GT06pkt):
336 class VIBRATION_RECEIVED(_GT06pkt):
340 class POSITION_UPLOAD_INTERVAL(_GT06pkt):
344 def from_packet(cls, length, proto, payload):
345 self = super().from_packet(length, proto, payload)
346 self.interval = unpack("!H", payload[:2])
350 return super().response(pack("!H", self.interval))
353 class SOS_ALARM(_GT06pkt):
357 # Build a dict protocol number -> class
359 if True: # just to indent the code, sorry!
362 for name, cls in globals().items()
364 and issubclass(cls, _GT06pkt)
365 and not name.startswith("_")
367 if hasattr(cls, "PROTO"):
368 CLASSES[cls.PROTO] = cls
371 def make_object(length, proto, payload):
373 return CLASSES[proto].from_packet(length, proto, payload)
375 return UNKNOWN.from_packet(length, proto, payload)
378 def handle_packet(packet, addr, when):
380 return UNKNOWN.from_packet(0, 0, packet)
382 xx, length, proto = unpack("!2sBB", packet[:4])
384 payload = packet[4:-2]
385 adjust = 2 if proto == STATUS.PROTO else 4 # Weird special case
388 not in (WIFI_POSITIONING.PROTO, WIFI_OFFLINE_POSITIONING.PROTO)
390 and len(payload) + adjust != length
393 "With proto %d length is %d but payload length is %d+%d",
399 if xx != b"xx" or crlf != b"\r\n":
400 return UNKNOWN.from_packet(length, proto, packet) # full packet
402 return make_object(length, proto, payload)
405 def make_response(msg, **kwargs):
406 return msg.response(**kwargs)
409 def set_config(config): # Note that we are setting _class_ attribute
410 _GT06pkt.CONFIG = config