|
|
|
|
@@ -1,6 +1,6 @@
|
|
|
|
|
"""ctypes port for the shared C99 ProtoCAN core.
|
|
|
|
|
"""ctypes port for the shared C99 SETProtocol core.
|
|
|
|
|
|
|
|
|
|
The protocol implementation lives in ``c/protocan-transport``. This module
|
|
|
|
|
The protocol implementation lives in ``c/set-protocol``. This module
|
|
|
|
|
only converts Python values to the stable ``pcan_abi.h`` interface.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
@@ -14,8 +14,12 @@ from pathlib import Path
|
|
|
|
|
from typing import Iterable
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class NativeCoreUnavailable(RuntimeError):
|
|
|
|
|
"""Raised when the SETCore shared library cannot be loaded."""
|
|
|
|
|
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):
|
|
|
|
|
@@ -53,18 +57,24 @@ class NativeGuiFrame:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _library_candidates() -> Iterable[Path | str]:
|
|
|
|
|
explicit = os.environ.get("SETCORE_LIBRARY")
|
|
|
|
|
if explicit:
|
|
|
|
|
yield Path(explicit)
|
|
|
|
|
for variable in ("SETPROTOCOL_LIBRARY", "SETCORE_LIBRARY"):
|
|
|
|
|
explicit = os.environ.get(variable)
|
|
|
|
|
if explicit:
|
|
|
|
|
yield Path(explicit)
|
|
|
|
|
here = Path(__file__).resolve()
|
|
|
|
|
names = ("setcore.dll", "libsetcore.so", "libsetcore.dylib")
|
|
|
|
|
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
|
|
|
|
|
discovered = ctypes.util.find_library("setcore")
|
|
|
|
|
if discovered:
|
|
|
|
|
yield discovered
|
|
|
|
|
for library_name in ("setprotocol", "setcore"):
|
|
|
|
|
discovered = ctypes.util.find_library(library_name)
|
|
|
|
|
if discovered:
|
|
|
|
|
yield discovered
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load_library() -> ctypes.CDLL:
|
|
|
|
|
@@ -74,13 +84,13 @@ def _load_library() -> ctypes.CDLL:
|
|
|
|
|
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)
|
|
|
|
|
raise NativeProtocolUnavailable(
|
|
|
|
|
"SETProtocol library not found. Build c/set-protocol with CMake or "
|
|
|
|
|
"set SETPROTOCOL_LIBRARY. Tried: " + "; ".join(errors)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class NativeCore:
|
|
|
|
|
class NativeProtocol:
|
|
|
|
|
"""Thin owner of the C ABI and its function signatures."""
|
|
|
|
|
|
|
|
|
|
def __init__(self, library: ctypes.CDLL | None = None) -> None:
|
|
|
|
|
@@ -88,7 +98,7 @@ class NativeCore:
|
|
|
|
|
self._bind()
|
|
|
|
|
version = int(self.lib.pcan_abi_version())
|
|
|
|
|
if version != 1:
|
|
|
|
|
raise NativeCoreUnavailable(f"unsupported SETCore ABI {version}")
|
|
|
|
|
raise NativeProtocolUnavailable(f"unsupported SETProtocol ABI {version}")
|
|
|
|
|
|
|
|
|
|
def _bind(self) -> None:
|
|
|
|
|
lib = self.lib
|
|
|
|
|
@@ -174,7 +184,7 @@ class NativeCore:
|
|
|
|
|
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")
|
|
|
|
|
raise ValueError("SETProtocol rejected the CAN frame")
|
|
|
|
|
return bytes(output[:size])
|
|
|
|
|
|
|
|
|
|
def parser(self) -> "NativeParser":
|
|
|
|
|
@@ -194,7 +204,7 @@ class NativeCore:
|
|
|
|
|
size = int(self.lib.pcan_abi_gui_frame_encode(
|
|
|
|
|
message_type, sequence, source, len(payload), output, len(output)))
|
|
|
|
|
if size == 0:
|
|
|
|
|
raise ValueError("SETCore rejected the GUI frame")
|
|
|
|
|
raise ValueError("SETProtocol rejected the GUI frame")
|
|
|
|
|
return bytes(output[:size])
|
|
|
|
|
|
|
|
|
|
def gui_parser(self) -> "NativeGuiParser":
|
|
|
|
|
@@ -202,12 +212,12 @@ class NativeCore:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class NativeParser:
|
|
|
|
|
def __init__(self, core: NativeCore) -> None:
|
|
|
|
|
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 NativeCoreUnavailable("SETCore parser initialization failed")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol parser initialization failed")
|
|
|
|
|
|
|
|
|
|
def feed(self, chunk: bytes) -> list[NativeFrame]:
|
|
|
|
|
frames: list[NativeFrame] = []
|
|
|
|
|
@@ -216,7 +226,7 @@ class NativeParser:
|
|
|
|
|
result = self._core.lib.pcan_abi_parser_push(
|
|
|
|
|
self._storage, byte, ctypes.byref(raw))
|
|
|
|
|
if result < 0:
|
|
|
|
|
raise NativeCoreUnavailable("SETCore parser rejected its context")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol parser rejected its context")
|
|
|
|
|
if result > 0:
|
|
|
|
|
frames.append(NativeFrame(
|
|
|
|
|
int(raw.sequence), int(raw.flags), int(raw.can_id),
|
|
|
|
|
@@ -227,7 +237,7 @@ class NativeParser:
|
|
|
|
|
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")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol parser stats unavailable")
|
|
|
|
|
return {
|
|
|
|
|
"frames": values[0].value,
|
|
|
|
|
"crc_errors": values[1].value,
|
|
|
|
|
@@ -237,12 +247,12 @@ class NativeParser:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class NativeGuiParser:
|
|
|
|
|
def __init__(self, core: NativeCore) -> None:
|
|
|
|
|
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 NativeCoreUnavailable("SETCore GUI parser initialization failed")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol GUI parser initialization failed")
|
|
|
|
|
|
|
|
|
|
def feed(self, chunk: bytes) -> list[NativeGuiFrame]:
|
|
|
|
|
frames: list[NativeGuiFrame] = []
|
|
|
|
|
@@ -251,7 +261,7 @@ class NativeGuiParser:
|
|
|
|
|
result = self._core.lib.pcan_abi_gui_parser_push(
|
|
|
|
|
self._storage, byte, ctypes.byref(raw))
|
|
|
|
|
if result < 0:
|
|
|
|
|
raise NativeCoreUnavailable("SETCore GUI parser rejected its context")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol GUI parser rejected its context")
|
|
|
|
|
if result > 0:
|
|
|
|
|
frames.append(NativeGuiFrame(
|
|
|
|
|
int(raw.message_type), int(raw.sequence),
|
|
|
|
|
@@ -262,7 +272,7 @@ class NativeGuiParser:
|
|
|
|
|
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 NativeCoreUnavailable("SETCore GUI parser stats unavailable")
|
|
|
|
|
raise NativeProtocolUnavailable("SETProtocol GUI parser stats unavailable")
|
|
|
|
|
return {
|
|
|
|
|
"frames": values[0].value,
|
|
|
|
|
"crc_errors": values[1].value,
|
|
|
|
|
@@ -272,11 +282,19 @@ class NativeGuiParser:
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_default_core: NativeCore | None = None
|
|
|
|
|
_default_protocol: NativeProtocol | None = None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_native_core() -> NativeCore:
|
|
|
|
|
global _default_core
|
|
|
|
|
if _default_core is None:
|
|
|
|
|
_default_core = NativeCore()
|
|
|
|
|
return _default_core
|
|
|
|
|
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()
|
|
|
|
|
|