templates/python/protocan/transport.py

# -*- 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)