Mercurial > ~darius > hgwebdir.cgi > ZigBee
annotate 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 |
| rev | line source |
|---|---|
| 4 | 1 import serial, inspect |
| 1 | 2 |
| 3 class PktBase(object): | |
| 4 | 4 def __init__(self, data = []): |
| 1 | 5 self.data = data |
| 6 print "Constructing " + self.__class__.__name__ | |
| 7 | |
| 8 def __repr__(self): | |
| 9 return str(self.__class__) + "(" + str([self.PKT_TYPE] + self.data) + ")" | |
| 10 | |
| 4 | 11 def Encapsulate(self): |
| 12 return Packets.Encapsulate([self.PKT_TYPE] + self.data) | |
| 1 | 13 |
| 4 | 14 def resize(self, dlen): |
| 15 """Ensure the data list can hold at least len elements (0 fill)""" | |
| 16 if (len(self.data) < dlen): | |
| 17 self.data = (self.data + [0] * dlen)[0:dlen] | |
| 18 | |
| 1 | 19 class AT_Cmd(PktBase): |
| 20 PKT_TYPE = 0x08 | |
| 21 PKT_DESC = "AT Command" | |
| 22 | |
| 4 | 23 def __init__(self, *args): |
| 24 if (len(args) == 1): | |
| 6 | 25 if (type(args[0]) == type(str())): |
| 26 super(AT_Cmd, self).__init__([]) | |
| 27 self.cmd = args[0] | |
| 28 else: | |
| 29 super(AT_Cmd, self).__init__(args[0]) | |
| 4 | 30 elif (len(args) == 2): |
| 31 super(AT_Cmd, self).__init__([]) | |
| 32 self.cmd = args[0] | |
| 6 | 33 self.cmdarg = args[1] |
| 4 | 34 else: |
| 6 | 35 raise TypeError("__init__ takes 1 list of ordinals, 1 string, or 2 strings") |
| 36 | |
| 37 self.data[0] = 1 # XXX: could be smarter about Frame ID | |
| 4 | 38 |
| 39 def setcmd(self, value): | |
| 6 | 40 self.resize(3) |
| 41 self.data[1] = ord(value[0]) | |
| 42 self.data[2] = ord(value[1]) | |
| 43 cmd = property(lambda s: chr(s.data[1]) + chr(s.data[2]), setcmd) | |
| 4 | 44 |
| 45 def setcmdarg(self, value): | |
| 6 | 46 self.resize(3 + len(value)) |
| 47 self.data = self.data[0:3] + map(ord, value) | |
| 4 | 48 def delcmdarg(self): |
| 6 | 49 self.data = self.data[0:3] |
| 50 cmdarg = property(lambda s: map(chr, s.data[3:]), setcmdarg, delcmdarg) | |
| 4 | 51 |
| 52 class AT_Cmd_Queue(AT_Cmd): | |
| 1 | 53 PKT_TYPE = 0x09 |
| 54 PKT_DESC = "AT Command (queued)" | |
| 55 | |
| 56 class AT_Response(PktBase): | |
| 57 PKT_TYPE = 0x88 | |
| 58 PKT_DESC = "AT Command response" | |
| 6 | 59 frameid = property(lambda s: s.data[0], None) |
| 60 cmd = property(lambda s: chr(s.data[1]) + chr(s.data[2]), None) | |
| 61 statusOK = property(lambda s: s.data[3] == 0, None) | |
| 62 payload = property(lambda s: s.data[4:], None) | |
| 1 | 63 |
| 64 class Modem_Status(PktBase): | |
| 65 PKT_TYPE = 0x8a | |
| 66 PKT_DESC = "Modem Status" | |
| 67 | |
| 68 class RX_16_Bit(PktBase): | |
| 69 PKT_TYPE = 0x81 | |
| 70 PKT_DESC = "RX Packet: 16 bit address" | |
| 4 | 71 ADDR_SIZE = 2 |
| 72 | |
| 73 def getsender(self): | |
| 74 value = 0 | |
| 75 for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): | |
| 76 value |= self.data[i] << j | |
| 77 return value | |
| 5 | 78 sender = property(getsender, None) |
| 79 | |
| 80 rssi = property(lambda s: -1 * s.data[s.ADDR_SIZE], None) | |
| 1 | 81 |
| 5 | 82 flags = property(lambda s: s.data[s.ADDR_SIZE + 1], None) |
| 4 | 83 |
| 5 | 84 payload = property(lambda s: s.data[s.ADDR_SIZE + 2:], None) |
| 85 | |
| 4 | 86 class RX_64_Bit(RX_16_Bit): |
| 87 PKT_TYPE = 0x80 | |
| 88 PKT_DESC = "RX Packet: 64 bit address" | |
| 89 ADDR_SIZE = 8 | |
| 90 | |
| 91 class RXIO_16_Bit(RX_16_Bit): | |
| 1 | 92 PKT_TYPE = 0x83 |
| 93 PKT_DESC = "RXIO Packet: 16 bit address" | |
| 94 | |
| 4 | 95 def setnsamples(self, value): |
| 96 self.resize(self.ADDR_SIZE + 3) | |
| 97 self.data[self.ADDR_SIZE + 2] = value | |
| 5 | 98 |
| 99 def getnsamples(self): | |
| 100 return self.data[self.ADDR_SIZE + 2] | |
| 101 nsamples = property(getnsamples, setnsamples) | |
| 4 | 102 |
| 103 def setmask(self, value): | |
| 104 self.resize(self.ADDR_SIZE + 5) | |
| 105 self.data[self.ADDR_SIZE + 3] = (value & 0xff00) >> 8 | |
| 106 self.data[self.ADDR_SIZE + 4] = value & 0xff | |
| 107 mask = property(lambda s: s.data[s.ADDR_SIZE + 3] << 8 | s.data[s.ADDR_SIZE + 4], setmask) | |
| 108 | |
| 1 | 109 def __str__(self): |
| 4 | 110 rtn = "0x%0*x (%ddBm) -> %d samples, mask 0x%04x" % (self.ADDR_SIZE * 2, self.sender, |
| 111 self.rssi, self.nsamples, self.mask) | |
| 1 | 112 # Any DIO lines enabled? |
| 4 | 113 if (self.mask | 0x01ff): |
| 114 rtn = rtn + ", DIO - 0x%03x" % (self.data[self.ADDR_SIZE + 5] << 8 | | |
| 115 self.data[self.ADDR_SIZE + 6]) | |
| 116 offs = self.ADDR_SIZE + 7 | |
| 1 | 117 else: |
| 4 | 118 offs = self.ADDR_SIZE + 5 |
| 1 | 119 |
| 120 # Any ADC lines enabled? | |
| 4 | 121 if (self.mask | 0x7e00): |
| 1 | 122 for i in range(6): |
| 4 | 123 if (self.mask & 1 << (i + 9)): |
| 1 | 124 rtn = rtn + ", ADC%d - 0x%02x" % (i, self.data[offs] << 8 | |
| 125 self.data[offs + 1]) | |
| 126 offs = offs + 2 | |
| 127 | |
| 128 return rtn | |
| 129 | |
| 5 | 130 class RXIO_64_Bit(RXIO_16_Bit): |
| 4 | 131 PKT_TYPE = 0x82 |
| 132 PKT_DESC = "RXIO Packet: 64 bit address" | |
| 133 ADDR_SIZE = 8 | |
| 134 | |
| 5 | 135 class TX_16_Bit(RX_64_Bit): |
| 6 | 136 PKT_TYPE = 0x01 |
| 4 | 137 PKT_DESC = "TX Packet: 16 bit address" |
| 6 | 138 ADDR_SIZE = 2 |
| 139 | |
| 140 def __init__(self, *args): | |
| 141 if (len(args) == 1): | |
| 142 if (type(args[0]) == type(int())): | |
| 143 super(TX_16_Bit, self).__init__([]) | |
| 144 self.recipient = args[0] | |
| 145 else: | |
| 146 super(TX_16_Bit, self).__init__(args[0]) | |
| 147 elif (len(args) == 2): | |
| 148 super(TX_16_Bit, self).__init__([]) | |
| 149 self.recipient = args[0] | |
| 150 self.payload = args[1] | |
| 151 else: | |
| 152 raise TypeError("__init__ takes 1 list of ordinals or 2 strings") | |
| 4 | 153 |
| 6 | 154 self.resize(self.ADDR_SIZE + 2) |
| 155 self.data[0] = 1 # XXX: could be smarter about Frame ID | |
| 156 self.data[3] = 0 # XXX: should be able to set flags | |
| 157 | |
| 158 def getrecipient(self): | |
| 159 value = 0 | |
| 5 | 160 for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): |
| 6 | 161 value |= self.data[i + 1] << j |
| 162 return value | |
| 163 | |
| 164 def setrecipient(self, value): | |
| 165 self.resize(self.ADDR_SIZE + 1) | |
| 166 self.data[0] = 1 # XXX: could be smarter about Frame ID | |
| 167 for i, j in zip(reversed(range(self.ADDR_SIZE)), range(0, self.ADDR_SIZE * 8, 8)): | |
| 168 self.data[i + 1] = (value & (0xff << j)) >> j | |
| 169 recipient = property(getrecipient, setrecipient) | |
| 5 | 170 |
| 6 | 171 def setpayload(self, value): |
| 172 self.resize(self.ADDR_SIZE + 2) | |
| 173 self.data[self.ADDR_SIZE + 2:] = value | |
| 174 payload = property(lambda s: s.data[self.ADDR_SIZE + 2:], setpayload) | |
| 175 | |
| 176 class TX_64_Bit(TX_16_Bit): | |
| 1 | 177 PKT_TYPE = 0x00 |
| 178 PKT_DESC = "TX Packet: 64 bit address" | |
| 6 | 179 ADDR_SIZE = 8 |
| 1 | 180 |
| 181 class TX_Status(PktBase): | |
| 182 PKT_TYPE = 0x89 | |
| 183 PKT_DESC = "TX Status" | |
| 6 | 184 statusTxt = ['OK', 'No Ack', 'CCA failure', 'Purged'] |
| 185 frameid = property(lambda s: s.data[0], None) | |
| 186 status = property(lambda s: s.data[1], None) | |
| 187 statusMsg = property(lambda s: s.statusTxt[s.data[1]], None) | |
| 188 | |
| 1 | 189 class Packets(object): |
| 4 | 190 PKT_CLASSES = None |
| 1 | 191 |
| 4 | 192 def Build(self, data): |
| 193 if (self.PKT_CLASSES == None): | |
| 194 m = inspect.getmodule(self) | |
| 195 # Generate list of objects from their names | |
| 196 mobjs = map(lambda n: m.__dict__[n], m.__dict__) | |
| 197 # Find all the classes | |
| 198 pktclasses = filter(inspect.isclass, mobjs) | |
| 199 # Find all subclasses of PktBase (but not PktBase itself) | |
| 200 pktclasses = filter(lambda s: issubclass(s, m.PktBase) and s != m.PktBase, pktclasses) | |
| 201 self.PKT_CLASSES = pktclasses | |
| 202 | |
| 203 for p in self.PKT_CLASSES: | |
| 1 | 204 if (p.PKT_TYPE == data[0]): |
| 205 return(p(data[1:])) | |
| 4 | 206 |
| 207 raise ValueError("Unknown packet type 0x%02x" % (data[0])) | |
| 208 Build = classmethod(Build) | |
| 1 | 209 |
| 210 def Encapsulate(data): | |
| 211 pktsum = reduce(lambda x, y: x + y, data) & 0xff | |
| 212 pkt = [0x7e] + [len(data) >> 8] + [len(data) & 0xff] + data + [0xff - pktsum] | |
| 213 return(map(chr, pkt)) | |
| 214 | |
| 215 Encapsulate = staticmethod(Encapsulate) | |
| 216 | |
| 6 | 217 def __init__(self, s): |
| 4 | 218 print str(inspect.getmodule(self)) |
| 1 | 219 self.buffer = [] |
| 220 self.state = 'init' | |
| 221 self.packets = [] | |
| 222 | |
| 223 self.bufmsb = 0 | |
| 224 self.dataleft = 0 | |
| 225 | |
| 226 self.fr_err = 0 # Framing error | |
| 227 self.ck_err = 0 # Checksum error | |
| 228 self.rx_cnt = 0 # Packet count | |
| 229 | |
| 230 self.pktq = [] | |
| 6 | 231 self.s = s |
| 232 | |
| 233 def writedata(self, data): | |
| 234 self.s.write("".join(map(str, data))) | |
| 5 | 235 |
| 6 | 236 def getdata(self): |
| 1 | 237 l = [] |
| 238 while (1): | |
| 6 | 239 a = self.s.read() |
| 1 | 240 if (a == ''): |
| 241 break | |
| 5 | 242 l.append(ord(a)) |
| 1 | 243 |
| 244 return self.process(l) | |
| 245 | |
| 246 def process(self, data): | |
| 247 pktcount = 0 | |
| 248 for d in data: | |
| 249 if (self.state == 'init'): | |
| 5 | 250 if (d != 0x7e): |
| 251 print "Framing error, got 0x%02x, expected 0x7e" % (d) | |
| 4 | 252 self.fr_err += 1 |
| 1 | 253 continue |
| 254 | |
| 255 self.state = 'sizemsb' | |
| 256 elif (self.state == 'sizemsb'): | |
| 5 | 257 self.bufmsb = d |
| 1 | 258 self.state = 'sizelsb' |
| 259 elif (self.state == 'sizelsb'): | |
| 260 self.dataleft = self.bufmsb << 8 | \ | |
| 5 | 261 d |
| 1 | 262 self.state = 'data' |
| 263 elif (self.state == 'data'): | |
| 5 | 264 self.buffer.append(d) |
| 1 | 265 self.dataleft = self.dataleft - 1 |
| 266 if (self.dataleft == 0): | |
| 267 self.state = 'cksum' | |
| 268 elif (self.state == 'cksum'): | |
| 269 pktsum = reduce(lambda x, y: x + y, self.buffer) & 0xff | |
| 5 | 270 rxcksum = d |
| 1 | 271 self.state = 'init' |
| 272 if (pktsum + rxcksum != 0xff): | |
| 273 self.buffer = [] | |
| 4 | 274 self.ck_err += 1 |
| 1 | 275 print "Checksum error, got 0x%02x, expected 0x%02x" % \ |
| 276 (rxcksum, 0xff - pktsum) | |
| 277 else: | |
| 6 | 278 print "Got a packet - " + str(self.buffer) |
| 1 | 279 p = Packets.Build(self.buffer) |
| 280 self.pktq.append(p) | |
| 281 self.buffer = [] | |
| 4 | 282 pktcount += 1 |
| 283 self.rx_cnt += 1 | |
| 1 | 284 else: |
| 285 print "Invalid state %s! Resetting" % (self.state) | |
| 286 self.state = 'init' | |
| 287 | |
| 288 return pktcount | |
| 289 | |
| 290 #for c in dir(): | |
| 291 # if (issubclass(c, PktBase)): | |
| 292 # print .. | |
|
0
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
293 |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
294 s = serial.Serial(port='/dev/cuad0', baudrate=9600, bytesize=8, parity='N', \ |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
295 stopbits=1, rtscts=0) |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
296 s.setTimeout(0.1) |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
297 #s.write('+++') |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
298 #s.readline(eol='\r') |
|
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
299 |
| 4 | 300 |
| 301 # 0x0001 (-36dBm) -> 1 samples, mask 0x000f, DIO - 0x00f | |
| 5 | 302 goodtest = [126, 0, 10, 131, 0, 1, 36, 0, 1, 0, 15, 0, 15, 56] |
| 4 | 303 |
| 304 # Checksum error | |
| 5 | 305 badtest = [126, 0, 10, 131, 0, 1, 36, 0, 1, 0, 15, 0, 14, 56] |
| 4 | 306 |
| 307 #0x0005 (-36dBm) -> 1 samples, mask 0x020e, DIO - 0x00e, ADC0 - 0x3ff | |
| 5 | 308 adctest = [126, 0, 12, 131, 0, 5, 36, 0, 1, 2, 14, 0, 14, 3, 255, 50] |
| 4 | 309 |
| 310 # Exception | |
| 5 | 311 badpkttypetest = [126, 0, 3, 10, 86, 76, 83] |
| 4 | 312 |
| 6 | 313 up = Packets(s) |
| 1 | 314 up.process(goodtest) |
|
3
7bf4d4265339
Pop off the packets we test with.
darius@inchoate.localdomain
parents:
1
diff
changeset
|
315 up.process(badtest) |
|
7bf4d4265339
Pop off the packets we test with.
darius@inchoate.localdomain
parents:
1
diff
changeset
|
316 up.process(adctest) |
|
7bf4d4265339
Pop off the packets we test with.
darius@inchoate.localdomain
parents:
1
diff
changeset
|
317 print up.pktq.pop(0) |
|
7bf4d4265339
Pop off the packets we test with.
darius@inchoate.localdomain
parents:
1
diff
changeset
|
318 print up.pktq.pop(0) |
|
0
bdcdd8380d94
Some test routines & framework for talking to MaxStream ZigBee modules.
darius@inchoate.localdomain
parents:
diff
changeset
|
319 |