回放叠加改用展开底图，并统一 2D/3D 投递。
箭头/采集图像与 live 的 processed_frame 对齐；deliver_measurement 先画 2D 再提交同一包给 3D。已用 overlay、scene3d 与 replay 相关测试验证。
This commit is contained in:
ykcheungab
@@ -0,0 +1,202 @@
|
||||
"""Session-scoped overlay base: UV unroll or box stitch without re-running flow.
|
||||
|
||||
Replay stores flow and depth, not the processed image. Live paints arrows on
|
||||
``SdkPacket.processed_frame`` (finger UV unroll or box stitch). This projector
|
||||
rebuilds that same base from a decoded camera frame using the result's profile,
|
||||
so scrubbing can match live without caching rendered pixels or reopening a video
|
||||
per frame.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = ["OverlayBaseProjector"]
|
||||
|
||||
_CWD_LOCK = threading.Lock()
|
||||
|
||||
|
||||
def _ensure_bgr(image: np.ndarray) -> np.ndarray:
|
||||
if image.ndim == 2:
|
||||
return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)
|
||||
if image.ndim == 3 and image.shape[2] == 1:
|
||||
return cv2.cvtColor(image[:, :, 0], cv2.COLOR_GRAY2BGR)
|
||||
if image.ndim == 3 and image.shape[2] == 4:
|
||||
return cv2.cvtColor(image, cv2.COLOR_BGRA2BGR)
|
||||
return np.ascontiguousarray(image)
|
||||
|
||||
|
||||
class OverlayBaseProjector:
|
||||
"""Maps one camera frame onto the measurement domain for overlay views."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
sensor_type: str,
|
||||
project,
|
||||
close=None,
|
||||
working: Path | None = None,
|
||||
) -> None:
|
||||
self.sensor_type = str(sensor_type)
|
||||
self._project = project
|
||||
self._close = close
|
||||
self._working = None if working is None else Path(working)
|
||||
self._cache_id: int | None = None
|
||||
self._cache: np.ndarray | None = None
|
||||
|
||||
@classmethod
|
||||
def open_from_provenance(cls, provenance: Any) -> OverlayBaseProjector | None:
|
||||
"""Build a projector from an analysis result's provenance, or ``None``."""
|
||||
kind = str(getattr(provenance, "sensor_type", "") or "").strip()
|
||||
config_ref = str(getattr(provenance, "sensor_config_ref", "") or "").strip()
|
||||
if kind == "finger":
|
||||
return cls._open_finger(config_ref=config_ref or None)
|
||||
if kind == "box":
|
||||
return cls._open_box(config_ref=config_ref or None)
|
||||
logger.warning("no overlay projector for sensor type %r", kind)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _open_finger(cls, *, config_ref: str | None) -> OverlayBaseProjector | None:
|
||||
from orisys.finger.runtime import resolve_finger_geometry
|
||||
from orisys.frame_processors import FrameUnwarper
|
||||
from orisys.unroll import build_or_load_domain
|
||||
from orisys_app_shared.analysis.service import resolve_finger_export_dir
|
||||
|
||||
if not config_ref:
|
||||
logger.warning("finger overlay needs sensor_config_ref from provenance")
|
||||
return None
|
||||
export_dir = resolve_finger_export_dir(config_ref)
|
||||
if export_dir is None:
|
||||
return None
|
||||
try:
|
||||
geometry = resolve_finger_geometry(
|
||||
export_dir=export_dir,
|
||||
config=str(config_ref),
|
||||
)
|
||||
mesh_path, intrinsics_path, pose_path = geometry.export_paths.as_tuple()
|
||||
domain = build_or_load_domain(
|
||||
mesh_path,
|
||||
intrinsics_path,
|
||||
pose_path,
|
||||
geometry.map_cache,
|
||||
geometry.source_K,
|
||||
geometry.source_width,
|
||||
geometry.source_height,
|
||||
geometry.out_width,
|
||||
geometry.out_height,
|
||||
edge_expand=geometry.edge_expand,
|
||||
include_surface=True,
|
||||
verbose=False,
|
||||
source_scale=geometry.source_scale,
|
||||
)
|
||||
domain.remap.ensure_fast_maps()
|
||||
unwarper = FrameUnwarper(
|
||||
domain.remap,
|
||||
camera_matrix=(
|
||||
geometry.source_K if geometry.dist_coeffs is not None else None
|
||||
),
|
||||
dist_coeffs=geometry.dist_coeffs,
|
||||
source_width=geometry.source_width,
|
||||
source_height=geometry.source_height,
|
||||
flip_source_vertical=True,
|
||||
)
|
||||
except Exception: # noqa: BLE001 - missing assets become "capture unavailable"
|
||||
logger.warning("could not open finger overlay projector", exc_info=True)
|
||||
return None
|
||||
|
||||
def project(frame: np.ndarray) -> np.ndarray | None:
|
||||
# Match live ``process_frame(..., color=False)``: grey unroll, then BGR.
|
||||
prepared = unwarper.prepare_source_gray(frame)
|
||||
unrolled = unwarper.flatten_prepared_gray(prepared)
|
||||
if unrolled is None:
|
||||
return None
|
||||
return _ensure_bgr(unrolled)
|
||||
|
||||
return cls(sensor_type="finger", project=project)
|
||||
|
||||
@classmethod
|
||||
def _open_box(cls, *, config_ref: str | None) -> OverlayBaseProjector | None:
|
||||
from orisys import Sensor
|
||||
|
||||
working = Path(tempfile.mkdtemp(prefix="orisys-overlay-box-"))
|
||||
try:
|
||||
with _CWD_LOCK:
|
||||
previous = Path.cwd()
|
||||
os.chdir(working)
|
||||
try:
|
||||
sensor = Sensor(
|
||||
"external",
|
||||
config_name=str(config_ref) if config_ref else "box",
|
||||
verbose=False,
|
||||
backend="auto",
|
||||
)
|
||||
finally:
|
||||
os.chdir(previous)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.warning("could not open box overlay projector", exc_info=True)
|
||||
shutil.rmtree(working, ignore_errors=True)
|
||||
return None
|
||||
|
||||
def project(frame: np.ndarray) -> np.ndarray | None:
|
||||
# Stitch only — no deformation / contact. Same grey domain live publishes.
|
||||
processed = sensor.get_img(frame=frame)
|
||||
if processed is None:
|
||||
return None
|
||||
return _ensure_bgr(processed)
|
||||
|
||||
def close() -> None:
|
||||
try:
|
||||
sensor.disconnect()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("box overlay sensor disconnect skipped", exc_info=True)
|
||||
shutil.rmtree(working, ignore_errors=True)
|
||||
|
||||
return cls(sensor_type="box", project=project, close=close, working=working)
|
||||
|
||||
def project(
|
||||
self, video_bgr: np.ndarray, *, frame_id: int | None = None
|
||||
) -> np.ndarray | None:
|
||||
"""Return the measurement-domain BGR base, or ``None`` when remap fails."""
|
||||
if frame_id is not None and self._cache is not None and self._cache_id == int(
|
||||
frame_id
|
||||
):
|
||||
return self._cache
|
||||
image = np.asarray(video_bgr)
|
||||
if image.ndim == 2:
|
||||
image = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)
|
||||
elif image.ndim == 3 and image.shape[2] == 4:
|
||||
image = cv2.cvtColor(image, cv2.COLOR_BGRA2BGR)
|
||||
try:
|
||||
projected = self._project(image)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.warning("overlay projection failed", exc_info=True)
|
||||
return None
|
||||
if projected is None:
|
||||
return None
|
||||
result = np.ascontiguousarray(projected)
|
||||
if frame_id is not None:
|
||||
self._cache_id = int(frame_id)
|
||||
self._cache = result
|
||||
return result
|
||||
|
||||
def close(self) -> None:
|
||||
self._cache = None
|
||||
self._cache_id = None
|
||||
if self._close is not None:
|
||||
try:
|
||||
self._close()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.debug("overlay projector close skipped", exc_info=True)
|
||||
self._close = None
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Overlay base projector: session cache and no fisheye resize fallback."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
|
||||
from orisys_app_shared.analysis.overlay_base import OverlayBaseProjector
|
||||
|
||||
|
||||
def test_overlay_projector_reuses_the_same_frame_id() -> None:
|
||||
calls: list[int] = []
|
||||
|
||||
def project(frame: np.ndarray) -> np.ndarray:
|
||||
calls.append(1)
|
||||
out = np.zeros((8, 8, 3), dtype=np.uint8)
|
||||
out[..., 1] = 40
|
||||
return out
|
||||
|
||||
projector = OverlayBaseProjector(sensor_type="finger", project=project)
|
||||
first = projector.project(np.zeros((32, 32, 3), dtype=np.uint8), frame_id=7)
|
||||
second = projector.project(np.ones((32, 32, 3), dtype=np.uint8) * 9, frame_id=7)
|
||||
assert len(calls) == 1
|
||||
assert first is second
|
||||
third = projector.project(np.zeros((32, 32, 3), dtype=np.uint8), frame_id=8)
|
||||
assert len(calls) == 2
|
||||
assert third is not second
|
||||
projector.close()
|
||||
|
||||
|
||||
def test_open_from_unknown_sensor_type_returns_none() -> None:
|
||||
class _Prov:
|
||||
sensor_type = "unknown"
|
||||
sensor_config_ref = ""
|
||||
|
||||
assert OverlayBaseProjector.open_from_provenance(_Prov()) is None
|
||||
Reference in New Issue
Block a user