[libsigrokdecode.git] / decoders / cc1101 / pd.py

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2019 Marco Geisler <m-sigrok@mageis.de>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <http://www.gnu.org/licenses/>.
##

import sigrokdecode as srd
from .lists import *

class Decoder(srd.Decoder):
    api_version = 3
    id = 'cc1101'
    name = 'CC1101'
    longname = 'Texas Instruments CC1101'
    desc = 'Low-power sub-1GHz RF transceiver chip.'
    license = 'gplv2+'
    inputs = ['spi']
    outputs = []
    tags = ['IC', 'Wireless/RF']
    annotations = (
        ('strobe', 'Command strobe'),
        ('single_read', 'Single register read'),
        ('single_write', 'Single register write'),
        ('burst_read', 'Burst register read'),
        ('burst_write', 'Burst register write'),
        ('status', 'Status register'),
        ('warning', 'Warning'),
    )
    ann_strobe        = 0
    ann_single_read   = 1
    ann_single_write  = 2
    ann_burst_read    = 3
    ann_burst_write   = 4
    ann_status_read   = 5
    ann_status        = 6
    ann_warn          = 7
    annotation_rows = (
        ('cmd', 'Commands', (ann_strobe,)),
        ('data', 'Data', (ann_single_read, ann_single_write, ann_burst_read, ann_burst_write, ann_status_read)),
        ('status', 'Status register', (ann_status,)),
        ('warnings', 'Warnings', (ann_warn,)),
    )

    def __init__(self):
        self.reset()

    def reset(self):
        self.next()
        self.requirements_met = True
        self.cs_was_released = False

    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)

    def warn(self, pos, msg):
        '''Put a warning message 'msg' at 'pos'.'''
        self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])

    def putp(self, pos, ann, msg):
        '''Put an annotation message 'msg' at 'pos'.'''
        self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])

    def putp2(self, pos, ann, msg1, msg2):
        '''Put an annotation message 'msg' at 'pos'.'''
        self.put(pos[0], pos[1], self.out_ann, [ann, [msg1, msg2]])

    def next(self):
        '''Resets the decoder after a complete command was decoded.'''
        # 'True' for the first byte after CS# went low.
        self.first = True

        # The current command, and the minimum and maximum number
        # of data bytes to follow.
        self.cmd = None
        self.min = 0
        self.max = 0

        # Used to collect the bytes after the command byte
        # (and the start/end sample number).
        self.mb = []
        self.ss_mb = -1
        self.es_mb = -1

    def mosi_bytes(self):
        '''Returns the collected MOSI bytes of a multi byte command.'''
        return [b[0] for b in self.mb]

    def miso_bytes(self):
        '''Returns the collected MISO bytes of a multi byte command.'''
        return [b[1] for b in self.mb]

    def decode_command(self, pos, b):
        '''Decodes the command byte 'b' at position 'pos' and prepares
        the decoding of the following data bytes.'''
        c = self.parse_command(b)
        if c is None:
            self.warn(pos, 'unknown command')
            return

        self.cmd, self.dat, self.min, self.max = c

        if self.cmd in ('STROBE_CMD',):
            self.putp(pos, self.ann_strobe, self.format_command())
        else:
            # Don't output anything now, the command is merged with
            # the data bytes following it.
            self.ss_mb = pos[0]

    def format_command(self):
        '''Returns the label for the current command.'''
        if self.cmd == 'SINGLE_READ':
            reg = regs[self.dat] if self.dat in regs else 'unknown register'
            return 'Read'
        if self.cmd == 'BURST_READ':
            reg = regs[self.dat] if self.dat in regs else 'unknown register'
            return 'Burst read'
        if self.cmd == 'SINGLE_WRITE':
            reg = regs[self.dat] if self.dat in regs else 'unknown register'
            return 'Write'
        if self.cmd == 'BURST_WRITE':
            reg = regs[self.dat] if self.dat in regs else 'unknown register'
            return 'Burst write'
        if self.cmd == 'STATUS_READ':
            reg = regs[self.dat] if self.dat in regs else 'unknown register'
            return 'Status read'
        if self.cmd == 'STROBE_CMD':
            reg = strobes[self.dat] if self.dat in strobes else 'unknown strobe'
            return 'STROBE "{}"'.format(reg)
        else:
            return 'TODO Cmd {}'.format(self.cmd)

    def parse_command(self, b):
        '''Parses the command byte.

        Returns a tuple consisting of:
        - the name of the command
        - additional data needed to dissect the following bytes
        - minimum number of following bytes
        - maximum number of following bytes (None for infinite)
        '''

        addr = b & 0x3F
        if (addr < 0x30) or (addr == 0x3E) or (addr == 0x3F):
            if (b & 0xC0) == 0x00:
                return ('SINGLE_WRITE', addr, 1, 1)
            if (b & 0xC0) == 0x40:
                return ('BURST_WRITE', addr, 1, 99999)
            if (b & 0xC0) == 0x80:
                return ('SINGLE_READ', addr, 1, 1)
            if (b & 0xC0) == 0xC0:
                return ('BURST_READ', addr, 1, 99999)
            else:
                self.warn(pos, 'unknown address/command combination')
        else:
            if (b & 0x40) == 0x00:
                return ('STROBE_CMD', addr, 0, 0)
            if (b & 0xC0) == 0xC0:
                return ('STATUS_READ', addr, 1, 99999)
            else:
                self.warn(pos, 'unknown address/command combination')

    def decode_register(self, pos, ann, regid, data):
        '''Decodes a register.

        pos   -- start and end sample numbers of the register
        ann   -- the annotation number that is used to output the register.
        regid -- may be either an integer used as a key for the 'regs'
                 dictionary, or a string directly containing a register name.'
        data  -- the register content.
        '''

        if type(regid) == int:
            # Get the name of the register.
            if regid not in regs:
                self.warn(pos, 'unknown register')
                return
            name = '{} (0x{:02X})'.format(regs[regid], regid)
        else:
            name = regid

        if regid == 'STATUS' and ann == self.ann_status:
            label = 'Status'
            self.decode_status_reg(pos, ann, data, label)
        else:
            if self.cmd in ('SINGLE_WRITE', 'SINGLE_READ', 'STATUS_READ', 'BURST_READ', 'BURST_WRITE'):
                label = '{}: {}'.format(self.format_command(), name)
            else:
                label = 'Reg ({}) {}'.format(self.cmd, name)
            self.decode_mb_data(pos, ann, data, label)

    def decode_status_reg(self, pos, ann, data, label):
        '''Decodes the data bytes 'data' of a status register at position
        'pos'. The decoded data is prefixed with 'label'.'''
        status = data[0]
        # bit 7 --> CHIP_RDYn
        if status & 0b10000000 == 0b10000000:
            longtext_chiprdy = 'CHIP_RDYn is high! '
        else:
            longtext_chiprdy = ''
        # bits 6:4 --> STATE
        state = (status & 0x70) >> 4
        longtext_state = 'STATE is {}, '.format(status_reg_states[state])
        # bits 3:0 --> FIFO_BYTES_AVAILABLE
        fifo_bytes = status & 0x0F
        if self.cmd in ('SINGLE_READ', 'STATUS_READ', 'BURST_READ'):
            longtext_fifo = '{} bytes available in RX FIFO'.format(fifo_bytes)
        else:
            longtext_fifo = '{} bytes free in TX FIFO'.format(fifo_bytes)

        text = '{} = "0x{:02X}"'.format(label, status)
        longtext = ''.join([text, '; ', longtext_chiprdy, longtext_state, longtext_fifo])
        self.putp2(pos, ann, longtext, text)

    def decode_mb_data(self, pos, ann, data, label):
        '''Decodes the data bytes 'data' of a multibyte command at position
        'pos'. The decoded data is prefixed with 'label'.'''

        def escape(b):
            return '{:02X}'.format(b)

        data = ' '.join([escape(b) for b in data])
        text = '{} = "0x{}"'.format(label, data)
        self.putp(pos, ann, text)

    def finish_command(self, pos):
        '''Decodes the remaining data bytes at position 'pos'.'''

        if self.cmd == 'SINGLE_WRITE':
            self.decode_register(pos, self.ann_single_write,
                                 self.dat, self.mosi_bytes())
        elif self.cmd == 'BURST_WRITE':
            self.decode_register(pos, self.ann_burst_write,
                                self.dat, self.mosi_bytes())
        elif self.cmd == 'SINGLE_READ':
            self.decode_register(pos, self.ann_single_read,
                                 self.dat, self.miso_bytes())
        elif self.cmd == 'BURST_READ':
            self.decode_register(pos, self.ann_burst_read,
                                self.dat, self.miso_bytes())
        elif self.cmd == 'STROBE_CMD':
            self.decode_register(pos, self.ann_strobe,
                                 self.dat, self.mosi_bytes())
        elif self.cmd == 'STATUS_READ':
            self.decode_register(pos, self.ann_status_read,
                                 self.dat, self.miso_bytes())
        else:
            self.warn(pos, 'unhandled command')

    def decode(self, ss, es, data):
        if not self.requirements_met:
            return

        ptype, data1, data2 = data

        if ptype == 'CS-CHANGE':
            if data1 is None:
                if data2 is None:
                    self.requirements_met = False
                    raise ChannelError('CS# pin required.')
                elif data2 == 1:
                    self.cs_was_released = True

            if data1 == 0 and data2 == 1:
                # Rising edge, the complete command is transmitted, process
                # the bytes that were sent after the command byte.
                if self.cmd:
                    # Check if we got the minimum number of data bytes
                    # after the command byte.
                    if len(self.mb) < self.min:
                        self.warn((ss, ss), 'missing data bytes')
                    elif self.mb:
                        self.finish_command((self.ss_mb, self.es_mb))

                self.next()
                self.cs_was_released = True

        elif ptype == 'DATA' and self.cs_was_released:
            mosi, miso = data1, data2
            pos = (ss, es)

            if miso is None or mosi is None:
                self.requirements_met = False
                raise ChannelError('Both MISO and MOSI pins required.')

            if self.first:
                self.first = False
                # First MOSI byte is always the command.
                self.decode_command(pos, mosi)
                # First MISO byte is always the status register.
                self.decode_register(pos, self.ann_status, 'STATUS', [miso])
            else:
                if not self.cmd or len(self.mb) >= self.max:
                    self.warn(pos, 'excess byte')
                else:
                    # Collect the bytes after the command byte.
                    if self.ss_mb == -1:
                        self.ss_mb = ss
                    self.es_mb = es
                    self.mb.append((mosi, miso))
Commit	Line	Data
2a743706 M	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))