115в. Python рядом с прибором: порт, байты, журнал, скрипт-испытатель

Прошивку пишут на C, а вот всё, что стоит на столе рядом с платой, удобнее писать на Python: отправить сотню команд, разобрать ответы, записать журнал, посчитать сводку и выдать протокол «годен — не годен». Раздел 115б провёл путь от программы до порта: мосты, имена портов, DTR и задержки. Раздел 194 разобрал порт и кадры со стороны Linux на C, раздел 196 — консоль термостата. Здесь тот же термостат проверяет скрипт на Python через библиотеку pyserial.

Опыт целиком в контейнерах. Логика термостата — _verify/proj196/thermo.c без единой правки, к ней новая платформа plat_qemu.c: USART1, SysTick и модель комнаты из 196. Прошивка собрана arm-none-eabi-gcc (образ stm-arm:1) с заголовком CMSIS для F407 и запущена в qemu-system-arm -M netduinoplus2 — это STM32F405 с той же периферией USART1 и SysTick. Скрипт bench.py работает в образе stm-pio:1 (Python 3.12.14, pyserial 3.5) и подключается к порту QEMU через временную сеть Docker.

Опоры: код раздела 194 (_verify/linux194/link194.c) — эталон кадров, раздел 126 (_verify/c126_host.txt) — эталон раскладки байтов, раздел 196 — консоль и её ответы, исходники pyserial 3.5 из образа. Файлы раздела — _verify/python115c/, сверяет _verify/python115c_check.py.

Как это называется в документации

Оригинал	Перевод
virtual environment (venv)	виртуальное окружение: свой набор библиотек для проекта
requirements file, pinning	список зависимостей с закреплёнными версиями
timeout, inter-byte timeout	тайм-аут чтения, тайм-аут между байтами
URL handler (socket://, loop://, spy://)	порт по адресу: сеть, петля, порт с журналом обмена
format string (struct)	строка формата: порядок байт, размеры и выравнивание полей
test fixture, test double	стенд, подставной прибор
unsolicited message	сообщение без запроса
clock drift	уход часов

115в.1 Что отдать скрипту, а что оставить прошивке

Правило одно: всё, что прибор должен делать без компьютера, остаётся в прошивке. Компьютер во время испытаний только спрашивает и записывает.

Для испытаний платформе QEMU добавлены две служебные команды, до логики 196 они не доходят: ms — время прибора в миллисекундах, sensor off и sensor on — «выдернуть» и вернуть датчик. Остальные строки платформа целиком передаёт в thermo_rx, как было на BluePill.

Если программу для стенда надо отдать технику без Python, её собирают в один исполняемый файл (PyInstaller). Если хочется Python на самом микроконтроллере — это MicroPython, другая история: интерпретатор в прошивке. Ни то ни другое здесь не проверялось.

115в.2 Окружение: venv и закреплённые версии

Скрипт стенда должен через год запускаться с теми же библиотеками. Для этого у проекта своё окружение и список версий:

pyserial==3.5        # как в stm-pio:1; обновлять осознанно, вместе с проверкой

Это весь requirements.txt раздела, и он сверен с образом: в stm-pio:1 стоит ровно pyserial 3.5. Обычный порядок работы:

python -m venv .venv
.venv\Scripts\activate            (Windows; в Linux — source .venv/bin/activate)
pip install -r requirements.txt

Что показал опыт без сети:

Вывод для стенда без интернета: пакеты заранее скачивают в каталог и ставят из него (pip install --no-index --find-links каталог -r requirements.txt), а версии закрепляют ==, иначе на новой машине приедет другая pyserial.

115в.3 pyserial: тайм-ауты, read против readline, порт по адресу

Порт открывается одной строкой, а поведение чтения задаёт timeout:

dev = serial.Serial("COM5", 115200, timeout=0.5)     # или "/dev/ttyUSB0"

Опыт на петле loop:// (записанное тут же читается обратно): после write(b"abcd") в in_waiting 4 байта; read(100) при тайм-ауте 0,3 с вернул 4 байта через 0,30 с — ждал недостающие; read(4) вернулся сразу; при timeout = 0 первое read(100) дало 2 байта, второе — 0, оба мгновенно.

readline и read_until ведут себя по-разному. Опыт на псевдотерминале: байты строки 0123456789\n приходят каждые 0,2 с, тайм-аут 0,5 с.

readline pyserial берёт у стандартного io: он читает по одному байту, и тайм-аут действует на каждый байт. Пока прибор присылает хоть что-то чаще раза в 0,5 с, readline не вернётся до конца строки, сколько бы она ни шла. read_until в pyserial 3.5 считает тайм-аут на всю строку. И оба при тайм-ауте отдают обрывок без \n — его надо отличать от целой строки.

Порт по адресу. serial.serial_for_url принимает вместо имени порта адрес, и остальной код не меняется:

Прибор в опыте запущен так:

qemu-system-arm -M netduinoplus2 -nographic -kernel fw.elf -serial tcp::5555,server=on,wait=on -monitor none

wait=on — QEMU не запускает прошивку, пока не подключится скрипт. Это важно: иначе первые строки прибора уходят в пустоту (115в.6).

spy:// в pyserial 3.5 — надстройка только над настоящим портом, к socket:// его не прицепить. В опыте он стоял на псевдотерминале: скрипт отправил get, «прибор» ответил строкой из 30 байт, в журнале 31 строка — одна на отправку и по одной на каждый принятый байт, потому что readline читает по байту:

TX 0000 67 65 74 0A get.
RX 0000 74 t
RX 0000 3D =

115в.4 Байты: struct, порядок байт, CRC и COBS

struct описывает пакет строкой формата. Первый символ — порядок байт и выравнивание, дальше поля: B — uint8_t, H — uint16_t, I — uint32_t, h — int16_t. Сверка с раскладками раздела 126 (там — то, что сделали компиляторы C):

Три урока. < и > — без выравнивания, как упакованная структура; для пакетов берут только их. @ выравнивает поля как компилятор машины, где запущен Python, но не добавляет хвост: @BIH — 10 байт, а sizeof той же структуры в C — 12; хвост дописывает 0I в конце. И заполнение struct забивает нулями, а C оставляет в нём старое содержимое памяти, поэтому сравнивать «сырую» структуру прибора с упакованной на ПК можно только по полям.

Знак задаёт буква: два байта FF 38 по >h — −200 (−20,0 °C в десятых), по >H — 65336. Число вне диапазона struct не обрежет, а откажет: 'H' format requires 0 <= number <= 65535. Битовых полей в struct нет — их собирают сдвигами: 1 | 2 << 1 | 17 << 3 = 0x8D, то же слово, что у компиляторов в разделе 126.

Кадры раздела 194 на Python — frames.py, тот же алгоритм строка в строку:

def frame(data: bytes) -> bytes:
    """кадр = COBS(данные + CRC старшим байтом вперёд) + 00"""
    return cobs_encode(data + crc16(data).to_bytes(2, "big")) + b"\x00"

Сверка с link194.c — байт в байт:

Сверх векторов 194 — 1000 случайных пакетов от 0 до 300 байт (в каждом десятом нет нулей вовсе): C-функция frame из link194.c и Python дали одинаковые кадры во всех 1000 случаях. Приёмник frames.Receiver принял обратно все 856 пакетов длиной от 1 до 254 байт; длиннее он, как приёмник 194, отбрасывает по длине.

CRC писать вручную не обязательно:

def crc16_fast(data: bytes) -> int:
    """То же из стандартной библиотеки: crc_hqx с начальным 0 — это XMODEM."""
    return binascii.crc_hqx(data, 0)

С начальным значением 0xFFFF та же функция даёт 0x29B1 — CRC-16/CCITT-FALSE настроек термостата 196. Один полином, разные начальные значения: путать их — частая причина «CRC не сходится». crc_hqx написан на C и в опыте быстрее цикла на чистом Python в сотни раз.

115в.5 Консоль прибора программой: команда, ответ, повтор, отказ

Консоль 196 писалась для человека: ответ — строка UTF-8 с \r\n в конце, номера команды в ответе нет, а авария приходит сама, без запроса. Скрипт должен это учитывать. Ядро bench.py:

    def line(self):
        """Одна строка без \\r\\n или None по тайм-ауту; обрывок без \\n тоже None."""
        raw = self.port.read_until(b"\n")
        if not raw.endswith(b"\n"):
            self.tail = raw                 # обрывок оставить для разбора ошибок
            return None
        return raw.rstrip(b"\r\n").decode("utf-8", errors="replace")

    def ask(self, cmd, expect, wait=1.0, retries=2):
        """Команда → ответ по образцу. «?» — Refused; чужие строки — в events; нет ответа — повтор."""
        self.drain()
        for attempt in range(1 + retries):
            self.port.write(cmd.encode("ascii") + b"\n")
            end = time.monotonic() + wait
            while time.monotonic() < end:
                s = self.line()
                if s is None:
                    continue
                if s.startswith("?"):
                    self.drain(quiet=True)          # у ответа нет номера команды: хвост отказа не должен
                    raise Refused(s)                # достаться следующей команде
                m = expect.match(s)
                if m:
                    self.attempts = attempt + 1
                    return m
                self.events.append((time.monotonic(), s))
        raise NoAnswer(cmd)

Ответ узнаётся по образцу: на get — t=15.1 set=21.0 нагрев, на ms — ms=451, на set — ok. Строка с ? — отказ прибора, и это не ошибка связи: повторять бессмысленно. Всё остальное — сообщения без запроса (АВАРИЯ: датчик, датчик в норме), они копятся в events с меткой времени.

Повтор — только при молчании. Проверен на подставном порте: прибор- заглушка «не услышал» первую команду — ответ пришёл со второй попытки, записей в порт 2; заглушка, которая молчит всегда, — три записи и NoAnswer через 0,9 с (три ожидания по 0,3 с). Повторять можно только команды, которые безопасно выполнить дважды: get и set 22.5 — можно, «прибавить градус» — нельзя.

Что было проверено на приборе в QEMU:

Две последние строки — главное. Одна команда может получить два ответа, и лишний, если его не забрать, достанется следующей команде. Поэтому после отказа ask дожидается тишины. А недописанная строка — от прошлого сеанса или от человека, набиравшего в терминале, — портит первую команду. Лечение — в начале сеанса отправить пустую строку: консоль 196 пустые строки пропускает, мусор получает свой отказ, и он уходит в events:

    def sync(self):
        """Начало сеанса: пустая строка закрывает то, что недописал прошлый сеанс; ответ на это — в events."""
        self.port.write(b"\n")
        self.drain(quiet=True)

В опыте после xx и sync в events лёг ровно один отказ, и следующий get получил свой ответ.

115в.6 Журнал, время ПК против времени прибора, протокол испытаний

Журнал — CSV: открывается чем угодно и читается модулем csv без библиотек. Раз в 0,25 с скрипт спрашивает ms и get:

            a = time.monotonic()
            ms = int(dev.ask("ms", RE_MS).group(1))
            b = time.monotonic()
            g = dev.ask("get", RE_GET)
            row = [round((a + b) / 2 - t0, 3), ms, float(g.group(1)), float(g.group(2)), g.group(3)]

Начало журнала из прогона:

pc_s,dev_ms,t_c,set_c,state
0.026,204,14.9,21.0,нагрев
0.25,451,14.9,21.0,нагрев
0.5,700,15.0,21.0,нагрев

Два времени в каждой строке. pc_s — часы ПК, dev_ms — часы прибора. Метка ПК — середина между запросом и ответом: ответ пришёл не мгновенно, и середина ближе всего к моменту, когда прибор читал свои часы. Для меток берут time.monotonic(), а не time.time(): системные часы ПК могут прыгнуть при синхронизации, монотонные — нет. Сводка считает прямую dev = a · pc + b по всем строкам: a — во сколько раз часы прибора быстрее, b — сдвиг. Если a не 1, прибор неправильно знает свою частоту, и все его «60 секунд» — не 60.

Сводка — без matplotlib: минимум, максимум, доля времени нагрева, a, и график journal.svg — одна ломаная из точек, двадцать строк кода на чистом Python.

Протокол — каждое испытание ГОДЕН или НЕ ГОДЕН с подробностью, в конце итог и код выхода 0 или 1, чтобы стенд можно было запускать из сборочного конвейера (раздел 110). Прибор A — как в 196, часы верные:

ГОДЕН    прибор загрузился с заводскими настройками
ГОДЕН    get: уставка по умолчанию 21.0 — t=14.9 set=21.0 нагрев
ГОДЕН    set 22.5 принят и читается обратно — t=15.2 set=22.5 нагрев
ГОДЕН    set 99 — отказ — ? get | set 5.0…35.0 | hyst 0.1…5.0
ГОДЕН    hyst abc — отказ — ? get | set 5.0…35.0 | hyst 0.1…5.0
ГОДЕН    строка длиннее 31 символа — отказ — ? строка длиннее 31 символа; ещё отказов вдогонку: 1
ГОДЕН    отказ датчика: авария и возврат — авария через 3.02 с
ГОДЕН    часы прибора совпадают с часами ПК в пределах 2 % — 1.0001
строк журнала 41, t 14.9…15.2 °C, нагрев 100%, часы прибора/ПК 1.0001
ИТОГ: ГОДЕН

За 10 с журнала — 41 строка; нагрев включён всё время: комната стартует с 15 °C при уставке 21. Авария пришла сама примерно через 3 с после sensor off — столько логика 196 ждёт правдоподобного показания. Часы прибора и ПК разошлись меньше чем на 0,1 %.

Прибор B — та же прошивка с двумя ошибками, которые ловит только стенд: SysTick посчитан на 16 МГц (частота HSI после сброса), а ядро в QEMU netduinoplus2 работает на 168 МГц; порт запущен с wait=off, и прошивка стартовала, не дожидаясь скрипта:

НЕ ГОДЕН прибор загрузился с заводскими настройками — AssertionError: нет строки о заводских настройках
НЕ ГОДЕН часы прибора совпадают с часами ПК в пределах 2 % — AssertionError: часы прибора идут в 10.341 раза быстрее ПК
строк журнала 13, t 18.2…18.5 °C, нагрев 100%, часы прибора/ПК 10.3411
ИТОГ: НЕ ГОДЕН

Строка cfg: заводские ушла в порт до подключения — и пропала. Часы прибора бегут примерно в 10 раз быстрее: по расчёту 168 / 16 = 10,5, в записанном прогоне — 10,34, от прогона к прогону число чуть гуляет, но остаётся меньше 10,5 — эмулятор не держит точный ход такого частого прерывания. Для стенда важно не точное число, а то, что расхождение в разы видно с первых секунд. Комната прибора B за 3 с «нагрелась» до 18 °C: его модель живёт по его неправильным часам.

115в.7 Грабли

readline висит, хотя тайм-аут 0,5 с. Тайм-аут действует на каждый байт, а прибор шлёт данные чаще. Для строк — read_until, у него тайм-аут на всю строку.

Строка «пришла», но без конца. И readline, и read_until по тайм-ауту отдают обрывок. Целая строка — только та, что кончается на \n.

Ответы «съехали» на одну команду. Прибор ответил дважды (строка длиннее 31 символа дала два отказа) или прислал аварию без запроса. Ответ узнавать по образцу, чужие строки — в сторону, после отказа — дождаться тишины.

Первая команда сеанса получает отказ. В строке прибора остался мусор от прошлого сеанса. Пустая строка в начале сеанса.

UnicodeDecodeError посреди ответа. Ответ t=15.1 set=21.0 нагрев\r\n — 30 байт, но 24 символа: кириллица занимает по два байта. Из 29 мест, где можно разрезать эти байты, 6 рвут букву пополам. Декодировать целую строку, а не куски от read(n), и с errors="replace".

Первые строки прибора пропали. Скрипт подключился позже, чем прибор начал говорить: у QEMU — wait=off, у платы — сброс раньше открытия порта. У QEMU — wait=on; у платы — открыть порт и только потом сбросить прибор (или спросить состояние командой).

@BIH в struct — 10 байт, а структура C — 12. @ не добавляет хвостовое заполнение. Для пакетов — только < или >.

CRC Python не сходится с прибором. Тот же полином с другим начальным значением: crc_hqx(данные, 0) — XMODEM (194), crc_hqx(данные, 0xFFFF) — CCITT-FALSE (196).

spy://socket://… не открывается. spy:// в pyserial 3.5 — только над настоящим портом.

В чистом venv ModuleNotFoundError: serial. У окружения свои библиотеки: pip install -r requirements.txt внутри него.

115в.8 Итоги

• Прошивке — всё, что прибор делает сам (авария выключает нагрев через 3 с без компьютера); скрипту — порядок испытаний, журнал, сводка и протокол.
• Окружение — venv и pyserial==3.5 в requirements.txt; версия сверена с образом, без сети пакет в чистый venv не поставить — готовить заранее.
• timeout определяет всё чтение: read(100) при 0,3 с вернул 4 байта через 0,30 с; readline при байте каждые 0,2 с ждал 2,0 с, read_until — 0,6 с.
• serial_for_url: socket:// к QEMU, loop:// для тестов, spy:// — журнал обмена настоящего порта.
• struct: <BIH — 7 байт, как упакованная структура 126; @BIH — 10, а не 12. Кадры COBS + CRC на Python совпали с C-кодом 194 на всех векторах и на 1000 случайных пакетах байт в байт.
• Консоль программой: ответ по образцу, отказ ? не повторять, чужие строки в events, пустая строка в начале сеанса; одна длинная строка дала два отказа.
• Журнал CSV с двумя временами: прибор A — часы совпали с ПК точнее 0,1 %, прибор B с SysTick на 16 МГц вместо 168 — примерно в 10 раз быстрее; стенд вынес «НЕ ГОДЕН».

Полные исходники

frames.py — кадры раздела 194 на Python: CRC-16/XMODEM и COBS.

"""Раздел 115в: кадры раздела 194 на Python — CRC-16/XMODEM, COBS, разделитель 00.
Байт в байт как link194.c: сверяет exp115.py на векторах 194 и на случайных пакетах."""
import binascii


def crc16(data: bytes) -> int:
    """CRC-16/XMODEM: полином 0x1021, начальное 0."""
    c = 0
    for b in data:
        c ^= b << 8
        for _ in range(8):
            c = ((c << 1) ^ 0x1021) & 0xFFFF if c & 0x8000 else (c << 1) & 0xFFFF
    return c


def crc16_fast(data: bytes) -> int:
    """То же из стандартной библиотеки: crc_hqx с начальным 0 — это XMODEM."""
    return binascii.crc_hqx(data, 0)


def cobs_encode(data: bytes) -> bytes:
    out = bytearray(b"\x00")          # место под первый код
    code_at, code = 0, 1
    for b in data:
        if b == 0:
            out[code_at] = code
            code_at, code = len(out), 1
            out.append(0)
        else:
            out.append(b)
            code += 1
            if code == 0xFF:
                out[code_at] = code
                code_at, code = len(out), 1
                out.append(0)
    out[code_at] = code
    return bytes(out)


def cobs_decode(data: bytes) -> bytes:
    out, r = bytearray(), 0
    while r < len(data):
        code = data[r]
        r += 1
        if code == 0 or r + code - 1 > len(data):
            raise ValueError("испорченный COBS")
        out += data[r:r + code - 1]
        r += code - 1
        if code != 0xFF and r < len(data):
            out.append(0)
    return bytes(out)


def frame(data: bytes) -> bytes:
    """кадр = COBS(данные + CRC старшим байтом вперёд) + 00"""
    return cobs_encode(data + crc16(data).to_bytes(2, "big")) + b"\x00"


class Receiver:
    """Копит байты до нуля, раскодирует COBS, сверяет CRC. Испорченное — выбросить и ждать следующий 00."""

    LIMIT = 258                       # 254 байта данных + CRC после COBS, как в 194

    def __init__(self):
        self.acc, self.bad = bytearray(), 0

    def feed(self, chunk: bytes) -> list:
        frames = []
        for b in chunk:
            if b:
                self.acc.append(b)
                continue
            raw, self.acc = bytes(self.acc), bytearray()
            try:
                d = cobs_decode(raw) if len(raw) <= self.LIMIT else b""
            except ValueError:
                d = b""
            if len(d) >= 3 and crc16(d[:-2]) == int.from_bytes(d[-2:], "big"):
                frames.append(d[:-2])
            else:
                self.bad += 1
        return frames

bench.py — скрипт-испытатель термостата через pyserial.

"""Раздел 115в: скрипт-испытатель термостата 196 через pyserial.

    python bench.py socket://localhost:5555 out/     # QEMU с -serial tcp::5555,server
    python bench.py COM5 out/                          # настоящий прибор через USB-UART
    python bench.py /dev/ttyUSB0 out/

Пишет out/protocol.txt (испытания: ГОДЕН / НЕ ГОДЕН), out/journal.csv (журнал) и out/journal.svg
(график без matplotlib). Код выхода: 0 — все испытания пройдены."""
import csv, os, re, sys, time

import serial

RE_GET = re.compile(r"t=(-?\d+\.\d) set=(-?\d+\.\d) (\S+)$")
RE_MS = re.compile(r"ms=(\d+)$")
RE_OK = re.compile(r"ok$")


class Refused(Exception):
    """Прибор ответил «?» — команду не понял или значение вне диапазона."""


class NoAnswer(Exception):
    """Ответа нет и после повторов."""


class Device:
    def __init__(self, port):
        self.port = port                    # объект pyserial или что угодно с read_until и write
        self.events = []                    # строки, пришедшие без запроса: (время ПК, текст)

    @classmethod
    def open(cls, url, timeout=0.2):
        dev = cls(serial.serial_for_url(url, baudrate=115200, timeout=timeout))
        dev.sync()
        return dev

    def sync(self):
        """Начало сеанса: пустая строка закрывает то, что недописал прошлый сеанс; ответ на это — в events."""
        self.port.write(b"\n")
        self.drain(quiet=True)

    def line(self):
        """Одна строка без \\r\\n или None по тайм-ауту; обрывок без \\n тоже None."""
        raw = self.port.read_until(b"\n")
        if not raw.endswith(b"\n"):
            self.tail = raw                 # обрывок оставить для разбора ошибок
            return None
        return raw.rstrip(b"\r\n").decode("utf-8", errors="replace")

    def drain(self, quiet=False):
        """Забрать то, что уже пришло, в events; quiet — ещё и дождаться тишины (строка не пришла за тайм-аут)."""
        while quiet or self.port.in_waiting:
            s = self.line()
            if s is None:
                break
            self.events.append((time.monotonic(), s))

    def ask(self, cmd, expect, wait=1.0, retries=2):
        """Команда → ответ по образцу. «?» — Refused; чужие строки — в events; нет ответа — повтор."""
        self.drain()
        for attempt in range(1 + retries):
            self.port.write(cmd.encode("ascii") + b"\n")
            end = time.monotonic() + wait
            while time.monotonic() < end:
                s = self.line()
                if s is None:
                    continue
                if s.startswith("?"):
                    self.drain(quiet=True)          # у ответа нет номера команды: хвост отказа не должен
                    raise Refused(s)                # достаться следующей команде
                m = expect.match(s)
                if m:
                    self.attempts = attempt + 1
                    return m
                self.events.append((time.monotonic(), s))
        raise NoAnswer(cmd)

    def wait_event(self, text, wait):
        """Дождаться строки без запроса, например «АВАРИЯ: датчик»."""
        end = time.monotonic() + wait
        while time.monotonic() < end:
            for i, (t, s) in enumerate(self.events):
                if s == text:
                    del self.events[i]
                    return t
            s = self.line()
            if s is not None:
                self.events.append((time.monotonic(), s))
        return None


def journal(dev, path, seconds, period):
    """Раз в period секунд: время прибора и показание. Метка ПК — середина между запросом и ответом."""
    rows, t0 = [], time.monotonic()
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["pc_s", "dev_ms", "t_c", "set_c", "state"])
        n = 0
        while True:
            due = t0 + n * period
            if due - t0 > seconds:
                break
            time.sleep(max(0.0, due - time.monotonic()))
            a = time.monotonic()
            ms = int(dev.ask("ms", RE_MS).group(1))
            b = time.monotonic()
            g = dev.ask("get", RE_GET)
            row = [round((a + b) / 2 - t0, 3), ms, float(g.group(1)), float(g.group(2)), g.group(3)]
            w.writerow(row)
            rows.append(row)
            n += 1
    return rows


def summary(rows):
    pc = [r[0] for r in rows]
    dev = [r[1] / 1000 for r in rows]
    t = [r[2] for r in rows]
    n = len(rows)
    mp, md = sum(pc) / n, sum(dev) / n                      # наклон прямой dev = a·pc + b
    a = sum((x - mp) * (y - md) for x, y in zip(pc, dev)) / sum((x - mp) ** 2 for x in pc)
    return {"rows": n, "t_min": min(t), "t_max": max(t), "heat_share": sum(r[4] == "нагрев" for r in rows) / n,
            "clock_ratio": a, "offset_s": md - a * mp}


def svg(rows, path, w=600, h=200):
    """График температуры по времени прибора: одна ломаная, оси подписаны. Без matplotlib."""
    xs = [r[1] / 1000 for r in rows]
    ys = [r[2] for r in rows]
    x0, x1, y0, y1 = min(xs), max(xs), min(ys) - 0.1, max(ys) + 0.1
    X = lambda x: 40 + (x - x0) / (x1 - x0 or 1) * (w - 50)
    Y = lambda y: h - 25 - (y - y0) / (y1 - y0) * (h - 40)
    pts = " ".join("%.1f,%.1f" % (X(x), Y(y)) for x, y in zip(xs, ys))
    with open(path, "w", encoding="utf-8") as f:
        f.write('<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" font-size="11">\n' % (w, h))
        f.write('<polyline fill="none" stroke="#c33" stroke-width="1.5" points="%s"/>\n' % pts)
        f.write('<text x="40" y="%d">%.1f с</text><text x="%d" y="%d" text-anchor="end">%.1f с (время прибора)</text>\n'
                % (h - 8, x0, w - 10, h - 8, x1))
        f.write('<text x="4" y="%.1f">%.1f</text><text x="4" y="%.1f">%.1f °C</text>\n</svg>\n' % (Y(y1), y1, Y(y0), y0))


def main(url, out):
    os.makedirs(out, exist_ok=True)
    dev = Device.open(url)
    results = []

    def test(name, fn):
        try:
            detail = fn()
            results.append(("ГОДЕН", name, detail or ""))
        except Exception as e:                              # испытание упало — записать и идти дальше
            results.append(("НЕ ГОДЕН", name, "%s: %s" % (type(e).__name__, e)))

    def boot():
        t = dev.wait_event("cfg: заводские", 2.0)
        assert t is not None, "нет строки о заводских настройках"

    def get_default():
        g = dev.ask("get", RE_GET)
        assert g.group(2) == "21.0", g.group(0)
        return g.group(0)

    def set_and_read():
        dev.ask("set 22.5", RE_OK)
        g = dev.ask("get", RE_GET)
        assert g.group(2) == "22.5", g.group(0)
        return g.group(0)

    def refused(cmd):
        def fn():
            try:
                dev.ask(cmd, RE_OK)
            except Refused as e:
                extra = [x for _, x in dev.events if x.startswith("?")]
                dev.events = [e for e in dev.events if not e[1].startswith("?")]
                return str(e) + ("; ещё отказов вдогонку: %d" % len(extra) if extra else "")
            raise AssertionError("прибор принял «%s»" % cmd)
        return fn

    def sensor_fault():
        t_off = time.monotonic()
        dev.ask("sensor off", RE_OK)
        t_alarm = dev.wait_event("АВАРИЯ: датчик", 5.0)
        assert t_alarm is not None, "нет аварии"
        state = dev.ask("get", RE_GET).group(3)
        assert state == "АВАРИЯ", state
        dev.ask("sensor on", RE_OK)
        assert dev.wait_event("датчик в норме", 2.0) is not None, "нет возврата"
        return "авария через %.2f с" % (t_alarm - t_off)

    only_clock = os.environ.get("BENCH_ONLY") == "clock"    # короткий прогон: загрузка и часы
    test("прибор загрузился с заводскими настройками", boot)
    if not only_clock:
        test("get: уставка по умолчанию 21.0", get_default)
    rows = journal(dev, os.path.join(out, "journal.csv"), float(os.environ.get("BENCH_SECONDS", "10")), 0.25)
    s = summary(rows)
    svg(rows, os.path.join(out, "journal.svg"))
    if not only_clock:
        test("set 22.5 принят и читается обратно", set_and_read)
        test("set 99 — отказ", refused("set 99"))
        test("hyst abc — отказ", refused("hyst abc"))
        test("строка длиннее 31 символа — отказ", refused("0123456789012345678901234567890123"))
        test("отказ датчика: авария и возврат", sensor_fault)

    def clock():
        assert abs(s["clock_ratio"] - 1) < 0.02, "часы прибора идут в %.3f раза быстрее ПК" % s["clock_ratio"]
        return "%.4f" % s["clock_ratio"]

    test("часы прибора совпадают с часами ПК в пределах 2 %", clock)
    with open(os.path.join(out, "protocol.txt"), "w", encoding="utf-8") as f:
        for verdict, name, detail in results:
            f.write("%-8s %s%s\n" % (verdict, name, " — " + detail if detail else ""))
        f.write("строк журнала %(rows)d, t %(t_min).1f…%(t_max).1f °C, нагрев %(heat_share).0f%%, "
                "часы прибора/ПК %(clock_ratio).4f\n" % dict(s, heat_share=100 * s["heat_share"]))
        f.write("ИТОГ: %s\n" % ("ГОДЕН" if all(r[0] == "ГОДЕН" for r in results) else "НЕ ГОДЕН"))
    return 0 if all(r[0] == "ГОДЕН" for r in results) else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv[1], sys.argv[2]))

Дальше: раздел 115г «Форматы обмена: TLV, protobuf (nanopb) и CBOR на микроконтроллере».