"""
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",
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
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):
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(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):
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):
class STATUS(GPS303Pkt):
PROTO = 0x13
+ DIR = Dir.EXT
@classmethod
def from_packet(cls, length, payload):
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):
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
- def response(self):
- return super().response(b"")
+ @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]
+
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):
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):
class POSITION_UPLOAD_INTERVAL(GPS303Pkt):
PROTO = 0x98
+ DIR = Dir.EXT
@classmethod
def from_packet(cls, length, payload):
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):
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)
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)
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
):
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