数据录制

pip install wuji-sdk
# 想在本地解析 mcap 再装
pip install mcap

QualitySummary(total_frames=8308, dropped=0, drop_rate=0.0000, sync_rate=0.8571)

from wuji_sdk import SdkManager
manager = SdkManager.instance()
glove = manager.connect(sn=serial_number, device_name="glove_0")

from wuji_sdk import TopicRecorder
recorder = TopicRecorder(compression="lz4")
recorder.record(glove.tactile().subscribe())
recorder.record(glove.emf_poses().subscribe())
recorder.record(glove.hand_skeleton().subscribe())

from pathlib import Path
Path("./data").mkdir(exist_ok=True)
handle = await recorder.start("./data/session.mcap")

import asyncio
try:
    await asyncio.sleep(10)
finally:
    stop_task = asyncio.ensure_future(handle.stop())
    try:
        summary = await asyncio.shield(stop_task)
    except asyncio.CancelledError:
        summary = await stop_task
    print(summary.total_frames, summary.file_size, summary.duration_s)
# 外层 finally 中执行 manager.disconnect_all()

python examples/python/wuji_glove/2.recording.py

  SN=WG1KA00XXXXXXXXX, Type=DeviceType.WujiGlove, Address=192.168.1.100:50001
Connected: WG1KA00XXXXXXXXX
Recording to ./data/<date>_<time>.mcap ...
Done — 3939 frames, 3.43 MB, 10.0s

stop_task = asyncio.ensure_future(handle.stop())
try:
    summary = await asyncio.shield(stop_task)
except asyncio.CancelledError:
    summary = await stop_task

mcap info ./data/<date>_<time>.mcap

from mcap.reader import make_reader

with open("./data/<date>_<time>.mcap", "rb") as fp:
    reader = make_reader(fp)
    summary = reader.get_summary()
    for cid, channel in summary.channels.items():
        n = summary.statistics.channel_message_counts.get(cid, 0)
        print(f"{channel.topic}: {n} 条")

    for schema, channel, message in reader.iter_messages(topics=["/WG1KA00XXXXXXXXX/tactile"]):
        print(message.log_time, len(message.data))