X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=0a222d05634930679995e6e4807d939d35104346;hb=057cce452eb53d5fbe365a66669bd8dec7dfe989;hp=5613bea455e3b2e9d66be24b173a10fddc325b35;hpb=7ff26a2079e56969c65937b1bae7f6d908aa0664;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index 5613bea..0a222d0 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -85,6 +85,10 @@ 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) @@ -299,7 +303,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 @@ -311,11 +315,17 @@ 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[:8].hex() + 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 @@ -374,6 +384,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: @@ -385,6 +402,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) @@ -392,6 +414,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 @@ -408,6 +433,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": @@ -432,6 +466,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 @@ -628,6 +684,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