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/nRF24L01+'
  66     desc = '2.4GHz transceiver chip.'
  67     license = 'gplv2+'
  68     inputs = ['spi']
  69     outputs = ['nrf24l01']
  70     options = (
  71         {'id': 'chip', 'desc': 'Chip type',
  72             'default': 'nrf24l01', 'values': ('nrf24l01', 'xn297')},
  73     )
  74     annotations = (
  75         # Sent from the host to the chip.
  76         ('cmd', 'Commands sent to the device'),
  77         ('tx-data', 'Payload sent to the device'),
  78
  79         # Returned by the chip.
  80         ('register', 'Registers read from the device'),
  81         ('rx-data', 'Payload read from the device'),
  82
  83         ('warning', 'Warnings'),
  84     )
  85     ann_cmd = 0
  86     ann_tx = 1
  87     ann_reg = 2
  88     ann_rx = 3
  89     ann_warn = 4
  90     annotation_rows = (
  91         ('commands', 'Commands', (ann_cmd, ann_tx)),
  92         ('responses', 'Responses', (ann_reg, ann_rx)),
  93         ('warnings', 'Warnings', (ann_warn,)),
  94     )
  95
  96     def __init__(self):
  97         self.reset()
  98
  99     def reset(self):
 100         self.next()
 101         self.requirements_met = True
 102         self.cs_was_released = False
 103
 104     def start(self):
 105         self.out_ann = self.register(srd.OUTPUT_ANN)
 106         if self.options['chip'] == 'xn297':
 107             regs.update(xn297_regs)
 108
 109     def warn(self, pos, msg):
 110         '''Put a warning message 'msg' at 'pos'.'''
 111         self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])
 112
 113     def putp(self, pos, ann, msg):
 114         '''Put an annotation message 'msg' at 'pos'.'''
 115         self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])
 116
 117     def next(self):
 118         '''Resets the decoder after a complete command was decoded.'''
 119         # 'True' for the first byte after CS went low.
 120         self.first = True
 121
 122         # The current command, and the minimum and maximum number
 123         # of data bytes to follow.
 124         self.cmd = None
 125         self.min = 0
 126         self.max = 0
 127
 128         # Used to collect the bytes after the command byte
 129         # (and the start/end sample number).
 130         self.mb = []
 131         self.mb_s = -1
 132         self.mb_e = -1
 133
 134     def mosi_bytes(self):
 135         '''Returns the collected MOSI bytes of a multi byte command.'''
 136         return [b[0] for b in self.mb]
 137
 138     def miso_bytes(self):
 139         '''Returns the collected MISO bytes of a multi byte command.'''
 140         return [b[1] for b in self.mb]
 141
 142     def decode_command(self, pos, b):
 143         '''Decodes the command byte 'b' at position 'pos' and prepares
 144         the decoding of the following data bytes.'''
 145         c = self.parse_command(b)
 146         if c is None:
 147             self.warn(pos, 'unknown command')
 148             return
 149
 150         self.cmd, self.dat, self.min, self.max = c
 151
 152         if self.cmd in ('W_REGISTER', 'ACTIVATE'):
 153             # Don't output anything now, the command is merged with
 154             # the data bytes following it.
 155             self.mb_s = pos[0]
 156         else:
 157             self.putp(pos, self.ann_cmd, self.format_command())
 158
 159     def format_command(self):
 160         '''Returns the label for the current command.'''
 161         if self.cmd == 'R_REGISTER':
 162             reg = regs[self.dat][0] if self.dat in regs else 'unknown register'
 163             return 'Cmd R_REGISTER "{}"'.format(reg)
 164         else:
 165             return 'Cmd {}'.format(self.cmd)
 166
 167     def parse_command(self, b):
 168         '''Parses the command byte.
 169
 170         Returns a tuple consisting of:
 171         - the name of the command
 172         - additional data needed to dissect the following bytes
 173         - minimum number of following bytes
 174         - maximum number of following bytes
 175         '''
 176
 177         if (b & 0xe0) in (0b00000000, 0b00100000):
 178             c = 'R_REGISTER' if not (b & 0xe0) else 'W_REGISTER'
 179             d = b & 0x1f
 180             m = regs[d][1] if d in regs else 1
 181             return (c, d, 1, m)
 182         if b == 0b01010000:
 183             # nRF24L01 only
 184             return ('ACTIVATE', None, 1, 1)
 185         if b == 0b01100001:
 186             return ('R_RX_PAYLOAD', None, 1, 32)
 187         if b == 0b01100000:
 188             return ('R_RX_PL_WID', None, 1, 1)
 189         if b == 0b10100000:
 190             return ('W_TX_PAYLOAD', None, 1, 32)
 191         if b == 0b10110000:
 192             return ('W_TX_PAYLOAD_NOACK', None, 1, 32)
 193         if (b & 0xf8) == 0b10101000:
 194             return ('W_ACK_PAYLOAD', b & 0x07, 1, 32)
 195         if b == 0b11100001:
 196             return ('FLUSH_TX', None, 0, 0)
 197         if b == 0b11100010:
 198             return ('FLUSH_RX', None, 0, 0)
 199         if b == 0b11100011:
 200             return ('REUSE_TX_PL', None, 0, 0)
 201         if b == 0b11111111:
 202             return ('NOP', None, 0, 0)
 203
 204     def decode_register(self, pos, ann, regid, data):
 205         '''Decodes a register.
 206
 207         pos   -- start and end sample numbers of the register
 208         ann   -- is the annotation number that is used to output the register.
 209         regid -- may be either an integer used as a key for the 'regs'
 210                  dictionary, or a string directly containing a register name.'
 211         data  -- is the register content.
 212         '''
 213
 214         if type(regid) == int:
 215             # Get the name of the register.
 216             if regid not in regs:
 217                 self.warn(pos, 'unknown register')
 218                 return
 219             name = regs[regid][0]
 220         else:
 221             name = regid
 222
 223         # Multi byte register come LSByte first.
 224         data = reversed(data)
 225
 226         if self.cmd == 'W_REGISTER' and ann == self.ann_cmd:
 227             # The 'W_REGISTER' command is merged with the following byte(s).
 228             label = '{}: {}'.format(self.format_command(), name)
 229         else:
 230             label = 'Reg {}'.format(name)
 231
 232         self.decode_mb_data(pos, ann, data, label, True)
 233
 234     def decode_mb_data(self, pos, ann, data, label, always_hex):
 235         '''Decodes the data bytes 'data' of a multibyte command at position
 236         'pos'. The decoded data is prefixed with 'label'. If 'always_hex' is
 237         True, all bytes are decoded as hex codes, otherwise only non
 238         printable characters are escaped.'''
 239
 240         if always_hex:
 241             def escape(b):
 242                 return '{:02X}'.format(b)
 243         else:
 244             def escape(b):
 245                 c = chr(b)
 246                 if not str.isprintable(c):
 247                     return '\\x{:02X}'.format(b)
 248                 return c
 249
 250         data = ''.join([escape(b) for b in data])
 251         text = '{} = "{}"'.format(label, data)
 252         self.putp(pos, ann, text)
 253
 254     def finish_command(self, pos):
 255         '''Decodes the remaining data bytes at position 'pos'.'''
 256
 257         if self.cmd == 'R_REGISTER':
 258             self.decode_register(pos, self.ann_reg,
 259                                  self.dat, self.miso_bytes())
 260         elif self.cmd == 'W_REGISTER':
 261             self.decode_register(pos, self.ann_cmd,
 262                                  self.dat, self.mosi_bytes())
 263         elif self.cmd == 'R_RX_PAYLOAD':
 264             self.decode_mb_data(pos, self.ann_rx,
 265                                 self.miso_bytes(), 'RX payload', False)
 266         elif (self.cmd == 'W_TX_PAYLOAD' or
 267               self.cmd == 'W_TX_PAYLOAD_NOACK'):
 268             self.decode_mb_data(pos, self.ann_tx,
 269                                 self.mosi_bytes(), 'TX payload', False)
 270         elif self.cmd == 'W_ACK_PAYLOAD':
 271             lbl = 'ACK payload for pipe {}'.format(self.dat)
 272             self.decode_mb_data(pos, self.ann_tx,
 273                                 self.mosi_bytes(), lbl, False)
 274         elif self.cmd == 'R_RX_PL_WID':
 275             msg = 'Payload width = {}'.format(self.mb[0][1])
 276             self.putp(pos, self.ann_reg, msg)
 277         elif self.cmd == 'ACTIVATE':
 278             self.putp(pos, self.ann_cmd, self.format_command())
 279             if self.mosi_bytes()[0] != 0x73:
 280                 self.warn(pos, 'wrong data for "ACTIVATE" command')
 281
 282     def decode(self, ss, es, data):
 283         if not self.requirements_met:
 284             return
 285
 286         ptype, data1, data2 = data
 287
 288         if ptype == 'CS-CHANGE':
 289             if data1 is None:
 290                 if data2 is None:
 291                     self.requirements_met = False
 292                     raise ChannelError('CS# pin required.')
 293                 elif data2 == 1:
 294                     self.cs_was_released = True
 295
 296             if data1 == 0 and data2 == 1:
 297                 # Rising edge, the complete command is transmitted, process
 298                 # the bytes that were send after the command byte.
 299                 if self.cmd:
 300                     # Check if we got the minimum number of data bytes
 301                     # after the command byte.
 302                     if len(self.mb) < self.min:
 303                         self.warn((ss, ss), 'missing data bytes')
 304                     elif self.mb:
 305                         self.finish_command((self.mb_s, self.mb_e))
 306
 307                 self.next()
 308                 self.cs_was_released = True
 309         elif ptype == 'DATA' and self.cs_was_released:
 310             mosi, miso = data1, data2
 311             pos = (ss, es)
 312
 313             if miso is None or mosi is None:
 314                 self.requirements_met = False
 315                 raise ChannelError('Both MISO and MOSI pins required.')
 316
 317             if self.first:
 318                 self.first = False
 319                 # First MOSI byte is always the command.
 320                 self.decode_command(pos, mosi)
 321                 # First MISO byte is always the status register.
 322                 self.decode_register(pos, self.ann_reg, 'STATUS', [miso])
 323             else:
 324                 if not self.cmd or len(self.mb) >= self.max:
 325                     self.warn(pos, 'excess byte')
 326                 else:
 327                     # Collect the bytes after the command byte.
 328                     if self.mb_s == -1:
 329                         self.mb_s = ss
 330                     self.mb_e = es
 331                     self.mb.append((mosi, miso))