2 ## This file is part of the libsigrokdecode project.
4 ## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
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.
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.
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/>.
20 import sigrokdecode as srd
22 class ChannelError(Exception):
26 # addr: ('name', size)
29 0x02: ('EN_RXADDR', 1),
30 0x03: ('SETUP_AW', 1),
31 0x04: ('SETUP_RETR', 1),
33 0x06: ('RF_SETUP', 1),
35 0x08: ('OBSERVE_TX', 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),
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),
56 0x19: ('DEMOD_CAL', 5),
61 class Decoder(srd.Decoder):
65 longname = 'Nordic Semiconductor nRF24L01/nRF24L01+'
66 desc = '2.4GHz transceiver chip.'
69 outputs = ['nrf24l01']
71 {'id': 'chip', 'desc': 'Chip type',
72 'default': 'nrf24l01', 'values': ('nrf24l01', 'xn297')},
75 # Sent from the host to the chip.
76 ('cmd', 'Commands sent to the device'),
77 ('tx-data', 'Payload sent to the device'),
79 # Returned by the chip.
80 ('register', 'Registers read from the device'),
81 ('rx-data', 'Payload read from the device'),
83 ('warning', 'Warnings'),
91 ('commands', 'Commands', (ann_cmd, ann_tx)),
92 ('responses', 'Responses', (ann_reg, ann_rx)),
93 ('warnings', 'Warnings', (ann_warn,)),
98 self.requirements_met = True
99 self.cs_was_released = False
102 self.out_ann = self.register(srd.OUTPUT_ANN)
103 if self.options['chip'] == 'xn297':
104 regs.update(xn297_regs)
106 def warn(self, pos, msg):
107 '''Put a warning message 'msg' at 'pos'.'''
108 self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])
110 def putp(self, pos, ann, msg):
111 '''Put an annotation message 'msg' at 'pos'.'''
112 self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])
115 '''Resets the decoder after a complete command was decoded.'''
116 # 'True' for the first byte after CS went low.
119 # The current command, and the minimum and maximum number
120 # of data bytes to follow.
125 # Used to collect the bytes after the command byte
126 # (and the start/end sample number).
131 def mosi_bytes(self):
132 '''Returns the collected MOSI bytes of a multi byte command.'''
133 return [b[0] for b in self.mb]
135 def miso_bytes(self):
136 '''Returns the collected MISO bytes of a multi byte command.'''
137 return [b[1] for b in self.mb]
139 def decode_command(self, pos, b):
140 '''Decodes the command byte 'b' at position 'pos' and prepares
141 the decoding of the following data bytes.'''
142 c = self.parse_command(b)
144 self.warn(pos, 'unknown command')
147 self.cmd, self.dat, self.min, self.max = c
149 if self.cmd in ('W_REGISTER', 'ACTIVATE'):
150 # Don't output anything now, the command is merged with
151 # the data bytes following it.
154 self.putp(pos, self.ann_cmd, self.format_command())
156 def format_command(self):
157 '''Returns the label for the current command.'''
158 if self.cmd == 'R_REGISTER':
159 reg = regs[self.dat][0] if self.dat in regs else 'unknown register'
160 return 'Cmd R_REGISTER "{}"'.format(reg)
162 return 'Cmd {}'.format(self.cmd)
164 def parse_command(self, b):
165 '''Parses the command byte.
167 Returns a tuple consisting of:
168 - the name of the command
169 - additional data needed to dissect the following bytes
170 - minimum number of following bytes
171 - maximum number of following bytes
174 if (b & 0xe0) in (0b00000000, 0b00100000):
175 c = 'R_REGISTER' if not (b & 0xe0) else 'W_REGISTER'
177 m = regs[d][1] if d in regs else 1
181 return ('ACTIVATE', None, 1, 1)
183 return ('R_RX_PAYLOAD', None, 1, 32)
185 return ('R_RX_PL_WID', None, 1, 1)
187 return ('W_TX_PAYLOAD', None, 1, 32)
189 return ('W_TX_PAYLOAD_NOACK', None, 1, 32)
190 if (b & 0xf8) == 0b10101000:
191 return ('W_ACK_PAYLOAD', b & 0x07, 1, 32)
193 return ('FLUSH_TX', None, 0, 0)
195 return ('FLUSH_RX', None, 0, 0)
197 return ('REUSE_TX_PL', None, 0, 0)
199 return ('NOP', None, 0, 0)
201 def decode_register(self, pos, ann, regid, data):
202 '''Decodes a register.
204 pos -- start and end sample numbers of the register
205 ann -- is the annotation number that is used to output the register.
206 regid -- may be either an integer used as a key for the 'regs'
207 dictionary, or a string directly containing a register name.'
208 data -- is the register content.
211 if type(regid) == int:
212 # Get the name of the register.
213 if regid not in regs:
214 self.warn(pos, 'unknown register')
216 name = regs[regid][0]
220 # Multi byte register come LSByte first.
221 data = reversed(data)
223 if self.cmd == 'W_REGISTER' and ann == self.ann_cmd:
224 # The 'W_REGISTER' command is merged with the following byte(s).
225 label = '{}: {}'.format(self.format_command(), name)
227 label = 'Reg {}'.format(name)
229 self.decode_mb_data(pos, ann, data, label, True)
231 def decode_mb_data(self, pos, ann, data, label, always_hex):
232 '''Decodes the data bytes 'data' of a multibyte command at position
233 'pos'. The decoded data is prefixed with 'label'. If 'always_hex' is
234 True, all bytes are decoded as hex codes, otherwise only non
235 printable characters are escaped.'''
239 return '{:02X}'.format(b)
243 if not str.isprintable(c):
244 return '\\x{:02X}'.format(b)
247 data = ''.join([escape(b) for b in data])
248 text = '{} = "{}"'.format(label, data)
249 self.putp(pos, ann, text)
251 def finish_command(self, pos):
252 '''Decodes the remaining data bytes at position 'pos'.'''
254 if self.cmd == 'R_REGISTER':
255 self.decode_register(pos, self.ann_reg,
256 self.dat, self.miso_bytes())
257 elif self.cmd == 'W_REGISTER':
258 self.decode_register(pos, self.ann_cmd,
259 self.dat, self.mosi_bytes())
260 elif self.cmd == 'R_RX_PAYLOAD':
261 self.decode_mb_data(pos, self.ann_rx,
262 self.miso_bytes(), 'RX payload', False)
263 elif (self.cmd == 'W_TX_PAYLOAD' or
264 self.cmd == 'W_TX_PAYLOAD_NOACK'):
265 self.decode_mb_data(pos, self.ann_tx,
266 self.mosi_bytes(), 'TX payload', False)
267 elif self.cmd == 'W_ACK_PAYLOAD':
268 lbl = 'ACK payload for pipe {}'.format(self.dat)
269 self.decode_mb_data(pos, self.ann_tx,
270 self.mosi_bytes(), lbl, False)
271 elif self.cmd == 'R_RX_PL_WID':
272 msg = 'Payload width = {}'.format(self.mb[0][1])
273 self.putp(pos, self.ann_reg, msg)
274 elif self.cmd == 'ACTIVATE':
275 self.putp(pos, self.ann_cmd, self.format_command())
276 if self.mosi_bytes()[0] != 0x73:
277 self.warn(pos, 'wrong data for "ACTIVATE" command')
279 def decode(self, ss, es, data):
280 if not self.requirements_met:
283 ptype, data1, data2 = data
285 if ptype == 'CS-CHANGE':
288 self.requirements_met = False
289 raise ChannelError('CS# pin required.')
291 self.cs_was_released = True
293 if data1 == 0 and data2 == 1:
294 # Rising edge, the complete command is transmitted, process
295 # the bytes that were send after the command byte.
297 # Check if we got the minimum number of data bytes
298 # after the command byte.
299 if len(self.mb) < self.min:
300 self.warn((ss, ss), 'missing data bytes')
302 self.finish_command((self.mb_s, self.mb_e))
305 self.cs_was_released = True
306 elif ptype == 'DATA' and self.cs_was_released:
307 mosi, miso = data1, data2
310 if miso is None or mosi is None:
311 self.requirements_met = False
312 raise ChannelError('Both MISO and MOSI pins required.')
316 # First MOSI byte is always the command.
317 self.decode_command(pos, mosi)
318 # First MISO byte is always the status register.
319 self.decode_register(pos, self.ann_reg, 'STATUS', [miso])
321 if not self.cmd or len(self.mb) >= self.max:
322 self.warn(pos, 'excess byte')
324 # Collect the bytes after the command byte.
328 self.mb.append((mosi, miso))