From 0f77a510e6edbaa6f81ecbd5104b01a8236277cb Mon Sep 17 00:00:00 2001 From: Andrey Date: Mon, 31 Aug 2026 20:46:52 +0300 Subject: [PATCH] =?UTF-8?q?feat(protocan):=20=D0=BE=D1=82=D0=BA=D1=80?= =?UTF-8?q?=D0=BE=D0=B9=20=D1=80=D0=B0=D0=B7=D0=B1=D0=BE=D1=80=20ID=20?= =?UTF-8?q?=D0=B8=20=D1=81=D1=82=D0=B0=D1=82=D0=B8=D1=81=D1=82=D0=B8=D0=BA?= =?UTF-8?q?=D1=83=20FFI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python/protocan/native.py | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/python/protocan/native.py b/python/protocan/native.py index 4f12a56..f5a824e 100644 --- a/python/protocan/native.py +++ b/python/protocan/native.py @@ -82,6 +82,12 @@ class NativeCore: 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 = [ @@ -96,12 +102,32 @@ class NativeCore: 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))) @@ -143,6 +169,18 @@ class NativeParser: 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