Mercurial > ~darius > hgwebdir.cgi > ZigBee

comparison zb.py @ 1:b5d8ec0ffd21

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Lots of changes.. - Create a class for each packet type, needs more fleshing out. - Make sure we reset state when we get a checksum error. - Add some more test cases. - OOify things a bit more.
author darius@inchoate.localdomain
date Tue, 30 Oct 2007 17:26:40 +1030
parents bdcdd8380d94
children 7bf4d4265339
comparison
equal deleted inserted replaced
0:bdcdd8380d94 1:b5d8ec0ffd21
1 import serial, struct 1 import serial
2
3 class PktBase(object):
4 def __init__(self, data):
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
11 #def __str__(self):
12 # return "%s: %s" % (self.PKT_DESC, str(self.data))
13
14 class AT_Cmd(PktBase):
15 PKT_TYPE = 0x08
16 PKT_DESC = "AT Command"
17
18 class AT_Cmd_Queue(PktBase):
19 PKT_TYPE = 0x09
20 PKT_DESC = "AT Command (queued)"
21
22 class AT_Response(PktBase):
23 PKT_TYPE = 0x88
24 PKT_DESC = "AT Command response"
25
26 class Modem_Status(PktBase):
27 PKT_TYPE = 0x8a
28 PKT_DESC = "Modem Status"
29
30 class RX_64_Bit(PktBase):
31 PKT_TYPE = 0x80
32 PKT_DESC = "RX Packet: 64 bit address"
33
34 class RX_16_Bit(PktBase):
35 PKT_TYPE = 0x81
36 PKT_DESC = "RX Packet: 16 bit address"
37
38 class RXIO_64_Bit(PktBase):
39 PKT_TYPE = 0x82
40 PKT_DESC = "RXIO Packet: 64 bit address"
41
42 class RXIO_16_Bit(PktBase):
43 PKT_TYPE = 0x83
44 PKT_DESC = "RXIO Packet: 16 bit address"
45
46 def __str__(self):
47 sender = self.data[0] << 8 | self.data[1]
48 rssi = -1 * self.data[2]
49 flags = self.data[3]
50 nsamples = self.data[4]
51 mask = self.data[5] << 8 | self.data[6]
52 rtn = "0x%04x (%ddBm) -> %d samples, mask 0x%04x" % (sender, rssi, nsamples, mask)
53 # Any DIO lines enabled?
54 if (mask | 0x01ff):
55 rtn = rtn + ", DIO - 0x%03x" % (self.data[7] << 8 | self.data[8])
56 offs = 9
57 else:
58 offs = 7
59
60 # Any ADC lines enabled?
61 if (mask | 0x7e00):
62 for i in range(6):
63 if (mask & 1 << (i + 9)):
64 rtn = rtn + ", ADC%d - 0x%02x" % (i, self.data[offs] << 8 |
65 self.data[offs + 1])
66 offs = offs + 2
67
68 return rtn
69
70 class TX_64_Bit(PktBase):
71 PKT_TYPE = 0x00
72 PKT_DESC = "TX Packet: 64 bit address"
73
74 class TX_16_Bit(PktBase):
75 PKT_TYPE = 0x00
76 PKT_DESC = "TX Packet: 16 bit address"
77
78 class TX_Status(PktBase):
79 PKT_TYPE = 0x89
80 PKT_DESC = "TX Status"
81
82 class Packets(object):
83 PKT_CLASSES = [AT_Cmd, AT_Cmd_Queue, AT_Response, Modem_Status, RX_64_Bit,
84 RX_16_Bit, RXIO_64_Bit, RXIO_16_Bit, TX_64_Bit, TX_16_Bit,
85 TX_Status]
86
87 def Build(data):
88 for p in Packets.PKT_CLASSES:
89 if (p.PKT_TYPE == data[0]):
90 return(p(data[1:]))
91
92 Build = staticmethod(Build)
93
94 def Encapsulate(data):
95 pktsum = reduce(lambda x, y: x + y, data) & 0xff
96 pkt = [0x7e] + [len(data) >> 8] + [len(data) & 0xff] + data + [0xff - pktsum]
97 return(map(chr, pkt))
98
99 Encapsulate = staticmethod(Encapsulate)
100
101 def __init__(self):
102 self.buffer = []
103 self.state = 'init'
104 self.packets = []
105
106 self.bufmsb = 0
107 self.dataleft = 0
108
109 self.fr_err = 0 # Framing error
110 self.ck_err = 0 # Checksum error
111 self.rx_cnt = 0 # Packet count
112
113 self.pktq = []
114
115 def getdata(self, s):
116 l = []
117 while (1):
118 a = s.read()
119 if (a == ''):
120 break
121 l.append(a)
122
123 return self.process(l)
124
125 def process(self, data):
126 pktcount = 0
127 for d in data:
128 dord = ord(d)
129 if (self.state == 'init'):
130 if (d != '\x7e'):
131 print "Framing error, got 0x%02x, expected 0x7f" % (dord)
132 self.fr_err = self.fr_err + 1
133 continue
134
135 self.state = 'sizemsb'
136 elif (self.state == 'sizemsb'):
137 self.bufmsb = dord
138 self.state = 'sizelsb'
139 elif (self.state == 'sizelsb'):
140 self.dataleft = self.bufmsb << 8 | \
141 dord
142 self.state = 'data'
143 elif (self.state == 'data'):
144 self.buffer.append(dord)
145 self.dataleft = self.dataleft - 1
146 if (self.dataleft == 0):
147 self.state = 'cksum'
148 elif (self.state == 'cksum'):
149 pktsum = reduce(lambda x, y: x + y, self.buffer) & 0xff
150 rxcksum = dord
151 self.state = 'init'
152 if (pktsum + rxcksum != 0xff):
153 self.buffer = []
154 self.ck_err = self.ck_err + 1
155 print "Checksum error, got 0x%02x, expected 0x%02x" % \
156 (rxcksum, 0xff - pktsum)
157 else:
158 print "Got a packet - " + str(self.buffer)
159 p = Packets.Build(self.buffer)
160 self.pktq.append(p)
161 self.buffer = []
162 pktcount = pktcount + 1
163 self.rx_cnt = self.rx_cnt + 1
164 else:
165 print "Invalid state %s! Resetting" % (self.state)
166 self.state = 'init'
167
168 return pktcount
169
170 #for c in dir():
171 # if (issubclass(c, PktBase)):
172 # print ..
2 173
3 s = serial.Serial(port='/dev/cuad0', baudrate=9600, bytesize=8, parity='N', \ 174 s = serial.Serial(port='/dev/cuad0', baudrate=9600, bytesize=8, parity='N', \
4 stopbits=1, rtscts=0) 175 stopbits=1, rtscts=0)
5 s.setTimeout(0.1) 176 s.setTimeout(0.1)
6 #s.write('+++') 177 #s.write('+++')
7 #s.readline(eol='\r') 178 #s.readline(eol='\r')
8 179
9 testdata = ['~', '\x00', '\n', '\x83', '\x00', '\x01', '$', '\x00', '\x01', '\x00', '\x0f', '\x00', '\x07', '@', '~', '\x00', '\n', '\x83', '\x00', '\x01', '$', '\x00', '\x01', '\x00', '\x0f', '\x00', '\x0f', '8'] 180 goodtest = ['~', '\x00', '\n', '\x83', '\x00', '\x01', '$', '\x00', '\x01', '\x00', '\x0f', '\x00', '\x0f', '8']
181 badtest = ['~', '\x00', '\n', '\x83', '\x00', '\x01', '$', '\x00', '\x01', '\x00', '\x0f', '\x00', '\x0e', '8']
182 adctest = ['~', '\x00', '\x0c', '\x83', '\x00', '\x05', '$', '\x00', '\x01', '\x02', '\x0e', '\x00', '\x0e', '\x03', '\xff', '2' ]
183 up = Packets()
184 up.process(goodtest)
185 #up.process(badtest)
186 p = up.pktq.pop(0)
10 187
11 def readpkt(s):
12 l = []
13 while (1):
14 a = s.read()
15 if (a == ''):
16 break
17 l.append(a)
18
19 return l
20
21 class Packet:
22 def __init__(self):
23 pass
24
25 class unPacker:
26 def __init__(self):
27 self.buffer = []
28 self.state = 'init'
29 self.packets = []
30
31 self.bufmsb = 0
32 self.dataleft = 0
33
34 self.fr_err = 0 # Framing error
35 self.ck_err = 0 # Checksum error
36
37 def process(self, data):
38 for d in data:
39 dint = struct.unpack('B', d)[0]
40 if (self.state == 'init'):
41 if (d != '\x7e'):
42 print "Framing error, got 0x%02x, expected 0x7f" % (dint)
43 self.fr_err = self.fr_err + 1
44 continue
45
46 self.state = 'sizemsb'
47 elif (self.state == 'sizemsb'):
48 self.bufmsb = dint
49 self.state = 'sizelsb'
50 elif (self.state == 'sizelsb'):
51 self.dataleft = self.bufmsb << 8 | \
52 dint
53 self.state = 'data'
54 elif (self.state == 'data'):
55 self.buffer.append(dint)
56 self.dataleft = self.dataleft - 1
57 if (self.dataleft == 0):
58 self.state = 'cksum'
59 elif (self.state == 'cksum'):
60 pktsum = reduce(lambda x, y: x + y, self.buffer) & 0xff
61 rxcksum = dint
62 if (pktsum + rxcksum != 0xff):
63 self.ck_err = self.ck_err + 1
64 print "Checksum error, got 0x%02x, expected 0x%02x" % \
65 (rxcksum, 0xff - pktsum)
66 else:
67 print "Got a packet - " + str(self.buffer)
68 self.state = 'init'
69 self.buffer = []
70 else:
71 print "Invalid state %s! Resetting" % (self.state)
72 self.state = 'init'