ykcheungab
199 lines
6.1 KiB
Python
199 lines
6.1 KiB
Python
"""A manually confirmed site produces one traceable trial directory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from orisys_app_shared.acquisition.models import SdkPacket
|
|
from orisys_autocal.trial import (
|
|
REFERENCE_HEADER,
|
|
ManualTrial,
|
|
load_path,
|
|
site_at,
|
|
uv_origin,
|
|
)
|
|
|
|
|
|
class Worker:
|
|
def __init__(self) -> None:
|
|
self.is_running = True
|
|
self.sinks = []
|
|
self.packet = SdkPacket(
|
|
raw_frame=np.zeros((32, 32, 3), dtype=np.uint8),
|
|
processed_frame=np.zeros((32, 32, 3), dtype=np.uint8),
|
|
timestamp=time.time(),
|
|
fps=30,
|
|
process_ms=1,
|
|
session_id=2,
|
|
frame_id=1,
|
|
mono_t=1.0,
|
|
capture_mono_ns=time.perf_counter_ns(),
|
|
capture_utc_ns=time.time_ns(),
|
|
source_frame_id=7,
|
|
time_basis="host_camera",
|
|
normal_force_raw=1.25,
|
|
)
|
|
|
|
def latest(self):
|
|
return self.packet
|
|
|
|
def add_sink(self, sink):
|
|
self.sinks.append(sink)
|
|
|
|
def remove_sink(self, sink):
|
|
self.sinks.remove(sink)
|
|
|
|
def emit(self):
|
|
for sink in self.sinks:
|
|
sink(self.packet)
|
|
|
|
|
|
def test_uv_coordinates_are_centered_and_masked():
|
|
mask = np.full((5, 7), -1, dtype=np.int32)
|
|
mask[1:4, 2:5] = 2
|
|
flat = SimpleNamespace(
|
|
width=7,
|
|
height=5,
|
|
face_index=mask,
|
|
sample=lambda x, y: (
|
|
(np.array([x, y, 3.0]), np.array([0, 0, 1]), 2)
|
|
if 0 <= x < 7 and 0 <= y < 5 and mask[y, x] >= 0
|
|
else None
|
|
),
|
|
)
|
|
assert uv_origin(flat) == (3, 3)
|
|
site = site_at(flat, -1, 2)
|
|
assert site.raster_xy_px == (2, 1)
|
|
assert site.position_sensor_mm == (2.0, 1.0, 3.0)
|
|
with pytest.raises(ValueError):
|
|
site_at(flat, -3, 0)
|
|
|
|
|
|
def test_manual_trial_records_video_stamps_and_load_unload_events(tmp_dir):
|
|
worker = Worker()
|
|
trial = ManualTrial(worker)
|
|
bundle = tmp_dir / "bundle"
|
|
bundle.mkdir()
|
|
(bundle / "mesh.npz").write_bytes(b"mesh")
|
|
config = tmp_dir / "sensor.json"
|
|
config.write_text("{}", encoding="utf-8")
|
|
site = SimpleNamespace(
|
|
u_px=0,
|
|
v_px=1,
|
|
position_sensor_mm=(1.0, 2.0, 3.0),
|
|
face_index=2,
|
|
raster_xy_px=(3, 2),
|
|
uv_origin_raster_xy_px=(3, 3),
|
|
)
|
|
assert load_path([5, 10]) == [
|
|
("load", 5),
|
|
("load", 10),
|
|
("unload", 5),
|
|
("unload", 0),
|
|
]
|
|
directory = trial.start(
|
|
root=tmp_dir,
|
|
site=site,
|
|
levels=[5, 10],
|
|
hold_s=20,
|
|
sensor_model="Finger D40",
|
|
sensor_sn="F-001",
|
|
bundle_path=str(bundle),
|
|
source={"kind": "camera", "value": 0},
|
|
config_ref=str(config),
|
|
fps=30,
|
|
)
|
|
worker.emit()
|
|
for value in (5, 10, 5, 0):
|
|
trial.confirm(value)
|
|
trial.finish()
|
|
report = trial.recorder.stop(wait=True)
|
|
assert report.ok
|
|
assert trial.poll() == "complete"
|
|
assert "Finger-D40_F-001" in directory.name
|
|
assert len(list(directory.glob("raw_*.mp4"))) == 1
|
|
with (directory / "frames.csv").open(encoding="utf-8", newline="") as handle:
|
|
row = next(csv.DictReader(handle))
|
|
assert row["t_host_mono_ns"] == str(worker.packet.capture_mono_ns)
|
|
assert row["source_frame_id"] == "7"
|
|
assert row["time_basis"] == "host_camera"
|
|
assert row["capture_dropped"] == "0"
|
|
with (directory / "reference.csv").open(encoding="utf-8", newline="") as handle:
|
|
reader = csv.DictReader(handle)
|
|
assert tuple(reader.fieldnames) == REFERENCE_HEADER
|
|
references = list(reader)
|
|
assert [row["branch"] for row in references] == ["load", "load", "unload", "unload"]
|
|
assert [row["F_ref_normal_N"] for row in references] == [
|
|
"5.0",
|
|
"10.0",
|
|
"5.0",
|
|
"0.0",
|
|
]
|
|
assert all(
|
|
row["F_ref_x_N"] == row["F_ref_y_N"] == row["F_ref_z_N"] == ""
|
|
for row in references
|
|
)
|
|
assert all(row["stability_verified"] == "0" for row in references)
|
|
assert all(
|
|
int(row["t_start_host_mono_ns"]) < int(row["t_end_host_mono_ns"])
|
|
for row in references
|
|
)
|
|
assert all(
|
|
previous["t_end_host_mono_ns"] == following["t_start_host_mono_ns"]
|
|
for previous, following in zip(references, references[1:])
|
|
)
|
|
with (directory / "events.csv").open(encoding="utf-8", newline="") as handle:
|
|
events = list(csv.DictReader(handle))
|
|
reached = [event for event in events if event["kind"] == "step_reached"]
|
|
assert len(reached) == 4
|
|
assert [json.loads(event["detail"])["branch"] for event in reached] == [
|
|
"load",
|
|
"load",
|
|
"unload",
|
|
"unload",
|
|
]
|
|
with (directory / "session.json").open(encoding="utf-8") as handle:
|
|
session = json.load(handle)
|
|
assert session["site"]["uv_xy_px"] == [0, 1]
|
|
assert session["sensor_model"] == "Finger D40"
|
|
assert session["reference_file"] == "reference.csv"
|
|
assert session["site"]["mesh_sha256"]
|
|
|
|
|
|
def test_calibration_page_keeps_precise_uv_site_when_switching_to_live_view(
|
|
make_window,
|
|
):
|
|
from orisys_app_shared.i18n import t
|
|
from orisys_autocal.sensor.uv_flat import FlatSurface
|
|
|
|
window = make_window()
|
|
mask = np.full((5, 7), -1, dtype=np.int32)
|
|
mask[1:4, 2:5] = 2
|
|
positions = np.zeros((5, 7, 3), dtype=np.float64)
|
|
positions[1, 2] = (2, 1, 3)
|
|
flat = FlatSurface(
|
|
size_hw=(5, 7),
|
|
image_rgb=np.zeros((5, 7, 3), dtype=np.uint8),
|
|
positions_mm=positions,
|
|
normals=np.zeros_like(positions),
|
|
face_index=mask,
|
|
uv_min=(0, 0),
|
|
uv_max=(1, 1),
|
|
)
|
|
panel = window.calibration_panel
|
|
assert panel.start_button.text() == t("calibration.start")
|
|
assert panel.target_label.text() == t("calibration.before_start")
|
|
panel.set_flat(flat, "bundle")
|
|
panel._picked(2, 1)
|
|
assert (panel.u_spin.value(), panel.v_spin.value()) == (-1, 2)
|
|
assert panel._site.raster_xy_px == (2, 1)
|
|
assert "2.000" in panel.position_value.text()
|
|
panel.views.setCurrentWidget(panel.source_panel)
|
|
assert panel._site.raster_xy_px == (2, 1)
|
|
window.close()
|