"""ctypes port for the shared C99 SETProtocol core. The protocol implementation lives in ``c/set-protocol``. 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 NativeProtocolUnavailable(RuntimeError): """Raised when the SETProtocol shared library cannot be loaded.""" # Compatibility name for applications written against ABI v1. NativeCoreUnavailable = NativeProtocolUnavailable 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), ] class _AbiGuiFrame(ctypes.Structure): _fields_ = [ ("sequence", ctypes.c_uint16), ("size", ctypes.c_uint16), ("message_type", ctypes.c_uint8), ("payload", ctypes.c_uint8 * 512), ] @dataclass(frozen=True) class NativeFrame: sequence: int flags: int can_id: int data: bytes @dataclass(frozen=True) class NativeGuiFrame: message_type: int sequence: int payload: bytes def _library_candidates() -> Iterable[Path | str]: for variable in ("SETPROTOCOL_LIBRARY", "SETCORE_LIBRARY"): explicit = os.environ.get(variable) if explicit: yield Path(explicit) here = Path(__file__).resolve() names = ( "setprotocol.dll", "libsetprotocol.so", "libsetprotocol.dylib", # One-release fallback for already packaged SETCore binaries. "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 for library_name in ("setprotocol", "setcore"): discovered = ctypes.util.find_library(library_name) 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 NativeProtocolUnavailable( "SETProtocol library not found. Build c/set-protocol with CMake or " "set SETPROTOCOL_LIBRARY. Tried: " + "; ".join(errors) ) class NativeProtocol: """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 NativeProtocolUnavailable(f"unsupported SETProtocol 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 lib.pcan_abi_gui_crc32.argtypes = [ctypes.c_void_p, ctypes.c_size_t] lib.pcan_abi_gui_crc32.restype = ctypes.c_uint32 lib.pcan_abi_gui_frame_encode.argtypes = [ ctypes.c_uint8, ctypes.c_uint16, ctypes.c_void_p, ctypes.c_uint16, ctypes.c_void_p, ctypes.c_size_t, ] lib.pcan_abi_gui_frame_encode.restype = ctypes.c_size_t lib.pcan_abi_gui_parser_size.restype = ctypes.c_size_t lib.pcan_abi_gui_parser_init.argtypes = [ctypes.c_void_p, ctypes.c_size_t] lib.pcan_abi_gui_parser_init.restype = ctypes.c_int lib.pcan_abi_gui_parser_push.argtypes = [ ctypes.c_void_p, ctypes.c_uint8, ctypes.POINTER(_AbiGuiFrame), ] lib.pcan_abi_gui_parser_push.restype = ctypes.c_int lib.pcan_abi_gui_parser_stats.argtypes = [ ctypes.c_void_p, *(ctypes.POINTER(ctypes.c_uint32) for _ in range(5)), ] lib.pcan_abi_gui_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("SETProtocol rejected the CAN frame") return bytes(output[:size]) def parser(self) -> "NativeParser": return NativeParser(self) def gui_crc32(self, data: bytes) -> int: source = (ctypes.c_uint8 * len(data)).from_buffer_copy(data) if data else None return int(self.lib.pcan_abi_gui_crc32(source, len(data))) def gui_encode(self, message_type: int, sequence: int, payload: bytes) -> bytes: if len(payload) > 512: raise ValueError("GUI payload cannot exceed 512 bytes") source = ((ctypes.c_uint8 * len(payload)).from_buffer_copy(payload) if payload else None) output = (ctypes.c_uint8 * (8 + 512 + 4))() size = int(self.lib.pcan_abi_gui_frame_encode( message_type, sequence, source, len(payload), output, len(output))) if size == 0: raise ValueError("SETProtocol rejected the GUI frame") return bytes(output[:size]) def gui_parser(self) -> "NativeGuiParser": return NativeGuiParser(self) class NativeParser: def __init__(self, core: NativeProtocol) -> 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 NativeProtocolUnavailable("SETProtocol 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 NativeProtocolUnavailable("SETProtocol 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 NativeProtocolUnavailable("SETProtocol parser stats unavailable") return { "frames": values[0].value, "crc_errors": values[1].value, "bad_len": values[2].value, "stray_bytes": values[3].value, } class NativeGuiParser: def __init__(self, core: NativeProtocol) -> None: self._core = core size = int(core.lib.pcan_abi_gui_parser_size()) self._storage = ctypes.create_string_buffer(size) if not core.lib.pcan_abi_gui_parser_init(self._storage, size): raise NativeProtocolUnavailable("SETProtocol GUI parser initialization failed") def feed(self, chunk: bytes) -> list[NativeGuiFrame]: frames: list[NativeGuiFrame] = [] raw = _AbiGuiFrame() for byte in chunk: result = self._core.lib.pcan_abi_gui_parser_push( self._storage, byte, ctypes.byref(raw)) if result < 0: raise NativeProtocolUnavailable("SETProtocol GUI parser rejected its context") if result > 0: frames.append(NativeGuiFrame( int(raw.message_type), int(raw.sequence), bytes(raw.payload[:raw.size]))) return frames def stats(self) -> dict[str, int]: values = [ctypes.c_uint32() for _ in range(5)] if not self._core.lib.pcan_abi_gui_parser_stats( self._storage, *(ctypes.byref(value) for value in values)): raise NativeProtocolUnavailable("SETProtocol GUI parser stats unavailable") return { "frames": values[0].value, "crc_errors": values[1].value, "version_errors": values[2].value, "length_errors": values[3].value, "stray_bytes": values[4].value, } _default_protocol: NativeProtocol | None = None def get_native_protocol() -> NativeProtocol: global _default_protocol if _default_protocol is None: _default_protocol = NativeProtocol() return _default_protocol # Source compatibility for existing SETGUI code during the rename. NativeCore = NativeProtocol def get_native_core() -> NativeProtocol: return get_native_protocol()