##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2022 Gerhard Sittig <gerhard.sittig@gmx.net>
##
## 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/>.
##
"""
OUTPUT_PYTHON format:
Packet:
(<ptype>, <pdata>)
This is the list of <ptype> codes and their respective <pdata> values:
- 'HEADER': The data is the header byte's value.
- 'PROPORTIONAL': The data is a tuple of the channel number (1-based)
and the channel's value.
- 'DIGITAL': The data is a tuple of the channel number (1-based)
and the channel's value.
- 'FLAG': The data is a tuple of the flag's name, and the flag's value.
- 'FOOTER': The data is the footer byte's value.
"""
import sigrokdecode as srd
from common.srdhelper import bitpack_lsb
class Ann:
HEADER, PROPORTIONAL, DIGITAL, FRAME_LOST, FAILSAFE, FOOTER, \
WARN = range(7)
FLAG_LSB = FRAME_LOST
class Decoder(srd.Decoder):
api_version = 3
id = 'sbus_futaba'
name = 'SBUS (Futaba)'
longname = 'Futaba SBUS (Serial bus)'
desc = 'Serial bus for hobby remote control by Futaba'
license = 'gplv2+'
inputs = ['uart']
outputs = ['sbus_futaba']
tags = ['Remote Control']
options = (
{'id': 'prop_val_min', 'desc': 'Proportional value lower boundary', 'default': 0},
{'id': 'prop_val_max', 'desc': 'Proportional value upper boundary', 'default': 2047},
)
annotations = (
('header', 'Header'),
('proportional', 'Proportional'),
('digital', 'Digital'),
('framelost', 'Frame Lost'),
('failsafe', 'Failsafe'),
('footer', 'Footer'),
('warning', 'Warning'),
)
annotation_rows = (
('framing', 'Framing', (Ann.HEADER, Ann.FOOTER,
Ann.FRAME_LOST, Ann.FAILSAFE)),
('channels', 'Channels', (Ann.PROPORTIONAL, Ann.DIGITAL)),
('warnings', 'Warnings', (Ann.WARN,)),
)
def __init__(self):
self.bits_accum = []
self.sent_fields = None
self.msg_complete = None
self.failed = None
self.reset()
def reset(self):
self.bits_accum.clear()
self.sent_fields = 0
self.msg_complete = False
self.failed = None
def start(self):
self.out_ann = self.register(srd.OUTPUT_ANN)
self.out_py = self.register(srd.OUTPUT_PYTHON)
def putg(self, ss, es, data):
# Put a graphical annotation.
self.put(ss, es, self.out_ann, data)
def putpy(self, ss, es, data):
# Pass Python to upper layers.
self.put(ss, es, self.out_py, data)
def get_ss_es_bits(self, bitcount):
# Get start/end times, and bit values of given length.
# Gets all remaining data when 'bitcount' is None.
if bitcount is None:
bitcount = len(self.bits_accum)
if len(self.bits_accum) < bitcount:
return None, None, None
bits = self.bits_accum[:bitcount]
self.bits_accum = self.bits_accum[bitcount:]
ss, es = bits[0][1], bits[-1][2]
bits = [b[0] for b in bits]
return ss, es, bits
def flush_accum_bits(self):
# Valid data was queued. See if we got full SBUS fields so far.
# Annotate them early, cease inspection of failed messages. The
# implementation is phrased to reduce the potential for clipboard
# errors: 'upto' is the next supported field count, 'want' is one
# field's bit count. Grab as many as we find in an invocation.
upto = 0
if self.failed:
return
# Annotate the header byte. Not seeing the expected bit pattern
# emits a warning annotation, but by design won't fail the SBUS
# message. It's considered more useful to present the channels'
# values instead. The warning still raises awareness.
upto += 1
want = 8
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
text = ['0x{:02x}'.format(value)]
self.putg(ss, es, [Ann.HEADER, text])
if value != 0x0f:
text = ['Unexpected header', 'Header']
self.putg(ss, es, [Ann.WARN, text])
self.putpy(ss, es, ['HEADER', value])
self.sent_fields += 1
# Annotate the proportional channels' data. Check for user
# provided value range violations. Channel numbers are in
# the 1..18 range (1-based).
upto += 16
want = 11
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
text = ['{:d}'.format(value)]
self.putg(ss, es, [Ann.PROPORTIONAL, text])
if value < self.options['prop_val_min']:
text = ['Low proportional value', 'Low value', 'Low']
self.putg(ss, es, [Ann.WARN, text])
if value > self.options['prop_val_max']:
text = ['High proportional value', 'High value', 'High']
self.putg(ss, es, [Ann.WARN, text])
idx = self.sent_fields - (upto - 16)
ch_nr = 1 + idx
self.putpy(ss, es, ['PROPORTIONAL', (ch_nr, value)])
self.sent_fields += 1
# Annotate the digital channels' data.
upto += 2
want = 1
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
text = ['{:d}'.format(value)]
self.putg(ss, es, [Ann.DIGITAL, text])
idx = self.sent_fields - (upto - 2)
ch_nr = 17 + idx
self.putpy(ss, es, ['DIGITAL', (ch_nr, value)])
self.sent_fields += 1
# Annotate the flags' state. Index starts from LSB.
flag_names = ['framelost', 'failsafe', 'msb']
upto += 2
want = 1
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
text = ['{:d}'.format(value)]
idx = self.sent_fields - (upto - 2)
cls = Ann.FLAG_LSB + idx
self.putg(ss, es, [cls, text])
flg_name = flag_names[idx]
self.putpy(ss, es, ['FLAG', (flg_name, value)])
self.sent_fields += 1
# Warn when flags' padding (bits [7:4]) is unexpexted.
upto += 1
want = 4
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
if value != 0x0:
text = ['Unexpected MSB flags', 'Flags']
self.putg(ss, es, [Ann.WARN, text])
flg_name = flag_names[-1]
self.putpy(ss, es, ['FLAG', (flg_name, value)])
self.sent_fields += 1
# Annotate the footer byte. Warn when unexpected.
upto += 1
want = 8
while self.sent_fields < upto:
if len(self.bits_accum) < want:
return
ss, es, bits = self.get_ss_es_bits(want)
value = bitpack_lsb(bits)
text = ['0x{:02x}'.format(value)]
self.putg(ss, es, [Ann.FOOTER, text])
if value != 0x00:
text = ['Unexpected footer', 'Footer']
self.putg(ss, es, [Ann.WARN, text])
self.putpy(ss, es, ['FOOTER', value])
self.sent_fields += 1
# Check for the completion of an SBUS message. Warn when more
# UART data is seen after the message. Defer the warning until
# more bits were collected, flush at next IDLE or BREAK, which
# spans all unprocessed data, and improves perception.
if self.sent_fields >= upto:
self.msg_complete = True
if self.msg_complete and self.bits_accum:
self.failed = ['Excess data bits', 'Excess']
def handle_bits(self, ss, es, bits):
# UART data bits were seen. Store them, validity is yet unknown.
self.bits_accum.extend(bits)
def handle_frame(self, ss, es, value, valid):
# A UART frame became complete. Get its validity. Process its bits.
if not valid:
self.failed = ['Invalid data', 'Invalid']
self.flush_accum_bits()
def handle_idle(self, ss, es):
# An IDLE period was seen in the UART level. Flush, reset state.
if self.bits_accum and not self.failed:
self.failed = ['Unprocessed data bits', 'Unprocessed']
if self.bits_accum and self.failed:
ss, es, _ = self.get_ss_es_bits(None)
self.putg(ss, es, [Ann.WARN, self.failed])
self.reset()
def handle_break(self, ss, es):
# A BREAK period was seen in the UART level. Warn, reset state.
break_ss, break_es = ss, es
if not self.failed:
self.failed = ['BREAK condition', 'Break']
# Re-use logic for "annotated bits warning".
self.handle_idle(None, None)
# Unconditionally annotate BREAK as warning.
text = ['BREAK condition', 'Break']
self.putg(ss, es, [Ann.WARN, text])
self.reset()
def decode(self, ss, es, data):
# Implementor's note: Expects DATA bits to arrive before FRAME
# validity. Either of IDLE or BREAK terminates an SBUS message.
ptype, rxtx, pdata = data
if ptype == 'DATA':
_, bits = pdata
self.handle_bits(ss, es, bits)
elif ptype == 'FRAME':
value, valid = pdata
self.handle_frame(ss, es, value, valid)
elif ptype == 'IDLE':
self.handle_idle(ss, es)
elif ptype == 'BREAK':
self.handle_break(ss, es)