Mercurial > ~darius > hgwebdir.cgi > ZigBee
diff zb.py @ 9:d147529ad2db
Switch to only reading RX pkts and writing TX pkts to make things simpler.
Add AT reply test case.
| author | darius@inchoate.localdomain |
|---|---|
| date | Mon, 05 Nov 2007 23:35:52 +1030 |
| parents | 9f0808b13454 |
| children | 4c91fdfc862e |
line wrap: on
line diff
--- a/zb.py Sun Nov 04 21:10:21 2007 +1030 +++ b/zb.py Mon Nov 05 23:35:52 2007 +1030 @@ -1,22 +1,46 @@ import serial, inspect +def easyord(i): + if (type(i) != type(str())): + return i + else: + return ord(i) + class PktBase(object): + PKT_MAXLEN = 2 ** 16 + def __init__(self, data = []): - self.data = data + self._data = data print "Constructing " + self.__class__.__name__ - def __repr__(self): - return str(self.__class__) + "(" + str([self.PKT_TYPE] + self.data) + ")" - def Encapsulate(self): return Packets.Encapsulate([self.PKT_TYPE] + self.data) def resize(self, dlen): """Ensure the data list can hold at least len elements (0 fill)""" - if (len(self.data) < dlen): - self.data = (self.data + [0] * dlen)[0:dlen] + if (len(self._data) < dlen): + self._data = (self._data + [0] * dlen)[0:dlen] + + def _checklist(list, min = 0, max = 255, maxlen = None): + if (maxlen != None and len(list) > maxlen): + raise ValueError("must have %d elements" % (maxlen)) + + for i in xrange(len(list)): + if (easyord(list[i]) < min or easyord(list[i]) > max): + raise ValueError("element %d (= %d) out of range must be between %d and %d inclusive" % + (i, ord(list[i]), min, max)) + _checklist = staticmethod(_checklist) + +class TXPkts(PktBase): + """Base class for all packets that go to the module""" + def setframeid(self, value): + if (value < 0 or value > 255): + raise ValueError("FrameID must be 0-255") + self._frameid = value + frameid = property(lambda s: s._frameid, setframeid) -class AT_Cmd(PktBase): + +class AT_Cmd(TXPkts): PKT_TYPE = 0x08 PKT_DESC = "AT Command" @@ -33,22 +57,25 @@ self.cmdarg = args[1] else: raise TypeError("__init__ takes 1 list of ordinals, 1 string, or 2 strings") - - self.data[0] = 1 # XXX: could be smarter about Frame ID - + self.frameid = 0 + self.cmdarg = [] + def setcmd(self, value): - self.resize(3) - self.data[1] = ord(value[0]) - self.data[2] = ord(value[1]) - cmd = property(lambda s: chr(s.data[1]) + chr(s.data[2]), setcmd) + if (len(value) != 2): + raise ValueError("must have 2 elements") + self._checklist(value, ord('0'), ord('z')) + self._cmd = value + cmd = property(lambda s: s._cmd, setcmd) def setcmdarg(self, value): - self.resize(3 + len(value)) - self.data = self.data[0:3] + map(ord, value) - def delcmdarg(self): - self.data = self.data[0:3] - cmdarg = property(lambda s: map(chr, s.data[3:]), setcmdarg, delcmdarg) - + self._checklist(value, maxlen = self.PKT_MAXLEN - 3) + self._cmdarg = value + cmdarg = property(lambda s: s._cmdarg, setcmdarg) + + def getdata(self): + return([self.frameid] + map(ord, self.cmd) + map(easyord, self.cmdarg)) + data = property(getdata) + class AT_Cmd_Queue(AT_Cmd): PKT_TYPE = 0x09 PKT_DESC = "AT Command (queued)" @@ -56,10 +83,10 @@ class AT_Response(PktBase): PKT_TYPE = 0x88 PKT_DESC = "AT Command response" - frameid = property(lambda s: s.data[0], None) - cmd = property(lambda s: chr(s.data[1]) + chr(s.data[2]), None) - statusOK = property(lambda s: s.data[3] == 0, None) - payload = property(lambda s: s.data[4:], None) + frameid = property(lambda s: s._data[0], None) + cmd = property(lambda s: chr(s._data[1]) + chr(s._data[2]), None) + statusOK = property(lambda s: s._data[3] == 0, None) + payload = property(lambda s: s._data[4:], None) class Modem_Status(PktBase): PKT_TYPE = 0x8a @@ -77,15 +104,15 @@ def getsender(self): value = 0 for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): - value |= self.data[i] << j + value |= self._data[i] << j return value sender = property(getsender, None) - rssi = property(lambda s: -1 * s.data[s.ADDR_SIZE], None) + rssi = property(lambda s: -1 * s._data[s.ADDR_SIZE], None) - flags = property(lambda s: s.data[s.ADDR_SIZE + 1], None) + flags = property(lambda s: s._data[s.ADDR_SIZE + 1], None) - payload = property(lambda s: s.data[s.ADDR_SIZE + 2:], None) + payload = property(lambda s: s._data[s.ADDR_SIZE + 2:], None) class RX_64_Bit(RX_16_Bit): PKT_TYPE = 0x80 @@ -96,27 +123,17 @@ PKT_TYPE = 0x83 PKT_DESC = "RXIO Packet: 16 bit address" - def setnsamples(self, value): - self.resize(self.ADDR_SIZE + 3) - self.data[self.ADDR_SIZE + 2] = value + nsamples = property(lambda s: s._data[s.ADDR_SIZE + 2]) - def getnsamples(self): - return self.data[self.ADDR_SIZE + 2] - nsamples = property(getnsamples, setnsamples) - - def setmask(self, value): - self.resize(self.ADDR_SIZE + 5) - self.data[self.ADDR_SIZE + 3] = (value & 0xff00) >> 8 - self.data[self.ADDR_SIZE + 4] = value & 0xff - mask = property(lambda s: s.data[s.ADDR_SIZE + 3] << 8 | s.data[s.ADDR_SIZE + 4], setmask) + mask = property(lambda s: s._data[s.ADDR_SIZE + 3] << 8 | s._data[s.ADDR_SIZE + 4]) def __str__(self): rtn = "0x%0*x (%ddBm) -> %d samples, mask 0x%04x" % (self.ADDR_SIZE * 2, self.sender, self.rssi, self.nsamples, self.mask) # Any DIO lines enabled? if (self.mask | 0x01ff): - rtn = rtn + ", DIO - 0x%03x" % (self.data[self.ADDR_SIZE + 5] << 8 | - self.data[self.ADDR_SIZE + 6]) + rtn = rtn + ", DIO - 0x%03x" % (self._data[self.ADDR_SIZE + 5] << 8 | + self._data[self.ADDR_SIZE + 6]) offs = self.ADDR_SIZE + 7 else: offs = self.ADDR_SIZE + 5 @@ -125,10 +142,9 @@ if (self.mask | 0x7e00): for i in range(6): if (self.mask & 1 << (i + 9)): - rtn = rtn + ", ADC%d - 0x%02x" % (i, self.data[offs] << 8 | - self.data[offs + 1]) + rtn = rtn + ", ADC%d - 0x%02x" % (i, self._data[offs] << 8 | + self._data[offs + 1]) offs = offs + 2 - return rtn class RXIO_64_Bit(RXIO_16_Bit): @@ -136,18 +152,18 @@ PKT_DESC = "RXIO Packet: 64 bit address" ADDR_SIZE = 8 -class TX_16_Bit(RX_64_Bit): +class TX_16_Bit(TXPkts): PKT_TYPE = 0x01 PKT_DESC = "TX Packet: 16 bit address" ADDR_SIZE = 2 - + FLG_DISABLE_ACK = 0x01 + FLG_BCAST_PANID = 0x04 + PKT_MAX_PAYLOAD = 100 + def __init__(self, *args): if (len(args) == 1): - if (type(args[0]) == type(int())): - super(TX_16_Bit, self).__init__([]) - self.recipient = args[0] - else: - super(TX_16_Bit, self).__init__(args[0]) + super(TX_16_Bit, self).__init__() + self.recipient = args[0] elif (len(args) == 2): super(TX_16_Bit, self).__init__([]) self.recipient = args[0] @@ -155,28 +171,38 @@ else: raise TypeError("__init__ takes 1 list of ordinals or 2 strings") - self.resize(self.ADDR_SIZE + 2) - self.data[0] = 1 # XXX: could be smarter about Frame ID - self.data[3] = 0 # XXX: should be able to set flags + self.frameid = 0 + self.flags = 0 + self.payload = [] + + def setrecipient(self, value): + if (value < 0 or value > 2 ** (self.ADDR_SIZE * 8)): + raise ValueError("value out of range must be between 0 and %d" % (2 ** self.ADDR_SIZE)) - def getrecipient(self): - value = 0 - for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): - value |= self.data[i + 1] << j - return value - - def setrecipient(self, value): - self.resize(self.ADDR_SIZE + 1) - self.data[0] = 1 # XXX: could be smarter about Frame ID - for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): - self.data[i + 1] = (value & (0xff << j)) >> j - recipient = property(getrecipient, setrecipient) + self._recipient = value + recipient = property(lambda s: s._recipient, setrecipient) + + def setflags(self, value): + if (value < 0 or value > 255): + raise ValueError("Value must be between 0 and 255 inclusive") + + self._flags = value + flags = property(lambda s: s._flags, setflags) def setpayload(self, value): - self.resize(self.ADDR_SIZE + 2) - self.data[self.ADDR_SIZE + 2:] = value - payload = property(lambda s: s.data[self.ADDR_SIZE + 2:], setpayload) - + self._checklist(value, maxlen = self.PKT_MAX_PAYLOAD) + self._payload = value + payload = property(lambda s: s._payload, setpayload) + + def getdata(self): + data = [self.frameid] + for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): + data.append((self.recipient & (0xff << j)) >> j) + data.append(self.flags) + data.extend(map(easyord, self.payload)) + return(data) + data = property(getdata) + class TX_64_Bit(TX_16_Bit): PKT_TYPE = 0x00 PKT_DESC = "TX Packet: 64 bit address" @@ -186,9 +212,9 @@ PKT_TYPE = 0x89 PKT_DESC = "TX Status" statusTxt = ['OK', 'No Ack', 'CCA failure', 'Purged'] - frameid = property(lambda s: s.data[0], None) - status = property(lambda s: s.data[1], None) - statusMsg = property(lambda s: s.statusTxt[s.data[1]], None) + frameid = property(lambda s: s._data[0], None) + status = property(lambda s: s._data[1], None) + statusMsg = property(lambda s: s.statusTxt[s._data[1]], None) class Packets(object): PKT_CLASSES = None @@ -225,7 +251,7 @@ self.packets = [] self.bufmsb = 0 - self.dataleft = 0 + self._dataleft = 0 self.fr_err = 0 # Framing error self.ck_err = 0 # Checksum error @@ -314,6 +340,9 @@ # Exception badpkttypetest = [126, 0, 3, 10, 86, 76, 83] +# Frame ID = 0, Cmd = 'VL', Status = OK, Value = 'VL Result' +atreply = [126, 0, 14, 136, 0, 86, 76, 0, 86, 76, 32, 82, 101, 115, 117, 108, 116, 148] + up = Packets(s) up.process(goodtest) up.process(badtest)