diff --git a/README.md b/README.md index a1de056..43ecc06 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,27 @@ occlusion, and it is exactly the non-overlapping-coverage problem this sim exists to benchmark. Reproduce: `uv run --with 'imageio[ffmpeg]' python scripts/record_multiview.py`.* +## Coverage and handoff reports + +Any saved manifest can be reduced to one entity-frame sample at a time and +reported as JSON. Add `--panel` to render the same ground-truth metrics as a +headless PNG; matplotlib remains a lazy script-only dependency. + +```bash +uv run python scripts/coverage_metrics.py scene.json +uv run --with matplotlib python scripts/coverage_metrics.py scene.json \ + --panel docs/assets/coverage_metrics.png +``` + +The panel combines per-camera coverage bars with overlap, handoff and blind-gap +totals. Its timeline makes the difference between one-camera coverage, overlapping +coverage and a true blind frame visible without using rendered pixels or an +external tracker. + +![Coverage panel for the handoff_ltr scene: per-camera fractions, overlap and handoff totals, and a frame-by-frame coverage timeline.](docs/assets/handoff_ltr_coverage_metrics.png) + +![Coverage panel for the assembly-line scene: complementary overview and worktop camera coverage across the operator and parts.](docs/assets/assembly_line_coverage_metrics.png) + **Can you recover a 3D point, or a human joint, when it is hidden in some camera views but still seen in others?** multicam-sim builds the synthetic multi-camera scenes you need to ask that question with ground truth in hand. diff --git a/docs/assets/assembly_line_coverage_metrics.png b/docs/assets/assembly_line_coverage_metrics.png new file mode 100644 index 0000000..e008e8b Binary files /dev/null and b/docs/assets/assembly_line_coverage_metrics.png differ diff --git a/docs/assets/handoff_ltr_coverage_metrics.png b/docs/assets/handoff_ltr_coverage_metrics.png new file mode 100644 index 0000000..c3ba11a Binary files /dev/null and b/docs/assets/handoff_ltr_coverage_metrics.png differ diff --git a/scripts/coverage_metrics.py b/scripts/coverage_metrics.py new file mode 100644 index 0000000..4ed69a4 --- /dev/null +++ b/scripts/coverage_metrics.py @@ -0,0 +1,210 @@ +"""Report public-safe coverage metrics for a multicam-sim manifest. + +Usage:: + + uv run python scripts/coverage_metrics.py path/to/manifest.json + uv run --with matplotlib python scripts/coverage_metrics.py \ + path/to/manifest.json --panel docs/assets/coverage_metrics.png + +JSON is always printed to stdout. ``--panel`` additionally writes a headless, +static PNG with per-camera coverage, scene-level totals, and an entity timeline +showing blind, single-camera, overlap, and handoff frames. Matplotlib is imported +lazily, so the JSON-only path keeps the package and CI dependency-free. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from multicam_sim.coverage import CoverageReport, compute_coverage_metrics +from multicam_sim.manifest import Manifest + + +def _coverage_matrix(manifest: Manifest) -> tuple[list[int], list[str], list[list[int]]]: + """Return frame ids, entity ids, and camera counts for the panel timeline. + + ``-1`` means that an entity has no sample at that frame, ``0`` is a blind + frame, ``1`` is single-camera coverage, and values of two or more are overlap. + Like :func:`compute_coverage_metrics`, named points are reduced to one + entity-frame sample so skeletons do not receive extra weight. + """ + frames = sorted({frame.frame for entity in manifest.entities for frame in entity.frames}) + entity_ids: list[str] = [] + matrix: list[list[int]] = [] + + for entity in manifest.entities: + counts: dict[int, int] = {} + for frame in entity.frames: + cameras = { + observation.cam + for point in frame.points.values() + for observation in point.per_cam + if observation.in_view + } + counts[frame.frame] = len(cameras) + entity_ids.append(entity.id) + matrix.append([counts.get(frame, -1) for frame in frames]) + + return frames, entity_ids, matrix + + +def render_coverage_panel( + manifest: Manifest, + report: CoverageReport, + output: Path, + *, + title: str, +) -> Path: + """Render a deterministic, headless coverage-and-handoff panel to ``output``.""" + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt + from matplotlib.colors import BoundaryNorm, ListedColormap + from matplotlib.lines import Line2D + from matplotlib.patches import Patch + + frames, entity_ids, matrix = _coverage_matrix(manifest) + output.parent.mkdir(parents=True, exist_ok=True) + + figure = plt.figure(figsize=(12, 6.75), facecolor="#111318", layout="constrained") + grid = figure.add_gridspec(2, 2, height_ratios=(1.0, 1.15), width_ratios=(1.7, 1.0)) + bars = figure.add_subplot(grid[0, 0]) + summary = figure.add_subplot(grid[0, 1]) + timeline = figure.add_subplot(grid[1, :]) + + for axis in (bars, summary, timeline): + axis.set_facecolor("#191c23") + axis.tick_params(colors="#c9d1d9") + for spine in axis.spines.values(): + spine.set_color("#343a46") + + camera_labels = [f"camera {row.camera_id}" for row in report.per_camera] + fractions = [row.fraction for row in report.per_camera] + positions = list(range(len(camera_labels))) + bar_colors = ["#58a6ff" if fraction < 1.0 else "#3fb950" for fraction in fractions] + bars.barh(positions, fractions, color=bar_colors, height=0.62) + bars.set_yticks(positions, camera_labels) + bars.invert_yaxis() + bars.set_xlim(0.0, 1.08) + bars.set_xlabel("fraction of entity-frame samples in view", color="#c9d1d9") + bars.set_title("Per-camera coverage", color="#f0f6fc", loc="left", weight="bold") + bars.grid(axis="x", color="#30363d", alpha=0.7, linewidth=0.8) + bars.set_axisbelow(True) + for position, row in zip(positions, report.per_camera, strict=True): + bars.text( + min(row.fraction + 0.018, 1.01), + position, + f"{row.fraction:.3f} ({row.in_view_frames}/{row.total_frames})", + va="center", + color="#f0f6fc", + fontsize=9, + ) + + sample_count = report.per_camera[0].total_frames if report.per_camera else 0 + summary.axis("off") + summary.set_title("Scene summary", color="#f0f6fc", loc="left", weight="bold") + summary_rows = ( + ("Entity-frame samples", sample_count, "#c9d1d9"), + ("Overlap frames", report.overlap_count, "#39d0c8"), + ("Handoff events", len(report.handoff_points), "#f2cc60"), + ("Blind-gap frames", report.blind_gap_count, "#ff7b72"), + ) + for row_index, (label, value, color) in enumerate(summary_rows): + y = 0.82 - row_index * 0.2 + summary.text(0.02, y, label, color="#8b949e", fontsize=10, transform=summary.transAxes) + summary.text( + 0.96, + y, + str(value), + color=color, + fontsize=18, + weight="bold", + ha="right", + transform=summary.transAxes, + ) + + if frames and entity_ids: + clipped = [[min(value, 2) for value in row] for row in matrix] + colors = ListedColormap(["#30363d", "#da3633", "#388bfd", "#2ea043"]) + norm = BoundaryNorm([-1.5, -0.5, 0.5, 1.5, 2.5], colors.N) + timeline.imshow(clipped, aspect="auto", interpolation="nearest", cmap=colors, norm=norm) + timeline.set_yticks(range(len(entity_ids)), entity_ids) + tick_step = max(1, len(frames) // 10) + tick_positions = list(range(0, len(frames), tick_step)) + if tick_positions[-1] != len(frames) - 1: + tick_positions.append(len(frames) - 1) + timeline.set_xticks(tick_positions, [str(frames[index]) for index in tick_positions]) + frame_positions = {frame: index for index, frame in enumerate(frames)} + entity_positions = {entity_id: index for index, entity_id in enumerate(entity_ids)} + for handoff in report.handoff_points: + x = frame_positions.get(handoff.frame) + y = entity_positions.get(handoff.entity_id) + if x is not None and y is not None: + timeline.scatter(x, y, marker="v", s=58, color="#f2cc60", edgecolor="#111318") + timeline.set_xlabel("frame", color="#c9d1d9") + timeline.set_title( + "Entity timeline (triangle = camera-set change)", + color="#f0f6fc", + loc="left", + weight="bold", + ) + timeline.set_xticks([index - 0.5 for index in range(1, len(frames))], minor=True) + timeline.grid(which="minor", axis="x", color="#111318", linewidth=0.35, alpha=0.5) + else: + timeline.text( + 0.5, + 0.5, + "No entity frames in manifest", + ha="center", + va="center", + color="#8b949e", + transform=timeline.transAxes, + ) + + legend = [ + Patch(facecolor="#30363d", label="no sample"), + Patch(facecolor="#da3633", label="blind (0 cameras)"), + Patch(facecolor="#388bfd", label="1 camera"), + Patch(facecolor="#2ea043", label="overlap (2+ cameras)"), + Line2D([0], [0], marker="v", color="none", markerfacecolor="#f2cc60", label="handoff"), + ] + figure.legend( + handles=legend, + loc="outside lower center", + ncols=5, + frameon=False, + labelcolor="#c9d1d9", + ) + figure.suptitle(title, color="#f0f6fc", fontsize=16, weight="bold") + figure.savefig(output, dpi=160, facecolor=figure.get_facecolor()) + plt.close(figure) + return output + + +def main() -> None: + """Load a manifest, print JSON, and optionally render a static PNG panel.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("manifest", type=Path, help="multicam-sim manifest JSON") + parser.add_argument("--panel", type=Path, help="also write a static coverage panel PNG") + parser.add_argument("--title", help="panel title (defaults to the manifest filename)") + args = parser.parse_args() + + manifest = Manifest.model_validate_json(args.manifest.read_text()) + report = compute_coverage_metrics(manifest) + if args.panel is not None: + render_coverage_panel( + manifest, + report, + args.panel, + title=args.title or f"Coverage metrics — {args.manifest.stem}", + ) + print(f"wrote coverage panel: {args.panel}", file=sys.stderr) + print(json.dumps(report.to_dict(), indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/src/multicam_sim/__init__.py b/src/multicam_sim/__init__.py index 31f6d32..1209885 100644 --- a/src/multicam_sim/__init__.py +++ b/src/multicam_sim/__init__.py @@ -29,6 +29,13 @@ ) from .appearance import AppearanceTable, EntityAppearance, write_appearance_json from .cameras import Camera, Intrinsics +from .coverage import ( + CameraCoverage, + CoverageReport, + FrameRef, + HandoffPoint, + compute_coverage_metrics, +) from .dropout import SensorDropout from .entities import Entity, EntityFrame from .groups import ( @@ -99,7 +106,9 @@ "BackgroundSpec", "Box", "CalibrationDrift", + "CameraCoverage", "CausalTiming", + "CoverageReport", "Cylinder", "Camera", "CameraManifest", @@ -115,10 +124,12 @@ "EntityFrame", "EntityManifest", "FrameObs", + "FrameRef", "GroupFrameMembership", "GroupMembership", "HandKeyframe", "HandOccluder", + "HandoffPoint", "InteractionEvent", "Intrinsics", "Light", @@ -157,6 +168,7 @@ "build_pose_smoke_scene", "build_smoke_scene", "compute_group_membership", + "compute_coverage_metrics", "export_coco", "export_overlay", "export_yolo", diff --git a/src/multicam_sim/coverage.py b/src/multicam_sim/coverage.py new file mode 100644 index 0000000..4adcad4 --- /dev/null +++ b/src/multicam_sim/coverage.py @@ -0,0 +1,140 @@ +"""Coverage and handoff metrics computed only from manifest ground truth.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from typing import Any + +from .manifest import Manifest + + +@dataclass(frozen=True) +class CameraCoverage: + """Coverage of one camera across all entity-frame samples.""" + + camera_id: int + in_view_frames: int + total_frames: int + + @property + def fraction(self) -> float: + """Return the fraction of entity-frame samples seen by this camera.""" + return self.in_view_frames / self.total_frames if self.total_frames else 0.0 + + +@dataclass(frozen=True) +class FrameRef: + """A frame belonging to one manifest entity.""" + + entity_id: str + frame: int + + +@dataclass(frozen=True) +class HandoffPoint: + """A change in the cameras covering an entity between adjacent frames.""" + + entity_id: str + frame: int + entered_cameras: tuple[int, ...] + exited_cameras: tuple[int, ...] + + +@dataclass(frozen=True) +class CoverageReport: + """Public-safe multi-camera coverage metrics for one manifest.""" + + per_camera: tuple[CameraCoverage, ...] + overlap_frames: tuple[FrameRef, ...] + handoff_points: tuple[HandoffPoint, ...] + blind_gap_frames: tuple[FrameRef, ...] + + @property + def overlap_count(self) -> int: + """Return the number of entity-frame samples seen by at least two cameras.""" + return len(self.overlap_frames) + + @property + def blind_gap_count(self) -> int: + """Return the number of entity-frame samples seen by no cameras.""" + return len(self.blind_gap_frames) + + def to_dict(self) -> dict[str, Any]: + """Return a deterministic, JSON-ready representation.""" + return { + "per_camera": [{**asdict(row), "fraction": row.fraction} for row in self.per_camera], + "overlap_count": self.overlap_count, + "overlap_frames": [asdict(frame) for frame in self.overlap_frames], + "handoff_points": [asdict(point) for point in self.handoff_points], + "blind_gap_count": self.blind_gap_count, + "blind_gap_frames": [asdict(frame) for frame in self.blind_gap_frames], + } + + +def _in_view_cameras(manifest: Manifest, entity_index: int, frame_index: int) -> set[int]: + frame = manifest.entities[entity_index].frames[frame_index] + return { + observation.cam + for point in frame.points.values() + for observation in point.per_cam + if observation.in_view + } + + +def compute_coverage_metrics(manifest: Manifest) -> CoverageReport: + """Compute camera coverage, overlaps, handoffs, and blind gaps. + + One sample is one entity at one frame. An entity is considered in view on a + camera when any of its named points has ``in_view=True`` on that camera. This + avoids overweighting skeletons merely because they have more named points + than single-point objects. + """ + camera_ids = sorted(camera.id for camera in manifest.cameras) + counts = {camera_id: 0 for camera_id in camera_ids} + total_frames = 0 + overlaps: list[FrameRef] = [] + handoffs: list[HandoffPoint] = [] + blind_gaps: list[FrameRef] = [] + + for entity_index, entity in enumerate(manifest.entities): + previous: set[int] | None = None + previous_frame: int | None = None + for frame_index, frame in enumerate(entity.frames): + in_view = _in_view_cameras(manifest, entity_index, frame_index) + frame_ref = FrameRef(entity.id, frame.frame) + total_frames += 1 + for camera_id in in_view: + counts[camera_id] = counts.get(camera_id, 0) + 1 + + if len(in_view) >= 2: + overlaps.append(frame_ref) + if not in_view: + blind_gaps.append(frame_ref) + + if ( + previous is not None + and previous_frame is not None + and frame.frame == previous_frame + 1 + and in_view != previous + and in_view + and previous + ): + handoffs.append( + HandoffPoint( + entity_id=entity.id, + frame=frame.frame, + entered_cameras=tuple(sorted(in_view - previous)), + exited_cameras=tuple(sorted(previous - in_view)), + ) + ) + previous = in_view + previous_frame = frame.frame + + return CoverageReport( + per_camera=tuple( + CameraCoverage(camera_id, counts[camera_id], total_frames) for camera_id in camera_ids + ), + overlap_frames=tuple(overlaps), + handoff_points=tuple(handoffs), + blind_gap_frames=tuple(blind_gaps), + ) diff --git a/tests/test_coverage_metrics.py b/tests/test_coverage_metrics.py new file mode 100644 index 0000000..ae9dbc8 --- /dev/null +++ b/tests/test_coverage_metrics.py @@ -0,0 +1,88 @@ +"""Ground-truth coverage metrics for overlapping and complementary scenes.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import ModuleType + +import pytest + +from multicam_sim import build_handoff_ltr_scene, build_manifest, compute_coverage_metrics +from multicam_sim.handoff_ltr import CAM_WIDE + + +def _assembly_example() -> ModuleType: + example_path = Path(__file__).resolve().parents[1] / "examples" / "assembly_station.py" + spec = importlib.util.spec_from_file_location("assembly_station", example_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _coverage_script() -> ModuleType: + script_path = Path(__file__).resolve().parents[1] / "scripts" / "coverage_metrics.py" + spec = importlib.util.spec_from_file_location("coverage_metrics", script_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_handoff_scene_reports_overlap_without_blind_gaps() -> None: + report = compute_coverage_metrics(build_manifest(build_handoff_ltr_scene())) + by_camera = {row.camera_id: row for row in report.per_camera} + + assert by_camera[CAM_WIDE].fraction == 1.0 + assert report.overlap_count > 0 + assert report.blind_gap_count == 0 + assert report.handoff_points + + +def test_assembly_scene_reports_complementary_camera_coverage() -> None: + example = _assembly_example() + report = compute_coverage_metrics(build_manifest(example.build_scene())) + by_camera = {row.camera_id: row for row in report.per_camera} + + assert by_camera[0].fraction == pytest.approx(0.25) + assert by_camera[1].fraction == pytest.approx(0.75) + assert report.overlap_count == 0 + assert report.blind_gap_count == 0 + + +def test_report_dict_is_json_ready_and_includes_frame_identity() -> None: + report = compute_coverage_metrics(build_manifest(build_handoff_ltr_scene())) + payload = report.to_dict() + + assert payload["overlap_count"] == len(payload["overlap_frames"]) + assert payload["handoff_points"][0]["entity_id"] == "parcel-1" + assert isinstance(payload["handoff_points"][0]["entered_cameras"], tuple) + + +def test_panel_matrix_uses_one_camera_count_per_entity_frame() -> None: + example = _assembly_example() + manifest = build_manifest(example.build_scene()) + script = _coverage_script() + + frames, entity_ids, matrix = script._coverage_matrix(manifest) + + assert frames == list(range(manifest.num_frames)) + assert entity_ids == [entity.id for entity in manifest.entities] + assert len(matrix) == len(manifest.entities) + assert all(len(row) == manifest.num_frames for row in matrix) + assert all(value == 1 for row in matrix for value in row) + + +def test_panel_renders_headless_png(tmp_path: Path) -> None: + pytest.importorskip("matplotlib") + manifest = build_manifest(build_handoff_ltr_scene()) + report = compute_coverage_metrics(manifest) + script = _coverage_script() + output = tmp_path / "nested" / "coverage.png" + + rendered = script.render_coverage_panel(manifest, report, output, title="handoff_ltr") + + assert rendered == output + assert output.read_bytes().startswith(b"\x89PNG\r\n\x1a\n") + assert output.stat().st_size > 10_000