"""ctypes port for the shared C99 ProtoCAN core. The protocol implementation lives in ``c/protocan-transport``. This module only converts Python values to the stable ``pcan_abi.h`` interface. """ from __future__ import annotations import ctypes import ctypes.util import os from dataclasses import dataclass from pathlib import Path from typing import Iterable class NativeCoreUnavailable(RuntimeError): """Raised when the SETCore shared library cannot be loaded.""" class _AbiFrame(ctypes.Structure): _fields_ = [ ("can_id", ctypes.c_uint32), ("sequence", ctypes.c_uint8), ("flags", ctypes.c_uint8), ("dlc", ctypes.c_uint8), ("data", ctypes.c_uint8 * 8), ] @dataclass(frozen=True) class NativeFrame: sequence: int flags: int can_id: int data: bytes def _library_candidates() -> Iterable[Path | str]: explicit = os.environ.get("SETCORE_LIBRARY") if explicit: yield Path(explicit) here = Path(__file__).resolve() names = ("setcore.dll", "libsetcore.so", "libsetcore.dylib") for parent in (here.parent, *here.parents[:5]): for name in names: yield parent / "native" / name yield parent / name discovered = ctypes.util.find_library("setcore") if discovered: yield discovered def _load_library() -> ctypes.CDLL: errors: list[str] = [] for candidate in _library_candidates(): try: return ctypes.CDLL(str(candidate)) except OSError as exc: errors.append(f"{candidate}: {exc}") raise NativeCoreUnavailable( "SETCore library not found. Build c/protocan-transport with CMake or " "set SETCORE_LIBRARY. Tried: " + "; ".join(errors) ) class NativeCore: """Thin owner of the C ABI and its function signatures.""" def __init__(self, library: ctypes.CDLL | None = None) -> None: self.lib = library or _load_library() self._bind() version = int(self.lib.pcan_abi_version()) if version != 1: raise NativeCoreUnavailable(f"unsupported SETCore ABI {version}") def _bind(self) -> None: lib = self.lib lib.pcan_abi_version.restype = ctypes.c_uint32 lib.pcan_abi_id_pack.argtypes = [ ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint16, ] lib.pcan_abi_id_pack.restype = ctypes.c_uint32 lib.pcan_abi_id_unpack.argtypes = [ ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(ctypes.c_uint16), ] lib.pcan_abi_crc16.argtypes = [ctypes.c_void_p, ctypes.c_size_t] lib.pcan_abi_crc16.restype = ctypes.c_uint16 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, ] lib.pcan_abi_frame_encode.restype = ctypes.c_size_t lib.pcan_abi_parser_size.restype = ctypes.c_size_t lib.pcan_abi_parser_init.argtypes = [ctypes.c_void_p, ctypes.c_size_t] lib.pcan_abi_parser_init.restype = ctypes.c_int lib.pcan_abi_parser_push.argtypes = [ ctypes.c_void_p, ctypes.c_uint8, ctypes.POINTER(_AbiFrame), ] lib.pcan_abi_parser_push.restype = ctypes.c_int lib.pcan_abi_parser_stats.argtypes = [ ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint32), ] lib.pcan_abi_parser_stats.restype = ctypes.c_int def id_pack(self, priority: int, route: int, device_type: int, device_id: int, message_type: int, body: int) -> int: return int(self.lib.pcan_abi_id_pack( priority, route, device_type, device_id, message_type, body)) def id_unpack(self, raw: int) -> tuple[int, int, int, int, int, int]: priority = ctypes.c_uint8() route = ctypes.c_uint8() device_type = ctypes.c_uint8() device_id = ctypes.c_uint8() message_type = ctypes.c_uint8() body = ctypes.c_uint16() self.lib.pcan_abi_id_unpack( raw, ctypes.byref(priority), ctypes.byref(route), ctypes.byref(device_type), ctypes.byref(device_id), ctypes.byref(message_type), ctypes.byref(body)) return (priority.value, route.value, device_type.value, device_id.value, message_type.value, body.value) def crc16(self, data: bytes) -> int: source = (ctypes.c_uint8 * len(data)).from_buffer_copy(data) if data else None return int(self.lib.pcan_abi_crc16(source, len(data))) def encode(self, sequence: int, flags: int, can_id: int, data: bytes) -> bytes: if len(data) > 8: raise ValueError("DLC cannot exceed 8 bytes") source = (ctypes.c_uint8 * len(data)).from_buffer_copy(data) if data else None output = (ctypes.c_uint8 * 19)() size = int(self.lib.pcan_abi_frame_encode( sequence, flags, can_id, source, len(data), output, len(output))) if size == 0: raise ValueError("SETCore rejected the CAN frame") return bytes(output[:size]) def parser(self) -> "NativeParser": return NativeParser(self) class NativeParser: def __init__(self, core: NativeCore) -> None: self._core = core size = int(core.lib.pcan_abi_parser_size()) self._storage = ctypes.create_string_buffer(size) if not core.lib.pcan_abi_parser_init(self._storage, size): raise NativeCoreUnavailable("SETCore parser initialization failed") def feed(self, chunk: bytes) -> list[NativeFrame]: frames: list[NativeFrame] = [] raw = _AbiFrame() for byte in chunk: result = self._core.lib.pcan_abi_parser_push( self._storage, byte, ctypes.byref(raw)) if result < 0: raise NativeCoreUnavailable("SETCore parser rejected its context") if result > 0: frames.append(NativeFrame( int(raw.sequence), int(raw.flags), int(raw.can_id), bytes(raw.data[:raw.dlc]))) return frames def stats(self) -> dict[str, int]: values = [ctypes.c_uint32() for _ in range(4)] if not self._core.lib.pcan_abi_parser_stats( self._storage, *(ctypes.byref(value) for value in values)): raise NativeCoreUnavailable("SETCore parser stats unavailable") return { "frames": values[0].value, "crc_errors": values[1].value, "bad_len": values[2].value, "stray_bytes": values[3].value, } _default_core: NativeCore | None = None def get_native_core() -> NativeCore: global _default_core if _default_core is None: _default_core = NativeCore() return _default_core