feat: update examples and README with streaming support and uint32 force types

- Change CForce3D fx/fy/fz from int16 to uint32 to match hardware
- Add independent C++ example with command and streaming modes
- Rewrite Python example with threaded streaming (read_loop + consumer pattern)
- Add ROS2 C++ publisher/subscriber examples
- Update README with streaming APIs, ROS2 docs, data type definitions, and full FFI table
- Add CHANGELOG
This commit is contained in:
lenn
2026-05-08 17:41:46 +08:00
parent c195234771
commit 705375085f
15 changed files with 903 additions and 149 deletions

View File

@@ -1,7 +1,7 @@
import ctypes
from ctypes import (
Structure, POINTER, c_void_p, c_char, c_char_p, c_uint8, c_uint16,
c_uint32, c_uint64, c_int16, c_bool
c_uint32, c_uint64, c_bool
)
LIB_PATH = "./libeskin_finger_sdk.so"
@@ -14,6 +14,29 @@ class EskinSdkVersion(Structure):
]
class CForce3D(Structure):
_fields_ = [
("fx", c_uint32),
("fy", c_uint32),
("fz", c_uint32),
]
class CCombinedForce(Structure):
_fields_ = [
("module", c_uint32),
("force", CForce3D),
]
class CFingerSample(Structure):
_fields_ = [
("timestamp_us", c_uint64),
("sequence", c_uint32),
("combined_force", CCombinedForce),
]
class EskinDevice:
def __init__(self):
self._lib = ctypes.CDLL(LIB_PATH)
@@ -90,6 +113,22 @@ class EskinDevice:
c_void_p, c_uint8, POINTER(c_uint16)
]
# Streaming bindings
lib.eskin_start_stream.restype = c_uint32
lib.eskin_start_stream.argtypes = [c_void_p]
lib.eskin_stop_stream.restype = c_uint32
lib.eskin_stop_stream.argtypes = [c_void_p]
lib.eskin_read_sample.restype = c_uint32
lib.eskin_read_sample.argtypes = [
c_void_p, c_uint32, POINTER(CFingerSample)
]
lib.eskin_get_mode.restype = c_uint32
lib.eskin_get_mode.argtypes = [c_void_p, POINTER(c_uint32)]
def version(self) -> tuple:
v = self._lib.eskin_version()
return (v.major, v.minor, v.patch)
@@ -210,8 +249,38 @@ class EskinDevice:
raise RuntimeError(f"write_matrix_col failed: error={err}")
return ret.value
# Streaming methods
def start_stream(self):
"""启动流式采集"""
err = self._lib.eskin_start_stream(self._handle)
if err != 0:
raise RuntimeError(f"start_stream failed: error={err}")
def stop_stream(self):
"""停止流式采集"""
err = self._lib.eskin_stop_stream(self._handle)
if err != 0:
raise RuntimeError(f"stop_stream failed: error={err}")
def read_sample(self, timeout_ms: int = 200) -> CFingerSample:
"""读取一个采样数据(流模式下调用)"""
sample = CFingerSample()
err = self._lib.eskin_read_sample(self._handle, timeout_ms, ctypes.byref(sample))
if err != 0:
raise RuntimeError(f"read_sample failed: error={err}")
return sample
def get_mode(self) -> int:
"""查询当前设备模式0=Command, 1=Streaming"""
out = c_uint32(0)
err = self._lib.eskin_get_mode(self._handle, ctypes.byref(out))
if err != 0:
raise RuntimeError(f"get_mode failed: error={err}")
return out.value
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
self.close()