templates/tools/dsview/decoders/set_uart/pd.py

# Декодирование потоков UART протоколов SET по накопленным байтам. Поиск сигнатуры и проверка
# сообщения должны переживать границу входного фрагмента; неверная контрольная сумма не
# выдаётся как успешный пакет. Этот адаптер переводит события libsigrokdecode в аннотации
# DSView; координаты выборок пересчитываются с учётом частоты захвата.

import sigrokdecode as srd
from common.setgui_decoders.set_uart import StreamParser
from common.setgui_decoders.template_protocols import LEGACY_PROJECTS  # Проекты CAN внутри моста выбираются из того же каталога templates.


class Decoder(srd.Decoder):
    api_version = 3
    id = 'set_uart'
    name = 'SET UART'
    longname = 'SET v2 / SETGUI v1 / ProtoCAN bridge'
    desc = 'SET frames, fields and CRC over UART.'
    license = 'gplv2+'
    inputs = ['uart']
    outputs = ['set']
    tags = ['Embedded/industrial']
    options = (
        {'id': 'protocol', 'desc': 'Protocol', 'default': 'auto',
         'values': ('auto', 'v2', 'v1', 'bridge')},
        {'id': 'gap_ms', 'desc': 'Reset incomplete frame after gap (ms, 0 disables)', 'default': 100},
        {'id': 'can_protocol', 'desc': 'Bridge CAN protocol', 'default': 'protocan', 'values': ('protocan', 'auto', 'devboard', 'balsam', 'ump-protocan', 'deif', 'legacy')},  # Профиль применяется к прикладным кадрам, переданным транспортом AA55.
        {'id': 'project', 'desc': 'Bridge legacy CAN project', 'default': LEGACY_PROJECTS[0], 'values': LEGACY_PROJECTS},  # Исторический проект выбирается явно, поскольку его ID могут пересекаться с другими форматами.
        {'id': 'ump_map', 'desc': 'Bridge UMP map', 'default': 'gas167', 'values': ('gas167', 'assigned')},  # Разделяет текущую локальную карту и назначенные глобальные GAS ID.
        {'id': 'ump_node', 'desc': 'Bridge UMP node', 'default': 1, 'values': (1, 2)},  # Номер УМП задаётся отдельно от локального адреса регистра.
    )
    annotations = (('frame', 'Frame'), ('payload', 'Payload'), ('error', 'Error'), ('field', 'Field'))  # Индексы прежних аннотаций сохранены, подробные поля добавлены в конец.
    annotation_rows = (('frames', 'Frames', (0,)), ('payload', 'Payload', (1,)), ('errors', 'Errors', (2,)), ('fields', 'Fields', (3,)))  # Поля прикладного парсера видны на отдельной дорожке.

    def __init__(self):
        self.reset()

    # Состояние принадлежит одному запуску декодера. Сброс не должен
    # переносить незавершённый кадр или частоту из предыдущей записи.
    def reset(self):
        self.parsers = {}
        self.last = {}
        self.samplerate = None

    # DSView передаёт время в номерах отсчётов. Частота нужна общему ядру
    # для перевода тайм-аутов и измерений в физические единицы.
    def metadata(self, key, value):
        if key == srd.SRD_CONF_SAMPLERATE:
            self.samplerate = value

    # Выходы регистрируются в жизненном цикле sigrok, а не в конструкторе.
    # Аннотации предназначены для дорожек, структурированные события — для стека.
    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)
        self.out_python = self.register(srd.OUTPUT_PYTHON)

    def emit(self, events, direction):
        for ss, es, frame, error in events:
            prefix = ('RX', 'TX')[direction] + ' '
            self.put(ss, es, self.out_ann, [2 if error else 0, [prefix + (error or frame['summary'])]])
            if frame:
                if frame['protocol'] == 'ProtoCAN bridge' and not frame['flags'] & 10:  # Флаги RTR и ERR исключают прикладные данные; STD допускается для профиля DEIF.
                    # Imported only at decode time: both packages are installed together.
                    from common.setgui_decoders.template_protocols import decode_can  # Мост использует те же прикладные парсеры, что анализатор SET CAN.
                    try:
                        application = decode_can(frame['can_id'], frame['payload'], self.options['can_protocol'], extended=bool(frame['flags'] & 1), project=self.options['project'], ump_map=self.options['ump_map'], ump_node=self.options['ump_node'])  # Флаг IDE и выбранные настройки передаются общему парсеру без изменения wire-данных.
                        frame['application'] = application
                        if application:  # Выбранный профиль может не распознать чужой CAN ID.
                            self.put(ss, es, self.out_ann, [0, [application['summary']]])  # Прикладная сводка дополняет сведения о транспортном кадре моста.
                            for title, value in application.get('fields', []):  # Детали CAN внутри UART берутся из того же результата templates.
                                self.put(ss, es, self.out_ann, [3, ['%s: %s' % (title, value)]])  # Поля доступны без отдельного импорта записи в CAN-монитор.
                    except ValueError as exc:
                        self.put(ss, es, self.out_ann, [2, [str(exc)]])
                for title, value in frame.get('fields', []):  # Стековые UART-профили могут возвращать готовые прикладные поля.
                    self.put(ss, es, self.out_ann, [3, [prefix + '%s: %s' % (title, value)]])  # Префикс RX/TX сохраняет направление каждого значения.
                frame['direction'] = direction
                self.put(ss, es, self.out_python, ['FRAME', frame])
                if frame['payload']:
                    self.put(ss, es, self.out_ann, [1, [prefix + ' '.join('%02X' % b for b in frame['payload'])]])

    def make_parser(self):  # Фабрика позволяет использовать ту же UART-границу с другим переносимым потоковым парсером.
        return StreamParser(self.options['protocol'])  # Каждый экземпляр хранит байты только одного направления.

    def decode(self, ss, es, data):
        kind, direction, payload = data
        if direction not in (0, 1):
            return
        # RX и TX имеют независимые парсеры: перемежающиеся байты двух направлений
        # нельзя объединять в один прикладной кадр.
        if direction not in self.parsers:  # В отличие от setdefault, проверка не создаёт и не отбрасывает парсер на каждом UART-событии.
            self.parsers[direction] = self.make_parser()  # Парсер создаётся один раз на направление и сохраняет незавершённый кадр.
        parser = self.parsers[direction]  # RX и TX продолжают использовать независимые буферы.
        if kind in ('INVALID STOPBIT', 'INVALID STARTBIT', 'PARITY ERROR', 'BREAK'):
            self.emit(parser.flush(), direction)
            self.put(ss, es, self.out_ann, [2, [kind]])
            return
        if kind != 'FRAME':
            return
        if not payload[1]:
            self.emit(parser.flush(), direction)
            self.put(ss, es, self.out_ann, [2, ['Invalid UART frame']])
            return
        gap = float(self.options['gap_ms'])
        if self.samplerate and gap > 0 and direction in self.last and ss - self.last[direction] > self.samplerate * gap / 1000:
            self.emit(parser.flush(), direction)
        self.last[direction] = es
        if not 0 <= payload[0] <= 255:
            self.emit(parser.flush(), direction)
            self.put(ss, es, self.out_ann, [2, ['Configure UART for 8 data bits']])
            return
        self.emit(parser.feed(payload[0], ss, es), direction)