X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fgps303proto.py;h=31073d8ceacb6ff541d926fb78c72587ff91acf9;hb=f2fe626a313b77e3807bf1cfd1c83d0c89f3a905;hp=db8fd014637279ca256b14d44466b3d7e5eb3a0f;hpb=9b5436a2e568931f38e5721cdcf2e42bab151619;p=loctrkd.git diff --git a/gps303/gps303proto.py b/gps303/gps303proto.py index db8fd01..31073d8 100755 --- a/gps303/gps303proto.py +++ b/gps303/gps303proto.py @@ -40,7 +40,18 @@ __all__ = ( "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", @@ -52,11 +63,46 @@ __all__ = ( "CHARGER_DISCONNECTED", "VIBRATION_RECEIVED", "POSITION_UPLOAD_INTERVAL", + "SOS_ALARM", ) 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 @@ -71,23 +117,17 @@ class MetaPkt(type): def __new__(cls, name, bases, attrs): newcls = super().__new__(cls, name, bases, attrs) - nestattrs = {"encode": lambda self: self.in_encode()} - if "IN_KWARGS" in attrs: - nestattrs["KWARGS"] = attrs["IN_KWARGS"] newcls.In = super().__new__( cls, name + ".In", (newcls,) + bases, - nestattrs, + {"KWARGS": newcls.IN_KWARGS, "encode": newcls.in_encode}, ) - nestattrs = {"encode": lambda self: self.out_encode()} - if "OUT_KWARGS" in attrs: - nestattrs["KWARGS"] = attrs["OUT_KWARGS"] newcls.Out = super().__new__( cls, name + ".Out", (newcls,) + bases, - nestattrs, + {"KWARGS": newcls.OUT_KWARGS, "encode": newcls.out_encode}, ) return newcls @@ -103,6 +143,7 @@ class GPS303Pkt(metaclass=MetaPkt): PROTO: int # 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 @@ -246,7 +287,7 @@ class STATUS(GPS303Pkt): return self def out_encode(self): # Set interval in minutes - return cls.make_packet(pack("B", self.upload_interval)) + return pack("B", self.upload_interval) class HIBERNATION(GPS303Pkt): # Server can send to send devicee to sleep @@ -366,7 +407,11 @@ class STOP_UPLOAD(GPS303Pkt): # Server response to LOGIN to thwart the device class GPS_OFF_PERIOD(GPS303Pkt): PROTO = 0x46 - OUT_KWARGS = (("onoff", int, 0), ("fm", str, "0000"), ("to", str, "2359")) + OUT_KWARGS = ( + ("onoff", int, 0), + ("fm", hhmm, "0000"), + ("to", hhmm, "2359"), + ) def out_encode(self): return ( @@ -381,10 +426,10 @@ class DND_PERIOD(GPS303Pkt): OUT_KWARGS = ( ("onoff", int, 0), ("week", int, 3), - ("fm1", str, "0000"), - ("to1", str, "2359"), - ("fm2", str, "0000"), - ("to2", str, "2359"), + ("fm1", hhmm, "0000"), + ("to1", hhmm, "2359"), + ("fm2", hhmm, "0000"), + ("to2", hhmm, "2359"), ) def out_endode(self): @@ -400,7 +445,7 @@ class DND_PERIOD(GPS303Pkt): class RESTART_SHUTDOWN(GPS303Pkt): PROTO = 0x48 - OUT_KWARGS = (("flag", int, 2),) + OUT_KWARGS = (("flag", int, 0),) def out_encode(self): # 1 - restart @@ -442,16 +487,16 @@ class STOP_ALARM(GPS303Pkt): class SETUP(GPS303Pkt): PROTO = 0x57 RESPOND = Respond.EXT - OUT_KWARGS = ( # TODO handle properly - ("uploadintervalseconds", int, 0x0300), - ("binaryswitch", int, 0b00110001), - ("alarms", int, [0, 0, 0]), + OUT_KWARGS = ( + ("uploadintervalseconds", intx, 0x0300), + ("binaryswitch", intx, 0b00110001), + ("alarms", l3int, [0, 0, 0]), ("dndtimeswitch", int, 0), - ("dndtimes", int, [0, 0, 0]), + ("dndtimes", l3int, [0, 0, 0]), ("gpstimeswitch", int, 0), ("gpstimestart", int, 0), ("gpstimestop", int, 0), - ("phonenumbers", int, ["", "", ""]), + ("phonenumbers", l3str, ["", "", ""]), ) def out_encode(self):