ykcheungab
06e06126ca
选点、对准、加载与录像分散在两个面板，状态靠隐藏相位判断。现改由不依赖 Qt 的采集控制器驱动，并加入录像选项、汇总平铺副本、保持到时自动结束、会话结束回到空闲，以及无设备时的模拟采集。已运行 Autocal 非 Qt 测试并通过范围内门禁；界面测试待 Windows 复核，未做实机验收。
589 lines
20 KiB
Python
589 lines
20 KiB
Python
"""Synthetic recording evidence for the acquisition, summary and review tests.
|
|
|
|
Two rules govern this module:
|
|
|
|
* a fixture writes the **shared recorder's real headers**, so a reader is tested
|
|
against the files it will meet rather than against a convenient subset;
|
|
* a video fixture is a **real decodable clip**. A file of fake MP4 bytes proves
|
|
nothing about seeking, decoding or frame identity.
|
|
|
|
Nothing here touches the app's preference file, its ``outputs/`` directory or a
|
|
real camera. Fixtures are built inside the test's ``tmp_path``; the run's
|
|
``--basetemp`` points at the task scratch directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from orisys_app_shared.acquisition.models import SdkPacket
|
|
from orisys_app_shared.record import FORCES_HEADER, FRAMES_HEADER
|
|
|
|
#: One working day in nanoseconds: holds inside a fixture are this far apart.
|
|
HOLD_STRIDE_NS = 10_000_000_000
|
|
#: A fixed, arbitrary session-relative origin for synthetic host timestamps.
|
|
BASE_NS = 1_700_000_000_000_000_000
|
|
|
|
FRAMES_FIELDS = list(FRAMES_HEADER)
|
|
FORCES_FIELDS = list(FORCES_HEADER)
|
|
|
|
|
|
def write_csv(path: Path, fields: list[str], rows: list[dict]) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
return path
|
|
|
|
|
|
def read_csv(path: Path) -> list[dict]:
|
|
with path.open(encoding="utf-8-sig", newline="") as handle:
|
|
return list(csv.DictReader(handle))
|
|
|
|
|
|
def frame_row(
|
|
frame_id: int,
|
|
t_host_mono_ns: int,
|
|
*,
|
|
segment: int = 1,
|
|
segment_frame: int | None = None,
|
|
in_video: int = 1,
|
|
dropped: int = 0,
|
|
time_basis: str = "host_camera",
|
|
t_rel_s: float = 0.0,
|
|
) -> dict:
|
|
return {
|
|
"frame_id": frame_id,
|
|
"source_frame_id": frame_id,
|
|
"t_rel_s": repr(float(t_rel_s)),
|
|
"t_host_mono_ns": int(t_host_mono_ns),
|
|
"t_host_utc_ns": int(t_host_mono_ns),
|
|
"acquired_at": "2026-10-17T00:00:00.000+00:00",
|
|
"time_basis": time_basis,
|
|
"media_time_ms": "",
|
|
"segment": segment if in_video else 0,
|
|
"segment_frame": (segment_frame if segment_frame is not None else frame_id)
|
|
if in_video
|
|
else 0,
|
|
"in_video": in_video,
|
|
"dropped": dropped,
|
|
"capture_dropped": 0,
|
|
}
|
|
|
|
|
|
def force_row(
|
|
frame_id: int,
|
|
*,
|
|
fn: float | None,
|
|
fx: float | None = 0.0,
|
|
fy: float | None = 0.0,
|
|
valid: int = 1,
|
|
session_id: int = 7,
|
|
t_rel_s: float = 0.0,
|
|
) -> dict:
|
|
def cell(value):
|
|
return "" if value is None else repr(float(value))
|
|
|
|
return {
|
|
"session_id": session_id,
|
|
"frame_id": frame_id,
|
|
"t_rel_s": repr(float(t_rel_s)),
|
|
"FN_raw": cell(fn),
|
|
"FX_raw": cell(fx),
|
|
"FY_raw": cell(fy),
|
|
"FN_N": "",
|
|
"FX_N": "",
|
|
"FY_N": "",
|
|
"valid": valid,
|
|
"calibration_id": "",
|
|
"FN_output_raw": "",
|
|
"FX_output_raw": "",
|
|
"FY_output_raw": "",
|
|
"FN_output_N": "",
|
|
"FX_output_N": "",
|
|
"FY_output_N": "",
|
|
"Fmag_raw": "",
|
|
}
|
|
|
|
|
|
def write_video(path: Path, *, frames: int = 6, fps: float = 10.0) -> Path:
|
|
"""A real, short, decodable clip whose frame number is visible in the picture.
|
|
|
|
Frame ``i`` carries a white 8x8 marker at column ``4 + (i % 6) * 8``. A test
|
|
can therefore ask the decoder which frame it got instead of trusting a seek.
|
|
"""
|
|
import cv2
|
|
import numpy as np
|
|
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
size = (64, 48) # width, height
|
|
writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), float(fps), size)
|
|
if not writer.isOpened():
|
|
raise RuntimeError("the MP4 writer is unavailable in this environment")
|
|
try:
|
|
for index in range(frames):
|
|
image = np.zeros((size[1], size[0], 3), dtype=np.uint8)
|
|
column = 4 + (index % 6) * 8
|
|
image[4:12, column : column + 8] = 255
|
|
writer.write(image)
|
|
finally:
|
|
writer.release()
|
|
return path
|
|
|
|
|
|
def decoded_frame_number(path: Path, *, segment_frame: int | None = None) -> int:
|
|
"""Decode one frame of a :func:`write_video` clip and read its marker column."""
|
|
import cv2
|
|
|
|
capture = cv2.VideoCapture(str(path))
|
|
try:
|
|
if segment_frame is not None:
|
|
capture.set(cv2.CAP_PROP_POS_FRAMES, max(0, int(segment_frame) - 1))
|
|
ok, image = capture.read()
|
|
if not ok:
|
|
raise RuntimeError(f"could not decode {path.name}")
|
|
column = int(image[4:12].sum(axis=(0, 2)).argmax())
|
|
return (column - 4) // 8
|
|
finally:
|
|
capture.release()
|
|
|
|
|
|
def write_capture(
|
|
directory: Path,
|
|
frames: list[dict],
|
|
forces: list[dict],
|
|
*,
|
|
fields: list[str] | None = None,
|
|
video_segments: int = 1,
|
|
video_frames: int = 6,
|
|
) -> Path:
|
|
"""Write ``frames.csv``, ``forces.csv`` and one real MP4 per video segment."""
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
write_csv(directory / "frames.csv", fields or FRAMES_FIELDS, frames)
|
|
write_csv(directory / "forces.csv", FORCES_FIELDS, forces)
|
|
for segment in range(1, video_segments + 1):
|
|
write_video(directory / f"raw_{segment:04d}.mp4", frames=video_frames)
|
|
write_json_file(
|
|
directory / "session.json",
|
|
{
|
|
"format_version": "1.4",
|
|
"session_id": forces[0]["session_id"] if forces else 7,
|
|
"files": {"frames": "frames.csv", "forces": "forces.csv", "video": "raw_%04d.mp4"},
|
|
},
|
|
)
|
|
return directory
|
|
|
|
|
|
def write_json_file(path: Path, payload: dict) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Hold:
|
|
"""One declared reference window as the campaign writes it."""
|
|
|
|
step_index: int
|
|
branch: str
|
|
target_load_n: float
|
|
reference_n: float
|
|
start_ns: int
|
|
end_ns: int
|
|
complete: int = 1
|
|
|
|
@property
|
|
def stability_verified(self) -> int:
|
|
return 0
|
|
|
|
|
|
def reference_row(hold: Hold, *, site_id: str, repeat_index: int, repeat_count: int) -> dict:
|
|
row = {name: "" for name in (
|
|
"step_index", "branch", "target_load_n", "t_start_host_mono_ns",
|
|
"t_end_host_mono_ns", "F_ref_normal_N", "site_id", "x", "y", "z",
|
|
"coordinate_frame", "repeat_index", "repeat_count", "hold_complete",
|
|
"reference_source", "fx_N", "fy_N", "fz_N", "force_to_tool_rotation",
|
|
"nx", "ny", "nz", "pose_x_mm", "pose_y_mm", "pose_yaw_deg",
|
|
"pose_pitch_deg", "pose_z_mm", "normal_aligned_to",
|
|
)}
|
|
row.update({
|
|
"step_index": hold.step_index,
|
|
"branch": hold.branch,
|
|
"target_load_n": hold.target_load_n,
|
|
"t_start_host_mono_ns": hold.start_ns,
|
|
"t_end_host_mono_ns": hold.end_ns,
|
|
"F_ref_normal_N": hold.reference_n,
|
|
"site_id": site_id,
|
|
"x": 0.0, "y": 0.0, "z": 0.0,
|
|
"coordinate_frame": "sensor",
|
|
"repeat_index": repeat_index,
|
|
"repeat_count": repeat_count,
|
|
"hold_complete": hold.complete,
|
|
"reference_source": "weight",
|
|
})
|
|
return row
|
|
|
|
|
|
REFERENCE_FIELDS = list(reference_row(
|
|
Hold(0, "baseline", 0.0, 0.0, 0, 1), site_id="", repeat_index=1, repeat_count=1,
|
|
).keys())
|
|
|
|
|
|
def holds_for_levels(
|
|
levels: list[float], *, first_start_ns: int = BASE_NS, hold_ns: int = 4_000_000_000,
|
|
) -> list[Hold]:
|
|
"""baseline, one load per level, then unload back to zero."""
|
|
steps: list[tuple[str, float]] = [("baseline", 0.0)]
|
|
steps += [("load", float(level)) for level in levels]
|
|
steps += [("unload", value) for value in [*reversed(levels[:-1]), 0.0]]
|
|
holds = []
|
|
for index, (branch, target) in enumerate(steps):
|
|
start = first_start_ns + index * HOLD_STRIDE_NS
|
|
holds.append(Hold(index, branch, target, target, start, start + hold_ns))
|
|
return holds
|
|
|
|
|
|
def hold_samples(
|
|
hold: Hold,
|
|
*,
|
|
count: int = 3,
|
|
first_frame_id: int,
|
|
raw_base: float,
|
|
raw_per_n: float,
|
|
raw_per_axis: dict[str, float] | None = None,
|
|
) -> tuple[list[dict], list[dict]]:
|
|
"""Evenly spaced samples inside a hold, raw linear in the declared force."""
|
|
frames: list[dict] = []
|
|
forces: list[dict] = []
|
|
span = (hold.end_ns - hold.start_ns) // (count + 1)
|
|
fn = raw_base + raw_per_n * hold.reference_n
|
|
for offset in range(count):
|
|
frame_id = first_frame_id + offset
|
|
stamp = hold.start_ns + span * (offset + 1)
|
|
frames.append(frame_row(frame_id, stamp, t_rel_s=(stamp - BASE_NS) / 1e9))
|
|
values = {"fn": fn}
|
|
if raw_per_axis:
|
|
values.update({
|
|
"fx": raw_per_axis["FX"] * hold.reference_n,
|
|
"fy": raw_per_axis["FY"] * hold.reference_n,
|
|
})
|
|
forces.append(force_row(frame_id, **values))
|
|
return frames, forces
|
|
|
|
|
|
def _write_multipoint_trial(
|
|
path: Path,
|
|
*,
|
|
session_id: str,
|
|
site_id: str,
|
|
repeat_index: int,
|
|
repeat_count: int,
|
|
holds: list[Hold],
|
|
with_capture: bool,
|
|
video: bool,
|
|
raw_base: float,
|
|
raw_per_n: float,
|
|
samples_per_hold: int,
|
|
frame_start: int,
|
|
capture_status: str | None = None,
|
|
) -> list[dict]:
|
|
write_csv(
|
|
path / "reference.csv",
|
|
REFERENCE_FIELDS,
|
|
[reference_row(hold, site_id=site_id, repeat_index=repeat_index,
|
|
repeat_count=repeat_count) for hold in holds],
|
|
)
|
|
write_csv(
|
|
path / "events.csv",
|
|
["t_host_mono_ns", "t_host_utc_ns", "kind", "detail"],
|
|
[],
|
|
)
|
|
manifest = {
|
|
"trial_id": f"{site_id}-repeat-{repeat_index:03d}",
|
|
"site_id": site_id,
|
|
"status": "complete",
|
|
"uv_px": {"u": 128, "v": 64},
|
|
"images": {"captured": with_capture,
|
|
"reason": "" if with_capture else "camera not attached yet"},
|
|
}
|
|
if capture_status is not None:
|
|
manifest["capture_status"] = capture_status
|
|
write_json_file(path / "session.json", manifest)
|
|
if not with_capture:
|
|
return []
|
|
frames: list[dict] = []
|
|
forces: list[dict] = []
|
|
for hold in holds:
|
|
hold_frames, hold_forces = hold_samples(
|
|
hold,
|
|
count=samples_per_hold,
|
|
first_frame_id=frame_start + len(frames),
|
|
raw_base=raw_base,
|
|
raw_per_n=raw_per_n,
|
|
)
|
|
frames.extend(hold_frames)
|
|
forces.extend(hold_forces)
|
|
if not video:
|
|
# A data-only recording: the rows exist, no frame was written to a video.
|
|
frames = [
|
|
{**row, "segment": 0, "segment_frame": 0, "in_video": 0} for row in frames
|
|
]
|
|
write_capture(path / "capture", frames, forces, fields=FRAMES_FIELDS,
|
|
video_segments=1 if video else 0)
|
|
return []
|
|
|
|
|
|
def multipoint_session(
|
|
root: Path,
|
|
*,
|
|
sites: tuple[str, ...] = ("P1",),
|
|
repeats: int = 1,
|
|
levels: tuple[float, ...] = (5.0,),
|
|
with_capture: bool = True,
|
|
video: bool = True,
|
|
raw_base: float = 100.0,
|
|
raw_per_n: float = 40.0,
|
|
samples_per_hold: int = 3,
|
|
simulate: bool = False,
|
|
trial_status: str = "complete",
|
|
extra_holds: list[Hold] | None = None,
|
|
recording_block: dict | None = None,
|
|
capture_status: str | None = None,
|
|
) -> Path:
|
|
"""A complete multi-point session directory through the review contract.
|
|
|
|
``video=False`` writes a data-only recording: frame rows exist with
|
|
``in_video=0`` and no MP4 is created. ``recording_block`` writes the root
|
|
manifest's ``recording`` block; ``None`` keeps the fixture shaped like a
|
|
session recorded before that block existed. ``capture_status`` puts the
|
|
lifecycle into the trial records, as the campaign does.
|
|
"""
|
|
session_id = f"finger-SN1_{root.name}_test0001"
|
|
trials: list[dict] = []
|
|
completion: list[dict] = []
|
|
site_records = []
|
|
for site_id in sites:
|
|
site_records.append({
|
|
"site_id": site_id,
|
|
"position": {"x": 1.0, "y": 2.0, "z": 3.0},
|
|
"position_frame": "sensor",
|
|
"uv_px": {"u": 128, "v": 64},
|
|
"levels_n": list(levels),
|
|
"hold_s": 4.0,
|
|
"repeats": repeats,
|
|
})
|
|
complete = 0
|
|
for repeat in range(1, repeats + 1):
|
|
site_dir = root / "sites" / site_id / f"repeat-{repeat:03d}"
|
|
holds = holds_for_levels(list(levels))
|
|
if extra_holds:
|
|
holds = [*holds, *extra_holds]
|
|
directory = f"sites/{site_id}/repeat-{repeat:03d}"
|
|
_write_multipoint_trial(
|
|
site_dir,
|
|
session_id=session_id,
|
|
site_id=site_id,
|
|
repeat_index=repeat,
|
|
repeat_count=repeats,
|
|
holds=holds,
|
|
with_capture=with_capture,
|
|
video=video,
|
|
raw_base=raw_base,
|
|
raw_per_n=raw_per_n,
|
|
samples_per_hold=samples_per_hold,
|
|
frame_start=1,
|
|
capture_status=capture_status,
|
|
)
|
|
trial_record = {
|
|
"trial_id": f"{site_id}-repeat-{repeat:03d}",
|
|
"site_id": site_id,
|
|
"repeat_index": repeat,
|
|
"directory": directory,
|
|
"status": trial_status,
|
|
"reason": "" if trial_status == "complete" else "stopped by the operator",
|
|
}
|
|
if capture_status is not None:
|
|
trial_record["capture_status"] = capture_status
|
|
trials.append(trial_record)
|
|
if trial_status == "complete":
|
|
complete += 1
|
|
completion.append({"site_id": site_id, "repeats_planned": repeats,
|
|
"repeats_complete": complete,
|
|
"complete": complete >= repeats})
|
|
manifest = {
|
|
"format_version": "autocal-multipoint-1.0",
|
|
"session_id": session_id,
|
|
"sensor_model": "finger",
|
|
"sensor_sn": "SN1",
|
|
"coordinate_frame": "sensor",
|
|
"max_force_n": 20.0,
|
|
"min_force_n": 0.0,
|
|
"simulate": simulate,
|
|
"status": "complete" if trial_status == "complete" else "incomplete",
|
|
"sites": site_records,
|
|
"trials": trials,
|
|
"completion": completion,
|
|
"started_at": "2026-10-17T00:00:00+00:00",
|
|
"sync": {"association": "A frame belongs to a hold when its t_host_mono_ns is in "
|
|
"[t_start_host_mono_ns, t_end_host_mono_ns). Rows are not paired."},
|
|
"platform_motion": False,
|
|
"automatic_force_control": False,
|
|
}
|
|
if recording_block is not None:
|
|
manifest["recording"] = recording_block
|
|
write_json_file(root / "session.json", manifest)
|
|
return root
|
|
|
|
|
|
def single_session(
|
|
root: Path,
|
|
*,
|
|
sites: tuple[str, ...] = ("site-001",),
|
|
trials_per_site: int = 1,
|
|
levels: tuple[float, ...] = (5.0,),
|
|
with_capture: bool = True,
|
|
raw_base: float = 100.0,
|
|
raw_per_n: float = 40.0,
|
|
samples_per_hold: int = 3,
|
|
trial_status: str = "complete",
|
|
reference_alignment_possible: bool = True,
|
|
) -> Path:
|
|
"""A complete single-trial calibration session (``autocal-session-1.0``)."""
|
|
session_id = f"finger-SN1_{root.name}_test0001"
|
|
uv_definition = {
|
|
"unit": "px", "u_positive": "right", "v_positive": "up",
|
|
"origin_raster_xy_px": [0, 0], "raster_size_hw": [64, 64],
|
|
"mapping_sha256": "test-mapping",
|
|
}
|
|
site_records = [
|
|
{"site_id": site_id, "uv_xy_px": [10 + index, 20 + index],
|
|
"position_sensor_mm": [1.0, 2.0, 3.0]}
|
|
for index, site_id in enumerate(sites)
|
|
]
|
|
trials: list[dict] = []
|
|
for site_index, site_id in enumerate(sites):
|
|
for repeat in range(1, trials_per_site + 1):
|
|
trial_id = f"trial-{len(trials) + 1:03d}"
|
|
directory = f"trials/{trial_id}"
|
|
path = root / directory
|
|
holds = holds_for_levels(list(levels))
|
|
write_csv(path / "reference.csv", REFERENCE_FIELDS, [
|
|
reference_row(hold, site_id=site_id, repeat_index=repeat, repeat_count=1)
|
|
for hold in holds
|
|
])
|
|
frames: list[dict] = []
|
|
forces: list[dict] = []
|
|
if with_capture:
|
|
for hold in holds:
|
|
hold_frames, hold_forces = hold_samples(
|
|
hold, count=samples_per_hold,
|
|
first_frame_id=1 + len(frames),
|
|
raw_base=raw_base, raw_per_n=raw_per_n,
|
|
)
|
|
frames.extend(hold_frames)
|
|
forces.extend(hold_forces)
|
|
write_csv(path / "frames.csv", FRAMES_FIELDS, frames)
|
|
write_csv(path / "forces.csv", FORCES_FIELDS, forces)
|
|
write_json_file(path / "session.json", {
|
|
"trial_id": trial_id,
|
|
"site_id": site_id,
|
|
"session_id": 7,
|
|
"calibration_session_id": session_id,
|
|
"trial_status": trial_status,
|
|
"trial_reason": "" if trial_status == "complete" else "application closed",
|
|
"reference_alignment_possible": reference_alignment_possible,
|
|
"uv_definition": uv_definition,
|
|
"steps": [{"step_index": hold.step_index, "branch": hold.branch,
|
|
"target_load_n": hold.target_load_n} for hold in holds],
|
|
})
|
|
trials.append({
|
|
"trial_id": trial_id, "site_id": site_id, "repeat_index": repeat,
|
|
"directory": directory, "status": trial_status,
|
|
"reason": "" if trial_status == "complete" else "application closed",
|
|
})
|
|
write_json_file(root / "session.json", {
|
|
"format_version": "autocal-session-1.0",
|
|
"session_id": session_id,
|
|
"status": "complete" if trial_status == "complete" else "incomplete",
|
|
"sites": site_records,
|
|
"trials": trials,
|
|
"uv_definition": uv_definition,
|
|
"sensor_model": "finger",
|
|
"sensor_sn": "SN1",
|
|
"started_at": "2026-10-17T00:00:00+00:00",
|
|
})
|
|
return root
|
|
|
|
|
|
class FakeWorker:
|
|
"""A fingerprint worker stand-in whose session can be restarted mid-recording."""
|
|
|
|
def __init__(self, *, session_id: int = 1, frame_id: int = 1, time_basis: str = "host_camera",
|
|
clock=None):
|
|
self.session_id = int(session_id)
|
|
self.is_running = True
|
|
self.sinks: list = []
|
|
self.time_basis = time_basis
|
|
self._clock = clock
|
|
self._frame_id = int(frame_id)
|
|
self._started = self._now_ns()
|
|
self.packet = self._packet()
|
|
|
|
def _now_ns(self) -> int:
|
|
return time.perf_counter_ns() if self._clock is None else int(self._clock())
|
|
|
|
def _packet(self) -> SdkPacket:
|
|
import numpy as np
|
|
|
|
now = self._now_ns()
|
|
return SdkPacket(
|
|
raw_frame=np.zeros((16, 16, 3), dtype=np.uint8),
|
|
processed_frame=np.zeros((16, 16, 3), dtype=np.uint8),
|
|
timestamp=time.time(),
|
|
fps=30,
|
|
process_ms=1,
|
|
session_id=self.session_id,
|
|
frame_id=self._frame_id,
|
|
mono_t=(now - self._started) / 1e9,
|
|
capture_mono_ns=now,
|
|
capture_utc_ns=time.time_ns(),
|
|
source_frame_id=self._frame_id,
|
|
time_basis=self.time_basis,
|
|
normal_force_raw=1.0,
|
|
)
|
|
|
|
def advance(self) -> SdkPacket:
|
|
self._frame_id += 1
|
|
self.packet = self._packet()
|
|
return self.packet
|
|
|
|
def latest(self):
|
|
return self.packet
|
|
|
|
def add_sink(self, sink) -> None:
|
|
if sink not in self.sinks:
|
|
self.sinks.append(sink)
|
|
|
|
def remove_sink(self, sink) -> None:
|
|
if sink in self.sinks:
|
|
self.sinks.remove(sink)
|
|
|
|
def emit(self) -> None:
|
|
for sink in list(self.sinks):
|
|
sink(self.packet)
|
|
|
|
def restart(self) -> None:
|
|
"""A new acquisition session with fresh frame numbers, sinks still attached."""
|
|
self.session_id += 1
|
|
self._frame_id = 1
|
|
self._started = self._now_ns()
|
|
self.packet = self._packet()
|
|
|
|
def stop(self, *, wait: bool = True) -> None:
|
|
self.is_running = False
|