Mercurial > ~darius > hgwebdir.cgi > ZigBee
diff zb.py @ 6:9c499d923544
Many fixes.
- Make AT_Cmd constructor smarter/useful.
- Flesh out AT_Response, TX_*_Bit and TX_Status
- Make writedata() and getdata() simpler to use (hold reference to
serial port inside the class)
author | darius@inchoate.localdomain |
---|---|
date | Thu, 01 Nov 2007 16:30:36 +1030 |
parents | 5d5963d542bc |
children | 579dedf5a1f1 |
line wrap: on
line diff
--- a/zb.py Wed Oct 31 20:47:32 2007 +1030 +++ b/zb.py Thu Nov 01 16:30:36 2007 +1030 @@ -22,27 +22,32 @@ def __init__(self, *args): if (len(args) == 1): - super(AT_Cmd, self).__init__(args[0]) + if (type(args[0]) == type(str())): + super(AT_Cmd, self).__init__([]) + self.cmd = args[0] + else: + super(AT_Cmd, self).__init__(args[0]) elif (len(args) == 2): super(AT_Cmd, self).__init__([]) self.cmd = args[0] - if (args[1] != None): - self.cmdarg = args[1] + self.cmdarg = args[1] else: - raise TypeError("__init__ takes 1 list of ordinals or 2 strings") + raise TypeError("__init__ takes 1 list of ordinals, 1 string, or 2 strings") + + self.data[0] = 1 # XXX: could be smarter about Frame ID def setcmd(self, value): - self.resize(2) - self.data[0] = ord(value[0]) - self.data[1] = ord(value[1]) - cmd = property(lambda s: chr(s.data[0]) + chr(s.data[1]), setcmd) + 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) def setcmdarg(self, value): - self.resize(2 + len(value)) - self.data = self.data[0:2] + map(ord, value) + self.resize(3 + len(value)) + self.data = self.data[0:3] + map(ord, value) def delcmdarg(self): - self.data = self.data[0:2] - cmdarg = property(lambda s: map(chr, s.data[2:]), setcmdarg, delcmdarg) + self.data = self.data[0:3] + cmdarg = property(lambda s: map(chr, s.data[3:]), setcmdarg, delcmdarg) class AT_Cmd_Queue(AT_Cmd): PKT_TYPE = 0x09 @@ -51,6 +56,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) class Modem_Status(PktBase): PKT_TYPE = 0x8a @@ -124,23 +133,59 @@ ADDR_SIZE = 8 class TX_16_Bit(RX_64_Bit): - PKT_TYPE = 0x00 + PKT_TYPE = 0x01 PKT_DESC = "TX Packet: 16 bit address" + ADDR_SIZE = 2 + + 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]) + elif (len(args) == 2): + super(TX_16_Bit, self).__init__([]) + self.recipient = args[0] + self.payload = args[1] + else: + raise TypeError("__init__ takes 1 list of ordinals or 2 strings") - def setsender(self, value): - self.resize(self.ADDR_SIZE) + 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 + + def getrecipient(self): + value = 0 for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): - self.data[i] = (value & (0xff << j)) >> j - sender = property(RX_64_Bit.getsender, setsender) + 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) -class TX_64_Bit(RX_64_Bit): + 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) + +class TX_64_Bit(TX_16_Bit): PKT_TYPE = 0x00 PKT_DESC = "TX Packet: 64 bit address" + ADDR_SIZE = 8 class TX_Status(PktBase): 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) + class Packets(object): PKT_CLASSES = None @@ -169,7 +214,7 @@ Encapsulate = staticmethod(Encapsulate) - def __init__(self): + def __init__(self, s): print str(inspect.getmodule(self)) self.buffer = [] self.state = 'init' @@ -183,14 +228,15 @@ self.rx_cnt = 0 # Packet count self.pktq = [] - - def writedata(self, s, data): - s.write("".join(map(str, data))) + self.s = s + + def writedata(self, data): + self.s.write("".join(map(str, data))) - def getdata(self, s): + def getdata(self): l = [] while (1): - a = s.read() + a = self.s.read() if (a == ''): break l.append(ord(a)) @@ -229,7 +275,7 @@ print "Checksum error, got 0x%02x, expected 0x%02x" % \ (rxcksum, 0xff - pktsum) else: - #print "Got a packet - " + str(self.buffer) + print "Got a packet - " + str(self.buffer) p = Packets.Build(self.buffer) self.pktq.append(p) self.buffer = [] @@ -264,7 +310,7 @@ # Exception badpkttypetest = [126, 0, 3, 10, 86, 76, 83] -up = Packets() +up = Packets(s) up.process(goodtest) up.process(badtest) up.process(adctest)