303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""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",
|
|
)
|
|
# Python 3.8 (used by the Windows 7 build) does not support slicing
|
|
# Path.parents directly. Convert the sequence before taking the slice.
|
|
for parent in (here.parent, *tuple(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()
|