把摄像头协商结果记进 session.json,并在达不到配置帧率时明说 - What: sdk_worker 新增 capture_status_of(sensor)(把采集源的 config 与 status 合成 {requested:{width,height,fps,fourcc}, negotiated:{width,height,fps,fourcc,backend}, dropped_frames})与 short_capture_fallacy(status)(当协商帧率低于申请值 90% 时给出一句话描述,含分辨率/实际帧率/申请帧率/格式/后端;两者都齐全时才说话);两个 worker 都加 capture_status() 访问器(box 取当前 sensor,finger 取 runtime.tactile.raw_sensor)。app_window:session.json 新增顶层 capture 块(_capture_metadata(),走同一个访问器);新增 _check_capture_negotiation()——每个会话在驱动第一次报出速率后检查一次,达不到就通过 _notify 明确提示(不是每帧刷屏),新键 status.capture_slow(中英)写明"软件改不了、记录按实际帧率保存、请降分辨率/换格式或换摄像头"。tests/test_worker_frame_rate.py:假设备补上 CaptureConfig 与 CaptureStatus 的 width/height/fps/fourcc/backend_name 字段,新增 test_a_low_camera_rate_is_recorded_as_a_negotiation(1920x1080、申请 30、协商 16.67 时 capture_status() 两半都在、描述里含 1920x1080/16.67/30;协商到位时描述为空串)。 - Why: 用户按上一轮的读数报告"还是摄像头 16.7",即摄像头自身只发布 16.7 —— 这排除了流水线(每帧只 12 ms)与应用侧节拍,剩下的只能是采集格式协商:配置申请的是 resolution_raw 1920x1080 @ 30 fps(fourcc 走 CaptureConfig 默认 MJPG)。这类"驱动只提供这个格式"的情况软件无法改变,但记录里必须留下证据、操作员必须被告知,否则一个 16.7 的采集和一次正常的 30 fps 采集在文件里长得一样,事后再也无法解释。 - Verification: shared 全量 161 passed(新增 1 项);studio 90 passed;ui_check 554 项检查 0 失败;locale_check 287 字面键全部命中、中英键集一致(343+1=344? 见下方键数)。 - Not in this change: 未改 SDK 的采集参数(仍是 1920x1080/30/MJPG、后端 auto);frames.csv 仍未记录 process_ms/capture_fps(下次改记录格式时一并加);未做自动降级(例如检测到 16.7 就自动改用更低分辨率)——那会改变传感器几何,必须由人决定。

This commit is contained in:
ykcheungab
2026-09-14 13:27:33 +08:00
parent a1314a02cb
commit 50122d19a2
6 changed files with 209 additions and 1 deletions
@@ -27,6 +27,7 @@ from orisys_app_shared.acquisition.sdk_worker import (
STOP_JOIN_TIMEOUT_S,
VIDEO_EOF_TIMEOUT_S,
calibration_digest,
capture_status_of,
loaded_calibration_version,
loaded_path,
process_chain,
@@ -126,6 +127,14 @@ class FingerAcquisitionWorker(RecordSinkMixin):
with self._lock:
return tuple(self._filters_unsupported)
def capture_status(self) -> dict:
"""What the camera was asked for, and what the driver actually gave."""
with self._lock:
runtime = next(iter(self._runtimes.values()), None)
tactile = getattr(runtime, "tactile", None) if runtime is not None else None
raw_sensor = getattr(tactile, "raw_sensor", None) if tactile is not None else None
return capture_status_of(raw_sensor)
@property
def is_running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
@@ -11,7 +11,7 @@ import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
from typing import Any, Callable, Mapping, Optional
import cv2
import numpy as np
@@ -84,6 +84,7 @@ __all__ = [
"SdkPacket",
"SdkAcquisitionWorker",
"calibration_digest",
"capture_status_of",
"load_fixed_mode",
"loaded_calibration_version",
"loaded_path",
@@ -91,6 +92,7 @@ __all__ = [
"process_chain",
"read_depth",
"read_forces_from_sensor",
"short_capture_fallacy",
]
@@ -155,6 +157,77 @@ def _capture_source(sensor):
return source
def capture_status_of(sensor) -> dict:
"""Requested vs negotiated capture parameters, or ``{}`` when unknown.
``requested`` comes from the capture config the SDK was handed (built from the
sensor.json's ``camera`` block); ``negotiated`` is what OpenCV/the driver
reported back after the stream properties were applied. They disagree exactly
when a camera cannot do what the config asks for.
"""
source = _capture_source(sensor) if sensor is not None else None
if source is None:
return {}
status = getattr(source, "status", None)
config = getattr(source, "config", None)
if status is None and config is None:
return {}
def _value(obj, name, default=None):
return getattr(obj, name, default) if obj is not None else default
dropped = _value(status, "dropped_frames", 0)
return {
"requested": {
"width": _value(config, "width"),
"height": _value(config, "height"),
"fps": _value(config, "fps"),
"fourcc": _value(config, "fourcc"),
},
"negotiated": {
"width": _value(status, "width"),
"height": _value(status, "height"),
"fps": _value(status, "fps"),
"fourcc": _value(status, "fourcc"),
"backend": _value(status, "backend_name"),
},
"dropped_frames": int(dropped or 0),
}
def short_capture_fallacy(status: Mapping[str, Any] | None) -> str:
"""A one-line description of a capture negotiation that fell short.
Empty when the driver gave what was asked for (or when either half is
unknown): this exists to be *said*, not to fill a field.
"""
if not isinstance(status, Mapping):
return ""
requested = dict(status.get("requested") or {})
negotiated = dict(status.get("negotiated") or {})
wanted = requested.get("fps")
got = negotiated.get("fps")
try:
wanted_value = float(wanted)
got_value = float(got)
except (TypeError, ValueError):
return ""
if wanted_value <= 0 or got_value <= 0 or got_value >= wanted_value * 0.9:
return ""
size = ""
width, height = negotiated.get("width"), negotiated.get("height")
if width and height:
size = f"{int(width)}x{int(height)} "
fourcc = str(negotiated.get("fourcc") or "").strip()
backend = str(negotiated.get("backend") or "").strip()
detail = "/".join(part for part in (fourcc, backend) if part)
return (
f"the camera negotiated {size}@{got_value:g} fps "
f"(the config asks for {wanted_value:g})"
+ (f" [{detail}]" if detail else "")
)
def process_chain(sensor) -> tuple[Optional[np.ndarray], str]:
"""Run the live processing chain for one already-read frame.
@@ -466,6 +539,18 @@ class SdkAcquisitionWorker(RecordSinkMixin):
with self._lock:
return tuple(self._filters_unsupported)
def capture_status(self) -> dict:
"""What the camera was asked for, and what the driver actually gave.
The negotiation is where a camera that cannot do the configured frame
rate is decided, and nothing downstream can turn 16.7 fps back into 30.
Recording both halves is what lets a recording explain its own frame
rate instead of leaving the operator to guess.
"""
with self._lock:
sensor = next(iter(self._sensors.values()), None)
return capture_status_of(sensor)
@property
def is_running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
+57
View File
@@ -31,6 +31,23 @@ class _Status:
def __init__(self) -> None:
self.last_seq = 0
self.dropped_frames = 0
#: What the driver reported back for the negotiated stream.
self.width = 1920
self.height = 1080
self.fps = 30.0
self.fourcc = "MJPG"
self.backend_name = "MSMF"
class _CaptureConfig:
"""The requested stream, as ``CaptureConfig`` carries it."""
def __init__(self) -> None:
self.width = 1920
self.height = 1080
self.fps = 30.0
self.fourcc = "MJPG"
self.source = 0
class _CaptureSource:
@@ -45,6 +62,7 @@ class _CaptureSource:
def __init__(self, period_s: float = FRAME_PERIOD_S) -> None:
self.status = _Status()
self.config = _CaptureConfig()
self.running = True
self._period = float(period_s)
self._stop = threading.Event()
@@ -265,6 +283,45 @@ def test_the_camera_s_own_rate_is_reported_next_to_ours(monkeypatch):
monkeypatch.setattr(sys.modules[__name__], "PROCESS_COST_S", 0.0)
def test_a_low_camera_rate_is_recorded_as_a_negotiation(monkeypatch):
"""A camera that cannot do the configured rate must be visible as such.
The pipeline cannot turn 16.7 fps into 30, so the only honest thing left is to
record what was asked for and what the driver gave, and to have a one-line
description ready for the operator.
"""
from orisys_app_shared.acquisition.sdk_worker import (
SdkAcquisitionWorker,
short_capture_fallacy,
)
CREATED.clear()
saved = _install()
worker = SdkAcquisitionWorker()
try:
worker.start(0, config_ref=BOX_CONFIG)
deadline = time.perf_counter() + 2.0
while time.perf_counter() < deadline and not CREATED:
time.sleep(0.01)
assert CREATED, "the fake sensor was never constructed"
source = CREATED[-1].capture_source
source.status.fps = 16.67 # what the driver actually negotiated
status = worker.capture_status()
assert status["requested"]["fps"] == 30.0
assert status["requested"]["width"] == 1920
assert status["negotiated"]["fps"] == 16.67, status
assert status["negotiated"]["fourcc"] == "MJPG"
assert status["negotiated"]["backend"] == "MSMF"
detail = short_capture_fallacy(status)
assert "1920x1080" in detail and "16.67" in detail and "30" in detail, detail
# A driver that gives what was asked for says nothing at all.
source.status.fps = 30.0
assert short_capture_fallacy(worker.capture_status()) == ""
finally:
worker.stop(wait=True)
_restore(saved)
def test_a_file_source_runs_on_the_video_s_own_timeline():
"""A pass over a video is stamped with the video's time, not the wall clock."""
from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker