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