decoders/cc1101/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2019 Marco Geisler <m-sigrok@mageis.de>
   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 from .lists import *
  22
  23 class Decoder(srd.Decoder):
  24     api_version = 3
  25     id = 'cc1101'
  26     name = 'CC1101'
  27     longname = 'Texas Instruments CC1101'
  28     desc = 'Low-power sub-1GHz RF transceiver chip.'
  29     license = 'gplv2+'
  30     inputs = ['spi']
  31     outputs = []
  32     tags = ['IC', 'Wireless/RF']
  33     annotations = (
  34         ('strobe', 'Command strobe'),
  35         ('single_read', 'Single register read'),
  36         ('single_write', 'Single register write'),
  37         ('burst_read', 'Burst register read'),
  38         ('burst_write', 'Burst register write'),
  39         ('status', 'Status register'),
  40         ('warning', 'Warning'),
  41     )
  42     ann_strobe        = 0
  43     ann_single_read   = 1
  44     ann_single_write  = 2
  45     ann_burst_read    = 3
  46     ann_burst_write   = 4
  47     ann_status_read   = 5
  48     ann_status        = 6
  49     ann_warn          = 7
  50     annotation_rows = (
  51         ('cmd', 'Commands', (ann_strobe,)),
  52         ('data', 'Data', (ann_single_read, ann_single_write, ann_burst_read, ann_burst_write, ann_status_read)),
  53         ('status', 'Status register', (ann_status,)),
  54         ('warnings', 'Warnings', (ann_warn,)),
  55     )
  56
  57     def __init__(self):
  58         self.reset()
  59
  60     def reset(self):
  61         self.next()
  62         self.requirements_met = True
  63         self.cs_was_released = False
  64
  65     def start(self):
  66         self.out_ann = self.register(srd.OUTPUT_ANN)
  67
  68     def warn(self, pos, msg):
  69         '''Put a warning message 'msg' at 'pos'.'''
  70         self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])
  71
  72     def putp(self, pos, ann, msg):
  73         '''Put an annotation message 'msg' at 'pos'.'''
  74         self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])
  75
  76     def putp2(self, pos, ann, msg1, msg2):
  77         '''Put an annotation message 'msg' at 'pos'.'''
  78         self.put(pos[0], pos[1], self.out_ann, [ann, [msg1, msg2]])
  79
  80     def next(self):
  81         '''Resets the decoder after a complete command was decoded.'''
  82         # 'True' for the first byte after CS# went low.
  83         self.first = True
  84
  85         # The current command, and the minimum and maximum number
  86         # of data bytes to follow.
  87         self.cmd = None
  88         self.min = 0
  89         self.max = 0
  90
  91         # Used to collect the bytes after the command byte
  92         # (and the start/end sample number).
  93         self.mb = []
  94         self.ss_mb = -1
  95         self.es_mb = -1
  96
  97     def mosi_bytes(self):
  98         '''Returns the collected MOSI bytes of a multi byte command.'''
  99         return [b[0] for b in self.mb]
 100
 101     def miso_bytes(self):
 102         '''Returns the collected MISO bytes of a multi byte command.'''
 103         return [b[1] for b in self.mb]
 104
 105     def decode_command(self, pos, b):
 106         '''Decodes the command byte 'b' at position 'pos' and prepares
 107         the decoding of the following data bytes.'''
 108         c = self.parse_command(b)
 109         if c is None:
 110             self.warn(pos, 'unknown command')
 111             return
 112
 113         self.cmd, self.dat, self.min, self.max = c
 114
 115         if self.cmd in ('STROBE_CMD',):
 116             self.putp(pos, self.ann_strobe, self.format_command())
 117         else:
 118             # Don't output anything now, the command is merged with
 119             # the data bytes following it.
 120             self.ss_mb = pos[0]
 121
 122     def format_command(self):
 123         '''Returns the label for the current command.'''
 124         if self.cmd == 'SINGLE_READ':
 125             reg = regs[self.dat] if self.dat in regs else 'unknown register'
 126             return 'Read'
 127         if self.cmd == 'BURST_READ':
 128             reg = regs[self.dat] if self.dat in regs else 'unknown register'
 129             return 'Burst read'
 130         if self.cmd == 'SINGLE_WRITE':
 131             reg = regs[self.dat] if self.dat in regs else 'unknown register'
 132             return 'Write'
 133         if self.cmd == 'BURST_WRITE':
 134             reg = regs[self.dat] if self.dat in regs else 'unknown register'
 135             return 'Burst write'
 136         if self.cmd == 'STATUS_READ':
 137             reg = regs[self.dat] if self.dat in regs else 'unknown register'
 138             return 'Status read'
 139         if self.cmd == 'STROBE_CMD':
 140             reg = strobes[self.dat] if self.dat in strobes else 'unknown strobe'
 141             return 'STROBE "{}"'.format(reg)
 142         else:
 143             return 'TODO Cmd {}'.format(self.cmd)
 144
 145     def parse_command(self, b):
 146         '''Parses the command byte.
 147
 148         Returns a tuple consisting of:
 149         - the name of the command
 150         - additional data needed to dissect the following bytes
 151         - minimum number of following bytes
 152         - maximum number of following bytes (None for infinite)
 153         '''
 154
 155         addr = b & 0x3F
 156         if (addr < 0x30) or (addr == 0x3E) or (addr == 0x3F):
 157             if (b & 0xC0) == 0x00:
 158                 return ('SINGLE_WRITE', addr, 1, 1)
 159             if (b & 0xC0) == 0x40:
 160                 return ('BURST_WRITE', addr, 1, 99999)
 161             if (b & 0xC0) == 0x80:
 162                 return ('SINGLE_READ', addr, 1, 1)
 163             if (b & 0xC0) == 0xC0:
 164                 return ('BURST_READ', addr, 1, 99999)
 165             else:
 166                 self.warn(pos, 'unknown address/command combination')
 167         else:
 168             if (b & 0x40) == 0x00:
 169                 return ('STROBE_CMD', addr, 0, 0)
 170             if (b & 0xC0) == 0xC0:
 171                 return ('STATUS_READ', addr, 1, 99999)
 172             else:
 173                 self.warn(pos, 'unknown address/command combination')
 174
 175     def decode_register(self, pos, ann, regid, data):
 176         '''Decodes a register.
 177
 178         pos   -- start and end sample numbers of the register
 179         ann   -- the annotation number that is used to output the register.
 180         regid -- may be either an integer used as a key for the 'regs'
 181                  dictionary, or a string directly containing a register name.'
 182         data  -- the register content.
 183         '''
 184
 185         if type(regid) == int:
 186             # Get the name of the register.
 187             if regid not in regs:
 188                 self.warn(pos, 'unknown register')
 189                 return
 190             name = '{} (0x{:02X})'.format(regs[regid], regid)
 191         else:
 192             name = regid
 193
 194         if regid == 'STATUS' and ann == self.ann_status:
 195             label = 'Status'
 196             self.decode_status_reg(pos, ann, data, label)
 197         else:
 198             if self.cmd in ('SINGLE_WRITE', 'SINGLE_READ', 'STATUS_READ', 'BURST_READ', 'BURST_WRITE'):
 199                 label = '{}: {}'.format(self.format_command(), name)
 200             else:
 201                 label = 'Reg ({}) {}'.format(self.cmd, name)
 202             self.decode_mb_data(pos, ann, data, label)
 203
 204     def decode_status_reg(self, pos, ann, data, label):
 205         '''Decodes the data bytes 'data' of a status register at position
 206         'pos'. The decoded data is prefixed with 'label'.'''
 207         status = data[0]
 208         # bit 7 --> CHIP_RDYn
 209         if status & 0b10000000 == 0b10000000:
 210             longtext_chiprdy = 'CHIP_RDYn is high! '
 211         else:
 212             longtext_chiprdy = ''
 213         # bits 6:4 --> STATE
 214         state = (status & 0x70) >> 4
 215         longtext_state = 'STATE is {}, '.format(status_reg_states[state])
 216         # bits 3:0 --> FIFO_BYTES_AVAILABLE
 217         fifo_bytes = status & 0x0F
 218         if self.cmd in ('SINGLE_READ', 'STATUS_READ', 'BURST_READ'):
 219             longtext_fifo = '{} bytes available in RX FIFO'.format(fifo_bytes)
 220         else:
 221             longtext_fifo = '{} bytes free in TX FIFO'.format(fifo_bytes)
 222
 223         text = '{} = "0x{:02X}"'.format(label, status)
 224         longtext = ''.join([text, '; ', longtext_chiprdy, longtext_state, longtext_fifo])
 225         self.putp2(pos, ann, longtext, text)
 226
 227     def decode_mb_data(self, pos, ann, data, label):
 228         '''Decodes the data bytes 'data' of a multibyte command at position
 229         'pos'. The decoded data is prefixed with 'label'.'''
 230
 231         def escape(b):
 232             return '{:02X}'.format(b)
 233
 234         data = ' '.join([escape(b) for b in data])
 235         text = '{} = "0x{}"'.format(label, data)
 236         self.putp(pos, ann, text)
 237
 238     def finish_command(self, pos):
 239         '''Decodes the remaining data bytes at position 'pos'.'''
 240
 241         if self.cmd == 'SINGLE_WRITE':
 242             self.decode_register(pos, self.ann_single_write,
 243                                  self.dat, self.mosi_bytes())
 244         elif self.cmd == 'BURST_WRITE':
 245             self.decode_register(pos, self.ann_burst_write,
 246                                 self.dat, self.mosi_bytes())
 247         elif self.cmd == 'SINGLE_READ':
 248             self.decode_register(pos, self.ann_single_read,
 249                                  self.dat, self.miso_bytes())
 250         elif self.cmd == 'BURST_READ':
 251             self.decode_register(pos, self.ann_burst_read,
 252                                 self.dat, self.miso_bytes())
 253         elif self.cmd == 'STROBE_CMD':
 254             self.decode_register(pos, self.ann_strobe,
 255                                  self.dat, self.mosi_bytes())
 256         elif self.cmd == 'STATUS_READ':
 257             self.decode_register(pos, self.ann_status_read,
 258                                  self.dat, self.miso_bytes())
 259         else:
 260             self.warn(pos, 'unhandled command')
 261
 262     def decode(self, ss, es, data):
 263         if not self.requirements_met:
 264             return
 265
 266         ptype, data1, data2 = data
 267
 268         if ptype == 'CS-CHANGE':
 269             if data1 is None:
 270                 if data2 is None:
 271                     self.requirements_met = False
 272                     raise ChannelError('CS# pin required.')
 273                 elif data2 == 1:
 274                     self.cs_was_released = True
 275
 276             if data1 == 0 and data2 == 1:
 277                 # Rising edge, the complete command is transmitted, process
 278                 # the bytes that were sent after the command byte.
 279                 if self.cmd:
 280                     # Check if we got the minimum number of data bytes
 281                     # after the command byte.
 282                     if len(self.mb) < self.min:
 283                         self.warn((ss, ss), 'missing data bytes')
 284                     elif self.mb:
 285                         self.finish_command((self.ss_mb, self.es_mb))
 286
 287                 self.next()
 288                 self.cs_was_released = True
 289
 290         elif ptype == 'DATA' and self.cs_was_released:
 291             mosi, miso = data1, data2
 292             pos = (ss, es)
 293
 294             if miso is None or mosi is None:
 295                 self.requirements_met = False
 296                 raise ChannelError('Both MISO and MOSI pins required.')
 297
 298             if self.first:
 299                 self.first = False
 300                 # First MOSI byte is always the command.
 301                 self.decode_command(pos, mosi)
 302                 # First MISO byte is always the status register.
 303                 self.decode_register(pos, self.ann_status, 'STATUS', [miso])
 304             else:
 305                 if not self.cmd or len(self.mb) >= self.max:
 306                     self.warn(pos, 'excess byte')
 307                 else:
 308                     # Collect the bytes after the command byte.
 309                     if self.ss_mb == -1:
 310                         self.ss_mb = ss
 311                     self.es_mb = es
 312                     self.mb.append((mosi, miso))