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

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
##
## 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

class ChannelError(Exception):
    pass

regs = {
#   addr: ('name',        size)
    0x00: ('CONFIG',      1),
    0x01: ('EN_AA',       1),
    0x02: ('EN_RXADDR',   1),
    0x03: ('SETUP_AW',    1),
    0x04: ('SETUP_RETR',  1),
    0x05: ('RF_CH',       1),
    0x06: ('RF_SETUP',    1),
    0x07: ('STATUS',      1),
    0x08: ('OBSERVE_TX',  1),
    0x09: ('RPD',         1),
    0x0a: ('RX_ADDR_P0',  5),
    0x0b: ('RX_ADDR_P1',  5),
    0x0c: ('RX_ADDR_P2',  1),
    0x0d: ('RX_ADDR_P3',  1),
    0x0e: ('RX_ADDR_P4',  1),
    0x0f: ('RX_ADDR_P5',  1),
    0x10: ('TX_ADDR',     5),
    0x11: ('RX_PW_P0',    1),
    0x12: ('RX_PW_P1',    1),
    0x13: ('RX_PW_P2',    1),
    0x14: ('RX_PW_P3',    1),
    0x15: ('RX_PW_P4',    1),
    0x16: ('RX_PW_P5',    1),
    0x17: ('FIFO_STATUS', 1),
    0x1c: ('DYNPD',       1),
    0x1d: ('FEATURE',     1),
}

xn297_regs = {
    0x19: ('DEMOD_CAL',   5),
    0x1e: ('RF_CAL',      7),
    0x1f: ('BB_CAL',      5),
}

class Decoder(srd.Decoder):
    api_version = 2
    id = 'nrf24l01'
    name = 'nRF24L01(+)'
    longname = 'Nordic Semiconductor nRF24L01/nRF24L01+'
    desc = '2.4GHz transceiver chip.'
    license = 'gplv2+'
    inputs = ['spi']
    outputs = ['nrf24l01']
    options = (
        {'id': 'chip', 'desc': 'Chip type',
            'default': 'nrf24l01', 'values': ('nrf24l01', 'xn297')},
    )
    annotations = (
        # Sent from the host to the chip.
        ('cmd', 'Commands sent to the device'),
        ('tx-data', 'Payload sent to the device'),

        # Returned by the chip.
        ('register', 'Registers read from the device'),
        ('rx-data', 'Payload read from the device'),

        ('warning', 'Warnings'),
    )
    ann_cmd = 0
    ann_tx = 1
    ann_reg = 2
    ann_rx = 3
    ann_warn = 4
    annotation_rows = (
        ('commands', 'Commands', (ann_cmd, ann_tx)),
        ('responses', 'Responses', (ann_reg, ann_rx)),
        ('warnings', 'Warnings', (ann_warn,)),
    )

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

    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)
        if self.options['chip'] == 'xn297':
            regs.update(xn297_regs)

    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 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.mb_s = -1
        self.mb_e = -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 ('W_REGISTER', 'ACTIVATE'):
            # Don't output anything now, the command is merged with
            # the data bytes following it.
            self.mb_s = pos[0]
        else:
            self.putp(pos, self.ann_cmd, self.format_command())

    def format_command(self):
        '''Returns the label for the current command.'''
        if self.cmd == 'R_REGISTER':
            reg = regs[self.dat][0] if self.dat in regs else 'unknown register'
            return 'Cmd R_REGISTER "{}"'.format(reg)
        else:
            return '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
        '''

        if (b & 0xe0) in (0b00000000, 0b00100000):
            c = 'R_REGISTER' if not (b & 0xe0) else 'W_REGISTER'
            d = b & 0x1f
            m = regs[d][1] if d in regs else 1
            return (c, d, 1, m)
        if b == 0b01010000:
            # nRF24L01 only
            return ('ACTIVATE', None, 1, 1)
        if b == 0b01100001:
            return ('R_RX_PAYLOAD', None, 1, 32)
        if b == 0b01100000:
            return ('R_RX_PL_WID', None, 1, 1)
        if b == 0b10100000:
            return ('W_TX_PAYLOAD', None, 1, 32)
        if b == 0b10110000:
            return ('W_TX_PAYLOAD_NOACK', None, 1, 32)
        if (b & 0xf8) == 0b10101000:
            return ('W_ACK_PAYLOAD', b & 0x07, 1, 32)
        if b == 0b11100001:
            return ('FLUSH_TX', None, 0, 0)
        if b == 0b11100010:
            return ('FLUSH_RX', None, 0, 0)
        if b == 0b11100011:
            return ('REUSE_TX_PL', None, 0, 0)
        if b == 0b11111111:
            return ('NOP', None, 0, 0)

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

        pos   -- start and end sample numbers of the register
        ann   -- is 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  -- is 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 = regs[regid][0]
        else:
            name = regid

        # Multi byte register come LSByte first.
        data = reversed(data)

        if self.cmd == 'W_REGISTER' and ann == self.ann_cmd:
            # The 'W_REGISTER' command is merged with the following byte(s).
            label = '{}: {}'.format(self.format_command(), name)
        else:
            label = 'Reg {}'.format(name)

        self.decode_mb_data(pos, ann, data, label, True)

    def decode_mb_data(self, pos, ann, data, label, always_hex):
        '''Decodes the data bytes 'data' of a multibyte command at position
        'pos'. The decoded data is prefixed with 'label'. If 'always_hex' is
        True, all bytes are decoded as hex codes, otherwise only non
        printable characters are escaped.'''

        if always_hex:
            def escape(b):
                return '{:02X}'.format(b)
        else:
            def escape(b):
                c = chr(b)
                if not str.isprintable(c):
                    return '\\x{:02X}'.format(b)
                return c

        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 == 'R_REGISTER':
            self.decode_register(pos, self.ann_reg,
                                 self.dat, self.miso_bytes())
        elif self.cmd == 'W_REGISTER':
            self.decode_register(pos, self.ann_cmd,
                                 self.dat, self.mosi_bytes())
        elif self.cmd == 'R_RX_PAYLOAD':
            self.decode_mb_data(pos, self.ann_rx,
                                self.miso_bytes(), 'RX payload', False)
        elif (self.cmd == 'W_TX_PAYLOAD' or
              self.cmd == 'W_TX_PAYLOAD_NOACK'):
            self.decode_mb_data(pos, self.ann_tx,
                                self.mosi_bytes(), 'TX payload', False)
        elif self.cmd == 'W_ACK_PAYLOAD':
            lbl = 'ACK payload for pipe {}'.format(self.dat)
            self.decode_mb_data(pos, self.ann_tx,
                                self.mosi_bytes(), lbl, False)
        elif self.cmd == 'R_RX_PL_WID':
            msg = 'Payload width = {}'.format(self.mb[0][1])
            self.putp(pos, self.ann_reg, msg)
        elif self.cmd == 'ACTIVATE':
            self.putp(pos, self.ann_cmd, self.format_command())
            if self.mosi_bytes()[0] != 0x73:
                self.warn(pos, 'wrong data for "ACTIVATE" 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 send 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.mb_s, self.mb_e))

                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_reg, '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.mb_s == -1:
                        self.mb_s = ss
                    self.mb_e = es
                    self.mb.append((mosi, miso))
Commit	Line	Data
	1	##
	2	## This file is part of the libsigrokdecode project.
	3	##
	4	## Copyright (C) 2014 Jens Steinhauser <jens.steinhauser@gmail.com>
	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
	22	class ChannelError(Exception):
	23	pass
	24
	25	regs = {
	26	# addr: ('name', size)
	27	0x00: ('CONFIG', 1),
	28	0x01: ('EN_AA', 1),
	29	0x02: ('EN_RXADDR', 1),
	30	0x03: ('SETUP_AW', 1),
	31	0x04: ('SETUP_RETR', 1),
	32	0x05: ('RF_CH', 1),
	33	0x06: ('RF_SETUP', 1),
	34	0x07: ('STATUS', 1),
	35	0x08: ('OBSERVE_TX', 1),
	36	0x09: ('RPD', 1),
	37	0x0a: ('RX_ADDR_P0', 5),
	38	0x0b: ('RX_ADDR_P1', 5),
	39	0x0c: ('RX_ADDR_P2', 1),
	40	0x0d: ('RX_ADDR_P3', 1),
	41	0x0e: ('RX_ADDR_P4', 1),
	42	0x0f: ('RX_ADDR_P5', 1),
	43	0x10: ('TX_ADDR', 5),
	44	0x11: ('RX_PW_P0', 1),
	45	0x12: ('RX_PW_P1', 1),
	46	0x13: ('RX_PW_P2', 1),
	47	0x14: ('RX_PW_P3', 1),
	48	0x15: ('RX_PW_P4', 1),
	49	0x16: ('RX_PW_P5', 1),
	50	0x17: ('FIFO_STATUS', 1),
	51	0x1c: ('DYNPD', 1),
	52	0x1d: ('FEATURE', 1),
	53	}
	54
	55	xn297_regs = {
	56	0x19: ('DEMOD_CAL', 5),
	57	0x1e: ('RF_CAL', 7),
	58	0x1f: ('BB_CAL', 5),
	59	}
	60
	61	class Decoder(srd.Decoder):
	62	api_version = 2
	63	id = 'nrf24l01'
	64	name = 'nRF24L01(+)'
	65	longname = 'Nordic Semiconductor nRF24L01/nRF24L01+'
	66	desc = '2.4GHz transceiver chip.'
	67	license = 'gplv2+'
	68	inputs = ['spi']
	69	outputs = ['nrf24l01']
	70	options = (
	71	{'id': 'chip', 'desc': 'Chip type',
	72	'default': 'nrf24l01', 'values': ('nrf24l01', 'xn297')},
	73	)
	74	annotations = (
	75	# Sent from the host to the chip.
	76	('cmd', 'Commands sent to the device'),
	77	('tx-data', 'Payload sent to the device'),
	78
	79	# Returned by the chip.
	80	('register', 'Registers read from the device'),
	81	('rx-data', 'Payload read from the device'),
	82
	83	('warning', 'Warnings'),
	84	)
	85	ann_cmd = 0
	86	ann_tx = 1
	87	ann_reg = 2
	88	ann_rx = 3
	89	ann_warn = 4
	90	annotation_rows = (
	91	('commands', 'Commands', (ann_cmd, ann_tx)),
	92	('responses', 'Responses', (ann_reg, ann_rx)),
	93	('warnings', 'Warnings', (ann_warn,)),
	94	)
	95
	96	def __init__(self):
	97	self.next()
	98	self.requirements_met = True
	99	self.cs_was_released = False
	100
	101	def start(self):
	102	self.out_ann = self.register(srd.OUTPUT_ANN)
	103	if self.options['chip'] == 'xn297':
	104	regs.update(xn297_regs)
	105
	106	def warn(self, pos, msg):
	107	'''Put a warning message 'msg' at 'pos'.'''
	108	self.put(pos[0], pos[1], self.out_ann, [self.ann_warn, [msg]])
	109
	110	def putp(self, pos, ann, msg):
	111	'''Put an annotation message 'msg' at 'pos'.'''
	112	self.put(pos[0], pos[1], self.out_ann, [ann, [msg]])
	113
	114	def next(self):
	115	'''Resets the decoder after a complete command was decoded.'''
	116	# 'True' for the first byte after CS went low.
	117	self.first = True
	118
	119	# The current command, and the minimum and maximum number
	120	# of data bytes to follow.
	121	self.cmd = None
	122	self.min = 0
	123	self.max = 0
	124
	125	# Used to collect the bytes after the command byte
	126	# (and the start/end sample number).
	127	self.mb = []
	128	self.mb_s = -1
	129	self.mb_e = -1
	130
	131	def mosi_bytes(self):
	132	'''Returns the collected MOSI bytes of a multi byte command.'''
	133	return [b[0] for b in self.mb]
	134
	135	def miso_bytes(self):
	136	'''Returns the collected MISO bytes of a multi byte command.'''
	137	return [b[1] for b in self.mb]
	138
	139	def decode_command(self, pos, b):
	140	'''Decodes the command byte 'b' at position 'pos' and prepares
	141	the decoding of the following data bytes.'''
	142	c = self.parse_command(b)
	143	if c is None:
	144	self.warn(pos, 'unknown command')
	145	return
	146
	147	self.cmd, self.dat, self.min, self.max = c
	148
	149	if self.cmd in ('W_REGISTER', 'ACTIVATE'):
	150	# Don't output anything now, the command is merged with
	151	# the data bytes following it.
	152	self.mb_s = pos[0]
	153	else:
	154	self.putp(pos, self.ann_cmd, self.format_command())
	155
	156	def format_command(self):
	157	'''Returns the label for the current command.'''
	158	if self.cmd == 'R_REGISTER':
	159	reg = regs[self.dat][0] if self.dat in regs else 'unknown register'
	160	return 'Cmd R_REGISTER "{}"'.format(reg)
	161	else:
	162	return 'Cmd {}'.format(self.cmd)
	163
	164	def parse_command(self, b):
	165	'''Parses the command byte.
	166
	167	Returns a tuple consisting of:
	168	- the name of the command
	169	- additional data needed to dissect the following bytes
	170	- minimum number of following bytes
	171	- maximum number of following bytes
	172	'''
	173
	174	if (b & 0xe0) in (0b00000000, 0b00100000):
	175	c = 'R_REGISTER' if not (b & 0xe0) else 'W_REGISTER'
	176	d = b & 0x1f
	177	m = regs[d][1] if d in regs else 1
	178	return (c, d, 1, m)
	179	if b == 0b01010000:
	180	# nRF24L01 only
	181	return ('ACTIVATE', None, 1, 1)
	182	if b == 0b01100001:
	183	return ('R_RX_PAYLOAD', None, 1, 32)
	184	if b == 0b01100000:
	185	return ('R_RX_PL_WID', None, 1, 1)
	186	if b == 0b10100000:
	187	return ('W_TX_PAYLOAD', None, 1, 32)
	188	if b == 0b10110000:
	189	return ('W_TX_PAYLOAD_NOACK', None, 1, 32)
	190	if (b & 0xf8) == 0b10101000:
	191	return ('W_ACK_PAYLOAD', b & 0x07, 1, 32)
	192	if b == 0b11100001:
	193	return ('FLUSH_TX', None, 0, 0)
	194	if b == 0b11100010:
	195	return ('FLUSH_RX', None, 0, 0)
	196	if b == 0b11100011:
	197	return ('REUSE_TX_PL', None, 0, 0)
	198	if b == 0b11111111:
	199	return ('NOP', None, 0, 0)
	200
	201	def decode_register(self, pos, ann, regid, data):
	202	'''Decodes a register.
	203
	204	pos -- start and end sample numbers of the register
	205	ann -- is the annotation number that is used to output the register.
	206	regid -- may be either an integer used as a key for the 'regs'
	207	dictionary, or a string directly containing a register name.'
	208	data -- is the register content.
	209	'''
	210
	211	if type(regid) == int:
	212	# Get the name of the register.
	213	if regid not in regs:
	214	self.warn(pos, 'unknown register')
	215	return
	216	name = regs[regid][0]
	217	else:
	218	name = regid
	219
	220	# Multi byte register come LSByte first.
	221	data = reversed(data)
	222
	223	if self.cmd == 'W_REGISTER' and ann == self.ann_cmd:
	224	# The 'W_REGISTER' command is merged with the following byte(s).
	225	label = '{}: {}'.format(self.format_command(), name)
	226	else:
	227	label = 'Reg {}'.format(name)
	228
	229	self.decode_mb_data(pos, ann, data, label, True)
	230
	231	def decode_mb_data(self, pos, ann, data, label, always_hex):
	232	'''Decodes the data bytes 'data' of a multibyte command at position
	233	'pos'. The decoded data is prefixed with 'label'. If 'always_hex' is
	234	True, all bytes are decoded as hex codes, otherwise only non
	235	printable characters are escaped.'''
	236
	237	if always_hex:
	238	def escape(b):
	239	return '{:02X}'.format(b)
	240	else:
	241	def escape(b):
	242	c = chr(b)
	243	if not str.isprintable(c):
	244	return '\\x{:02X}'.format(b)
	245	return c
	246
	247	data = ''.join([escape(b) for b in data])
	248	text = '{} = "{}"'.format(label, data)
	249	self.putp(pos, ann, text)
	250
	251	def finish_command(self, pos):
	252	'''Decodes the remaining data bytes at position 'pos'.'''
	253
	254	if self.cmd == 'R_REGISTER':
	255	self.decode_register(pos, self.ann_reg,
	256	self.dat, self.miso_bytes())
	257	elif self.cmd == 'W_REGISTER':
	258	self.decode_register(pos, self.ann_cmd,
	259	self.dat, self.mosi_bytes())
	260	elif self.cmd == 'R_RX_PAYLOAD':
	261	self.decode_mb_data(pos, self.ann_rx,
	262	self.miso_bytes(), 'RX payload', False)
	263	elif (self.cmd == 'W_TX_PAYLOAD' or
	264	self.cmd == 'W_TX_PAYLOAD_NOACK'):
	265	self.decode_mb_data(pos, self.ann_tx,
	266	self.mosi_bytes(), 'TX payload', False)
	267	elif self.cmd == 'W_ACK_PAYLOAD':
	268	lbl = 'ACK payload for pipe {}'.format(self.dat)
	269	self.decode_mb_data(pos, self.ann_tx,
	270	self.mosi_bytes(), lbl, False)
	271	elif self.cmd == 'R_RX_PL_WID':
	272	msg = 'Payload width = {}'.format(self.mb[0][1])
	273	self.putp(pos, self.ann_reg, msg)
	274	elif self.cmd == 'ACTIVATE':
	275	self.putp(pos, self.ann_cmd, self.format_command())
	276	if self.mosi_bytes()[0] != 0x73:
	277	self.warn(pos, 'wrong data for "ACTIVATE" command')
	278
	279	def decode(self, ss, es, data):
	280	if not self.requirements_met:
	281	return
	282
	283	ptype, data1, data2 = data
	284
	285	if ptype == 'CS-CHANGE':
	286	if data1 is None:
	287	if data2 is None:
	288	self.requirements_met = False
	289	raise ChannelError('CS# pin required.')
	290	elif data2 == 1:
	291	self.cs_was_released = True
	292
	293	if data1 == 0 and data2 == 1:
	294	# Rising edge, the complete command is transmitted, process
	295	# the bytes that were send after the command byte.
	296	if self.cmd:
	297	# Check if we got the minimum number of data bytes
	298	# after the command byte.
	299	if len(self.mb) < self.min:
	300	self.warn((ss, ss), 'missing data bytes')
	301	elif self.mb:
	302	self.finish_command((self.mb_s, self.mb_e))
	303
	304	self.next()
	305	self.cs_was_released = True
	306	elif ptype == 'DATA' and self.cs_was_released:
	307	mosi, miso = data1, data2
	308	pos = (ss, es)
	309
	310	if miso is None or mosi is None:
	311	self.requirements_met = False
	312	raise ChannelError('Both MISO and MOSI pins required.')
	313
	314	if self.first:
	315	self.first = False
	316	# First MOSI byte is always the command.
	317	self.decode_command(pos, mosi)
	318	# First MISO byte is always the status register.
	319	self.decode_register(pos, self.ann_reg, 'STATUS', [miso])
	320	else:
	321	if not self.cmd or len(self.mb) >= self.max:
	322	self.warn(pos, 'excess byte')
	323	else:
	324	# Collect the bytes after the command byte.
	325	if self.mb_s == -1:
	326	self.mb_s = ss
	327	self.mb_e = es
	328	self.mb.append((mosi, miso))