"""
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",
- "make_object",
- "make_response",
- "set_config",
+ "class_by_prefix",
+ "inline_response",
+ "parse_message",
+ "proto_by_name",
+ "Respond",
+ "GPS303Pkt",
"UNKNOWN",
"LOGIN",
"SUPERVISION",
"WHITELIST_TOTAL",
"WIFI_OFFLINE_POSITIONING",
"TIME",
+ "PROHIBIT_LBS",
+ "GPS_LBS_SWITCH_TIMES",
+ "REMOTE_MONITOR_PHONE",
+ "SOS_PHONE",
+ "DAD_PHONE",
"MOM_PHONE",
+ "STOP_UPLOAD",
+ "GPS_OFF_PERIOD",
+ "DND_PERIOD",
+ "RESTART_SHUTDOWN",
+ "DEVICE",
+ "ALARM_CLOCK",
"STOP_ALARM",
"SETUP",
"SYNCHRONOUS_WHITELIST",
"CHARGER_DISCONNECTED",
"VIBRATION_RECEIVED",
"POSITION_UPLOAD_INTERVAL",
+ "SOS_ALARM",
+ "UNKNOWN_B3",
)
-log = getLogger("gps303")
+
+def intx(x):
+ if isinstance(x, str):
+ x = int(x, 0)
+ return x
+
+
+def hhmm(x):
+ """Check for the string that represents hours and minutes"""
+ if not isinstance(x, str) or len(x) != 4:
+ raise ValueError(str(x) + " is not a four-character string")
+ hh = int(x[:2])
+ mm = int(x[2:])
+ if hh < 0 or hh > 23 or mm < 0 or mm > 59:
+ raise ValueError(str(x) + " does not contain valid hours and minutes")
+ return x
+
+
+def l3str(x):
+ if isinstance(x, str):
+ x = x.split(",")
+ if len(x) != 3 or not all(isinstance(el, str) for el in x):
+ raise ValueError(str(x) + " is not a list of three strings")
+ return x
+
+
+def l3int(x):
+ if isinstance(x, str):
+ x = x.split(",")
+ x = [int(el) for el in x]
+ if len(x) != 3 or not all(isinstance(el, int) for el in x):
+ raise ValueError(str(x) + " is not a list of three integers")
+ return x
+
+
+class MetaPkt(type):
+ """
+ For each class corresponding to a message, automatically create
+ two nested classes `In` and `Out` that also inherit from their
+ "nest". Class attribute `IN_KWARGS` defined in the "nest" is
+ copied to the `In` nested class under the name `KWARGS`, and
+ likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS`
+ to the nested class `Out`. In addition, method `encode` is
+ defined in both classes equal to `in_encode()` and `out_encode()`
+ respectively.
+ """
+
+ def __new__(cls, name, bases, attrs):
+ newcls = super().__new__(cls, name, bases, attrs)
+ newcls.In = super().__new__(
+ cls,
+ name + ".In",
+ (newcls,) + bases,
+ {"KWARGS": newcls.IN_KWARGS, "encode": newcls.in_encode},
+ )
+ newcls.Out = super().__new__(
+ cls,
+ name + ".Out",
+ (newcls,) + bases,
+ {"KWARGS": newcls.OUT_KWARGS, "encode": newcls.out_encode},
+ )
+ return newcls
+
+
+class Respond(Enum):
+ NON = 0 # Incoming, no response needed
+ INL = 1 # Birirectional, use `inline_response()`
+ EXT = 2 # Birirectional, use external responder
-class _GT06pkt:
+class GPS303Pkt(metaclass=MetaPkt):
+ RESPOND = Respond.NON # Do not send anything back by default
PROTO: int
- CONFIG = None
+ # Have these kwargs for now, TODO redo
+ IN_KWARGS = (("length", int, 0), ("payload", bytes, b""))
+ OUT_KWARGS = ()
def __init__(self, *args, **kwargs):
assert len(args) == 0
- for k, v in kwargs.items():
- setattr(self, k, v)
+ for kw, typ, dfl in self.KWARGS:
+ setattr(self, kw, typ(kwargs.pop(kw, dfl)))
+ if kwargs:
+ print("KWARGS", self.KWARGS)
+ print("kwargs", kwargs)
+ raise TypeError(
+ self.__class__.__name__ + " stray kwargs " + str(kwargs)
+ )
def __repr__(self):
return "{}({})".format(
),
)
- @classmethod
- def from_packet(cls, length, proto, payload):
- return cls(proto=proto, payload=payload, length=length)
-
- def response(self, *args):
- if len(args) == 0:
- return None
- assert len(args) == 1 and isinstance(args[0], bytes)
- payload = args[0]
+ def in_encode(self):
+ raise NotImplementedError(
+ self.__class__.__name__ + ".encode() not implemented"
+ )
+
+ def out_encode(self):
+ return b""
+
+ @property
+ def packed(self):
+ payload = self.encode()
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, self.PROTO) + payload
+
+ @classmethod
+ def from_packet(cls, length, payload):
+ return cls.In(payload=payload, length=length)
-class UNKNOWN(_GT06pkt):
- pass
+class UNKNOWN(GPS303Pkt):
+ PROTO = 256 # > 255 is impossible in real packets
-class LOGIN(_GT06pkt):
+class LOGIN(GPS303Pkt):
PROTO = 0x01
+ RESPOND = Respond.INL
+ # Default response for ACK, can also respond with STOP_UPLOAD
@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):
- return super().response(b"")
-
-class SUPERVISION(_GT06pkt): # Server sends supervision number status
+class SUPERVISION(GPS303Pkt):
PROTO = 0x05
+ OUT_KWARGS = (("status", int, 1),)
- def response(self, supnum=0):
+ def out_encode(self):
# 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 pack("B", self.status)
-class HEARTBEAT(_GT06pkt):
+class HEARTBEAT(GPS303Pkt):
PROTO = 0x08
+ RESPOND = Respond.INL
-class _GPS_POSITIONING(_GT06pkt):
+class _GPS_POSITIONING(GPS303Pkt):
+ RESPOND = Respond.INL
+
@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
self.flags = flags
return self
- def response(self):
- return super().response(self.dtime)
+ def out_encode(self):
+ tup = datetime.utcnow().timetuple()
+ ttup = (tup[0] % 100,) + tup[1:6]
+ return pack("BBBBBB", *ttup)
class GPS_POSITIONING(_GPS_POSITIONING):
PROTO = 0x11
-class STATUS(_GT06pkt):
+class STATUS(GPS303Pkt):
PROTO = 0x13
+ RESPOND = Respond.EXT
+ OUT_KWARGS = (("upload_interval", int, 25),)
@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,
self.signal = None
return self
- def response(self, upload_interval=25): # Set interval in minutes
- return super().response(pack("B", upload_interval))
+ def out_encode(self): # Set interval in minutes
+ return pack("B", self.upload_interval)
-class HIBERNATION(_GT06pkt):
+class HIBERNATION(GPS303Pkt): # Server can send to send devicee to sleep
PROTO = 0x14
+ RESPOND = Respond.INL
-class RESET(_GT06pkt): # Device sends when it got reset SMS
+class RESET(GPS303Pkt):
+ # Device sends when it got reset SMS
+ # Server can send to initiate factory reset
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
+ OUT_KWARGS = (("number", int, 3),)
- def response(self, number=3): # Number of whitelist entries
- return super().response(pack("B", number))
+ def out_encode(self): # Number of whitelist entries
+ return 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
class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING):
PROTO = 0x17
+ RESPOND = Respond.INL
- def response(self):
- return super().response(self.dtime)
+ def out_encode(self):
+ return bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S"))
-class TIME(_GT06pkt):
+class TIME(GPS303Pkt):
PROTO = 0x30
+ RESPOND = Respond.INL
- def response(self):
- payload = pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
- return super().response(payload)
+ def out_encode(self):
+ return pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
-class PROHIBIT_LBS(_GT06pkt):
+class PROHIBIT_LBS(GPS303Pkt):
PROTO = 0x33
+ OUT_KWARGS = (("status", int, 1),)
+
+ def out_encode(self): # Server sent, 0-off, 1-on
+ return pack("B", self.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
+ # 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
+ def out_encode(self):
+ return b"" # TODO
-class MOM_PHONE(_GT06pkt):
+
+class _SET_PHONE(GPS303Pkt):
+ OUT_KWARGS = (("phone", str, ""),)
+
+ def out_encode(self):
+ return self.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(_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 GPS_OFF_PERIOD(GPS303Pkt):
+ PROTO = 0x46
+ OUT_KWARGS = (
+ ("onoff", int, 0),
+ ("fm", hhmm, "0000"),
+ ("to", hhmm, "2359"),
+ )
+
+ def out_encode(self):
+ return (
+ pack("B", self.onoff)
+ + bytes.fromhex(self.fm)
+ + bytes.fromhex(self.to)
+ )
+
+
+class DND_PERIOD(GPS303Pkt):
+ PROTO = 0x47
+ OUT_KWARGS = (
+ ("onoff", int, 0),
+ ("week", int, 3),
+ ("fm1", hhmm, "0000"),
+ ("to1", hhmm, "2359"),
+ ("fm2", hhmm, "0000"),
+ ("to2", hhmm, "2359"),
+ )
+
+ def out_endode(self):
+ return (
+ pack("B", self.onoff)
+ + pack("B", self.week)
+ + bytes.fromhex(self.fm1)
+ + bytes.fromhex(self.to1)
+ + bytes.fromhex(self.fm2)
+ + bytes.fromhex(self.to2)
+ )
+
+
+class RESTART_SHUTDOWN(GPS303Pkt):
+ PROTO = 0x48
+ OUT_KWARGS = (("flag", int, 0),)
-class STOP_ALARM(_GT06pkt):
+ def out_encode(self):
+ # 1 - restart
+ # 2 - shutdown
+ return pack("B", self.flag)
+
+
+class DEVICE(GPS303Pkt):
+ PROTO = 0x49
+ OUT_KWARGS = (("flag", int, 0),)
+
+ # 0 - Stop looking for equipment
+ # 1 - Start looking for equipment
+ def out_encode(self):
+ return pack("B", self.flag)
+
+
+class ALARM_CLOCK(GPS303Pkt):
+ PROTO = 0x50
+
+ def out_encode(self):
+ # TODO implement parsing kwargs
+ 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]
+ return self
-class SETUP(_GT06pkt):
- PROTO = 0x57
- def response(
- self,
- uploadIntervalSeconds=0x0300,
- binarySwitch=0b00110001,
- alarms=[0, 0, 0],
- dndTimeSwitch=0,
- dndTimes=[0, 0, 0],
- gpsTimeSwitch=0,
- gpsTimeStart=0,
- gpsTimeStop=0,
- phoneNumbers=["", "", ""],
- ):
+class SETUP(GPS303Pkt):
+ PROTO = 0x57
+ RESPOND = Respond.EXT
+ OUT_KWARGS = (
+ ("uploadintervalseconds", intx, 0x0300),
+ ("binaryswitch", intx, 0b00110001),
+ ("alarms", l3int, [0, 0, 0]),
+ ("dndtimeswitch", int, 0),
+ ("dndtimes", l3int, [0, 0, 0]),
+ ("gpstimeswitch", int, 0),
+ ("gpstimestart", int, 0),
+ ("gpstimestop", int, 0),
+ ("phonenumbers", l3str, ["", "", ""]),
+ )
+
+ def out_encode(self):
def pack3b(x):
return pack("!I", x)[1:]
- payload = b"".join(
+ return b"".join(
[
- pack("!H", uploadIntervalSeconds),
- pack("B", binarySwitch),
+ pack("!H", self.uploadintervalseconds),
+ pack("B", self.binaryswitch),
]
- + [pack3b(el) for el in alarms]
+ + [pack3b(el) for el in self.alarms]
+ [
- pack("B", dndTimeSwitch),
+ pack("B", self.dndtimeswitch),
]
- + [pack3b(el) for el in dndTimes]
+ + [pack3b(el) for el in self.dndtimes]
+ [
- pack("B", gpsTimeSwitch),
- pack("!H", gpsTimeStart),
- pack("!H", gpsTimeStop),
+ pack("B", self.gpstimeswitch),
+ pack("!H", self.gpstimestart),
+ pack("!H", self.gpstimestop),
]
- + [b";".join([el.encode() for el in phoneNumbers])]
+ + [b";".join([el.encode() for el in self.phonenumbers])]
)
- return super().response(payload)
-class SYNCHRONOUS_WHITELIST(_GT06pkt):
+class SYNCHRONOUS_WHITELIST(GPS303Pkt):
PROTO = 0x58
-class RESTORE_PASSWORD(_GT06pkt):
+class RESTORE_PASSWORD(GPS303Pkt):
PROTO = 0x67
class WIFI_POSITIONING(_WIFI_POSITIONING):
PROTO = 0x69
+ RESPOND = Respond.EXT
+ OUT_KWARGS = (("lat", float, None), ("lon", float, None))
- def response(self, 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)
+ def out_encode(self):
+ if self.lat is None or self.lon is None:
+ return b""
+ return "{:+#010.8g},{:+#010.8g}".format(self.lat, self.lon).encode()
-class MANUAL_POSITIONING(_GT06pkt):
+class MANUAL_POSITIONING(GPS303Pkt):
PROTO = 0x80
+ @classmethod
+ def from_packet(cls, length, payload):
+ self = super().from_packet(length, payload)
+ self.flag = payload[0] if len(payload) > 0 else None
+ 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")
+ return self
-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
+ RESPOND = Respond.EXT
+ OUT_KWARGS = (("interval", int, 10),)
@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
- def response(self):
- return super().response(pack("!H", self.interval))
+ def out_encode(self):
+ return pack("!H", interval)
-class SOS_ALARM(_GT06pkt):
+class SOS_ALARM(GPS303Pkt):
PROTO = 0x99
-# Build a dict protocol number -> class
+class UNKNOWN_B3(GPS303Pkt):
+ PROTO = 0xb3
+ IN_KWARGS = (("asciidata", str, ""),)
+
+ @classmethod
+ def from_packet(cls, length, payload):
+ self = super().from_packet(length, payload)
+ self.asciidata = payload.decode()
+ return self
+
+
+# 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 make_object(length, proto, payload):
- if proto in CLASSES:
- return CLASSES[proto].from_packet(length, proto, payload)
- else:
- return UNKNOWN.from_packet(length, proto, payload)
+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 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 proto_by_name(name):
+ return PROTOS.get(name, -1)
+
+def proto_of_message(packet):
+ return unpack("B", packet[1:2])[0]
-def make_response(msg, **kwargs):
- return msg.response(**kwargs)
+
+def inline_response(packet):
+ proto = proto_of_message(packet)
+ if proto in CLASSES:
+ cls = CLASSES[proto]
+ if cls.RESPOND is Respond.INL:
+ return cls.Out().packed
+ return None
-def set_config(config): # Note that we are setting _class_ attribute
- _GT06pkt.CONFIG = config
+def parse_message(packet):
+ """ From a packet (without framing bytes) derive the XXX.In object """
+ length, proto = unpack("BB", packet[:2])
+ payload = packet[2:]
+ if proto in CLASSES:
+ return CLASSES[proto].from_packet(length, payload)
+ else:
+ retobj = UNKNOWN.from_packet(length, payload)
+ retobj.PROTO = proto # Override class attr with object attr
+ return retobj
projects / loctrkd.git / blobdiff