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