from datetime import datetime, timezone
from enum import Enum
from inspect import isclass
-from struct import pack, unpack
+from struct import error, pack, unpack
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ List,
+ Optional,
+ Tuple,
+ Type,
+ TYPE_CHECKING,
+ Union,
+)
__all__ = (
"class_by_prefix",
"inline_response",
"parse_message",
"proto_by_name",
+ "DecodeError",
"Respond",
"GPS303Pkt",
"UNKNOWN",
)
-def intx(x):
+class DecodeError(Exception):
+ def __init__(self, e: Exception, **kwargs: Any) -> None:
+ super().__init__(e)
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
+
+def maybe(typ: type) -> Callable[[Any], Any]:
+ return lambda x: None if x is None else typ(x)
+
+
+def intx(x: Union[str, int]) -> int:
if isinstance(x, str):
x = int(x, 0)
return x
-def hhmm(x):
+def boolx(x: Union[str, bool]) -> bool:
+ if isinstance(x, str):
+ if x.upper() in ("ON", "TRUE", "1"):
+ return True
+ if x.upper() in ("OFF", "FALSE", "0"):
+ return False
+ raise ValueError(str(x) + " could not be parsed as a Boolean")
+ return x
+
+
+def hhmm(x: str) -> str:
"""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")
return x
-def l3str(x):
+def hhmmhhmm(x: str) -> str:
+ """Check for the string that represents hours and minutes twice"""
+ if not isinstance(x, str) or len(x) != 8:
+ raise ValueError(str(x) + " is not an eight-character string")
+ return hhmm(x[:4]) + hhmm(x[4:])
+
+
+def l3str(x: Union[str, List[str]]) -> List[str]:
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
+ lx = x.split(",")
+ else:
+ lx = x
+ if len(lx) != 3 or not all(isinstance(el, str) for el in x):
+ raise ValueError(str(lx) + " is not a list of three strings")
+ return lx
-def l3int(x):
+def l3alarms(x: Union[str, List[Tuple[int, str]]]) -> List[Tuple[int, str]]:
+ def alrmspec(sub: str) -> Tuple[int, str]:
+ if len(sub) != 7:
+ raise ValueError(sub + " does not represent day and time")
+ return (
+ {
+ "MON": 1,
+ "TUE": 2,
+ "WED": 3,
+ "THU": 4,
+ "FRI": 5,
+ "SAT": 6,
+ "SUN": 7,
+ }[sub[:3].upper()],
+ sub[3:],
+ )
+
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
+ lx = [alrmspec(sub) for sub in x.split(",")]
+ else:
+ lx = x
+ lx.extend([(0, "0000") for _ in range(3 - len(lx))])
+ if len(lx) != 3 or any(d < 0 or d > 7 for d, tm in lx):
+ raise ValueError(str(lx) + " is a wrong alarms specification")
+ return [(d, hhmm(tm)) for d, tm in lx]
+
+
+def l3int(x: Union[str, List[int]]) -> List[int]:
+ if isinstance(x, str):
+ lx = [int(el) for el in x.split(",")]
+ else:
+ lx = x
+ if len(lx) != 3 or not all(isinstance(el, int) for el in lx):
+ raise ValueError(str(lx) + " is not a list of three integers")
+ return lx
class MetaPkt(type):
respectively.
"""
- def __new__(cls, name, bases, attrs):
+ if TYPE_CHECKING:
+
+ def __getattr__(self, name: str) -> Any:
+ pass
+
+ def __setattr__(self, name: str, value: Any) -> None:
+ pass
+
+ def __new__(
+ cls: Type["MetaPkt"],
+ name: str,
+ bases: Tuple[type, ...],
+ attrs: Dict[str, Any],
+ ) -> "MetaPkt":
newcls = super().__new__(cls, name, bases, attrs)
newcls.In = super().__new__(
cls,
class GPS303Pkt(metaclass=MetaPkt):
RESPOND = Respond.NON # Do not send anything back by default
PROTO: int
- IN_KWARGS = ()
- OUT_KWARGS = ()
+ IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
+ OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
+ KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = ()
+ In: Type["GPS303Pkt"]
+ Out: Type["GPS303Pkt"]
+
+ if TYPE_CHECKING:
+
+ def __getattr__(self, name: str) -> Any:
+ pass
+
+ def __setattr__(self, name: str, value: Any) -> None:
+ pass
- def __init__(self, *args, **kwargs):
+ def __init__(self, *args: Any, **kwargs: Any):
"""
Construct the object _either_ from (length, payload),
_or_ from the values of individual fields
assert not args or (len(args) == 2 and not kwargs)
if args: # guaranteed to be two arguments at this point
self.length, self.payload = args
- self.decode(self.length, self.payload)
+ try:
+ self.decode(self.length, self.payload)
+ except error as e:
+ raise DecodeError(e, obj=self)
else:
for kw, typ, dfl in self.KWARGS:
setattr(self, kw, typ(kwargs.pop(kw, dfl)))
self.__class__.__name__ + " stray kwargs " + str(kwargs)
)
- def __repr__(self):
+ def __repr__(self) -> str:
return "{}({})".format(
self.__class__.__name__,
", ".join(
),
)
- def in_decode(self, length, packet):
+ decode: Callable[["GPS303Pkt", int, bytes], None]
+
+ def in_decode(self, length: int, packet: bytes) -> None:
# Overridden in subclasses, otherwise do not decode payload
return
- def out_decode(self, length, packet):
+ def out_decode(self, length: int, packet: bytes) -> None:
# Overridden in subclasses, otherwise do not decode payload
return
- def in_encode(self):
+ encode: Callable[["GPS303Pkt"], bytes]
+
+ def in_encode(self) -> bytes:
# Necessary to emulate terminal, which is not implemented
raise NotImplementedError(
self.__class__.__name__ + ".encode() not implemented"
)
- def out_encode(self):
+ def out_encode(self) -> bytes:
# Overridden in subclasses, otherwise make empty payload
return b""
@property
- def packed(self):
+ def packed(self) -> bytes:
payload = self.encode()
- length = len(payload) + 1
+ length = getattr(self, "length", len(payload) + 1)
return pack("BB", length, self.PROTO) + payload
PROTO = 0x01
RESPOND = Respond.INL
# Default response for ACK, can also respond with STOP_UPLOAD
+ IN_KWARGS = (("imei", str, "0000000000000000"), ("ver", int, 0))
- def in_decode(self, length, payload):
- self.imei = payload[:-1].hex()
- self.ver = unpack("B", payload[-1:])[0]
- return self
+ def in_decode(self, length: int, payload: bytes) -> None:
+ self.imei = payload[:8].ljust(8, b"\0").hex()
+ self.ver = payload[8]
+
+ def in_encode(self) -> bytes:
+ return bytes.fromhex(self.imei).ljust(8, b"\0")[:8] + pack(
+ "B", self.ver
+ )
class SUPERVISION(GPS303Pkt):
PROTO = 0x05
OUT_KWARGS = (("status", int, 1),)
- def out_encode(self):
+ def out_encode(self) -> bytes:
# 1: The device automatically answers Pickup effect
# 2: Automatically Answering Two-way Calls
# 3: Ring manually answer the two-way call
class _GPS_POSITIONING(GPS303Pkt):
RESPOND = Respond.INL
- def in_decode(self, length, payload):
+ def in_decode(self, length: int, payload: bytes) -> None:
self.dtime = payload[:6]
if self.dtime == b"\0\0\0\0\0\0":
self.devtime = None
self.longitude = lon / (30000 * 60) * (-1 if flip_lon else 1)
self.speed = speed
self.flags = flags
- return self
- def out_encode(self):
+ def out_encode(self) -> bytes:
tup = datetime.utcnow().timetuple()
ttup = (tup[0] % 100,) + tup[1:6]
return pack("BBBBBB", *ttup)
class STATUS(GPS303Pkt):
PROTO = 0x13
RESPOND = Respond.EXT
+ IN_KWARGS = (
+ ("batt", int, 100),
+ ("ver", int, 0),
+ ("timezone", int, 0),
+ ("intvl", int, 0),
+ ("signal", maybe(int), None),
+ )
OUT_KWARGS = (("upload_interval", int, 25),)
- def in_decode(self, length, payload):
+ def in_decode(self, length: int, payload: bytes) -> None:
self.batt, self.ver, self.timezone, self.intvl = unpack(
"BBBB", payload[:4]
)
if len(payload) > 4:
- self.signal = payload[4]
+ self.signal: Optional[int] = payload[4]
else:
self.signal = None
- return self
- def out_encode(self): # Set interval in minutes
+ def in_encode(self) -> bytes:
+ return pack("BBBB", self.batt, self.ver, self.timezone, self.intvl) + (
+ b"" if self.signal is None else pack("B", self.signal)
+ )
+
+ def out_encode(self) -> bytes: # Set interval in minutes
return pack("B", self.upload_interval)
class HIBERNATION(GPS303Pkt): # Server can send to send devicee to sleep
PROTO = 0x14
+ def in_encode(self) -> bytes:
+ return b""
+
class RESET(GPS303Pkt):
# Device sends when it got reset SMS
PROTO = 0x16
OUT_KWARGS = (("number", int, 3),)
- def out_encode(self): # Number of whitelist entries
- return pack("B", number)
+ def out_encode(self) -> bytes: # Number of whitelist entries
+ return pack("B", self.number)
class _WIFI_POSITIONING(GPS303Pkt):
- def in_decode(self, length, payload):
+ IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = (
+ # IN_KWARGS = (
+ ("dtime", bytes, b"\0\0\0\0\0\0"),
+ ("wifi_aps", list, []),
+ ("mcc", int, 0),
+ ("mnc", int, 0),
+ ("gsm_cells", list, []),
+ )
+
+ def in_decode(self, length: int, payload: bytes) -> None:
self.dtime = payload[:6]
if self.dtime == b"\0\0\0\0\0\0":
self.devtime = None
"!HHB", gsm_slice[4 + i * 5 : 9 + i * 5]
)
self.gsm_cells.append((locac, cellid, -sigstr))
- return self
+
+ def in_encode(self) -> bytes:
+ self.length = len(self.wifi_aps)
+ return b"".join(
+ [
+ self.dtime,
+ b"".join(
+ [
+ bytes.fromhex(mac.replace(":", "")).ljust(6, b"\0")[:6]
+ + pack("B", -sigstr)
+ for mac, sigstr in self.wifi_aps
+ ]
+ ),
+ pack("!BHB", len(self.gsm_cells), self.mcc, self.mnc),
+ b"".join(
+ [
+ pack("!HHB", locac, cellid, -sigstr)
+ for locac, cellid, sigstr in self.gsm_cells
+ ]
+ ),
+ ]
+ )
class WIFI_OFFLINE_POSITIONING(_WIFI_POSITIONING):
PROTO = 0x17
RESPOND = Respond.INL
- def out_encode(self):
+ def out_encode(self) -> bytes:
return bytes.fromhex(datetime.utcnow().strftime("%y%m%d%H%M%S"))
PROTO = 0x30
RESPOND = Respond.INL
- def out_encode(self):
+ def out_encode(self) -> bytes:
return pack("!HBBBBB", *datetime.utcnow().timetuple()[:6])
PROTO = 0x33
OUT_KWARGS = (("status", int, 1),)
- def out_encode(self): # Server sent, 0-off, 1-on
+ def out_encode(self) -> bytes: # Server sent, 0-off, 1-on
return pack("B", self.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
+ OUT_KWARGS = (
+ ("gps_off", boolx, False), # Clarify the meaning of 0/1
+ ("gps_interval_set", boolx, False),
+ ("gps_interval", hhmmhhmm, "00000000"),
+ ("lbs_off", boolx, False), # Clarify the meaning of 0/1
+ ("boot_time_set", boolx, False),
+ ("boot_time", hhmm, "0000"),
+ ("shut_time_set", boolx, False),
+ ("shut_time", hhmm, "0000"),
+ )
+
+ def out_encode(self) -> bytes:
+ return (
+ pack("B", self.gps_off)
+ + pack("B", self.gps_interval_set)
+ + bytes.fromhex(self.gps_interval)
+ + pack("B", self.lbs_off)
+ + pack("B", self.boot_time_set)
+ + bytes.fromhex(self.boot_time)
+ + pack("B", self.shut_time_set)
+ + bytes.fromhex(self.shut_time)
+ )
class _SET_PHONE(GPS303Pkt):
OUT_KWARGS = (("phone", str, ""),)
- def out_encode(self):
- return self.phone.encode()
+ def out_encode(self) -> bytes:
+ self.phone: str
+ return self.phone.encode("")
class REMOTE_MONITOR_PHONE(_SET_PHONE):
("to", hhmm, "2359"),
)
- def out_encode(self):
+ def out_encode(self) -> bytes:
return (
pack("B", self.onoff)
+ bytes.fromhex(self.fm)
("to2", hhmm, "2359"),
)
- def out_endode(self):
+ def out_encode(self) -> bytes:
return (
pack("B", self.onoff)
+ pack("B", self.week)
PROTO = 0x48
OUT_KWARGS = (("flag", int, 0),)
- def out_encode(self):
+ def out_encode(self) -> bytes:
# 1 - restart
# 2 - shutdown
return pack("B", self.flag)
# 0 - Stop looking for equipment
# 1 - Start looking for equipment
- def out_encode(self):
+ def out_encode(self) -> bytes:
return pack("B", self.flag)
class ALARM_CLOCK(GPS303Pkt):
PROTO = 0x50
+ OUT_KWARGS: Tuple[
+ Tuple[str, Callable[[Any], Any], List[Tuple[int, str]]], ...
+ ] = (
+ ("alarms", l3alarms, []),
+ )
- def out_encode(self):
- # TODO implement parsing kwargs
- alarms = ((0, "0000"), (0, "0000"), (0, "0000"))
+ def out_encode(self) -> bytes:
return b"".join(
- cls("B", day) + bytes.fromhex(tm) for day, tm in alarms
+ pack("B", day) + bytes.fromhex(tm) for day, tm in self.alarms
)
class STOP_ALARM(GPS303Pkt):
PROTO = 0x56
- def in_decode(self, length, payload):
+ def in_decode(self, length: int, payload: bytes) -> None:
self.flag = payload[0]
- return self
class SETUP(GPS303Pkt):
("phonenumbers", l3str, ["", "", ""]),
)
- def out_encode(self):
- def pack3b(x):
+ def out_encode(self) -> bytes:
+ def pack3b(x: int) -> bytes:
return pack("!I", x)[1:]
return b"".join(
+ [b";".join([el.encode() for el in self.phonenumbers])]
)
+ def in_encode(self) -> bytes:
+ return b""
+
class SYNCHRONOUS_WHITELIST(GPS303Pkt):
PROTO = 0x58
RESPOND = Respond.EXT
OUT_KWARGS = (("latitude", float, None), ("longitude", float, None))
- def out_encode(self):
+ def out_encode(self) -> bytes:
if self.latitude is None or self.longitude is None:
return b""
return "{:+#010.8g},{:+#010.8g}".format(
self.latitude, self.longitude
).encode()
- def out_decode(self, length, payload):
+ def out_decode(self, length: int, payload: bytes) -> None:
lat, lon = payload.decode().split(",")
self.latitude = float(lat)
self.longitude = float(lon)
class MANUAL_POSITIONING(GPS303Pkt):
PROTO = 0x80
- def in_decode(self, length, payload):
- self.flag = payload[0] if len(payload) > 0 else None
+ def in_decode(self, length: int, payload: bytes) -> None:
+ self.flag = payload[0] if len(payload) > 0 else -1
self.reason = {
1: "Incorrect time",
2: "LBS less",
6: "LBS prohibited, WiFi absent",
7: "GPS spacing < 50 m",
}.get(self.flag, "Unknown")
- return self
class BATTERY_CHARGE(GPS303Pkt):
RESPOND = Respond.EXT
OUT_KWARGS = (("interval", int, 10),)
- def in_decode(self, length, payload):
+ def in_decode(self, length: int, payload: bytes) -> None:
self.interval = unpack("!H", payload[:2])
- return self
- def out_encode(self):
- return pack("!H", interval)
+ def out_encode(self) -> bytes:
+ return pack("!H", self.interval)
class SOS_ALARM(GPS303Pkt):
PROTO = 0xB3
IN_KWARGS = (("asciidata", str, ""),)
- def in_decode(self, length, payload):
+ def in_decode(self, length: int, payload: bytes) -> None:
self.asciidata = payload.decode()
- return self
# Build dicts protocol number -> class and class name -> protocol number
PROTOS[cls.__name__] = cls.PROTO
-def class_by_prefix(prefix):
+def class_by_prefix(
+ prefix: str,
+) -> Union[Type[GPS303Pkt], List[Tuple[str, int]]]:
lst = [
(name, proto)
for name, proto in PROTOS.items()
return CLASSES[proto]
-def proto_by_name(name):
+def proto_by_name(name: str) -> int:
return PROTOS.get(name, -1)
-def proto_of_message(packet):
- return unpack("B", packet[1:2])[0]
+def proto_of_message(packet: bytes) -> int:
+ return packet[1]
-def inline_response(packet):
+def inline_response(packet: bytes) -> Optional[bytes]:
proto = proto_of_message(packet)
if proto in CLASSES:
cls = CLASSES[proto]
return None
-def parse_message(packet, is_incoming=True):
+def parse_message(packet: bytes, is_incoming: bool = True) -> GPS303Pkt:
"""From a packet (without framing bytes) derive the XXX.In object"""
length, proto = unpack("BB", packet[:2])
payload = packet[2:]
- if proto in CLASSES:
- if is_incoming:
- return CLASSES[proto].In(length, payload)
- else:
- return CLASSES[proto].Out(length, payload)
+ if proto not in CLASSES:
+ cause: Union[DecodeError, ValueError, IndexError] = ValueError(
+ f"Proto {proto} is unknown"
+ )
else:
+ try:
+ if is_incoming:
+ return CLASSES[proto].In(length, payload)
+ else:
+ return CLASSES[proto].Out(length, payload)
+ except (DecodeError, ValueError, IndexError) as e:
+ cause = e
+ if is_incoming:
retobj = UNKNOWN.In(length, payload)
- retobj.PROTO = proto # Override class attr with object attr
- return retobj
+ else:
+ retobj = UNKNOWN.Out(length, payload)
+ retobj.PROTO = proto # Override class attr with object attr
+ retobj.cause = cause
+ return retobj