templates/python/set_devices/can_monitor.py

"""Shared CAN protocol selection and disk-backed session buffer (no Qt)."""
import csv
import os
import tempfile
from collections import Counter, deque
from datetime import datetime
from pathlib import Path
from protocan import protocan as pc
from .can_bridge import decode_frame

PROTOCOLS = (("auto", "Авто (ProtoCAN / Balsam)"),  # Автоматический разбор выбирает формат по кадру, не требуя ручного указания прибора.
             ("devboard", "DevBoard_V1 / SETCAN"),  # Формат DevBoard сохраняет самостоятельный пункт для существующих мониторов платы.
             ("deif", "DEIF XDi-192 / CANopen"),  # Прибор DEIF использует собственные правила CANopen и остаётся отдельным форматом.
             ("protocan", "ProtoCAN / GAS"),  # Общий ProtoCAN разбирает кадры без ручного назначения карты технологических слов УМП.
             ("ump_balsam", "УМП / Balsam 167 (legacy CAN)"),  # Старый Balsam имеет другой формат кадра и требует отдельного декодера.
             ("ump_protocan", "УМП / ProtoCAN GAS"))  # Единый формат УМП отделяет выбор карты адресов и номера устройства от протокола.

UMP_MAPS = (("gas167", "167/v1 (текущая)"),  # Действующая карта использует слова 0…127 и окно 0x5000…0x507F из профиля прошивки 167/v1.
            ("assigned", "Новые ID 5000/6000 (планируется)"))  # Десятичные базы 5000 и 6000 назначены документацией, но ещё не реализованы прошивкой.


def decode_protocol(frame, protocol="auto", *, ump_map="gas167", ump_node=1):  # Карта 167/v1 и УМП 1 служат исходным выбором единого формата; параметры карты и узла передаются по имени.
    if protocol == "deif":
        from .deif_xdi import decode
        return decode(frame)
    if protocol == "ump_balsam":
        from .balsam_can import decode
        result = decode(frame.can_id, frame.data, ide=frame.ide, rtr=frame.rtr)
        if result is not None:
            return result
        result = pc.Decoded(pc.ProtoCanId.parse(frame.can_id), len(frame.data),
                            bytes(frame.data), ide=frame.ide, rtr=frame.rtr,
                            protocol="BALZAM 167", message_label="Не Balsam 167")
        result.summary = "УМП / Balsam 167: CAN ID или формат кадра не соответствует протоколу"
        result.warnings.append(result.summary)
        return result
    if protocol in ("ump1_gas167", "ump2_gas167"):
        from .ump_can import decode_gas167
        return decode_gas167(frame, node=1 if protocol == "ump1_gas167" else 2)
    if protocol == "ump_protocan":  # Новые элементы интерфейса используют единый формат, а старые ключи выше сохраняют прежний разбор.
        if ump_map == "gas167":  # Действующие локальные адреса сами не определяют номер УМП, поэтому он задаётся отдельно.
            from .ump_can import decode_gas167  # Словарь 167/v1 подключается только при явном выборе этой карты.
            return decode_gas167(frame, node=ump_node)  # Один декодер разбирает технологические слова обоих УМП с выбранной подписью устройства.
        if ump_map == "assigned":  # Планируемая сквозная карта определяет УМП по десятичным диапазонам 5000…5127 и 6000…6127.
            return pc.decode(frame.can_id, frame.data, ide=frame.ide, rtr=frame.rtr)  # Общий ProtoCAN уже содержит имена назначенных ID и не требует ручного номера УМП.
        raise ValueError("Неизвестная карта адресов УМП: %s" % ump_map)  # Ошибочный ключ карты не должен молча давать расшифровку другого профиля.
    if protocol in ("devboard", "protocan", "ump_gas"):
        return pc.decode(frame.can_id, frame.data, ide=frame.ide, rtr=frame.rtr)
    return decode_frame(frame)


def protocol_values(identifier, extended):
    if not extended:
        return ("",) * 7
    value = pc.ProtoCanId.parse(identifier)
    return (str(value.priority), str(value.pm), str(value.device_type),
            str(value.device), value.device_name,
            "0x%X — %s" % (value.msg_type, value.msg_type_name),
            "0x%04X" % value.body)


def parse_filter(text):
    if not text.strip():
        return None
    try:
        ids = {int(part.strip(), 16) for part in text.replace(";", ",").split(",")}
        if any(value < 0 or value > 0x1FFFFFFF for value in ids):
            raise ValueError()
        return ids
    except ValueError:
        raise ValueError("Введите CAN ID в hex, например 701 или 0x701, 201") from None


def parse_gas_filter(text):  # GAS ID вводятся десятичными числами либо в HEX с явным префиксом 0x.
    if not text.strip():  # Пустое поле не добавляет ограничений к независимому фильтру CAN ID.
        return None  # Значение None обозначает отсутствие условия, а не пустой набор разрешённых адресов.
    try:  # Ошибки ввода превращаются в понятное сообщение общей панели монитора.
        parts = [part.strip() for part in text.replace(";", ",").split(",")]  # Поддерживаем те же разделители списка, что и у существующего CAN-фильтра.
        addresses = {int(part, 16 if part.lower().startswith("0x") else 10) for part in parts}  # Число 5000 означает назначенный десятичный GAS ID, а 0x5000 — адрес локального окна.
        if any(address < 0 or address > 0xFFFF for address in addresses):  # Поле Body ProtoCAN содержит 16-битный адрес слова GAS в диапазоне 0…65535.
            raise ValueError()  # Выход за пределы адресного пространства отклоняется вместе с остальными ошибками формата.
        return addresses  # Повторяющиеся адреса объединяются без изменения логики отбора кадров.
    except ValueError:  # Неполные, отрицательные и некорректно записанные значения не должны проходить фильтр молча.
        raise ValueError("Введите GAS ID: 5000, 5127 или 0x5000; диапазон 0…65535") from None  # Сообщаем систему счисления и допустимые границы непосредственно рядом с фильтром.


def matches_filters(identifier, data, extended, rtr=False, *, can_ids=None, gas_ids=None):  # Оба условия объединяются по И, а значения внутри одного списка — по ИЛИ.
    if can_ids is not None and identifier not in can_ids:  # Быстрый отбор по CAN ID выполняется до декодирования полезной нагрузки.
        return False  # Несовпадение любого включённого ограничения исключает кадр.
    if gas_ids is None:  # При выключенном или пустом GAS-фильтре сохраняется прежний отбор CAN-кадров всех форматов.
        return True  # CAN ID уже проверен либо не ограничен.
    if not extended or rtr or len(data) > 8 or len(data) % 2:  # GAS использует расширенный кадр данных с полными 16-битными словами и DLC не больше 8 байт.
        return False  # STD, RTR и повреждённая нагрузка не обозначают достоверные GAS-регистры.
    if pc.ProtoCanId.parse(identifier).msg_type != pc.MsgType.GENERAL_ADDRESS_SPACE:  # Совпадение младших битов ID другого типа сообщения не является совпадением адреса GAS.
        return False  # Остальные протоколы остаются видимыми только при выключенном GAS-условии.
    decoded = pc.decode(identifier, data, ide=extended, rtr=rtr)  # Общий декодер определяет все регистры кадра и отличает их от текстового GAS-отклика.
    if not data:  # Пустой кадр данных GAS представляет запрос по начальному адресу Body.
        return decoded.id.body in gas_ids  # Запрос учитывается только для указанного в нём адреса.
    if not decoded.registers or any(address > 0xFFFF for address, _value in decoded.registers):  # Отклоняем служебный текст и пакет, выходящий за конец 16-битного пространства.
        return False  # Такие кадры не содержат допустимого набора регистров для адресного фильтра.
    return any(address in gas_ids for address, _value in decoded.registers)  # Совпадение второго, третьего или четвёртого слова выбирает весь исходный кадр.


class CanSession:
    """All frames spool to disk; retain recent frames and unique-frame totals.

    Starting recording includes the current session buffer. Reset clears the
    buffer/counters but never removes frames already written to a recording.
    """
    HEADER = ("Время", "Канал", "Тип", "ID", "RTR", "DLC", "Данные")

    def __init__(self):
        self.recent = deque(maxlen=5000)
        self.counts = Counter()
        self.unique_frames = {}
        self.output = None
        self.path = None
        self.buffer = tempfile.TemporaryFile(mode="w+", encoding="utf-8", newline="")

    def append(self, identifier, data, extended, rtr=False, source="CAN", timestamp=None):
        row = ((datetime.fromtimestamp(timestamp) if timestamp is not None else datetime.now()).isoformat(timespec="milliseconds"),
               source, "EXT" if extended else "STD", ("%08X" if extended else "%03X") % identifier,
               int(rtr), len(data), bytes(data).hex(" ").upper())
        csv.writer(self.buffer, delimiter=";").writerow(row)
        self.recent.append((identifier, row))
        self.counts[identifier] += 1
        # Timestamp is not part of a frame's identity. Keep the latest arrival
        # and a session-wide count, including repeats outside the recent buffer.
        key = row[1:]
        previous = self.unique_frames.get(key)
        self.unique_frames[key] = (row, previous[1] + 1 if previous else 1)
        if self.output is not None:
            csv.writer(self.output, delimiter=";").writerow(row)
        return key  # Возвращаем ключ уникального кадра, чтобы интерфейс обновлял только поступившие записи без повторного обхода сессии.

    def start(self, directory):
        directory = Path(directory) / datetime.now().strftime("session_%Y%m%d_%H%M%S_%f")
        directory.mkdir(parents=True, exist_ok=False)
        path = directory / "frames.csv"
        output = path.open("w", encoding="utf-8-sig", newline="")
        try:
            csv.writer(output, delimiter=";").writerow(self.HEADER)
            self.buffer.seek(0)
            while chunk := self.buffer.read(65536):
                output.write(chunk)
            output.flush()
        except Exception:
            output.close()
            raise
        finally:
            self.buffer.seek(0, 2)
        self.output, self.path = output, path
        return path

    def save(self, path):
        """Save a full buffer snapshot without stopping an active recording."""
        path = Path(path)
        if self.output is not None and (path.resolve() == self.path.resolve()
                or (path.exists() and path.samefile(self.path))):
            raise ValueError("Выберите другой файл: в этот CSV сейчас идёт запись.")
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8-sig",
                    newline="", dir=path.parent, delete=False) as output:
                temporary = Path(output.name)
                csv.writer(output, delimiter=";").writerow(self.HEADER)
                self.buffer.seek(0)
                while chunk := self.buffer.read(65536):
                    output.write(chunk)
            os.replace(temporary, path)
        finally:
            self.buffer.seek(0, 2)
            if temporary is not None and temporary.exists():
                temporary.unlink()
        return path

    def flush(self):
        if self.output is not None:
            self.output.flush()

    def stop(self):
        if self.output is not None:
            self.output.close()
            self.output = None

    def reset(self):
        self.buffer.seek(0)
        self.buffer.truncate()
        self.recent.clear()
        self.counts.clear()
        self.unique_frames.clear()

    def close(self):
        self.stop()
        self.buffer.close()


def format_data(data: bytes, decimal: bool = False) -> str:
    """Format individual payload bytes without changing their order or width."""
    return " ".join(str(byte) if decimal else "%02X" % byte for byte in data)