decoders/nrf905/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2020 Jorge Solla Rubiales <jorgesolla@gmail.com>
   5 ##
   6 ## Permission is hereby granted, free of charge, to any person obtaining a copy
   7 ## of this software and associated documentation files (the "Software"), to deal
   8 ## in the Software without restriction, including without limitation the rights
   9 ## to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10 ## copies of the Software, and to permit persons to whom the Software is
  11 ## furnished to do so, subject to the following conditions:
  12 ##
  13 ## The above copyright notice and this permission notice shall be included in all
  14 ## copies or substantial portions of the Software.
  15 ##
  16 ## THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17 ## IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18 ## FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19 ## AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20 ## LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21 ## OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  22 ## SOFTWARE.
  23
  24 import sigrokdecode as srd
  25 from common.srdhelper import SrdIntEnum
  26
  27 CFG_REGS = {
  28     0: [{'name': 'CH_NO', 'stbit': 7, 'nbits': 8}],
  29     1: [
  30         {'name': 'AUTO_RETRAN', 'stbit': 5, 'nbits': 1,
  31          'opts': {0: 'No retransmission', 1: 'Retransmission of data packet'}},
  32         {'name': 'RX_RED_PWR', 'stbit': 4, 'nbits': 1,
  33          'opts': {0: 'Normal operation', 1: 'Reduced power'}},
  34         {'name': 'PA_PWR', 'stbit': 3, 'nbits': 2,
  35          'opts': {0: '-10 dBm', 1: '-2 dBm', 2: '+6 dBm', 3: '+10 dBm'}},
  36         {'name': 'HFREQ_PLL', 'stbit': 1, 'nbits': 1,
  37          'opts': {0: '433 MHz', 1: '868 / 915 MHz'}},
  38         {'name': 'CH_NO_8', 'stbit': 0, 'nbits': 1},
  39     ],
  40     2: [
  41         {'name': 'TX_AFW (TX addr width)', 'stbit': 6, 'nbits': 3},
  42         {'name': 'RX_AFW (RX addr width)', 'stbit': 2, 'nbits': 3},
  43     ],
  44     3: [{'name': 'RW_PW (RX payload width)', 'stbit': 5, 'nbits': 6}],
  45     4: [{'name': 'TX_PW (TX payload width)', 'stbit': 5, 'nbits': 6}],
  46     5: [{'name': 'RX_ADDR_0', 'stbit': 7, 'nbits': 8}],
  47     6: [{'name': 'RX_ADDR_1', 'stbit': 7, 'nbits': 8}],
  48     7: [{'name': 'RX_ADDR_2', 'stbit': 7, 'nbits': 8}],
  49     8: [{'name': 'RX_ADDR_3', 'stbit': 7, 'nbits': 8}],
  50     9: [
  51         {'name': 'CRC_MODE', 'stbit': 7, 'nbits': 1,
  52          'opts': {0: '8 CRC check bit', 1: '16 CRC check bit'}},
  53         {'name': 'CRC_EN', 'stbit': 6, 'nbits': 1,
  54          'opts': {0: 'Disabled', 1: 'Enabled'}},
  55         {'name': 'XOR', 'stbit': 5, 'nbits': 3,
  56          'opts': {0: '4 MHz', 1: '8 MHz', 2: '12 MHz',
  57                   3: '16 MHz', 4: '20 MHz'}},
  58         {'name': 'UP_CLK_EN', 'stbit': 2, 'nbits': 1,
  59          'opts': {0: 'No external clock signal avail.',
  60                   1: 'External clock signal enabled'}},
  61         {'name': 'UP_CLK_FREQ', 'stbit': 1, 'nbits': 2,
  62          'opts': {0: '4 MHz', 1: '2 MHz', 2: '1 MHz', 3: '500 kHz'}},
  63     ],
  64 }
  65
  66 CHN_CFG = [
  67     {'name': 'PA_PWR', 'stbit': 3, 'nbits': 2,
  68      'opts': {0: '-10 dBm', 1: '-2 dBm', 2: '+6 dBm', 3: '+10 dBm'}},
  69     {'name': 'HFREQ_PLL', 'stbit': 1, 'nbits': 1,
  70      'opts': {0: '433 MHz', 1: '868 / 915 MHz'}},
  71 ]
  72
  73 STAT_REG = [
  74     {'name': 'AM', 'stbit': 7, 'nbits': 1},
  75     {'name': 'DR', 'stbit': 5, 'nbits': 1},
  76 ]
  77
  78 Ann = SrdIntEnum.from_str('Ann', 'CMD REG_WR REG_RD TX RX RESP WARN')
  79
  80 class Decoder(srd.Decoder):
  81     api_version = 3
  82     id = 'nrf905'
  83     name = 'nRF905'
  84     longname = 'Nordic Semiconductor nRF905'
  85     desc = '433/868/933MHz transceiver chip.'
  86     license = 'mit'
  87     inputs = ['spi']
  88     outputs = []
  89     tags = ['IC', 'Wireless/RF']
  90     annotations = (
  91         ('cmd', 'Command sent to the device'),
  92         ('reg-write', 'Config register written to the device'),
  93         ('reg-read', 'Config register read from the device'),
  94         ('tx-data', 'Payload sent to the device'),
  95         ('rx-data', 'Payload read from the device'),
  96         ('resp', 'Response to commands received from the device'),
  97         ('warning', 'Warning'),
  98     )
  99     annotation_rows = (
 100         ('commands', 'Commands', (Ann.CMD,)),
 101         ('responses', 'Responses', (Ann.RESP,)),
 102         ('registers', 'Registers', (Ann.REG_WR, Ann.REG_RD)),
 103         ('tx', 'Transmitted data', (Ann.TX,)),
 104         ('rx', 'Received data', (Ann.RX,)),
 105         ('warnings', 'Warnings', (Ann.WARN,)),
 106     )
 107
 108     def __init__(self):
 109         self.ss_cmd, self.es_cmd = 0, 0
 110         self.cs_asserted = False
 111         self.reset()
 112
 113     def reset(self):
 114         self.mosi_bytes, self.miso_bytes = [], []
 115         self.cmd_samples = {'ss': 0, 'es': 0}
 116
 117     def start(self):
 118         self.out_ann = self.register(srd.OUTPUT_ANN)
 119
 120     def extract_bits(self, byte, start_bit, num_bits):
 121         begin = 7 - start_bit
 122         end = begin + num_bits
 123         if begin < 0 or end > 8:
 124             return 0
 125         binary = format(byte, '08b')[begin:end]
 126         return int(binary, 2)
 127
 128     def extract_vars(self, reg_vars, reg_value):
 129         # Iterate all vars on current register.
 130         data = ''
 131         for var in reg_vars:
 132             var_value = self.extract_bits(reg_value, var['stbit'], var['nbits'])
 133             data += var['name'] + ' = ' + str(var_value)
 134             opt = ''
 135
 136             # If var has options, just add the option meaning.
 137             if 'opts' in var:
 138                 opt = var['opts'].get(var_value, 'unknown')
 139                 data += ' (' + opt + ')'
 140
 141             # Add var separator.
 142             if reg_vars.index(var) != len(reg_vars) - 1:
 143                 data += ' | '
 144         return data
 145
 146     def parse_config_register(self, addr, value, is_write):
 147         reg_value = value[0]
 148         data = 'CFG_REG[' + hex(addr) + '] -> '
 149
 150         # Get register vars for this register.
 151         if addr in CFG_REGS:
 152             reg_vars = CFG_REGS[addr]
 153         else:
 154             # Invalid register address.
 155             self.put(value[1], value[2],
 156                      self.out_ann, [Ann.WARN, ['Invalid reg. addr']])
 157             return
 158
 159         data += self.extract_vars(reg_vars, reg_value)
 160
 161         ann = Ann.REG_WR if is_write else Ann.REG_RD
 162         self.put(value[1], value[2], self.out_ann, [ann, [data]])
 163
 164     def parse_config_registers(self, addr, registers, is_write):
 165         i = 0
 166         while i < len(registers):
 167             reg_addr = i + addr
 168             if reg_addr <= 9:
 169                 self.parse_config_register(reg_addr, registers[i], is_write)
 170             i += 1
 171
 172     def dump_cmd_bytes(self, prefix, cmd_bytes, ann):
 173         ss, es = cmd_bytes[1][1], 0
 174         data = ''
 175         for byte in cmd_bytes[1:]:
 176             data += format(byte[0], '02X') + ' '
 177             es = byte[2]
 178         self.put(ss, es, self.out_ann, [ann, [prefix + data]])
 179
 180     def handle_WC(self):
 181         start_addr = self.mosi_bytes[0][0] & 0x0F
 182         if start_addr > 9:
 183             return
 184         self.parse_config_registers(start_addr, self.mosi_bytes[1:], True)
 185
 186     def handle_RC(self):
 187         start_addr = self.mosi_bytes[0][0] & 0x0F
 188         if start_addr > 9:
 189             return
 190         self.parse_config_registers(start_addr, self.miso_bytes[1:], False)
 191
 192     def handle_WTP(self):
 193         self.dump_cmd_bytes('Write TX payload.: ', self.mosi_bytes, Ann.TX)
 194
 195     def handle_RTP(self):
 196         self.dump_cmd_bytes('Read TX payload: ', self.miso_bytes, Ann.RESP)
 197
 198     def handle_WTA(self):
 199         self.dump_cmd_bytes('Write TX addr: ', self.mosi_bytes, Ann.REG_WR)
 200
 201     def handle_RTA(self):
 202         self.dump_cmd_bytes('Read TX addr: ', self.miso_bytes, Ann.RESP)
 203
 204     def handle_RRP(self):
 205         self.dump_cmd_bytes('Read RX payload: ', self.miso_bytes, Ann.RX)
 206
 207     def handle_CC(self):
 208         cmd, dta = self.mosi_bytes[0], self.mosi_bytes[1]
 209         channel = ((cmd[0] & 0x01) << 8) + dta
 210         data = self.extract_vars(CHN_CFG, cmd[0])
 211         data += '| CHN = ' + str(channel)
 212         self.put(self.mosi_bytes[0][1], self.mosi_bytes[1][2],
 213                  self.out_ann, [Ann.REG_WR, [data]])
 214
 215     def handle_STAT(self):
 216         status = 'STAT = ' + self.extract_vars(STAT_REG, self.miso_bytes[0][0])
 217         self.put(self.miso_bytes[0][1], self.miso_bytes[0][2],
 218                  self.out_ann, [Ann.REG_RD, [status]])
 219
 220     def process_cmd(self):
 221         cmd, cmd_name, cmd_hnd = '', '', None
 222
 223         for byte in self.mosi_bytes:
 224             cmd += hex(byte[0]) + ' '
 225
 226         cmd = self.mosi_bytes[0][0]
 227
 228         if (cmd & 0xF0) == 0x00:
 229             cmd_name, cmd_hnd = 'CMD: W_CONFIG (WC)', self.handle_WC
 230         elif (cmd & 0xF0) == 0x10:
 231             cmd_name, cmd_hnd = 'CMD: R_CONFIG (RC)', self.handle_RC
 232         elif cmd == 0x20:
 233             cmd_name, cmd_hnd = 'CMD: W_TX_PAYLOAD (WTP)', self.handle_WTP
 234         elif cmd == 0x21:
 235             cmd_name, cmd_hnd = 'CMD: R_TX_PAYLOAD (RTP)', self.handle_RTP
 236         elif cmd == 0x22:
 237             cmd_name, cmd_hnd = 'CMD: W_TX_ADDRESS (WTA)', self.handle_WTA
 238         elif cmd == 0x23:
 239             cmd_name, cmd_hnd = 'CMD: R_TX_ADDRESS (RTA)', self.handle_RTA
 240         elif cmd == 0x24:
 241             cmd_name, cmd_hnd = 'CMD: R_RX_PAYLOAD (RRP)', self.handle_RRP
 242         elif (cmd & 0xF0 == 0x80):
 243             cmd_name, cmd_hnd = 'CMD: CHANNEL_CONFIG (CC)', self.handle_CC
 244
 245         # Report command name.
 246         self.put(self.cmd_samples['ss'], self.cmd_samples['es'],
 247                  self.out_ann, [Ann.CMD, [cmd_name]])
 248
 249         # Handle status byte.
 250         self.handle_STAT()
 251
 252         # Handle command.
 253         if cmd_hnd is not None:
 254             cmd_hnd()
 255
 256     def set_cs_status(self, sample, asserted):
 257         if self.cs_asserted == asserted:
 258             return
 259
 260         if asserted:
 261             self.cmd_samples['ss'] = sample
 262             self.cmd_samples['es'] = -1
 263         else:
 264             self.cmd_samples['es'] = sample
 265
 266         self.cs_asserted = asserted
 267
 268     def decode(self, ss, es, data):
 269         ptype, data1, data2 = data
 270
 271         if ptype == 'CS-CHANGE':
 272             if data1 is None and data2 is None:
 273                 self.requirements_met = False
 274                 raise ChannelError('CS# pin required.')
 275
 276             if data1 is None and data2 == 0:
 277                 self.set_cs_status(ss, True)
 278
 279             elif data1 is None and data2 == 1:
 280                 self.set_cs_status(ss, False)
 281
 282             elif data1 == 1 and data2 == 0:
 283                 self.set_cs_status(ss, True)
 284
 285             elif data1 == 0 and data2 == 1:
 286                 self.set_cs_status(ss, False)
 287                 if len(self.mosi_bytes):
 288                     self.process_cmd()
 289                     self.reset()
 290
 291         elif ptype == 'DATA':
 292             # Ignore traffic if CS is not asserted.
 293             if self.cs_asserted is False:
 294                 return
 295
 296             mosi, miso = data1, data2
 297             if miso is None or mosi is None:
 298                 raise ChannelError('Both MISO and MOSI pins required.')
 299
 300             self.mosi_bytes.append((mosi, ss, es))
 301             self.miso_bytes.append((miso, ss, es))