4dd9451e345693c843a93d231c2ab1d57821df46
[libsigrokdecode.git] / decoders / nrf24l01 / pd.py
1 ##
2 ## This file is part of the libsigrokdecode project.
3 ##
4 ## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
5 ##
6 ## This program is free software; you can redistribute it and/or modify
7 ## it under the terms of the GNU General Public License as published by
8 ## the Free Software Foundation; either version 2 of the License, or
9 ## (at your option) any later version.
10 ##
11 ## This program is distributed in the hope that it will be useful,
12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 ## GNU General Public License for more details.
15 ##
16 ## You should have received a copy of the GNU General Public License
17 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
18 ##
19
20 import sigrokdecode as srd
21
22 class ChannelError(Exception):
23     pass
24
25 regs = {
26 #   addr: ('name',        size)
27     0x00: ('CONFIG',      1),
28     0x01: ('EN_AA',       1),
29     0x02: ('EN_RXADDR',   1),
30     0x03: ('SETUP_AW',    1),
31     0x04: ('SETUP_RETR',  1),
32     0x05: ('RF_CH',       1),
33     0x06: ('RF_SETUP',    1),
34     0x07: ('STATUS',      1),
35     0x08: ('OBSERVE_TX',  1),
36     0x09: ('RPD',         1),
37     0x0a: ('RX_ADDR_P0',  5),
38     0x0b: ('RX_ADDR_P1',  5),
39     0x0c: ('RX_ADDR_P2',  1),
40     0x0d: ('RX_ADDR_P3',  1),
41     0x0e: ('RX_ADDR_P4',  1),
42     0x0f: ('RX_ADDR_P5',  1),
43     0x10: ('TX_ADDR',     5),
44     0x11: ('RX_PW_P0',    1),
45     0x12: ('RX_PW_P1',    1),
46     0x13: ('RX_PW_P2',    1),
47     0x14: ('RX_PW_P3',    1),
48     0x15: ('RX_PW_P4',    1),
49     0x16: ('RX_PW_P5',    1),
50     0x17: ('FIFO_STATUS', 1),
51     0x1c: ('DYNPD',       1),
52     0x1d: ('FEATURE',     1),
53 }
54
55 xn297_regs = {
56     0x19: ('DEMOD_CAL',   5),
57     0x1e: ('RF_CAL',      7),
58     0x1f: ('BB_CAL',      5),
59 }
60
61 class Decoder(srd.Decoder):
62     api_version = 3
63     id = 'nrf24l01'
64     name = 'nRF24L01(+)'
65     longname = 'Nordic Semiconductor nRF24L01(+)'
66     desc = '2.4GHz RF transceiver chip.'
67     license = 'gplv2+'
68     inputs = ['spi']
69     outputs = ['nrf24l01']
70     tags = ['IC', 'Wireless/RF']
71     options = (
72         {'id': 'chip', 'desc': 'Chip type',
73             'default': 'nrf24l01', 'values': ('nrf24l01', 'xn297')},
74     )
75     annotations = (
76         # Sent from the host to the chip.
77         ('cmd', 'Commands sent to the device'),
78         ('tx-data', 'Payload sent to the device'),
79
80         # Returned by the chip.
81         ('register', 'Registers read from the device'),
82         ('rx-data', 'Payload read from the device'),
83
84         ('warning', 'Warnings'),
85     )
86     ann_cmd = 0
87     ann_tx = 1
88     ann_reg = 2
89     ann_rx = 3
90     ann_warn = 4
91     annotation_rows = (
92         ('commands', 'Commands', (ann_cmd, ann_tx)),
93         ('responses', 'Responses', (ann_reg, ann_rx)),
94         ('warnings', 'Warnings', (ann_warn,)),
95     )
96
97     def __init__(self):
98         self.reset()
99
100     def reset(self):
101         self.next()
102         self.requirements_met = True
103         self.cs_was_released = False
104
105     def start(self):
106         self.out_ann = self.register(srd.OUTPUT_ANN)
107         if self.options['chip'] == 'xn297':
108             regs.update(xn297_regs)
109
110     def warn(self, pos, msg):
111         '''Put a warning message 'msg' at 'pos'.'''
112         self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])
113
114     def putp(self, pos, ann, msg):
115         '''Put an annotation message 'msg' at 'pos'.'''
116         self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])
117
118     def next(self):
119         '''Resets the decoder after a complete command was decoded.'''
120         # 'True' for the first byte after CS went low.
121         self.first = True
122
123         # The current command, and the minimum and maximum number
124         # of data bytes to follow.
125         self.cmd = None
126         self.min = 0
127         self.max = 0
128
129         # Used to collect the bytes after the command byte
130         # (and the start/end sample number).
131         self.mb = []
132         self.mb_s = -1
133         self.mb_e = -1
134
135     def mosi_bytes(self):
136         '''Returns the collected MOSI bytes of a multi byte command.'''
137         return [b[0] for b in self.mb]
138
139     def miso_bytes(self):
140         '''Returns the collected MISO bytes of a multi byte command.'''
141         return [b[1] for b in self.mb]
142
143     def decode_command(self, pos, b):
144         '''Decodes the command byte 'b' at position 'pos' and prepares
145         the decoding of the following data bytes.'''
146         c = self.parse_command(b)
147         if c is None:
148             self.warn(pos, 'unknown command')
149             return
150
151         self.cmd, self.dat, self.min, self.max = c
152
153         if self.cmd in ('W_REGISTER', 'ACTIVATE'):
154             # Don't output anything now, the command is merged with
155             # the data bytes following it.
156             self.mb_s = pos[0]
157         else:
158             self.putp(pos, self.ann_cmd, self.format_command())
159
160     def format_command(self):
161         '''Returns the label for the current command.'''
162         if self.cmd == 'R_REGISTER':
163             reg = regs[self.dat][0] if self.dat in regs else 'unknown register'
164             return 'Cmd R_REGISTER "{}"'.format(reg)
165         else:
166             return 'Cmd {}'.format(self.cmd)
167
168     def parse_command(self, b):
169         '''Parses the command byte.
170
171         Returns a tuple consisting of:
172         - the name of the command
173         - additional data needed to dissect the following bytes
174         - minimum number of following bytes
175         - maximum number of following bytes
176         '''
177
178         if (b & 0xe0) in (0b00000000, 0b00100000):
179             c = 'R_REGISTER' if not (b & 0xe0) else 'W_REGISTER'
180             d = b & 0x1f
181             m = regs[d][1] if d in regs else 1
182             return (c, d, 1, m)
183         if b == 0b01010000:
184             # nRF24L01 only
185             return ('ACTIVATE', None, 1, 1)
186         if b == 0b01100001:
187             return ('R_RX_PAYLOAD', None, 1, 32)
188         if b == 0b01100000:
189             return ('R_RX_PL_WID', None, 1, 1)
190         if b == 0b10100000:
191             return ('W_TX_PAYLOAD', None, 1, 32)
192         if b == 0b10110000:
193             return ('W_TX_PAYLOAD_NOACK', None, 1, 32)
194         if (b & 0xf8) == 0b10101000:
195             return ('W_ACK_PAYLOAD', b & 0x07, 1, 32)
196         if b == 0b11100001:
197             return ('FLUSH_TX', None, 0, 0)
198         if b == 0b11100010:
199             return ('FLUSH_RX', None, 0, 0)
200         if b == 0b11100011:
201             return ('REUSE_TX_PL', None, 0, 0)
202         if b == 0b11111111:
203             return ('NOP', None, 0, 0)
204
205     def decode_register(self, pos, ann, regid, data):
206         '''Decodes a register.
207
208         pos   -- start and end sample numbers of the register
209         ann   -- is the annotation number that is used to output the register.
210         regid -- may be either an integer used as a key for the 'regs'
211                  dictionary, or a string directly containing a register name.'
212         data  -- is the register content.
213         '''
214
215         if type(regid) == int:
216             # Get the name of the register.
217             if regid not in regs:
218                 self.warn(pos, 'unknown register')
219                 return
220             name = regs[regid][0]
221         else:
222             name = regid
223
224         # Multi byte register come LSByte first.
225         data = reversed(data)
226
227         if self.cmd == 'W_REGISTER' and ann == self.ann_cmd:
228             # The 'W_REGISTER' command is merged with the following byte(s).
229             label = '{}: {}'.format(self.format_command(), name)
230         else:
231             label = 'Reg {}'.format(name)
232
233         self.decode_mb_data(pos, ann, data, label, True)
234
235     def decode_mb_data(self, pos, ann, data, label, always_hex):
236         '''Decodes the data bytes 'data' of a multibyte command at position
237         'pos'. The decoded data is prefixed with 'label'. If 'always_hex' is
238         True, all bytes are decoded as hex codes, otherwise only non
239         printable characters are escaped.'''
240
241         if always_hex:
242             def escape(b):
243                 return '{:02X}'.format(b)
244         else:
245             def escape(b):
246                 c = chr(b)
247                 if not str.isprintable(c):
248                     return '\\x{:02X}'.format(b)
249                 return c
250
251         data = ''.join([escape(b) for b in data])
252         text = '{} = "{}"'.format(label, data)
253         self.putp(pos, ann, text)
254
255     def finish_command(self, pos):
256         '''Decodes the remaining data bytes at position 'pos'.'''
257
258         if self.cmd == 'R_REGISTER':
259             self.decode_register(pos, self.ann_reg,
260                                  self.dat, self.miso_bytes())
261         elif self.cmd == 'W_REGISTER':
262             self.decode_register(pos, self.ann_cmd,
263                                  self.dat, self.mosi_bytes())
264         elif self.cmd == 'R_RX_PAYLOAD':
265             self.decode_mb_data(pos, self.ann_rx,
266                                 self.miso_bytes(), 'RX payload', False)
267         elif (self.cmd == 'W_TX_PAYLOAD' or
268               self.cmd == 'W_TX_PAYLOAD_NOACK'):
269             self.decode_mb_data(pos, self.ann_tx,
270                                 self.mosi_bytes(), 'TX payload', False)
271         elif self.cmd == 'W_ACK_PAYLOAD':
272             lbl = 'ACK payload for pipe {}'.format(self.dat)
273             self.decode_mb_data(pos, self.ann_tx,
274                                 self.mosi_bytes(), lbl, False)
275         elif self.cmd == 'R_RX_PL_WID':
276             msg = 'Payload width = {}'.format(self.mb[0][1])
277             self.putp(pos, self.ann_reg, msg)
278         elif self.cmd == 'ACTIVATE':
279             self.putp(pos, self.ann_cmd, self.format_command())
280             if self.mosi_bytes()[0] != 0x73:
281                 self.warn(pos, 'wrong data for "ACTIVATE" command')
282
283     def decode(self, ss, es, data):
284         if not self.requirements_met:
285             return
286
287         ptype, data1, data2 = data
288
289         if ptype == 'CS-CHANGE':
290             if data1 is None:
291                 if data2 is None:
292                     self.requirements_met = False
293                     raise ChannelError('CS# pin required.')
294                 elif data2 == 1:
295                     self.cs_was_released = True
296
297             if data1 == 0 and data2 == 1:
298                 # Rising edge, the complete command is transmitted, process
299                 # the bytes that were send after the command byte.
300                 if self.cmd:
301                     # Check if we got the minimum number of data bytes
302                     # after the command byte.
303                     if len(self.mb) < self.min:
304                         self.warn((ss, ss), 'missing data bytes')
305                     elif self.mb:
306                         self.finish_command((self.mb_s, self.mb_e))
307
308                 self.next()
309                 self.cs_was_released = True
310         elif ptype == 'DATA' and self.cs_was_released:
311             mosi, miso = data1, data2
312             pos = (ss, es)
313
314             if miso is None or mosi is None:
315                 self.requirements_met = False
316                 raise ChannelError('Both MISO and MOSI pins required.')
317
318             if self.first:
319                 self.first = False
320                 # First MOSI byte is always the command.
321                 self.decode_command(pos, mosi)
322                 # First MISO byte is always the status register.
323                 self.decode_register(pos, self.ann_reg, 'STATUS', [miso])
324             else:
325                 if not self.cmd or len(self.mb) >= self.max:
326                     self.warn(pos, 'excess byte')
327                 else:
328                     # Collect the bytes after the command byte.
329                     if self.mb_s == -1:
330                         self.mb_s = ss
331                     self.mb_e = es
332                     self.mb.append((mosi, miso))