Объединить SETProtocol v2 и общие кодеки #1
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user