Andrey/templates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(protocan-py): хостовые кодеки ProtoCAN и каталога на Python

Browse Source 
Собрано из SETGUI/src/gui_desktop/core (protocan, can_transport, protocol,
gas_catalog); в CAN_to_RS485/template/python лежала такая же копия.

Только stdlib: ни Qt, ни pyserial — модулям передают bytes, порт и
таймауты остаются делом вызывающего кода. Кодировщики совпадают побайтово
с c/protocan-transport, что зафиксировано эталонами в его тестах.

Добавлен __init__.py: gas_catalog импортирует protocol относительным
импортом, без пакета копия в CAN_to_RS485/template не собиралась.
This commit is contained in:
Andrey Kruchinkin
2026-08-23 01:15:36 +03:00
parent 34cbd92808
commit 02da823a94
6 changed files with 1388 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
python/protocan/README.md Normal file
View File

@@ -0,0 +1,57 @@
# protocan (Python)
Хостовая сторона протоколов: разбор сообщений **ProtoCAN**, транспортный кадр
моста CAN↔RS485 и кодеки каталога общего адресного пространства.
Только stdlib, Python 3.9+. Ни Qt, ни pyserial: модули не открывают порт и не
знают, откуда пришли байты, — им передают `bytes`. Поэтому они проверяются
обычными host-тестами и одинаково работают в GUI, в скрипте и в CI.
## Состав
| Модуль | Что делает | Зеркало на C |
|---|---|---|
| `protocan.py` | разбор 29-битного идентификатора и тела всех типов сообщений в человекочитаемый вид | `pcan_id.c` |
| `transport.py` | кадр `AA 55 … CRC16`, потоковый разборщик с ресинхронизацией, CRC-16/CCITT-FALSE | `pcan_frame.c` |
| `protocol.py` | кадр `A5 5A` протокола SETGUI: сборка, разбор, CRC32 | `gui_frame.c` |
| `gas_catalog.py` | каталог регистров с именами и подписка на поток значений | `gui_catalog.c` |
Зависимость внутри пакета одна: `gas_catalog` использует `protocol`.
Остальные модули независимы.
Кодировщики C и Python дают побайтово одинаковый результат — это зафиксировано
эталонами в `c/protocan-transport/tests/test_transport.c`.
## Быстрый старт
```python
from protocan import protocan, transport
parser = transport.FrameParser()
for frame in parser.feed(port.read(256)): # bytes из любого источника
print(protocan.decode(frame.can_id, frame.data).summary)
port.write(transport.build_frame(0x18FF0102, b"\x01\x02").encode())
```
Каталог регистров:
```python
from protocan.gas_catalog import ObjectEntry, ObjectType, encode_entry, decode_entry
entry = ObjectEntry(address=0x0100, type=ObjectType.U16, flags=0,
scale_pow10=-1, unit=0, name="Температура")
raw = encode_entry(entry)
assert decode_entry(raw) == entry
```
## Границы
Разбор отделён от ввода-вывода намеренно: порт, таймауты и переподключение —
дело вызывающего кода. Байты внутрь, объекты наружу.
## Проверено в проектах
`SETGUI` (вкладки «Шина CAN» и «Адресное пространство»), `CAN_to_RS485`.
Там пакет лежит внутри `gui_desktop/core`, здесь — источник истины.

1
python/protocan/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Переносимые модули протоколов ProtoCAN и SETGUI: только stdlib."""

317
python/protocan/gas_catalog.py Normal file
View File

@@ -0,0 +1,317 @@
"""Каталог общего адресного пространства и поток выбранных значений.
Прибор объявляет, какие регистры у него есть и как они называются, GUI
выбирает подмножество и получает его пакетами. Схема повторяет реестр
регистров ST Motor Control Workbench.
Модуль не импортирует Qt: кодеки переносимы и проверяются host-тестами.
Двоичный контракт описан в ``protocan-transport/docs/GUI_CATALOG.md`` и
продублирован на C в ``gui/gui_catalog.c``.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import IntEnum
from .protocol import MAX_PAYLOAD_SIZE, ProtocolError, decode_u16, decode_u32, encode_u16
#: Длина одной записи каталога на линии.
ENTRY_SIZE = 32
#: Длина заголовка ответа GAS_CATALOG.
CATALOG_HEADER_SIZE = 6
#: Длина поля имени в записи. 24 байта - это 12 кириллических
#: символов в UTF-8; на 16 байтах не помещалось даже "Температура".
NAME_SIZE = 24
#: Максимум адресов в одной подписке.
WATCH_MAX = 64
#: Сколько записей каталога помещается в один кадр.
ENTRIES_PER_FRAME = (MAX_PAYLOAD_SIZE - CATALOG_HEADER_SIZE) // ENTRY_SIZE
#: Сколько значений помещается в один кадр потока.
VALUES_PER_FRAME = (MAX_PAYLOAD_SIZE - 6) // 2
class ObjectType(IntEnum):
"""Формат значения регистра."""
U16 = 0
I16 = 1
U32 = 2
I32 = 3
BITS = 4
@property
def registers(self) -> int:
"""Сколько подряд идущих адресов занимает значение."""
return 2 if self in (ObjectType.U32, ObjectType.I32) else 1
@property
def signed(self) -> bool:
return self in (ObjectType.I16, ObjectType.I32)
#: Доступ и признаки записи каталога.
FLAG_READABLE = 0x01
FLAG_WRITABLE = 0x02
FLAG_DEFAULT_WATCH = 0x04
#: Единицы измерения; коды входят в wire-контракт и не перенумеровываются.
UNITS: dict[int, str] = {
0: "",
1: "В",
2: "А",
3: "°C",
4: "%",
5: "Гц",
6: "мс",
7: "с",
8: "кбит/с",
9: "шт.",
10: "об/мин",
11: "Вт",
}
@dataclass(frozen=True, slots=True)
class ObjectEntry:
"""@brief Одна запись каталога общего адресного пространства.
@param address Адрес первого регистра значения.
@param type Формат значения.
@param flags Биты доступа ``FLAG_*``.
@param scale_pow10 Показатель степени: физическое = raw * 10^scale.
@param unit Код единицы измерения из ``UNITS``.
@param name Имя для оператора, до 24 байт в UTF-8.
"""
address: int
type: ObjectType
flags: int
scale_pow10: int
unit: int
name: str
@property
def readable(self) -> bool:
return bool(self.flags & FLAG_READABLE)
@property
def writable(self) -> bool:
return bool(self.flags & FLAG_WRITABLE)
@property
def default_watch(self) -> bool:
return bool(self.flags & FLAG_DEFAULT_WATCH)
@property
def unit_text(self) -> str:
return UNITS.get(self.unit, "?")
@property
def registers(self) -> int:
return self.type.registers
def scale(self, raw: int) -> float:
"""@brief Переводит сырое значение в физическую величину.
@param raw Слово или пара слов, уже собранные в целое.
@return Значение с учётом знака и множителя ``10^scale_pow10``.
"""
width = 32 if self.registers == 2 else 16
if self.type.signed and raw >= (1 << (width - 1)):
raw -= 1 << width
return raw * (10.0 ** self.scale_pow10)
def format(self, raw: int) -> str:
"""@brief Готовая подпись значения для таблицы GUI."""
if self.type is ObjectType.BITS:
width = 4 * self.registers
return f"0x{raw:0{width}X}"
value = self.scale(raw)
decimals = max(0, -self.scale_pow10)
text = f"{value:.{decimals}f}"
return f"{text} {self.unit_text}".strip()
def encode_entry(entry: ObjectEntry) -> bytes:
"""@brief Кодирует запись каталога в 32 байта.
@raises ProtocolError Если имя не влезает в 24 байта UTF-8.
"""
name = entry.name.encode("utf-8")
if len(name) > NAME_SIZE:
raise ProtocolError(f"имя '{entry.name}' длиннее {NAME_SIZE} байт в UTF-8")
if not 0 <= entry.address <= 0xFFFF:
raise ProtocolError("адрес вне диапазона 0..0xFFFF")
if not -128 <= entry.scale_pow10 <= 127:
raise ProtocolError("scale_pow10 вне диапазона int8")
return (
encode_u16(entry.address)
+ bytes((int(entry.type) & 0xFF, entry.flags & 0xFF,
entry.scale_pow10 & 0xFF, entry.unit & 0xFF))
+ b"\x00\x00"
+ name.ljust(NAME_SIZE, b"\x00")
)
def decode_entry(data: bytes, offset: int = 0) -> ObjectEntry:
"""@brief Разбирает 32 байта записи каталога.
Неизвестный код типа не роняет разбор: запись остаётся видимой
оператору как ``U16``, иначе новая прошивка сделала бы старый GUI
полностью слепым.
"""
if offset + ENTRY_SIZE > len(data):
raise ProtocolError("payload короче записи каталога")
chunk = data[offset:offset + ENTRY_SIZE]
raw_type = chunk[2]
try:
object_type = ObjectType(raw_type)
except ValueError:
object_type = ObjectType.U16
scale = chunk[4] - 256 if chunk[4] >= 128 else chunk[4]
name = chunk[8:8 + NAME_SIZE].split(b"\x00", 1)[0].decode("utf-8", "replace")
return ObjectEntry(
address=decode_u16(chunk),
type=object_type,
flags=chunk[3],
scale_pow10=scale,
unit=chunk[5],
name=name,
)
def build_catalog_request(start_index: int = 0, max_count: int = 0) -> bytes:
"""@brief Payload запроса каталога.
@param start_index Порядковый номер записи, а не адрес.
@param max_count Сколько записей вернуть; 0 — сколько влезет в кадр.
"""
return encode_u16(start_index) + encode_u16(max_count)
def build_catalog_response(total: int, start_index: int,
entries: list[ObjectEntry]) -> bytes:
"""@brief Payload ответа каталога (нужен mock-режиму и тестам)."""
if len(entries) > ENTRIES_PER_FRAME:
raise ProtocolError(f"в кадр входит не более {ENTRIES_PER_FRAME} записей")
return (encode_u16(total) + encode_u16(start_index) + encode_u16(len(entries))
+ b"".join(encode_entry(item) for item in entries))
@dataclass(frozen=True, slots=True)
class CatalogChunk:
"""@brief Часть каталога из одного кадра.
@param total Полный размер каталога прибора.
@param start_index Индекс первой записи в этом куске.
@param entries Записи в порядке прибора.
"""
total: int
start_index: int
entries: tuple[ObjectEntry, ...]
@property
def complete(self) -> bool:
"""Дошёл ли каталог до конца именно этим куском."""
return self.start_index + len(self.entries) >= self.total
def decode_catalog(payload: bytes) -> CatalogChunk:
"""@brief Разбирает кадр ``GAS_CATALOG`` от прибора."""
if len(payload) < CATALOG_HEADER_SIZE:
raise ProtocolError("GAS_CATALOG короче 6 байт")
total = decode_u16(payload)
start_index = decode_u16(payload, 2)
count = decode_u16(payload, 4)
if len(payload) != CATALOG_HEADER_SIZE + count * ENTRY_SIZE:
raise ProtocolError("GAS_CATALOG содержит неверное число записей")
if start_index + count > total:
raise ProtocolError("GAS_CATALOG выходит за объявленный размер каталога")
entries = tuple(
decode_entry(payload, CATALOG_HEADER_SIZE + index * ENTRY_SIZE)
for index in range(count)
)
return CatalogChunk(total=total, start_index=start_index, entries=entries)
def build_watch_set(period_ms: int, addresses: list[int]) -> bytes:
"""@brief Payload подписки на поток значений.
@param period_ms Период потока; 0 останавливает поток.
@param addresses Адреса в нужном порядке, не больше ``WATCH_MAX``.
"""
if not 0 <= period_ms <= 0xFFFF:
raise ProtocolError("период вне диапазона 0..65535 мс")
if len(addresses) > WATCH_MAX:
raise ProtocolError(f"в подписке не больше {WATCH_MAX} адресов")
return (encode_u16(period_ms) + encode_u16(len(addresses))
+ b"".join(encode_u16(item) for item in addresses))
def decode_watch_set(payload: bytes) -> tuple[int, list[int]]:
"""@brief Разбирает подписку; на стороне прибора и в mock-режиме."""
if len(payload) < 4:
raise ProtocolError("GAS_WATCH_SET короче 4 байт")
period_ms = decode_u16(payload)
count = decode_u16(payload, 2)
if len(payload) != 4 + count * 2:
raise ProtocolError("GAS_WATCH_SET содержит неверное число адресов")
return period_ms, [decode_u16(payload, 4 + index * 2) for index in range(count)]
def decode_watch_ack(payload: bytes) -> tuple[int, int]:
"""@brief Разбирает эхо прибора: принятый период и число адресов."""
if len(payload) < 4:
raise ProtocolError("эхо GAS_WATCH_SET короче 4 байт")
return decode_u16(payload), decode_u16(payload, 2)
def build_watch_data(timestamp_ms: int, values: list[int]) -> bytes:
"""@brief Payload пакета значений (нужен mock-режиму и тестам)."""
if len(values) > VALUES_PER_FRAME:
raise ProtocolError(f"в кадр входит не более {VALUES_PER_FRAME} значений")
return (int(timestamp_ms & 0xFFFFFFFF).to_bytes(4, "little")
+ encode_u16(len(values))
+ b"".join(encode_u16(item & 0xFFFF) for item in values))
def decode_watch_data(payload: bytes) -> tuple[int, list[int]]:
"""@brief Разбирает пакет потока.
@return Время прибора в миллисекундах и значения в порядке подписки.
"""
if len(payload) < 6:
raise ProtocolError("GAS_WATCH_DATA короче 6 байт")
timestamp_ms = decode_u32(payload)
count = decode_u16(payload, 4)
if len(payload) != 6 + count * 2:
raise ProtocolError("GAS_WATCH_DATA содержит неверное число значений")
return timestamp_ms, [decode_u16(payload, 6 + index * 2) for index in range(count)]
def assemble(entries: list[ObjectEntry], values: dict[int, int]) -> dict[int, int]:
"""@brief Собирает многословные значения из потока сырых слов.
@param entries Записи каталога, описывающие ширину каждого значения.
@param values Сырые слова по адресам, как пришли в потоке.
@return Значение по адресу первого слова; ширина учтена.
"""
result: dict[int, int] = {}
for entry in entries:
low = values.get(entry.address)
if low is None:
continue
if entry.registers == 1:
result[entry.address] = low
continue
high = values.get(entry.address + 1)
if high is None:
# Старшее слово не подписано - показывать половину числа хуже,
# чем не показывать ничего.
continue
result[entry.address] = low | (high << 16)
return result

519
python/protocan/protocan.py Normal file
View File

@@ -0,0 +1,519 @@
# -*- coding: utf-8 -*-
"""Разбор и сборка сообщений ProtoCAN — прикладного уровня шины CAN.
Раскладка полей повторяет ``ProtoCanId_t`` из SETCAN/Inc/protocan.h:
битовые поля укладываются от младших бит к старшим.
Модуль не импортирует Qt: разбор переносим и проверяется host-тестами.
Источник контракта — CAN_to_RS485/docs/PROTOCOL.md, §1.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import IntEnum
from typing import List, Optional
# --------------------------------------------------------------------------
# Перечисления протокола
# --------------------------------------------------------------------------
class Priority(IntEnum):
CRITICAL = 0
STANDARD = 1
class Route(IntEnum):
FROM_PM = 0
FROM_DEVICE = 1
class MsgType(IntEnum):
BROADCAST = 0b0000
DISCRETE = 0b0001
ANALOG = 0b0010
GENERAL_ADDRESS_SPACE = 0b0011
MODBUS_COIL = 0b0100
MODBUS_DISCRETE = 0b0101
MODBUS_HOLDING = 0b0110
MODBUS_INPUT = 0b0111
ERROR = 0b1000
PULSE = 0b1111
class BroadcastType(IntEnum):
STATUS = 0
ONOFF = 1
RESTARTDEVICE = 2
RTCSETUP = 3
END = 0xFFF
class DiscreteType(IntEnum):
ACCIDENT = 0
WARNING = 1
CONTROL_SIGNALS = 2
FLAGS = 3
RESET = 4
CHANGE_MODE = 5
REQUEST_LIST_OF_PARAMETERS = 6
END = 0xF
class AnalogType(IntEnum):
UNIVERSAL = 0
SETTINGS = 1
U = 2
I = 3
T = 4
END = 0xF
PRIORITY_RU = {
Priority.CRITICAL: "критический",
Priority.STANDARD: "стандартный",
}
ROUTE_RU = {
Route.FROM_PM: "от мастера (PM)",
Route.FROM_DEVICE: "от устройства",
}
MSGTYPE_RU = {
MsgType.BROADCAST: "Broadcast",
MsgType.DISCRETE: "Discrete",
MsgType.ANALOG: "Analog",
MsgType.GENERAL_ADDRESS_SPACE: "General Address Space",
MsgType.MODBUS_COIL: "Modbus Coil",
MsgType.MODBUS_DISCRETE: "Modbus Discrete",
MsgType.MODBUS_HOLDING: "Modbus Holding",
MsgType.MODBUS_INPUT: "Modbus Input",
MsgType.ERROR: "Error",
MsgType.PULSE: "Pulse",
}
BROADCAST_RU = {
BroadcastType.STATUS: "запрос статуса",
BroadcastType.ONOFF: "вкл/выкл пульса",
BroadcastType.RESTARTDEVICE: "перезапуск устройства",
BroadcastType.RTCSETUP: "установка RTC",
BroadcastType.END: "конец диапазона",
}
DISCRETE_RU = {
DiscreteType.ACCIDENT: "авария",
DiscreteType.WARNING: "предупреждение",
DiscreteType.CONTROL_SIGNALS: "управляющие сигналы",
DiscreteType.FLAGS: "флаги",
DiscreteType.RESET: "сброс",
DiscreteType.CHANGE_MODE: "смена режима",
DiscreteType.REQUEST_LIST_OF_PARAMETERS: "запрос списка параметров",
DiscreteType.END: "конец диапазона",
}
ANALOG_RU = {
AnalogType.UNIVERSAL: "универсальный",
AnalogType.SETTINGS: "уставки",
AnalogType.U: "напряжение U",
AnalogType.I: "ток I",
AnalogType.T: "температура T",
AnalogType.END: "конец диапазона",
}
WEEKDAY_RU = {
1: "Пн", 2: "Вт", 3: "Ср", 4: "Чт", 5: "Пт", 6: "Сб", 7: "Вс", 0: "?",
}
def _name(enum_cls, value, table):
"""Имя элемента перечисления либо «резерв» для неизвестного кода."""
try:
item = enum_cls(value)
except ValueError:
return "резерв 0x%X" % value
return "%s (%s)" % (item.name, table.get(item, ""))
# --------------------------------------------------------------------------
# Идентификатор
# --------------------------------------------------------------------------
@dataclass
class ProtoCanId:
"""Разобранный 29-битный идентификатор."""
raw: int
msg_body: int
msg_type: int
device_id: int
device_type: int
route: int
priority: int
@staticmethod
def parse(raw: int) -> "ProtoCanId":
raw &= 0x1FFFFFFF
return ProtoCanId(
raw=raw,
msg_body=raw & 0xFFFF,
msg_type=(raw >> 16) & 0xF,
device_id=(raw >> 20) & 0xF,
device_type=(raw >> 24) & 0x7,
route=(raw >> 27) & 0x1,
priority=(raw >> 28) & 0x1,
)
@staticmethod
def build(priority: int, route: int, device_type: int, device_id: int,
msg_type: int, msg_body: int) -> int:
return (((priority & 0x1) << 28)
| ((route & 0x1) << 27)
| ((device_type & 0x7) << 24)
| ((device_id & 0xF) << 20)
| ((msg_type & 0xF) << 16)
| (msg_body & 0xFFFF))
@property
def msg_type_name(self) -> str:
try:
return MSGTYPE_RU[MsgType(self.msg_type)]
except ValueError:
return "резерв 0x%X" % self.msg_type
# --------------------------------------------------------------------------
# Раскладки MsgBody
# --------------------------------------------------------------------------
def split_broadcast(body: int):
"""BROADCAST: Body[3:0], Type[15:4]."""
return body & 0xF, (body >> 4) & 0xFFF
def split_discrete(body: int):
"""DISCRETE: Body[11:0], Type[15:12]."""
return body & 0xFFF, (body >> 12) & 0xF
def split_analog(body: int):
"""ANALOG: SensorID[11:0], Type[15:12]."""
return body & 0xFFF, (body >> 12) & 0xF
def split_modbus(body: int):
"""MODBUS: RegCount[3:0], StrAdr[15:4]."""
return (body >> 4) & 0xFFF, body & 0xF
def split_error(body: int):
"""ERROR: ErrorCode[7:0], Info[15:8]."""
return body & 0xFF, (body >> 8) & 0xFF
def merge_broadcast(bcast_type: int, body: int) -> int:
return ((bcast_type & 0xFFF) << 4) | (body & 0xF)
def merge_discrete(disc_type: int, body: int) -> int:
return ((disc_type & 0xF) << 12) | (body & 0xFFF)
def merge_analog(an_type: int, sensor_id: int) -> int:
return ((an_type & 0xF) << 12) | (sensor_id & 0xFFF)
def merge_modbus(str_adr: int, reg_count: int) -> int:
return ((str_adr & 0xFFF) << 4) | (reg_count & 0xF)
def merge_error(info: int, code: int) -> int:
return ((info & 0xFF) << 8) | (code & 0xFF)
def sensor_to_modbus_register(sensor_type: int, sensor_id: int) -> int:
"""Макрос SensorToModbusRegister из protocan.h."""
return ((sensor_type << 11) | sensor_id) & 0xFFFF
def regs_le(data: bytes) -> List[int]:
"""Пары байт -> регистры u16 little-endian (порядок из PROTOCAN_SEND_*)."""
return [data[i] | (data[i + 1] << 8) for i in range(0, len(data) - 1, 2)]
# --------------------------------------------------------------------------
# Результат разбора
# --------------------------------------------------------------------------
@dataclass
class Decoded:
"""Полный разбор одного кадра."""
id: ProtoCanId
dlc: int
data: bytes
ide: bool = True
rtr: bool = False
#: Короткая сводка для колонки таблицы
summary: str = ""
#: Пары (поле, значение) для панели подробностей
fields: List[tuple] = field(default_factory=list)
#: Регистры, если сообщение их несёт
registers: Optional[List[tuple]] = None
#: Замечания о нарушениях протокола
warnings: List[str] = field(default_factory=list)
def _ascii(data: bytes) -> str:
return "".join(chr(b) if 32 <= b < 127 else "." for b in data)
def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Decoded:
"""Разбирает кадр ProtoCAN в структуру Decoded."""
pid = ProtoCanId.parse(raw_id)
data = bytes(data[:8])
res = Decoded(id=pid, dlc=len(data), data=data, ide=ide, rtr=rtr)
res.fields.append(("Priority", "%d — %s" % (
pid.priority, PRIORITY_RU.get(Priority(pid.priority), ""))))
res.fields.append(("Route", "%d — %s" % (
pid.route, ROUTE_RU.get(Route(pid.route), ""))))
res.fields.append(("DeviceType", "%d (0b{:03b})".format(pid.device_type) % pid.device_type))
res.fields.append(("DeviceID", "%d (0b{:04b})".format(pid.device_id) % pid.device_id))
res.fields.append(("MsgType", "0x%X — %s" % (pid.msg_type, pid.msg_type_name)))
res.fields.append(("MsgBody", "0x%04X" % pid.msg_body))
if not ide:
res.warnings.append("Стандартный ID: ProtoCAN использует расширенный (29 бит)")
if rtr:
res.warnings.append("Remote frame: полезная нагрузка отсутствует")
handler = {
MsgType.BROADCAST: _decode_broadcast,
MsgType.DISCRETE: _decode_discrete,
MsgType.ANALOG: _decode_analog,
MsgType.GENERAL_ADDRESS_SPACE: _decode_gas,
MsgType.MODBUS_COIL: _decode_modbus_bit,
MsgType.MODBUS_DISCRETE: _decode_modbus_bit,
MsgType.MODBUS_HOLDING: _decode_modbus_reg,
MsgType.MODBUS_INPUT: _decode_modbus_reg,
MsgType.ERROR: _decode_error,
MsgType.PULSE: _decode_pulse,
}
try:
fn = handler[MsgType(pid.msg_type)]
except ValueError:
res.summary = "Неизвестный MsgType 0x%X" % pid.msg_type
res.warnings.append("MsgType 0x%X не описан в protocan.h" % pid.msg_type)
if data:
res.fields.append(("Data ASCII", _ascii(data)))
return res
fn(res)
return res
def _decode_broadcast(res: Decoded) -> None:
body, btype = split_broadcast(res.id.msg_body)
res.fields.append((" Broadcast.Type", "0x%03X — %s" % (
btype, _name(BroadcastType, btype, BROADCAST_RU))))
res.fields.append((" Broadcast.Body", "0x%X (%d)" % (body, body)))
if btype == BroadcastType.STATUS:
res.summary = "BROADCAST STATUS"
if res.dlc == 7:
h, m, s, yy, mo, dd, wd = res.data
res.summary = "BROADCAST STATUS: 20%02d-%02d-%02d %02d:%02d:%02d" % (
yy, mo, dd, h, m, s)
res.fields += [
(" Время", "%02d:%02d:%02d" % (h, m, s)),
(" Дата", "20%02d-%02d-%02d" % (yy, mo, dd)),
(" День недели", "%d (%s)" % (wd, WEEKDAY_RU.get(wd, "?"))),
]
_check_datetime(res, h, m, s, yy, mo, dd, wd)
elif res.dlc == 0:
res.summary = "BROADCAST STATUS (запрос)"
else:
res.warnings.append("STATUS-ответ должен иметь DLC = 7, получено %d" % res.dlc)
elif btype == BroadcastType.ONOFF:
res.summary = "BROADCAST ONOFF (инверсия флага пульса)"
elif btype == BroadcastType.RESTARTDEVICE:
page = int.from_bytes(res.data, "little") if res.data else 0
ids = [i for i in range(res.dlc * 8) if (page >> i) & 1]
res.summary = "BROADCAST RESTART: устройства %s" % (ids if ids else "нет")
res.fields.append((" Битовая карта", "0x%X" % page))
res.fields.append((" Перезапустить ID", ", ".join(map(str, ids)) or "—"))
if res.dlc == 0:
res.warnings.append("RESTART с DLC = 0 отвергается устройством")
elif btype == BroadcastType.RTCSETUP:
res.summary = "BROADCAST RTCSETUP"
if res.dlc == 7:
h, m, s, yy, mo, dd, wd = res.data
res.summary = "BROADCAST RTCSETUP: 20%02d-%02d-%02d %02d:%02d:%02d" % (
yy, mo, dd, h, m, s)
res.fields += [
(" Время", "%02d:%02d:%02d" % (h, m, s)),
(" Дата", "20%02d-%02d-%02d" % (yy, mo, dd)),
(" День недели", "%d (%s)" % (wd, WEEKDAY_RU.get(wd, "?"))),
]
_check_datetime(res, h, m, s, yy, mo, dd, wd)
else:
res.warnings.append("RTCSETUP принимается только при DLC = 7, получено %d" % res.dlc)
else:
res.summary = "BROADCAST 0x%03X" % btype
def _is_leap(year: int) -> bool:
"""Совпадает с IsLeapYear() в protocan.c: год двузначный, 2000-е."""
y = 2000 + year
return (y % 4 == 0 and y % 100 != 0) or y % 400 == 0
_DAYS = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
def _check_datetime(res: Decoded, h, m, s, yy, mo, dd, wd) -> None:
if h > 23:
res.warnings.append("Часы %d > 23" % h)
if m > 59:
res.warnings.append("Минуты %d > 59" % m)
if s > 59:
res.warnings.append("Секунды %d > 59" % s)
if yy > 99:
res.warnings.append("Год %d > 99" % yy)
if mo == 0 or mo > 12:
res.warnings.append("Месяц %d вне диапазона 1..12" % mo)
else:
limit = _DAYS[mo - 1] + (1 if (mo == 2 and _is_leap(yy)) else 0)
if dd == 0 or dd > limit:
res.warnings.append("Число %d вне диапазона 1..%d" % (dd, limit))
if wd > 6:
res.warnings.append("День недели %d > 6" % wd)
def _decode_discrete(res: Decoded) -> None:
body, dtype = split_discrete(res.id.msg_body)
res.fields.append((" Discrete.Type", "0x%X — %s" % (
dtype, _name(DiscreteType, dtype, DISCRETE_RU))))
res.fields.append((" Discrete.Body", "0x%03X (%d)" % (body, body)))
res.summary = "DISCRETE %s, body=%d, %d байт" % (
_name(DiscreteType, dtype, DISCRETE_RU), body, res.dlc)
if res.data:
bits = " ".join("{:08b}".format(b) for b in reversed(res.data))
res.fields.append((" Data биты (MSB..LSB)", bits))
res.fields.append((" Data ASCII", _ascii(res.data)))
if dtype == DiscreteType.REQUEST_LIST_OF_PARAMETERS:
mask = int.from_bytes(res.data, "little") if res.data else 0
idx = [i for i in range(res.dlc * 8) if (mask >> i) & 1]
res.fields.append((" Запрошены параметры", ", ".join(map(str, idx)) or "—"))
def _decode_analog(res: Decoded) -> None:
sensor_id, atype = split_analog(res.id.msg_body)
res.fields.append((" Analog.Type", "0x%X — %s" % (
atype, _name(AnalogType, atype, ANALOG_RU))))
res.fields.append((" Analog.SensorID", "%d (0x%03X)" % (sensor_id, sensor_id)))
res.fields.append((" Modbus-регистр", "0x%04X" % sensor_to_modbus_register(atype, sensor_id)))
text = _ascii(res.data)
res.summary = "ANALOG %s, sensor %d" % (_name(AnalogType, atype, ANALOG_RU), sensor_id)
if res.data:
res.fields.append((" Data ASCII", text))
res.summary += ", '%s'" % text
if len(res.data) >= 6 and res.data[:2] in (b"US", b"IS", b"TS"):
digits = res.data[2:6].decode("ascii", "replace")
if digits.isdigit():
res.fields.append((" Ответ по датчику", "%s №%d" % (
res.data[:2].decode(), int(digits))))
if int(digits) != sensor_id:
res.warnings.append(
"SensorID в ASCII (%d) не совпадает с ID в MsgBody (%d)"
% (int(digits), sensor_id))
if res.dlc == 0:
res.summary += " (запрос)"
def _decode_gas(res: Decoded) -> None:
start = res.id.msg_body
res.fields.append((" Адрес первого регистра", "0x%04X (%d)" % (start, start)))
if res.dlc % 2:
res.warnings.append("DLC = %d нечётный: регистры GAS передаются парами байт" % res.dlc)
vals = regs_le(res.data)
if len(vals) > 4:
res.warnings.append("В одном кадре GAS не более 4 регистров, получено %d" % len(vals))
res.registers = [(start + i, v) for i, v in enumerate(vals)]
if res.data[:4] == b"GAS-":
res.summary = "GAS отклик '%s'" % _ascii(res.data)
res.fields.append((" Data ASCII", _ascii(res.data)))
res.registers = None
return
for adr, v in res.registers:
res.fields.append((" Reg 0x%04X" % adr, "0x%04X (%d)" % (v, v)))
if vals:
res.summary = "GAS: %d рег. с 0x%04X = %s" % (
len(vals), start, " ".join("%04X" % v for v in vals))
else:
res.summary = "GAS: запрос по адресу 0x%04X" % start
def _decode_modbus_bit(res: Decoded) -> None:
str_adr, reg_count = split_modbus(res.id.msg_body)
kind = "COIL" if res.id.msg_type == MsgType.MODBUS_COIL else "DISCRETE"
res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)))
res.fields.append((" Modbus.RegCount", "%d" % reg_count))
expected = reg_count % 8 + 1
if res.dlc and res.dlc != expected:
res.warnings.append(
"Для %s ожидается DLC = RegCount %% 8 + 1 = %d, получено %d"
% (kind, expected, res.dlc))
if len(res.data) >= 2:
val = res.data[0] | (res.data[1] << 8)
bits = ["%d" % ((val >> i) & 1) for i in range(16)]
res.registers = [(str_adr, val)]
res.fields.append((" Значение", "0x%04X" % val))
res.fields.append((" Биты 0..15", " ".join(bits)))
res.summary = "MODBUS %s @0x%03X x%d = 0x%04X" % (kind, str_adr, reg_count, val)
else:
res.summary = "MODBUS %s @0x%03X x%d (запрос)" % (kind, str_adr, reg_count)
def _decode_modbus_reg(res: Decoded) -> None:
str_adr, reg_count = split_modbus(res.id.msg_body)
kind = "HOLDING" if res.id.msg_type == MsgType.MODBUS_HOLDING else "INPUT"
res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)))
res.fields.append((" Modbus.RegCount", "%d" % reg_count))
if res.dlc and res.dlc != reg_count * 2:
res.warnings.append(
"Для %s ожидается DLC = 2 * RegCount = %d, получено %d"
% (kind, reg_count * 2, res.dlc))
if reg_count > 4:
res.warnings.append("В одном кадре не более 4 регистров, в ID указано %d" % reg_count)
vals = regs_le(res.data)
res.registers = [(str_adr + i, v) for i, v in enumerate(vals)]
for adr, v in res.registers:
res.fields.append((" Reg 0x%04X" % adr, "0x%04X (%d)" % (v, v)))
if vals:
res.summary = "MODBUS %s @0x%03X: %s" % (
kind, str_adr, " ".join("%04X" % v for v in vals))
else:
res.summary = "MODBUS %s @0x%03X x%d (запрос)" % (kind, str_adr, reg_count)
def _decode_error(res: Decoded) -> None:
code, info = split_error(res.id.msg_body)
res.fields.append((" Error.Code", "0x%02X (%d)" % (code, code)))
res.fields.append((" Error.Info", "0x%02X (%d)" % (info, info)))
res.summary = "ERROR code=0x%02X info=0x%02X" % (code, info)
if code == 0xFF and info == 0x00:
res.summary += " — необработанный запрос (PROTOCAN_RequestError)"
if res.dlc:
res.warnings.append("ERROR передаётся с DLC = 0, получено %d" % res.dlc)
def _decode_pulse(res: Decoded) -> None:
if res.dlc >= 1:
res.summary = "PULSE, счётчик %d" % res.data[0]
res.fields.append((" Счётчик пульса", "%d (0x%02X)" % (res.data[0], res.data[0])))
else:
res.summary = "PULSE (без счётчика)"
res.warnings.append("PULSE передаётся с DLC = 1")
if res.id.msg_body:
res.warnings.append("MsgBody у PULSE должен быть 0, получено 0x%04X" % res.id.msg_body)

285
python/protocan/protocol.py Normal file
View File

@@ -0,0 +1,285 @@
"""Совместимое с ``lib/gui_transport`` кадрирование GUI protocol v1."""
from __future__ import annotations
import binascii
from dataclasses import dataclass
from enum import IntEnum
SOF = b"\xA5\x5A"
PROTOCOL_VERSION = 0x01
MAX_PAYLOAD_SIZE = 512
HEADER_SIZE = 8
CRC_SIZE = 4
class ProtocolError(ValueError):
"""@brief Ошибка нарушения контракта транспортного кадра.
Возникает до передачи при неверных границах и при явной проверке кадра.
"""
class MessageType(IntEnum):
"""@brief Стабильные типы сообщений из ``gui_transport_protocol.h``.
Числовые значения являются частью wire-протокола и не перенумеровываются.
"""
PING = 0x01
DEVICE_INFO = 0x02
GET_OBJECT = 0x03
SET_OBJECT = 0x04
GET_OBJECT_LIST = 0x05
GET_OBJECT_INFO = 0x06
COMMAND_STATUS = 0x07
DIAGNOSTICS = 0x08
READ_REGISTERS = 0x09
WRITE_REGISTERS = 0x0A
FIRMWARE_BEGIN = 0x0B
FIRMWARE_DATA = 0x0C
FIRMWARE_END = 0x0D
FIRMWARE_ABORT = 0x0E
FIRMWARE_STATUS = 0x0F
READ_LOGS = 0x10
# Каталог общего адресного пространства и поток выбранных значений,
# см. protocan-transport/docs/GUI_CATALOG.md.
GAS_CATALOG = 0x11
GAS_WATCH_SET = 0x12
GAS_WATCH_DATA = 0x13
SENSOR_SCAN = 0x20
SENSOR_LIST = 0x21
SENSOR_READ = 0x22
SENSOR_DATA = 0x23
SET_USER_BYTES = 0x24
SET_RESOLUTION = 0x25
SET_POLL_PERIOD = 0x26
SEND_ID_CAN = 0x27
UI_KEY = 0x28
UI_READ = 0x29
UI_STATE = 0x2A
EEPROM_SCAN = 0x2B
EEPROM_INFO = 0x2C
EEPROM_READ = 0x2D
EEPROM_LIST = 0x2E
NACK = 0x80
ACK = 0x81
ERROR = 0x82
class ObjectResult(IntEnum):
"""Коды результата, общие с ``GUITransport_ObjectResult``."""
OK = 0
INVALID_ARGUMENT = 1
INVALID_LENGTH = 2
NOT_FOUND = 3
ACCESS_DENIED = 4
BUSY = 5
NO_PROVIDER = 6
INTERNAL = 7
# Расширение диапазона 0x10+ занято прошивкой STM32F103C8T6 (1-Wire).
BUS_ERROR = 0x10
UNSUPPORTED = 0x11
NO_DISPLAY = 0x12
NOT_APPLIED = 0x13
NO_MEMORY = 0x14
#: Причины отказа прибора на языке оператора.
RESULT_TEXT: dict[int, str] = {
ObjectResult.OK: "успех",
ObjectResult.INVALID_ARGUMENT: "недопустимый аргумент",
ObjectResult.INVALID_LENGTH: "неверная длина payload",
ObjectResult.NOT_FOUND: "объект или датчик не найден",
ObjectResult.ACCESS_DENIED: "доступ запрещён",
ObjectResult.BUSY: "прибор занят",
ObjectResult.NO_PROVIDER: "обработчик не назначен",
ObjectResult.INTERNAL: "внутренняя ошибка прошивки",
ObjectResult.BUS_ERROR: "ошибка шины 1-Wire: датчик не подтвердил запись",
ObjectResult.UNSUPPORTED: "команда не поддерживается",
ObjectResult.NO_DISPLAY: "панель недоступна",
ObjectResult.NOT_APPLIED: "датчик ответил, но оставил прежнее значение",
ObjectResult.NO_MEMORY: "внешняя память недоступна: отсутствует, переполнена или не приняла запись",
}
def describe_result(code: int) -> str:
"""@brief Переводит код результата в текст для журнала GUI.
@param code Значение из payload кадров NACK и ERROR.
@return Русское описание либо запись с неизвестным числовым кодом.
"""
return RESULT_TEXT.get(code, f"неизвестный код {code}")
def crc8_maxim(data: bytes) -> int:
"""@brief Вычисляет CRC8 Dallas/Maxim для ROM и scratchpad DS18B20.
@param data Байты без поля контрольной суммы.
@return Значение CRC8 с обратным полиномом ``0x8C``.
"""
crc = 0
for byte in data:
crc ^= byte
for _ in range(8):
crc = (crc >> 1) ^ 0x8C if crc & 0x01 else crc >> 1
return crc
def encode_u16(value: int) -> bytes:
if not 0 <= value <= 0xFFFF:
raise ProtocolError("u16 вне диапазона")
return value.to_bytes(2, "little")
def encode_u32(value: int) -> bytes:
if not 0 <= value <= 0xFFFFFFFF:
raise ProtocolError("u32 вне диапазона")
return value.to_bytes(4, "little")
def decode_u16(data: bytes, offset: int = 0) -> int:
if offset < 0 or offset + 2 > len(data):
raise ProtocolError("payload не содержит u16")
return int.from_bytes(data[offset : offset + 2], "little")
def decode_u32(data: bytes, offset: int = 0) -> int:
if offset < 0 or offset + 4 > len(data):
raise ProtocolError("payload не содержит u32")
return int.from_bytes(data[offset : offset + 4], "little")
@dataclass(frozen=True, slots=True)
class Frame:
"""@brief Проверенный кадр без служебных полей SOF и CRC.
@param message_type Тип сообщения из общего C/Python-контракта.
@param sequence Номер запроса в диапазоне 0..65535.
@param payload Полезная нагрузка не более ``MAX_PAYLOAD_SIZE`` байт.
"""
message_type: MessageType
sequence: int
payload: bytes = b""
def __post_init__(self) -> None:
"""@brief Проверяет границы до кодирования или передачи кадра.
@raises ProtocolError При неверном sequence или слишком длинном payload.
"""
if not 0 <= self.sequence <= 0xFFFF:
raise ProtocolError("sequence должен быть в диапазоне 0..65535")
if len(self.payload) > MAX_PAYLOAD_SIZE:
raise ProtocolError("payload превышает 512 байт")
def crc32_ieee(data: bytes) -> int:
"""@brief Вычисляет IEEE CRC32 как ``GUITransport_Protocol_Crc32``.
@param data Защищаемые байты от version до конца payload.
@return Беззнаковое 32-битное значение контрольной суммы.
"""
return binascii.crc32(data) & 0xFFFFFFFF
def build_frame(frame: Frame) -> bytes:
"""@brief Кодирует полный wire-кадр транспортного протокола.
@param frame Предварительно проверенная модель кадра.
@return SOF, big-endian header, payload и little-endian CRC32.
"""
payload_size = len(frame.payload)
protected = bytes(
(
PROTOCOL_VERSION,
int(frame.message_type),
(frame.sequence >> 8) & 0xFF,
frame.sequence & 0xFF,
(payload_size >> 8) & 0xFF,
payload_size & 0xFF,
)
) + frame.payload
return SOF + protected + crc32_ieee(protected).to_bytes(4, "little")
class FrameParser:
"""@brief Потоковый parser с восстановлением синхронизации.
Экземпляр владеет входным буфером и счётчиками ошибок. Он не использует
глобальное состояние и подходит для независимых последовательных каналов.
"""
def __init__(self) -> None:
self._buffer = bytearray()
self.crc_errors = 0
self.version_errors = 0
self.length_errors = 0
def reset(self) -> None:
"""@brief Удаляет незавершённые входные данные.
Счётчики диагностики сохраняются для анализа качества соединения.
"""
self._buffer.clear()
def feed(self, data: bytes) -> list[Frame]:
"""@brief Добавляет произвольный фрагмент последовательного потока.
@param data Новые байты; границы фрагмента не обязаны совпадать с кадром.
@return Ноль или несколько полностью проверенных кадров.
"""
if not data:
return []
self._buffer.extend(data)
frames: list[Frame] = []
while True:
sof_index = self._buffer.find(SOF)
if sof_index < 0:
# Последний A5 может быть началом SOF следующего фрагмента.
self._buffer[:] = self._buffer[-1:] if self._buffer[-1:] == SOF[:1] else b""
break
if sof_index:
del self._buffer[:sof_index]
if len(self._buffer) < HEADER_SIZE:
break
if self._buffer[2] != PROTOCOL_VERSION:
self.version_errors += 1
del self._buffer[0]
continue
payload_size = (self._buffer[6] << 8) | self._buffer[7]
if payload_size > MAX_PAYLOAD_SIZE:
self.length_errors += 1
del self._buffer[0]
continue
total_size = HEADER_SIZE + payload_size + CRC_SIZE
if len(self._buffer) < total_size:
break
packet = bytes(self._buffer[:total_size])
protected = packet[2:-CRC_SIZE]
received_crc = int.from_bytes(packet[-CRC_SIZE:], "little")
if crc32_ieee(protected) != received_crc:
self.crc_errors += 1
del self._buffer[0]
continue
try:
message_type = MessageType(packet[3])
except ValueError:
# Неизвестный тип остаётся протокольной ошибкой GUI.
del self._buffer[:total_size]
continue
frames.append(
Frame(
message_type=message_type,
sequence=(packet[4] << 8) | packet[5],
payload=packet[8 : 8 + payload_size],
)
)
del self._buffer[:total_size]
return frames

209
python/protocan/transport.py Normal file
View File

@@ -0,0 +1,209 @@
# -*- coding: utf-8 -*-
"""Транспортный кадр моста CAN <-> RS485.
Это не протокол SETGUI (``A5 5A``) из ``protocol.py``, а кадр полевого
канала моста: им приходят кадры самой шины CAN.
Формат описан в CAN_to_RS485/docs/PROTOCOL.md и реализован в прошивке
моста (lib/protocan-transport/src/pcan_frame.c):
AA 55 | LEN | SEQ | FLAGS | ID0..ID3 | DATA[0..8] | CRC_L CRC_H
LEN = 6 + DLC (длина участка SEQ..DATA)
CRC = CRC-16/CCITT-FALSE по байтам LEN..DATA включительно, little-endian
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterator, List, Optional
SOF0 = 0xAA
SOF1 = 0x55
FLAG_IDE = 0x01
FLAG_RTR = 0x02
FLAG_DIR = 0x04 # 0: CAN -> RS485, 1: RS485 -> CAN
FLAG_ERR = 0x08
MIN_LEN = 6
MAX_LEN = 14
FRAME_OVERHEAD = 5 # SOF(2) + LEN(1) + CRC(2)
def crc16_ccitt(data: bytes, crc: int = 0xFFFF) -> int:
"""CRC-16/CCITT-FALSE: poly 0x1021, init 0xFFFF, без рефлексии."""
for byte in data:
crc ^= byte << 8
for _ in range(8):
crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF
return crc
@dataclass
class Frame:
"""Один транспортный кадр."""
seq: int
flags: int
can_id: int
data: bytes
@property
def ide(self) -> bool:
return bool(self.flags & FLAG_IDE)
@property
def rtr(self) -> bool:
return bool(self.flags & FLAG_RTR)
@property
def to_can(self) -> bool:
"""True — кадр адресован в CAN (передан хостом)."""
return bool(self.flags & FLAG_DIR)
@property
def is_error(self) -> bool:
return bool(self.flags & FLAG_ERR)
@property
def direction(self) -> str:
if self.is_error:
return "ERR"
return "TX" if self.to_can else "RX"
def encode(self) -> bytes:
payload = bytearray()
payload.append(MIN_LEN + len(self.data))
payload.append(self.seq & 0xFF)
payload.append(self.flags & 0xFF)
payload += (self.can_id & 0x1FFFFFFF).to_bytes(4, "little")
payload += self.data
crc = crc16_ccitt(bytes(payload))
return bytes([SOF0, SOF1]) + bytes(payload) + crc.to_bytes(2, "little")
def build_frame(can_id: int, data: bytes, seq: int = 0, ide: bool = True,
rtr: bool = False, to_can: bool = True) -> Frame:
"""Собирает кадр для передачи в мост."""
if len(data) > 8:
raise ValueError("DLC не может превышать 8 байт")
flags = 0
if ide:
flags |= FLAG_IDE
if rtr:
flags |= FLAG_RTR
if to_can:
flags |= FLAG_DIR
return Frame(seq=seq & 0xFF, flags=flags, can_id=can_id, data=bytes(data))
@dataclass
class ParseError:
"""Отброшенный участок потока."""
reason: str
raw: bytes
class FrameParser:
"""Потоковый разборщик: накапливает байты и отдаёт готовые кадры.
Ресинхронизация — сдвигом на один байт от неудачной сигнатуры,
поэтому мусор в линии стоит не больше одного пропущенного кадра.
"""
def __init__(self) -> None:
self._buf = bytearray()
self.stats = {"frames": 0, "crc_errors": 0, "resync_bytes": 0, "seq_lost": 0}
self._last_seq: Optional[int] = None
self.errors: List[ParseError] = []
def reset(self) -> None:
self._buf.clear()
self._last_seq = None
def feed(self, chunk: bytes) -> List[Frame]:
"""Добавляет байты в буфер и возвращает все разобранные кадры."""
self._buf += chunk
return list(self._drain())
def _drain(self) -> Iterator[Frame]:
buf = self._buf
while True:
# 1. Ищем сигнатуру
start = -1
for i in range(len(buf) - 1):
if buf[i] == SOF0 and buf[i + 1] == SOF1:
start = i
break
if start < 0:
# Сигнатуры нет: оставляем последний байт (вдруг это 0xAA)
drop = max(0, len(buf) - 1)
if drop:
self.stats["resync_bytes"] += drop
del buf[:drop]
return
if start:
self.stats["resync_bytes"] += start
del buf[:start]
# 2. Ждём LEN
if len(buf) < 3:
return
length = buf[2]
if not (MIN_LEN <= length <= MAX_LEN):
self.errors.append(ParseError("LEN=%d вне 6..14" % length, bytes(buf[:3])))
self.stats["resync_bytes"] += 1
del buf[:1]
continue
total = 2 + 1 + length + 2
if len(buf) < total:
return
payload = bytes(buf[2:2 + 1 + length])
got = buf[2 + 1 + length] | (buf[3 + 1 + length] << 8)
want = crc16_ccitt(payload)
if got != want:
self.stats["crc_errors"] += 1
self.errors.append(ParseError(
"CRC 0x%04X, ожидалось 0x%04X" % (got, want), bytes(buf[:total])))
self.stats["resync_bytes"] += 1
del buf[:1]
continue
frame = Frame(
seq=payload[1],
flags=payload[2],
can_id=int.from_bytes(payload[3:7], "little") & 0x1FFFFFFF,
data=payload[7:],
)
del buf[:total]
self.stats["frames"] += 1
if self._last_seq is not None:
gap = (frame.seq - self._last_seq - 1) & 0xFF
if gap:
self.stats["seq_lost"] += gap
self._last_seq = frame.seq
yield frame
def parse_hex(text: str) -> bytes:
"""Разбирает строку вида 'AA 55 08' / 'aa5508' / '0xAA,0x55' в байты."""
cleaned = (text.replace("0x", " ").replace("0X", " ")
.replace(",", " ").replace(";", " ")
.replace("\r", " ").replace("\n", " ").replace("\t", " "))
tokens = cleaned.split()
if not tokens:
return b""
if all(len(t) <= 2 for t in tokens):
return bytes(int(t, 16) for t in tokens)
joined = "".join(tokens)
if len(joined) % 2:
raise ValueError("нечётное число hex-символов")
return bytes.fromhex(joined)
def hex_str(data: bytes) -> str:
return " ".join("%02X" % b for b in data)