155 lines
5.3 KiB
Python
155 lines
5.3 KiB
Python
"""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_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
|
|
|
|
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 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
|
|
|
|
|
|
_default_core: NativeCore | None = None
|
|
|
|
|
|
def get_native_core() -> NativeCore:
|
|
global _default_core
|
|
if _default_core is None:
|
|
_default_core = NativeCore()
|
|
return _default_core
|