Share PM67 upload protocol through C core

python/protocan/native.py

@@ -177,6 +177,24 @@ class NativeProtocol:
             ctypes.c_size_t, ctypes.c_size_t,
         ]
         lib.pcan_abi_periph28335_command_name.restype = ctypes.c_char_p
+        lib.pcan_abi_tms2812_crc16.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
+        lib.pcan_abi_tms2812_crc16.restype = ctypes.c_uint16
+        lib.pcan_abi_tms2812_build_upload.argtypes = [
+            ctypes.c_uint8, ctypes.c_uint32, ctypes.c_uint32,
+            ctypes.c_void_p, ctypes.c_size_t,
+        ]
+        lib.pcan_abi_tms2812_build_upload.restype = ctypes.c_size_t
+        lib.pcan_abi_tms2812_expected_upload_size.argtypes = [ctypes.c_uint32]
+        lib.pcan_abi_tms2812_expected_upload_size.restype = ctypes.c_size_t
+        lib.pcan_abi_tms2812_validate_upload.argtypes = [
+            ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint8, ctypes.c_uint32,
+        ]
+        lib.pcan_abi_tms2812_validate_upload.restype = ctypes.c_int
+        lib.pcan_abi_tms2812_decode_upload.argtypes = [
+            ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint8, ctypes.c_uint32,
+            ctypes.c_void_p, ctypes.c_size_t,
+        ]
+        lib.pcan_abi_tms2812_decode_upload.restype = ctypes.c_int
         lib.pcan_abi_frame_encode.argtypes = [
             ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint32,
             ctypes.c_void_p, ctypes.c_uint8, ctypes.c_void_p, ctypes.c_size_t,
@@ -335,6 +353,35 @@ class NativeProtocol:
             result[raw_name.decode("utf-8")] = tuple(commands)
         return result
 
+    def tms2812_crc16(self, data: bytes) -> int:
+        return int(self.lib.pcan_abi_tms2812_crc16(
+            self._bytes_buffer(data), len(data)))
+
+    def tms2812_build_upload(self, controller: int, word_address: int,
+                             byte_count: int) -> bytes:
+        output = (ctypes.c_uint8 * 12)()
+        size = int(self.lib.pcan_abi_tms2812_build_upload(
+            controller, word_address, byte_count, output, len(output)))
+        if size == 0:
+            raise ValueError("SETProtocol rejected PM67 upload request")
+        return bytes(output[:size])
+
+    def tms2812_expected_upload_size(self, byte_count: int) -> int:
+        return int(self.lib.pcan_abi_tms2812_expected_upload_size(byte_count))
+
+    def tms2812_validate_upload(self, data: bytes, controller: int,
+                                byte_count: int) -> int:
+        return int(self.lib.pcan_abi_tms2812_validate_upload(
+            self._bytes_buffer(data), len(data), controller, byte_count))
+
+    def tms2812_decode_upload(self, data: bytes, controller: int,
+                              byte_count: int) -> tuple[int, bytes]:
+        output = (ctypes.c_uint8 * byte_count)()
+        status = int(self.lib.pcan_abi_tms2812_decode_upload(
+            self._bytes_buffer(data), len(data), controller, byte_count,
+            output, len(output)))
+        return status, bytes(output) if status == 0 else b""
+
     def encode(self, sequence: int, flags: int, can_id: int, data: bytes) -> bytes:
         if len(data) > 8:
             raise ValueError("DLC cannot exceed 8 bytes")

python/protocan/tms2812.py

@@ -0,0 +1,60 @@
+"""Thin Python adapter for the shared C99 PM67/TMS320F2812 protocol."""
+
+from __future__ import annotations
+
+from .native import get_native_protocol
+
+
+CMD_UPLOAD = 52
+UPLOAD_RESPONSE_OVERHEAD = 8
+
+
+def _core():
+    return get_native_protocol()
+
+
+def crc16_modbus(data: bytes, initial: int = 0xFFFF) -> int:
+    if initial != 0xFFFF:
+        raise ValueError("Произвольное начальное значение CRC не входит в протокол ПМ67")
+    return _core().tms2812_crc16(bytes(data))
+
+
+def build_upload_request(controller: int, word_address: int,
+                         byte_count: int) -> bytes:
+    _range("адрес контроллера", controller, 0xFF)
+    _range("адрес слова", word_address, 0xFFFFFFFF)
+    if not 1 <= byte_count <= 0xFFFFFFFF:
+        raise ValueError("размер блока вне диапазона 1..0xFFFFFFFF")
+    return _core().tms2812_build_upload(controller, word_address, byte_count)
+
+
+def expected_upload_response_size(byte_count: int) -> int:
+    if not 1 <= byte_count <= 0xFFFFFFFF:
+        raise ValueError("размер блока вне диапазона 1..0xFFFFFFFF")
+    return _core().tms2812_expected_upload_size(byte_count)
+
+
+def decode_upload_reply(raw: bytes, controller: int, byte_count: int) -> bytes:
+    expected = expected_upload_response_size(byte_count)
+    if len(raw) != expected:
+        raise ValueError(
+            f"ответ CMD_UPLOAD: ожидалось {expected} байт, получено {len(raw)}"
+        )
+    status, data = _core().tms2812_decode_upload(
+        bytes(raw), controller, byte_count)
+    errors = {
+        -1: "неверные аргументы CMD_UPLOAD",
+        -2: "размер CMD_UPLOAD вне диапазона",
+        -3: "неверная длина ответа CMD_UPLOAD",
+        -4: "CRC ответа CMD_UPLOAD не совпадает",
+        -5: "ответ CMD_UPLOAD имеет неверный адрес или номер команды",
+        -6: "недостаточный буфер CMD_UPLOAD",
+    }
+    if status != 0:
+        raise ValueError(errors.get(status, f"ошибка ответа CMD_UPLOAD: {status}"))
+    return data
+
+
+def _range(name: str, value: int, maximum: int) -> None:
+    if not 0 <= value <= maximum:
+        raise ValueError(f"{name} вне диапазона 0..{maximum}")