24352f6232add525a9e1e9b6935e98e8dd61b751
[libsigrokdecode.git] / 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 in ('STROBE_CMD',):
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 == 'SINGLE_READ':
121             reg = regs[self.dat] if self.dat in regs else 'unknown register'
122             return 'Read'
123         if self.cmd == 'BURST_READ':
124             reg = regs[self.dat] if self.dat in regs else 'unknown register'
125             return 'Burst read'
126         if self.cmd == 'SINGLE_WRITE':
127             reg = regs[self.dat] if self.dat in regs else 'unknown register'
128             return 'Write'
129         if self.cmd == 'BURST_WRITE':
130             reg = regs[self.dat] if self.dat in regs else 'unknown register'
131             return 'Burst write'
132         if self.cmd == 'STATUS_READ':
133             reg = regs[self.dat] if self.dat in regs else 'unknown register'
134             return 'Status read'
135         if self.cmd == 'STROBE_CMD':
136             reg = strobes[self.dat] if self.dat in strobes else 'unknown strobe'
137             return 'STROBE "{}"'.format(reg)
138         else:
139             return 'TODO Cmd {}'.format(self.cmd)
140
141     def parse_command(self, b):
142         '''Parses the command byte.
143
144         Returns a tuple consisting of:
145         - the name of the command
146         - additional data needed to dissect the following bytes
147         - minimum number of following bytes
148         - maximum number of following bytes (None for infinite)
149         '''
150
151         addr = b & 0x3F
152         if (addr < 0x30) or (addr == 0x3E) or (addr == 0x3F):
153             if (b & 0xC0) == 0x00:
154                 return ('SINGLE_WRITE', addr, 1, 1)
155             if (b & 0xC0) == 0x40:
156                 return ('BURST_WRITE', addr, 1, 99999)
157             if (b & 0xC0) == 0x80:
158                 return ('SINGLE_READ', addr, 1, 1)
159             if (b & 0xC0) == 0xC0:
160                 return ('BURST_READ', addr, 1, 99999)
161             else:
162                 self.warn(pos, 'unknown address/command combination')
163         else:
164             if (b & 0x40) == 0x00:
165                 return ('STROBE_CMD', addr, 0, 0)
166             if (b & 0xC0) == 0xC0:
167                 return ('STATUS_READ', addr, 1, 99999)
168             else:
169                 self.warn(pos, 'unknown address/command combination')
170
171     def decode_register(self, pos, ann, regid, data):
172         '''Decodes a register.
173
174         pos   -- start and end sample numbers of the register
175         ann   -- the annotation number that is used to output the register.
176         regid -- may be either an integer used as a key for the 'regs'
177                  dictionary, or a string directly containing a register name.'
178         data  -- the register content.
179         '''
180
181         if type(regid) == int:
182             # Get the name of the register.
183             if regid not in regs:
184                 self.warn(pos, 'unknown register')
185                 return
186             name = '{} (0x{:02X})'.format(regs[regid], regid)
187         else:
188             name = regid
189
190         if regid == 'STATUS' and ann == ANN_STATUS:
191             label = 'Status'
192             self.decode_status_reg(pos, ann, data, label)
193         else:
194             if self.cmd in ('SINGLE_WRITE', 'SINGLE_READ', 'STATUS_READ', 'BURST_READ', 'BURST_WRITE'):
195                 label = '{}: {}'.format(self.format_command(), name)
196             else:
197                 label = 'Reg ({}) {}'.format(self.cmd, name)
198             self.decode_mb_data(pos, ann, data, label)
199
200     def decode_status_reg(self, pos, ann, data, label):
201         '''Decodes the data bytes 'data' of a status register at position
202         'pos'. The decoded data is prefixed with 'label'.'''
203         status = data[0]
204         # bit 7 --> CHIP_RDYn
205         if status & 0b10000000 == 0b10000000:
206             longtext_chiprdy = 'CHIP_RDYn is high! '
207         else:
208             longtext_chiprdy = ''
209         # bits 6:4 --> STATE
210         state = (status & 0x70) >> 4
211         longtext_state = 'STATE is {}, '.format(status_reg_states[state])
212         # bits 3:0 --> FIFO_BYTES_AVAILABLE
213         fifo_bytes = status & 0x0F
214         if self.cmd in ('SINGLE_READ', 'STATUS_READ', 'BURST_READ'):
215             longtext_fifo = '{} bytes available in RX FIFO'.format(fifo_bytes)
216         else:
217             longtext_fifo = '{} bytes free in TX FIFO'.format(fifo_bytes)
218
219         text = '{} = "0x{:02X}"'.format(label, status)
220         longtext = ''.join([text, '; ', longtext_chiprdy, longtext_state, longtext_fifo])
221         self.putp2(pos, ann, longtext, text)
222
223     def decode_mb_data(self, pos, ann, data, label):
224         '''Decodes the data bytes 'data' of a multibyte command at position
225         'pos'. The decoded data is prefixed with 'label'.'''
226
227         def escape(b):
228             return '{:02X}'.format(b)
229
230         data = ' '.join([escape(b) for b in data])
231         text = '{} = "0x{}"'.format(label, data)
232         self.putp(pos, ann, text)
233
234     def finish_command(self, pos):
235         '''Decodes the remaining data bytes at position 'pos'.'''
236
237         if self.cmd == 'SINGLE_WRITE':
238             self.decode_register(pos, ANN_SINGLE_WRITE,
239                                  self.dat, self.mosi_bytes())
240         elif self.cmd == 'BURST_WRITE':
241             self.decode_register(pos, ANN_BURST_WRITE,
242                                 self.dat, self.mosi_bytes())
243         elif self.cmd == 'SINGLE_READ':
244             self.decode_register(pos, ANN_SINGLE_READ,
245                                  self.dat, self.miso_bytes())
246         elif self.cmd == 'BURST_READ':
247             self.decode_register(pos, ANN_BURST_READ,
248                                 self.dat, self.miso_bytes())
249         elif self.cmd == 'STROBE_CMD':
250             self.decode_register(pos, ANN_STROBE,
251                                  self.dat, self.mosi_bytes())
252         elif self.cmd == 'STATUS_READ':
253             self.decode_register(pos, ANN_STATUS_READ,
254                                  self.dat, self.miso_bytes())
255         else:
256             self.warn(pos, 'unhandled command')
257
258     def decode(self, ss, es, data):
259         if not self.requirements_met:
260             return
261
262         ptype, data1, data2 = data
263
264         if ptype == 'CS-CHANGE':
265             if data1 is None:
266                 if data2 is None:
267                     self.requirements_met = False
268                     raise ChannelError('CS# pin required.')
269                 elif data2 == 1:
270                     self.cs_was_released = True
271
272             if data1 == 0 and data2 == 1:
273                 # Rising edge, the complete command is transmitted, process
274                 # the bytes that were sent after the command byte.
275                 if self.cmd:
276                     # Check if we got the minimum number of data bytes
277                     # after the command byte.
278                     if len(self.mb) < self.min:
279                         self.warn((ss, ss), 'missing data bytes')
280                     elif self.mb:
281                         self.finish_command((self.ss_mb, self.es_mb))
282
283                 self.next()
284                 self.cs_was_released = True
285
286         elif ptype == 'DATA' and self.cs_was_released:
287             mosi, miso = data1, data2
288             pos = (ss, es)
289
290             if miso is None or mosi is None:
291                 self.requirements_met = False
292                 raise ChannelError('Both MISO and MOSI pins required.')
293
294             if self.first:
295                 self.first = False
296                 # First MOSI byte is always the command.
297                 self.decode_command(pos, mosi)
298                 # First MISO byte is always the status register.
299                 self.decode_register(pos, ANN_STATUS, 'STATUS', [miso])
300             else:
301                 if not self.cmd or len(self.mb) >= self.max:
302                     self.warn(pos, 'excess byte')
303                 else:
304                     # Collect the bytes after the command byte.
305                     if self.ss_mb == -1:
306                         self.ss_mb = ss
307                     self.es_mb = es
308                     self.mb.append((mosi, miso))