ykcheungab
83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
"""Exercise real SDK instances for 1-4 recorded inputs, without UART/hardware."""
|
|
|
|
import argparse
|
|
import itertools
|
|
import os
|
|
import sys
|
|
from functools import partial
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "serial"))
|
|
from finger_api import FingerSensorBank
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--recording", required=True, type=Path)
|
|
parser.add_argument("--scratch", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
scratch = args.scratch.resolve()
|
|
scratch.mkdir(parents=True, exist_ok=True)
|
|
os.environ["NUMBA_CACHE_DIR"] = str(scratch / "numba")
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import orisys
|
|
from orisys.finger import FingerRuntime
|
|
|
|
print(f"SDK: {orisys.__file__}", flush=True)
|
|
capture = cv2.VideoCapture(str(args.recording.resolve()))
|
|
frames = []
|
|
try:
|
|
for _ in range(12):
|
|
ok, frame = capture.read()
|
|
if not ok:
|
|
raise RuntimeError("recording must contain at least 12 frames")
|
|
frames.append(frame)
|
|
finally:
|
|
capture.release()
|
|
height, width = frames[0].shape[:2]
|
|
sources = []
|
|
for sensor_id in range(1, 5):
|
|
path = scratch / f"sensor-{sensor_id}.avi"
|
|
writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"MJPG"), 30, (width, height))
|
|
try:
|
|
if not writer.isOpened():
|
|
raise RuntimeError("cannot create test recording")
|
|
for frame in frames:
|
|
writer.write(np.roll(frame, sensor_id, axis=1))
|
|
finally:
|
|
writer.release()
|
|
sources.append(str(path))
|
|
|
|
# Keep the SDK implementation intact; redirect only its generated UV cache.
|
|
open_runtime = partial(FingerRuntime.open, map_cache=str(scratch / "uv-map.npz"))
|
|
with patch.object(FingerRuntime, "open", side_effect=open_runtime):
|
|
for count in range(1, 5):
|
|
bank = FingerSensorBank()
|
|
try:
|
|
for source in sources[:count]:
|
|
existing = dict(bank.runtimes)
|
|
bank.add(source)
|
|
assert all(bank.runtimes[i] is runtime for i, runtime in existing.items())
|
|
assert len(bank.read_all()) == len(bank.runtimes)
|
|
runtimes = list(bank.runtimes.values())
|
|
for left, right in itertools.combinations(runtimes, 2):
|
|
assert left.capture is not right.capture
|
|
assert left.tactile.raw_sensor is not right.tactile.raw_sensor
|
|
assert not np.shares_memory(left.tactile.flow, right.tactile.flow)
|
|
for _ in range(3):
|
|
readings = bank.read_all()
|
|
assert [r.sensor_id for r in readings] == list(range(1, count + 1))
|
|
assert all(np.isfinite([r.normal, r.shear_x, r.shear_y]).all() for r in readings)
|
|
print(f"PASS {count} sources: incremental add/read, independent states, 3 final rounds", flush=True)
|
|
finally:
|
|
bank.close()
|
|
assert all(runtime.capture is None for runtime in runtimes)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|