From 01ffc0e496a4f20a5343f771e118667c7826441f Mon Sep 17 00:00:00 2001 From: Andrey Date: Fri, 4 Sep 2026 18:52:36 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20=D0=BE=D0=B1=D1=89=D0=B8=D0=B9=20Python=20API=20=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D1=80=D0=BE=D0=B3=D0=BE=20CAN=20terminal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python/protocan/legacycan.py | 227 +++++++++++++++++++++++++++++++++ python/tests/test_legacycan.py | 39 ++++++ 2 files changed, 266 insertions(+) create mode 100644 python/protocan/legacycan.py create mode 100644 python/tests/test_legacycan.py diff --git a/python/protocan/legacycan.py b/python/protocan/legacycan.py new file mode 100644 index 0000000..2c3b405 --- /dev/null +++ b/python/protocan/legacycan.py @@ -0,0 +1,227 @@ +"""UI-independent model of the historical CAN_terminal protocol. + +The catalog is migrated from ``CAN_terminal/Projects.ini``. Both desktop +SETGUI and Android keep only presentation and transport code in the apps; +wire layouts and project metadata live in the shared templates repository. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from enum import Enum + + +class LegacyCanFormat(Enum): + ROTATING_THREE_WORDS = "rotating_three_words" + ADDRESS_IN_IDENTIFIER = "address_in_identifier" + + +class LegacyCanSource(Enum): + TO_DEVICE = "TX" + FROM_DEVICE = "RX" + + +@dataclass(frozen=True) +class LegacyCanPacket: + address: int + mask: int + values: tuple[int, ...] + source: LegacyCanSource + + @property + def present_values(self) -> tuple[tuple[int, int], ...]: + return tuple( + (self.address + index, value) + for index, value in enumerate(self.values) + if self.mask == 0xFF or self.mask & (4 >> index) + ) + + +@dataclass(frozen=True) +class LegacyCanWireFrame: + can_id: int + data: bytes + + +@dataclass(frozen=True) +class LegacyCanRegisterValue: + address: int + value: int = 0 + source: LegacyCanSource | None = None + revision: int = 0 + + +@dataclass(frozen=True) +class LegacyCanNode: + index: int + rs_address: int + can_address: int + rx_id: int + tx_id: int + name: str + + +@dataclass(frozen=True) +class LegacyCanProject: + name: str + format: LegacyCanFormat + base_id: int + id_offset: int + nodes: tuple[LegacyCanNode, ...] + command_names: tuple[str, ...] + + def node_for(self, can_id: int) -> LegacyCanNode | None: + routed = routing_id(self.format, can_id) + return next( + (node for node in self.nodes + if node.rx_id == routed or node.tx_id == routed), + None, + ) + + +def empty_register_bank() -> tuple[LegacyCanRegisterValue, ...]: + return tuple(LegacyCanRegisterValue(index) for index in range(128)) + + +def apply_packet( + bank: tuple[LegacyCanRegisterValue, ...], + packet: LegacyCanPacket, + revision: int, +) -> tuple[LegacyCanRegisterValue, ...]: + if len(bank) != 128: + raise ValueError("Банк должен содержать 128 регистров") + updates = dict(packet.present_values) + return tuple( + replace(item, value=updates[item.address], source=packet.source, + revision=revision) + if item.address in updates else item + for item in bank + ) + + +def routing_id(format_: LegacyCanFormat, can_id: int) -> int: + if format_ is LegacyCanFormat.ROTATING_THREE_WORDS: + return can_id & 0x1FFFFFFF + return can_id & 0x1FF00000 + + +def decode( + format_: LegacyCanFormat, + node: LegacyCanNode, + can_id: int, + data: bytes, +) -> LegacyCanPacket | None: + route = routing_id(format_, can_id) + if route == node.tx_id: + source = LegacyCanSource.TO_DEVICE + elif route == node.rx_id: + source = LegacyCanSource.FROM_DEVICE + else: + return None + if format_ is LegacyCanFormat.ROTATING_THREE_WORDS: + if len(data) != 8: + return None + mask = (data[4] >> 5) & 7 + address = ((data[4] & 0x1F) << 8) | data[5] + values = (_u16be(data, 6), _u16be(data, 0), _u16be(data, 2)) + return LegacyCanPacket(address, mask, values, source) + if not data or len(data) > 8 or len(data) % 2: + return None + return LegacyCanPacket( + can_id & 0x7F, + 0xFF, + tuple(_u16be(data, offset) for offset in range(0, len(data), 2)), + source, + ) + + +def encode_write( + format_: LegacyCanFormat, + can_id: int, + address: int, + values: tuple[int, ...] | list[int], +) -> LegacyCanWireFrame: + if not 0 <= address <= 127: + raise ValueError("Адрес регистра должен быть в диапазоне 0..127") + maximum = 4 if format_ is LegacyCanFormat.ADDRESS_IN_IDENTIFIER else 3 + if not 1 <= len(values) <= maximum: + raise ValueError(f"Нужно от 1 до {maximum} слов данных") + if any(not 0 <= value <= 0xFFFF for value in values): + raise ValueError("Значение должно быть в диапазоне 0..65535") + if format_ is LegacyCanFormat.ADDRESS_IN_IDENTIFIER: + data = b"".join(int(value).to_bytes(2, "big") for value in values) + return LegacyCanWireFrame((can_id & 0x1FF00000) + address, data) + padded = tuple(values) + (0,) * (3 - len(values)) + mask = (4, 6, 7)[len(values) - 1] + data = ( + padded[1].to_bytes(2, "big") + + padded[2].to_bytes(2, "big") + + bytes(((mask << 5) | (address >> 8), address & 0xFF)) + + padded[0].to_bytes(2, "big") + ) + return LegacyCanWireFrame(can_id & 0x1FFFFFFF, data) + + +def encode_command( + project: LegacyCanProject, + node: LegacyCanNode, + command_index: int, +) -> LegacyCanWireFrame: + if not 0 <= command_index <= 16: + raise ValueError("Номер команды должен быть в диапазоне 0..16") + value = 1 << command_index if command_index < 16 else 0 + return encode_write(project.format, node.rx_id, 127, (value,)) + + +def signed_word(value: int) -> int: + return value if value < 0x8000 else value - 0x10000 + + +def _u16be(data: bytes, offset: int) -> int: + return int.from_bytes(data[offset:offset + 2], "big") + + +_DEFAULT_COMMANDS = ( + "Test", "Def", "Save", "Load", "Calibr", "Calcul", "Secret", + "Light", "Raw", "-", "-", "-", "-", "-", "-", "Reset", + "Nothing at all", +) + + +def _commands(**changes: str) -> tuple[str, ...]: + result = list(_DEFAULT_COMMANDS) + for index, name in changes.items(): + result[int(index)] = name + return tuple(result) + + +def _project( + name: str, + specs: tuple[tuple[int, int, int, str], ...], + base_id: int = 0, + offset: int = 0x10, + format_: LegacyCanFormat = LegacyCanFormat.ROTATING_THREE_WORDS, + commands: tuple[str, ...] = _DEFAULT_COMMANDS, +) -> LegacyCanProject: + shift = 20 if format_ is LegacyCanFormat.ADDRESS_IN_IDENTIFIER else 0 + actual_offset = 1 << 28 if format_ is LegacyCanFormat.ADDRESS_IN_IDENTIFIER else offset + nodes = tuple( + LegacyCanNode(index, rs, can, base_id + (can << shift), + base_id + actual_offset + (can << shift), node_name) + for index, can, rs, node_name in specs + ) + return LegacyCanProject(name, format_, base_id, actual_offset, nodes, commands) + + +PROJECTS = ( + _project("Буксир", ((0,0,1,"УКСС СБ"),(1,1,2,"БКСС ГД"),(2,2,3,"УКСВЭП"),(3,3,4,"Задатчик")), base_id=0x00318200), + _project("СЭДБМ", ((0,0,0,"УКСС СК1 СБ1"),(1,1,1,"УКСС СК2 СБ1"),(2,2,2,"УКСС СК3 СБ1"),(3,3,3,"УКСС СК4 СБ1"),(4,4,4,"УКССВЭП СБ1"),(5,5,5,"Задатчик СБ1"),(6,6,6,"БТР ИТЭС"),(8,0x20,8,"УКСС СК1 СБ2"),(9,0x21,9,"УКСС СК2 СБ2"),(10,0x22,10,"УКСС СК3 СБ2"),(11,0x23,11,"УКСС СК4 СБ2"),(12,0x24,12,"УКССВЭП СБ2"),(13,0x25,13,"Задатчик СБ2"),(14,0x26,14,"УКСС БОИН"),(15,0x27,15,"УКСВЭП БОИН")), base_id=0x01051020, commands=_commands(**{"7":"Raw","8":"HiVolt"})), + _project("Ледокол", ((0,0,1,"УКСС БВ1 ПЧ1"),(8,1,2,"УКСС БВ1 ПЧ2"),(1,2,3,"УКСС БВ1 ПЧ1"),(9,3,4,"УКСС БВ2 ПЧ2"),(2,4,5,"УКСС БИ1 ПЧ1"),(10,5,6,"УКСС БИ1 ПЧ2"),(3,6,7,"УКСС БИ2 ПЧ1"),(11,7,8,"УКСС БИ2 ПЧ2"),(4,8,9,"УКССВЭП1 ПЧ1"),(12,9,10,"УКССВЭП1 ПЧ2"),(5,10,11,"УКССВЭП2 ПЧ1"),(13,11,12,"УКССВЭП2 ПЧ2")), base_id=0x001CE020, offset=-0x20, commands=_commands(**{"4":"Raw","5":"Read","6":"ExtLamp","7":"ExtLite","8":"No log"})), + _project("Бальзам", ((0,0,1,"БКСС Тр1"),(8,1,2,"БКСС Тр2"),(1,2,3,"УКСС СБ1"),(9,3,4,"УКСС СБ2"),(2,4,5,"УКСС УМП1"),(10,5,6,"УКСС УМП2"),(3,6,7,"БКСС ГД"),(4,7,9,"Задатчик"),(5,8,11,"УКСС ВЭП")), base_id=0x00BA0000, commands=_commands(**{"6":"Stop","7":"Start","8":"Init","9":"Tune","10":"Secret","11":"Light","12":"Raw"})), + _project("23550", ((0,0,1,"Задатчик"),(1,1,2,"Выносной пульт"),(2,2,3,"УКСВЭП"),(3,3,4,"БКСС ГД")), base_id=0x00235500, commands=_commands(**{"5":"Read","7":"Send","8":"-"})), + _project("23550.X", ((0,0,1,"Задатчик"),(1,1,2,"Выносной пульт"),(2,2,3,"УКСВЭП"),(3,3,4,"БКСС ГД")), format_=LegacyCanFormat.ADDRESS_IN_IDENTIFIER, commands=_commands(**{"5":"Read","7":"Send","8":"-"})), + _project("23550.2", ((0,0,1,"Задатчик"),(1,1,2,"Выносной пульт"),(2,2,3,"БКСС ГД"),(3,4,4,"УКСС СИ СБ1"),(4,6,6,"УКСС СВФ СБ1"),(5,8,8,"УКСВЭП СБ1"),(11,5,5,"УКСС СИ СБ2"),(12,7,7,"УКСС СВФ СБ2"),(13,9,9,"УКСВЭП СБ2"),(16,0x1F,16,"BroadCast")), format_=LegacyCanFormat.ADDRESS_IN_IDENTIFIER, commands=_commands(**{"5":"Calc","7":"Send"})), + _project("Янтарь", ((0,0,1,"УКСС БВ"),(1,1,2,"УКСС БИ1"),(2,2,3,"УКСС БИ2"),(3,3,4,"БКСС ГД"),(4,4,5,"УКСВЭП"),(5,5,6,"Задатчик"),(6,6,7,"Выносной пульт")), base_id=0x00213000), + _project("23550 БСУ", ((0,0,0,"БСУ1"),(1,1,1,"БСУ2")), base_id=0x0CEB0F1, offset=-0x10, commands=("-",) * 16 + ("Nothing at all",)), +) + diff --git a/python/tests/test_legacycan.py b/python/tests/test_legacycan.py new file mode 100644 index 0000000..ae3c004 --- /dev/null +++ b/python/tests/test_legacycan.py @@ -0,0 +1,39 @@ +from protocan.legacycan import ( + LegacyCanFormat, LegacyCanSource, PROJECTS, apply_packet, decode, + empty_register_bank, encode_command, encode_write, signed_word, +) + + +def test_catalog_matches_legacy_projects(): + assert [project.name for project in PROJECTS] == [ + "Буксир", "СЭДБМ", "Ледокол", "Бальзам", "23550", "23550.X", + "23550.2", "Янтарь", "23550 БСУ", + ] + assert len(next(project for project in PROJECTS if project.name == "СЭДБМ").nodes) == 15 + + +def test_rotating_three_word_codec_and_bank(): + project = next(project for project in PROJECTS if project.name == "Бальзам") + node = project.nodes[0] + wire = encode_write(project.format, node.rx_id, 24, (41, 0xFFFE, 0x1234)) + assert wire.data == bytes.fromhex("FF FE 12 34 E0 18 00 29") + packet = decode(project.format, node, wire.can_id, wire.data) + assert packet is not None + assert packet.present_values == ((24, 41), (25, 0xFFFE), (26, 0x1234)) + assert packet.source is LegacyCanSource.FROM_DEVICE + bank = apply_packet(empty_register_bank(), packet, 7) + assert bank[25].value == 0xFFFE + assert bank[25].revision == 7 + assert signed_word(bank[25].value) == -2 + + +def test_address_in_identifier_codec_and_command(): + project = next(project for project in PROJECTS if project.name == "23550.2") + node = project.nodes[2] + assert project.format is LegacyCanFormat.ADDRESS_IN_IDENTIFIER + wire = encode_write(project.format, node.tx_id, 17, (1, 2, 0xFFFF, 4)) + assert wire.can_id & 0x7F == 17 + assert wire.data == bytes.fromhex("00 01 00 02 FF FF 00 04") + command = encode_command(project, node, 7) + assert command.can_id & 0x7F == 127 + assert command.data == b"\x00\x80"