115в. Python next to the device: port, bytes, log, test script

Firmware is written in C, but everything that sits on the desk next to the board is easier to write in Python: send a hundred commands, parse the answers, write a log, compute a summary and produce a "pass — fail" report. Section 115б traced the path from the program to the port: bridges, port names, DTR and latency. Section 194 covered the port and frames from the Linux side in C, section 196 the thermostat console. Here the same thermostat is tested by a Python script through the pyserial library.

The whole experiment runs in containers. The thermostat logic is _verify/proj196/thermo.c without a single change, plus a new platform plat_qemu.c: USART1, SysTick and the room model from 196. The firmware is built with arm-none-eabi-gcc (the stm-arm:1 image) with the CMSIS header for F407 and run in qemu-system-arm -M netduinoplus2 — an STM32F405 with the same USART1 and SysTick peripherals. The bench.py script runs in the stm-pio:1 image (Python 3.12.14, pyserial 3.5) and connects to the QEMU port over a temporary Docker network.

Sources: the code of section 194 (_verify/linux194/link194.c) as the reference for frames, section 126 (_verify/c126_host.txt) as the reference for byte layouts, section 196 for the console and its answers, the pyserial 3.5 sources from the image. The section's files are in _verify/python115c/, checked by _verify/python115c_check.py.

Terms used in the documentation

Term	Meaning
virtual environment (venv)	a separate set of libraries for one project
requirements file, pinning	a list of dependencies with fixed versions
timeout, inter-byte timeout	read timeout, timeout between bytes
URL handler (socket://, loop://, spy://)	a port given by an address: network, loopback, port with an exchange log
format string (struct)	a format string: byte order, sizes and alignment of fields
test fixture, test double	the test bench, a stand-in device
unsolicited message	a message sent without a request
clock drift	a clock running fast or slow

115в.1 What to give to the script and what to leave in the firmware

There is one rule: everything the device must do without a computer stays in the firmware. During testing the computer only asks and records.

For testing, two service commands were added to the QEMU platform; they never reach the 196 logic: ms — the device time in milliseconds, sensor off and sensor on — "unplug" and restore the sensor. The platform passes all other lines unchanged to thermo_rx, as it was on the BluePill.

If the bench program has to be handed to a technician without Python, it is packed into a single executable file (PyInstaller). If you want Python on the microcontroller itself, that is MicroPython, a different story: an interpreter in the firmware. Neither was tested here.

115в.2 Environment: venv and pinned versions

A bench script must still run a year later with the same libraries. For this the project has its own environment and a list of versions:

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

This is the section's entire requirements.txt, and it is checked against the image: stm-pio:1 has exactly pyserial 3.5. The usual workflow:

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

What the experiment without a network showed:

The lesson for a bench without internet: download the packages into a folder in advance and install from it (pip install --no-index --find-links folder -r requirements.txt), and pin versions with ==, otherwise a new machine gets a different pyserial.

115в.3 pyserial: timeouts, read versus readline, port by address

A port opens in one line, and read behaviour is set by timeout:

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

An experiment on the loop:// loopback (whatever is written is read back right away): after write(b"abcd"), in_waiting holds 4 bytes; read(100) with a 0.3 s timeout returned 4 bytes after 0.30 s — it waited for the missing ones; read(4) returned at once; with timeout = 0 the first read(100) gave 2 bytes, the second 0, both instantly.

readline and read_until behave differently. An experiment on a pseudo-terminal: the bytes of the line 0123456789\n arrive every 0.2 s, timeout 0.5 s.

pyserial takes readline from the standard io module: it reads one byte at a time, and the timeout applies to each byte. As long as the device sends anything at all more often than once every 0.5 s, readline will not return until the end of the line, however long it takes. read_until in pyserial 3.5 counts the timeout for the whole line. And both return a fragment without \n on timeout — you have to tell it apart from a complete line.

Port by address. serial.serial_for_url takes an address instead of a port name, and the rest of the code does not change:

The device in the experiment was started like this:

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

wait=on — QEMU does not start the firmware until the script connects. This matters: otherwise the device's first lines go nowhere (115в.6).

In pyserial 3.5, spy:// is a wrapper only over a real port; it cannot be attached to socket://. In the experiment it sat on a pseudo-terminal: the script sent get, the "device" answered with a line of 30 bytes, and the log had 31 lines — one for the send and one for each received byte, because readline reads byte by byte:

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

115в.4 Bytes: struct, byte order, CRC and COBS

struct describes a packet with a format string. The first character is the byte order and alignment, then come the fields: B — uint8_t, H — uint16_t, I — uint32_t, h — int16_t. A comparison with the layouts of section 126 (there, what the C compilers produced):

Three lessons. < and > mean no alignment, like a packed structure; for packets use only these. @ aligns fields like the compiler of the machine where Python runs, but does not add the tail: @BIH is 10 bytes, while sizeof of the same structure in C is 12; the tail is added by 0I at the end. And struct fills padding with zeros, while C leaves the old memory contents in it, so a "raw" structure from the device can be compared with one packed on the PC only field by field.

The letter sets the sign: the two bytes FF 38 read as >h give −200 (−20.0 °C in tenths), read as >H give 65336. struct will not truncate a number out of range, it refuses: 'H' format requires 0 <= number <= 65535. struct has no bit fields — they are assembled with shifts: 1 | 2 << 1 | 17 << 3 = 0x8D, the same word that the compilers produced in section 126.

The frames of section 194 in Python — frames.py, the same algorithm line for line:

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

Comparison with link194.c — byte for byte:

Beyond the 194 vectors — 1000 random packets from 0 to 300 bytes (every tenth has no zeros at all): the C function frame from link194.c and Python produced identical frames in all 1000 cases. The frames.Receiver receiver took back all 856 packets of length 1 to 254 bytes; longer ones it drops by length, just like the 194 receiver.

You do not have to write the CRC by hand:

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

With an initial value of 0xFFFF the same function gives 0x29B1 — the CRC-16/CCITT-FALSE of the thermostat settings in 196. One polynomial, different initial values: mixing them up is a common cause of "the CRC does not match". crc_hqx is written in C, and in the experiment it was hundreds of times faster than a loop in pure Python.

115в.5 Driving the device console from a program: command, answer, retry, refusal

The 196 console was written for a human: the answer is a UTF-8 line ending in \r\n, there is no command number in the answer, and an alarm arrives by itself, without a request. The script must take this into account. The core of 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)

The answer is recognised by a pattern: for get — t=15.1 set=21.0 нагрев, for ms — ms=451, for set — ok. A line starting with ? is a refusal from the device, and it is not a communication error: retrying is pointless. Everything else is unsolicited messages (АВАРИЯ: датчик, датчик в норме); they pile up in events with a timestamp.

Retry happens only on silence. It was tested on a stand-in port: the stub device "did not hear" the first command — the answer came on the second attempt, with 2 writes to the port; a stub that is always silent gave three writes and NoAnswer after 0.9 s (three waits of 0.3 s each). Only commands that are safe to execute twice may be retried: get and set 22.5 can be, "add one degree" cannot.

What was tested on the device in QEMU:

The last two rows are the key point. One command can get two answers, and the extra one, if not collected, goes to the next command. That is why after a refusal ask waits for silence. And an unfinished line — from a previous session or from a person typing in a terminal — spoils the first command. The cure is to send an empty line at the start of a session: the 196 console skips empty lines, the garbage gets its own refusal, and that goes to events:

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

In the experiment, after xx and sync, exactly one refusal landed in events, and the next get got its own answer.

115в.6 The log, PC time versus device time, the test report

The log is CSV: it opens in anything and is read by the csv module without extra libraries. Every 0.25 s the script asks ms and 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)]

The start of the log from a run:

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,нагрев

Two times in every row. pc_s is the PC clock, dev_ms is the device clock. The PC timestamp is the midpoint between the request and the answer: the answer did not arrive instantly, and the midpoint is closest to the moment the device read its clock. For timestamps use time.monotonic(), not time.time(): the PC system clock can jump during synchronisation, the monotonic one cannot. The summary fits the line dev = a · pc + b through all rows: a is how many times faster the device clock runs, b is the offset. If a is not 1, the device has the wrong idea of its own frequency, and all its "60 seconds" are not 60.

The summary — without matplotlib: minimum, maximum, share of time heating, a, and the chart journal.svg — a single polyline of points, twenty lines of code in pure Python.

The report marks each test ГОДЕН (pass) or НЕ ГОДЕН (fail) with details, and at the end gives the verdict and exit code 0 or 1, so that the bench can be run from a build pipeline (section 110). Device A is as in 196, with a correct clock:

ГОДЕН    прибор загрузился с заводскими настройками
ГОДЕН    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 s of logging gives 41 rows; heating is on the whole time: the room starts at 15 °C with a setpoint of 21. The alarm came by itself about 3 s after sensor off — that is how long the 196 logic waits for a plausible reading. The device and PC clocks diverged by less than 0.1 %.

Device B is the same firmware with two bugs that only the bench catches: SysTick is calculated for 16 MHz (the HSI frequency after reset), while the core in QEMU netduinoplus2 runs at 168 MHz; the port was started with wait=off, and the firmware started without waiting for the script:

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

The cfg: заводские line went to the port before the connection — and was lost. The device clock runs about 10 times too fast: by calculation 168 / 16 = 10.5, in the recorded run it was 10.34; the number varies slightly from run to run but stays below 10.5 — the emulator does not keep exact timing for such a frequent interrupt. For the bench the exact number does not matter; what matters is that a several-fold discrepancy is visible from the first seconds. Device B's room "warmed up" to 18 °C in 3 s: its model lives by its wrong clock.

115в.7 Pitfalls

readline hangs even though the timeout is 0.5 s. The timeout applies to each byte, and the device sends data more often. For lines use read_until; its timeout covers the whole line.

A line "arrived", but without an end. Both readline and read_until return a fragment on timeout. A complete line is only one that ends with \n.

The answers "slipped" by one command. The device answered twice (a line longer than 31 characters gave two refusals) or sent an alarm without a request. Recognise the answer by a pattern, set foreign lines aside, and after a refusal wait for silence.

The first command of a session gets a refusal. Garbage from the previous session is left in the device's line buffer. Send an empty line at the start of the session.

UnicodeDecodeError in the middle of an answer. The answer t=15.1 set=21.0 нагрев\r\n is 30 bytes but 24 characters: each Cyrillic letter takes two bytes. Of the 29 places where these bytes can be cut, 6 split a letter in half. Decode the whole line, not pieces from read(n), and with errors="replace".

The device's first lines are lost. The script connected later than the device started talking: in QEMU — wait=off, on a board — a reset before the port was opened. In QEMU use wait=on; on a board open the port first and only then reset the device (or ask for its state with a command).

@BIH in struct is 10 bytes, but the C structure is 12. @ does not add tail padding. For packets use only < or >.

The Python CRC does not match the device. The same polynomial with a different initial value: crc_hqx(data, 0) is XMODEM (194), crc_hqx(data, 0xFFFF) is CCITT-FALSE (196).

spy://socket://… will not open. In pyserial 3.5, spy:// works only over a real port.

ModuleNotFoundError: serial in a clean venv. The environment has its own libraries: run pip install -r requirements.txt inside it.

115в.8 Summary

• The firmware gets everything the device does by itself (the alarm switches heating off after 3 s without a computer); the script gets the test order, the log, the summary and the report.
• Environment: venv and pyserial==3.5 in requirements.txt; the version is checked against the image; without a network the package cannot be installed into a clean venv — prepare in advance.
• timeout defines all reading: read(100) at 0.3 s returned 4 bytes after 0.30 s; readline with a byte every 0.2 s waited 2.0 s, read_until 0.6 s.
• serial_for_url: socket:// to QEMU, loop:// for tests, spy:// as an exchange log of a real port.
• struct: <BIH is 7 bytes, like the packed structure of 126; @BIH is 10, not 12. COBS + CRC frames in Python matched the C code of 194 on all vectors and on 1000 random packets byte for byte.
• Console from a program: answer by pattern, do not retry a ? refusal, foreign lines go to events, an empty line at the start of a session; one long line gave two refusals.
• A CSV log with two times: device A's clock matched the PC to better than 0.1 %, device B with SysTick set for 16 MHz instead of 168 runs about 10 times faster; the bench returned НЕ ГОДЕН (fail).

Full sources

frames.py — the frames of section 194 in Python: CRC-16/XMODEM and 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 — the thermostat test script using 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]))

Next: section 115г "Exchange formats: TLV, protobuf (nanopb) and CBOR on a microcontroller".