[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 collections import namedtuple
from common.srdhelper import SrdIntEnum
from .lists import *

Ann = SrdIntEnum.from_str('Ann', 'STROBE SINGLE_READ SINGLE_WRITE BURST_READ \
    BURST_WRITE STATUS_READ STATUS WARN')

Pos = namedtuple('Pos', ['ss', 'es'])
Data = namedtuple('Data', ['mosi', 'miso'])

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_read', 'Status read'),
        ('status_reg', 'Status register'),
        ('warning', 'Warning'),
    )
    annotation_rows = (
        ('cmds', 'Commands', (Ann.STROBE,)),
        ('data', 'Data', (Ann.prefixes('SINGLE_ BURST_ STATUS_'))),
        ('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.ss, pos.es, self.out_ann, [Ann.WARN, [msg]])

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

    def putp2(self, pos, ann, msg1, msg2):
        '''Put an annotation message 'msg' at 'pos'.'''
        self.put(pos.ss, pos.es, 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.mosi for b in self.mb]

    def miso_bytes(self):
        '''Returns the collected MISO bytes of a multi byte command.'''
        return [b.miso 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 == 'Strobe':
            self.putp(pos, 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.ss

    def format_command(self):
        '''Returns the label for the current command.'''
        if self.cmd in ('Read', 'Burst read', 'Write', 'Burst write', 'Status read'):
            return self.cmd
        if self.cmd == 'Strobe':
            reg = strobes.get(self.dat, 'unknown strobe')
            return '{} {}'.format(self.cmd, 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 ('Write', addr, 1, 1)
            if (b & 0xC0) == 0x40:
                return ('Burst write', addr, 1, 99999)
            if (b & 0xC0) == 0x80:
                return ('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', addr, 0, 0)
            if (b & 0xC0) == 0xC0:
                return ('Status read', addr, 1, 99999)
            else:
                self.warn(pos, 'unknown address/command combination')

    def decode_reg(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 = '{} ({:02X})'.format(regs[regid], regid)
        else:
            name = regid

        if regid == 'STATUS' and ann == Ann.STATUS:
            label = 'Status'
            self.decode_status_reg(pos, ann, data, label)
        else:
            if self.cmd in ('Write', '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 = '{} = {: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 = '{} = {}'.format(label, data)
        self.putp(pos, ann, text)

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

        if self.cmd == 'Write':
            self.decode_reg(pos, Ann.SINGLE_WRITE, self.dat, self.mosi_bytes())
        elif self.cmd == 'Burst write':
            self.decode_reg(pos, Ann.BURST_WRITE, self.dat, self.mosi_bytes())
        elif self.cmd == 'Read':
            self.decode_reg(pos, Ann.SINGLE_READ, self.dat, self.miso_bytes())
        elif self.cmd == 'Burst read':
            self.decode_reg(pos, Ann.BURST_READ, self.dat, self.miso_bytes())
        elif self.cmd == 'Strobe':
            self.decode_reg(pos, Ann.STROBE, self.dat, self.mosi_bytes())
        elif self.cmd == 'Status read':
            self.decode_reg(pos, 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(Pos(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 = 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_reg(pos, 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(Data(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
5ae47601	21	from collections import namedtuple
c5213e26	22	from common.srdhelper import SrdIntEnum
2a743706 M	23	from .lists import *
2a743706 M	24
c5213e26 UH	25	Ann = SrdIntEnum.from_str('Ann', 'STROBE SINGLE_READ SINGLE_WRITE BURST_READ \
c5213e26 UH	26	BURST_WRITE STATUS_READ STATUS WARN')
743457ac	27
5ae47601 UH	28	Pos = namedtuple('Pos', ['ss', 'es'])
	29	Data = namedtuple('Data', ['mosi', 'miso'])
	30
2a743706 M	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'),
bf126644	47	('status_read', 'Status read'),
e144452b	48	('status_reg', 'Status register'),
2a743706 M	49	('warning', 'Warning'),
2a743706 M	50	)
2a743706	51	annotation_rows = (
c5213e26 UH	52	('cmds', 'Commands', (Ann.STROBE,)),
	53	('data', 'Data', (Ann.prefixes('SINGLE_ BURST_ STATUS_'))),
	54	('status', 'Status register', (Ann.STATUS,)),
	55	('warnings', 'Warnings', (Ann.WARN,)),
2a743706 M	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'.'''
c5213e26	71	self.put(pos.ss, pos.es, self.out_ann, [Ann.WARN, [msg]])
2a743706 M	72
	73	def putp(self, pos, ann, msg):
	74	'''Put an annotation message 'msg' at 'pos'.'''
5ae47601	75	self.put(pos.ss, pos.es, self.out_ann, [ann, [msg]])
2a743706 M	76
	77	def putp2(self, pos, ann, msg1, msg2):
	78	'''Put an annotation message 'msg' at 'pos'.'''
5ae47601	79	self.put(pos.ss, pos.es, self.out_ann, [ann, [msg1, msg2]])
2a743706 M	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.'''
5ae47601	100	return [b.mosi for b in self.mb]
2a743706 M	101
	102	def miso_bytes(self):
	103	'''Returns the collected MISO bytes of a multi byte command.'''
5ae47601	104	return [b.miso for b in self.mb]
2a743706 M	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
26ec0881	116	if self.cmd == 'Strobe':
c5213e26	117	self.putp(pos, Ann.STROBE, self.format_command())
2a743706 M	118	else:
	119	# Don't output anything now, the command is merged with
	120	# the data bytes following it.
5ae47601	121	self.ss_mb = pos.ss
2a743706 M	122
	123	def format_command(self):
	124	'''Returns the label for the current command.'''
26ec0881 UH	125	if self.cmd in ('Read', 'Burst read', 'Write', 'Burst write', 'Status read'):
	126	return self.cmd
	127	if self.cmd == 'Strobe':
4c58713b	128	reg = strobes.get(self.dat, 'unknown strobe')
42ca52b5	129	return '{} {}'.format(self.cmd, reg)
2a743706 M	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:
26ec0881	146	return ('Write', addr, 1, 1)
2a743706	147	if (b & 0xC0) == 0x40:
26ec0881	148	return ('Burst write', addr, 1, 99999)
2a743706	149	if (b & 0xC0) == 0x80:
26ec0881	150	return ('Read', addr, 1, 1)
2a743706	151	if (b & 0xC0) == 0xC0:
26ec0881	152	return ('Burst read', addr, 1, 99999)
2a743706 M	153	else:
	154	self.warn(pos, 'unknown address/command combination')
	155	else:
	156	if (b & 0x40) == 0x00:
26ec0881	157	return ('Strobe', addr, 0, 0)
2a743706	158	if (b & 0xC0) == 0xC0:
26ec0881	159	return ('Status read', addr, 1, 99999)
2a743706 M	160	else:
	161	self.warn(pos, 'unknown address/command combination')
	162
a782243c	163	def decode_reg(self, pos, ann, regid, data):
2a743706 M	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
42ca52b5	178	name = '{} ({:02X})'.format(regs[regid], regid)
2a743706 M	179	else:
	180	name = regid
	181
c5213e26	182	if regid == 'STATUS' and ann == Ann.STATUS:
2a743706 M	183	label = 'Status'
	184	self.decode_status_reg(pos, ann, data, label)
	185	else:
26ec0881	186	if self.cmd in ('Write', 'Read', 'Status read', 'Burst read', 'Burst write'):
2a743706 M	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
26ec0881	206	if self.cmd in ('Single read', 'Status read', 'Burst read'):
2a743706 M	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
42ca52b5	211	text = '{} = {:02X}'.format(label, status)
2a743706 M	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])
42ca52b5	223	text = '{} = {}'.format(label, data)
2a743706 M	224	self.putp(pos, ann, text)
	225
	226	def finish_command(self, pos):
	227	'''Decodes the remaining data bytes at position 'pos'.'''
	228
26ec0881	229	if self.cmd == 'Write':
c5213e26	230	self.decode_reg(pos, Ann.SINGLE_WRITE, self.dat, self.mosi_bytes())
26ec0881	231	elif self.cmd == 'Burst write':
c5213e26	232	self.decode_reg(pos, Ann.BURST_WRITE, self.dat, self.mosi_bytes())
26ec0881	233	elif self.cmd == 'Read':
c5213e26	234	self.decode_reg(pos, Ann.SINGLE_READ, self.dat, self.miso_bytes())
26ec0881	235	elif self.cmd == 'Burst read':
c5213e26	236	self.decode_reg(pos, Ann.BURST_READ, self.dat, self.miso_bytes())
26ec0881	237	elif self.cmd == 'Strobe':
c5213e26	238	self.decode_reg(pos, Ann.STROBE, self.dat, self.mosi_bytes())
26ec0881	239	elif self.cmd == 'Status read':
c5213e26	240	self.decode_reg(pos, Ann.STATUS_READ, self.dat, self.miso_bytes())
2a743706 M	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:
5ae47601	267	self.finish_command(Pos(self.ss_mb, self.es_mb))
2a743706 M	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
5ae47601	274	pos = Pos(ss, es)
2a743706 M	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.
c5213e26	285	self.decode_reg(pos, Ann.STATUS, 'STATUS', [miso])
2a743706 M	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
5ae47601	294	self.mb.append(Data(mosi, miso))