X-Git-Url: http://www.average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=efb02d249f6c118185ce2a66b23675c9acf85efd;hb=c3bc6d5bbdc0d0bf10e338c6e3bad1a519d5afa0;hp=ab03c6829ccc94b025504ddb79208c5fcac4945b;hpb=bd031f87f4d3f640ce8862271aa69107f31a06d5;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index ab03c68..efb02d2 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -18,6 +18,7 @@ from datetime import datetime, timezone from enum import Enum from inspect import isclass from struct import error, pack, unpack +from time import time from typing import ( Any, Callable, @@ -31,10 +32,13 @@ from typing import ( ) __all__ = ( + "Stream", "class_by_prefix", "inline_response", "parse_message", + "probe_buffer", "proto_by_name", + "proto_name", "DecodeError", "Respond", "GPS303Pkt", @@ -77,6 +81,77 @@ __all__ = ( "UNKNOWN_B3", ) +PROTO_PREFIX = "ZX" + +### Deframer ### + +MAXBUFFER: int = 4096 + + +class Stream: + def __init__(self) -> None: + self.buffer = b"" + + @staticmethod + def enframe(buffer: bytes) -> bytes: + return b"xx" + buffer + b"\r\n" + + def recv(self, segment: bytes) -> List[Union[bytes, str]]: + """ + Process next segment of the stream. Return successfully deframed + packets as `bytes` and error messages as `str`. + """ + when = time() + self.buffer += segment + if len(self.buffer) > MAXBUFFER: + # We are receiving junk. Let's drop it or we run out of memory. + self.buffer = b"" + return [f"More than {MAXBUFFER} unparseable data, dropping"] + msgs: List[Union[bytes, str]] = [] + while True: + framestart = self.buffer.find(b"xx") + if framestart == -1: # No frames, return whatever we have + break + if framestart > 0: # Should not happen, report + msgs.append( + f'Undecodable data ({framestart}) "{self.buffer[:framestart][:64].hex()}"' + ) + self.buffer = self.buffer[framestart:] + # At this point, buffer starts with a packet + if len(self.buffer) < 6: # no len and proto - cannot proceed + break + exp_end = self.buffer[2] + 3 # Expect '\r\n' here + frameend = 0 + # Length field can legitimeely be much less than the + # length of the packet (e.g. WiFi positioning), but + # it _should not_ be greater. Still sometimes it is. + # Luckily, not by too much: by maybe two or three bytes? + # Do this embarrassing hack to avoid accidental match + # of some binary data in the packet against '\r\n'. + while True: + frameend = self.buffer.find(b"\r\n", frameend + 1) + if frameend == -1 or frameend >= ( + exp_end - 3 + ): # Found realistic match or none + break + if frameend == -1: # Incomplete frame, return what we have + break + packet = self.buffer[2:frameend] + self.buffer = self.buffer[frameend + 2 :] + if len(packet) < 2: # frameend comes too early + msgs.append(f"Packet too short: {packet.hex()}") + else: + msgs.append(packet) + return msgs + + def close(self) -> bytes: + ret = self.buffer + self.buffer = b"" + return ret + + +### Parser/Constructor ### + class DecodeError(Exception): def __init__(self, e: Exception, **kwargs: Any) -> None: @@ -85,12 +160,26 @@ class DecodeError(Exception): 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 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: @@ -102,6 +191,13 @@ def hhmm(x: str) -> str: return 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): lx = x.split(",") @@ -112,6 +208,33 @@ def l3str(x: Union[str, List[str]]) -> List[str]: return lx +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): + 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(",")] @@ -255,7 +378,7 @@ class GPS303Pkt(metaclass=MetaPkt): @property 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 @@ -267,10 +390,16 @@ class LOGIN(GPS303Pkt): 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: int, payload: bytes) -> None: - self.imei = payload[:-1].hex() - self.ver = unpack("B", payload[-1:])[0] + 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): @@ -330,6 +459,13 @@ class GPS_OFFLINE_POSITIONING(_GPS_POSITIONING): 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: int, payload: bytes) -> None: @@ -341,6 +477,11 @@ class STATUS(GPS303Pkt): else: self.signal = None + 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) @@ -348,6 +489,9 @@ class STATUS(GPS303Pkt): 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 @@ -364,6 +508,15 @@ class WHITELIST_TOTAL(GPS303Pkt): # Server sends to initiage sync (0x58) class _WIFI_POSITIONING(GPS303Pkt): + 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": @@ -388,6 +541,28 @@ class _WIFI_POSITIONING(GPS303Pkt): ) self.gsm_cells.append((locac, cellid, -sigstr)) + 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 @@ -416,17 +591,28 @@ class PROHIBIT_LBS(GPS303Pkt): 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 + 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 b"" # TODO + 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): @@ -517,12 +703,15 @@ class DEVICE(GPS303Pkt): class ALARM_CLOCK(GPS303Pkt): PROTO = 0x50 + OUT_KWARGS: Tuple[ + Tuple[str, Callable[[Any], Any], List[Tuple[int, str]]], ... + ] = ( + ("alarms", l3alarms, []), + ) def out_encode(self) -> bytes: - # TODO implement parsing kwargs - alarms = ((0, "0000"), (0, "0000"), (0, "0000")) return b"".join( - pack("B", day) + bytes.fromhex(tm) for day, tm in alarms + pack("B", day) + bytes.fromhex(tm) for day, tm in self.alarms ) @@ -570,6 +759,9 @@ class SETUP(GPS303Pkt): + [b";".join([el.encode() for el in self.phonenumbers])] ) + def in_encode(self) -> bytes: + return b"" + class SYNCHRONOUS_WHITELIST(GPS303Pkt): PROTO = 0x58 @@ -683,16 +875,40 @@ def class_by_prefix( return CLASSES[proto] +def proto_name(obj: Union[MetaPkt, GPS303Pkt]) -> str: + return ( + PROTO_PREFIX + + ":" + + ( + obj.__class__.__name__ + if isinstance(obj, GPS303Pkt) + else obj.__name__ + ) + ).ljust(16, "\0")[:16] + + def proto_by_name(name: str) -> int: return PROTOS.get(name, -1) -def proto_of_message(packet: bytes) -> int: - return packet[1] +def proto_of_message(packet: bytes) -> str: + return proto_name(CLASSES.get(packet[1], UNKNOWN)) + + +def imei_from_packet(packet: bytes) -> Optional[str]: + if packet[1] == LOGIN.PROTO: + msg = parse_message(packet) + if isinstance(msg, LOGIN): + return msg.imei + return None + + +def is_goodbye_packet(packet: bytes) -> bool: + return packet[1] == HIBERNATION.PROTO def inline_response(packet: bytes) -> Optional[bytes]: - proto = proto_of_message(packet) + proto = packet[1] if proto in CLASSES: cls = CLASSES[proto] if cls.RESPOND is Respond.INL: @@ -700,12 +916,21 @@ def inline_response(packet: bytes) -> Optional[bytes]: return None +def probe_buffer(buffer: bytes) -> bool: + framestart = buffer.find(b"xx") + if framestart < 0: + return False + if len(buffer) - framestart < 6: + return False + return 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 not in CLASSES: - cause: Union[DecodeError, ValueError] = ValueError( + cause: Union[DecodeError, ValueError, IndexError] = ValueError( f"Proto {proto} is unknown" ) else: @@ -714,7 +939,7 @@ def parse_message(packet: bytes, is_incoming: bool = True) -> GPS303Pkt: return CLASSES[proto].In(length, payload) else: return CLASSES[proto].Out(length, payload) - except DecodeError as e: + except (DecodeError, ValueError, IndexError) as e: cause = e if is_incoming: retobj = UNKNOWN.In(length, payload)