diff --git a/src/multicam_sim/__init__.py b/src/multicam_sim/__init__.py index ed9502c..13e68ad 100644 --- a/src/multicam_sim/__init__.py +++ b/src/multicam_sim/__init__.py @@ -6,6 +6,7 @@ from __future__ import annotations +from .activity import ActivitySegment, ActivityState, ActivityTimeline, write_activity_json from .annotations import ( CocoAnnotation, CocoCategory, @@ -66,6 +67,9 @@ __all__ = [ "COCO17_EDGES", "COCO17_JOINTS", + "ActivitySegment", + "ActivityState", + "ActivityTimeline", "AssumedCalibration", "Box", "CalibrationDrift", @@ -119,6 +123,7 @@ "export_overlay", "export_yolo", "validate_manifest", + "write_activity_json", "write_coco", "write_group_json", "write_manifest", diff --git a/src/multicam_sim/activity.py b/src/multicam_sim/activity.py new file mode 100644 index 0000000..9f206ba --- /dev/null +++ b/src/multicam_sim/activity.py @@ -0,0 +1,131 @@ +"""Activity-state ground-truth sidecar: what each entity is doing, when. + +An **activity timeline** records that an entity is in a typed activity state +(e.g. ``standing`` / ``crouching`` / ``reaching``) over a half-open frame +interval ``[start_frame, end_frame)``. It gives an activity-recognition head a +per-entity, per-frame label to score against: the state at any frame is +queryable via :meth:`ActivityTimeline.state_at_frame`. + +The label is a :class:`ActivityState` ``StrEnum``: adding a state later is a +new enum member whose serialized value is just another string, so extension is +additive and needs no schema fork. This channel is distinct from the skeletal +motion DSL (a motion *producer*); it never generates or alters motion — it +only *labels* frames. + +Like :mod:`multicam_sim.possession`, this module is pure typed models + logic. +The timeline rides in a JSON sidecar and is attached to +:class:`~multicam_sim.scene.Scene` via an optional field, so the byte-golden +analytic manifest is unchanged when activity GT is absent. +""" + +from __future__ import annotations + +import json +from enum import StrEnum +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict, field_validator, model_validator + + +class ActivityState(StrEnum): + """A typed activity label for an entity over a frame interval. + + Str-backed so a new state is a new member here and a new string value on + the wire — additive, with no schema fork for consumers. + """ + + standing = "standing" + crouching = "crouching" + reaching = "reaching" + + +class ActivitySegment(BaseModel): + """One labeled interval: ``entity_id`` is in ``state`` over + ``[start_frame, end_frame)``.""" + + model_config = ConfigDict(frozen=True) + + entity_id: str + state: ActivityState + start_frame: int + end_frame: int + + @field_validator("start_frame", "end_frame") + @classmethod + def _non_negative(cls, value: int) -> int: + if value < 0: + raise ValueError("frame indices must be >= 0") + return value + + @model_validator(mode="after") + def _start_before_end(self) -> ActivitySegment: + if self.end_frame <= self.start_frame: + raise ValueError( + f"end_frame {self.end_frame} must be strictly greater than " + f"start_frame {self.start_frame}" + ) + return self + + def contains(self, frame: int) -> bool: + """True when ``frame`` is inside the half-open labeled interval.""" + return self.start_frame <= frame < self.end_frame + + +class ActivityTimeline(BaseModel): + """A collection of activity segments for one or more entities. + + Segments are kept sorted by ``(entity_id, start_frame)`` and the model + rejects overlapping intervals for the same entity, so every + ``(entity_id, frame)`` pair has at most one state. Frames outside all of + an entity's segments are *unlabeled* — :meth:`state_at_frame` returns + ``None`` for them. + """ + + model_config = ConfigDict(frozen=True) + + segments: list[ActivitySegment] = [] + + @field_validator("segments") + @classmethod + def _no_overlap(cls, value: list[ActivitySegment]) -> list[ActivitySegment]: + by_entity: dict[str, list[ActivitySegment]] = {} + for seg in value: + by_entity.setdefault(seg.entity_id, []).append(seg) + for entity_id, segs in by_entity.items(): + ordered = sorted(segs, key=lambda s: s.start_frame) + for prev, cur in zip(ordered, ordered[1:], strict=False): + if cur.start_frame < prev.end_frame: + raise ValueError( + f"overlapping activity segments for entity {entity_id!r}: " + f"[{prev.start_frame}, {prev.end_frame}) and " + f"[{cur.start_frame}, {cur.end_frame})" + ) + return sorted(value, key=lambda s: (s.entity_id, s.start_frame)) + + def state_at_frame(self, entity_id: str, frame: int) -> ActivityState | None: + """Return the activity state for ``entity_id`` at ``frame``, or ``None`` + if the entity is unlabeled at that frame. + """ + for seg in self.segments: + if seg.entity_id != entity_id: + continue + if seg.contains(frame): + return seg.state + if seg.start_frame > frame: + break + return None + + def to_json(self, *, indent: int | None = 2) -> str: + """Serialise to a JSON string (the ``activity.json`` sidecar payload).""" + return self.model_dump_json(indent=indent) + + +def write_activity_json(timeline: ActivityTimeline, path: str | Path) -> dict[str, Any]: + """Write an activity timeline to ``path`` as JSON. + + Returns the dumped dict so a caller can assert on it without re-reading. + """ + data: dict[str, Any] = timeline.model_dump(mode="json") + Path(path).write_text(json.dumps(data, indent=2)) + return data diff --git a/src/multicam_sim/dsl/builder.py b/src/multicam_sim/dsl/builder.py index 1205b35..4a2b305 100644 --- a/src/multicam_sim/dsl/builder.py +++ b/src/multicam_sim/dsl/builder.py @@ -9,6 +9,7 @@ from dataclasses import dataclass +from ..activity import ActivitySegment, ActivityState, ActivityTimeline from ..cameras import Camera from ..entities import Entity, EntityFrame from ..occluders import OccluderUnion @@ -70,6 +71,7 @@ def __init__(self, fps: float, num_frames: int) -> None: self._hand_sweeps: list[HandSweep] = [] self._attachments: list[_AttachmentSpec] = [] self._interactions: list[InteractionEvent] = [] + self._activity_segments: list[ActivitySegment] = [] def cameras(self, cameras: list[Camera]) -> SceneBuilder: """Set the camera array (e.g. from :class:`multicam_sim.dsl.CameraRig`).""" @@ -218,6 +220,30 @@ def handoff( ) return self + def activity( + self, + entity_id: str, + state: ActivityState, + start: int, + end: int, + ) -> SceneBuilder: + """Label ``entity_id`` as being in activity ``state`` over ``[start, end)``. + + Pure ground truth: it records an + :class:`~multicam_sim.activity.ActivitySegment` in the activity GT + sidecar of the built :class:`Scene` and never touches the entity's + motion or geometry, so the byte-golden manifest is unchanged. Frames + outside all of an entity's segments are unlabeled. + """ + if start < 0 or end > self.num_frames or end <= start: + raise ValueError( + f"invalid activity window [{start}, {end}) for num_frames={self.num_frames}" + ) + self._activity_segments.append( + ActivitySegment(entity_id=entity_id, state=state, start_frame=start, end_frame=end) + ) + return self + def build(self) -> Scene: """Compile the DSL into a :class:`Scene` (cameras, entities, occluders).""" if not self._cameras: @@ -319,6 +345,13 @@ def build(self) -> Scene: else None ) + for seg in self._activity_segments: + if seg.entity_id not in frames_by_id: + raise ValueError(f"activity segment references unknown entity {seg.entity_id!r}") + activity = ( + ActivityTimeline(segments=self._activity_segments) if self._activity_segments else None + ) + return Scene( fps=self.fps, num_frames=self.num_frames, @@ -326,4 +359,5 @@ def build(self) -> Scene: entities=entities, occluders=occluders, possession=possession, + activity=activity, ) diff --git a/src/multicam_sim/scene.py b/src/multicam_sim/scene.py index a44712c..4a800d8 100644 --- a/src/multicam_sim/scene.py +++ b/src/multicam_sim/scene.py @@ -6,6 +6,7 @@ from pydantic import BaseModel +from .activity import ActivityTimeline from .cameras import Camera from .entities import Entity from .occluders import OccluderUnion @@ -25,6 +26,9 @@ class Scene(BaseModel): :mod:`multicam_sim.possession`). It is additive: absent by default and never read by the manifest builder, so scenes that do not use it keep the byte-golden analytic manifest. + + ``activity`` is an optional activity-state GT sidecar (see + :mod:`multicam_sim.activity`) — same additive contract as ``possession``. """ fps: float @@ -34,6 +38,7 @@ class Scene(BaseModel): occluders: list[OccluderUnion] = [] topology: CameraTopology | None = None possession: PossessionTimeline | None = None + activity: ActivityTimeline | None = None def model_post_init(self, __context: Any) -> None: if self.topology is None: diff --git a/tests/test_activity.py b/tests/test_activity.py new file mode 100644 index 0000000..cd8d575 --- /dev/null +++ b/tests/test_activity.py @@ -0,0 +1,264 @@ +"""Activity-state ground-truth channel (#65). + +A typed per-entity activity label (``standing`` / ``crouching`` / ``reaching``) +over half-open frame intervals, riding in an additive ``activity.json`` sidecar +following the ``order.py`` / ``possession.py`` precedent. The byte-golden +analytic manifest is unchanged unless the channel is opted into — and even then +the labels never enter the manifest. +""" + +from __future__ import annotations + +import json +import math +from pathlib import Path + +import pytest + +from multicam_sim import build_manifest, build_smoke_scene +from multicam_sim.activity import ( + ActivitySegment, + ActivityState, + ActivityTimeline, + write_activity_json, +) +from multicam_sim.dsl import CameraRig, SceneBuilder +from multicam_sim.dsl import Path as MotionPath +from multicam_sim.scene import Scene + +_GOLDEN = Path(__file__).parent / "fixtures" / "manifest_golden" + + +def _same_shape(got: object, ref: object, path: str = "") -> None: + """Assert identical JSON structure — key order, list length, field presence, + and value types. Float values are compared with a tolerance (mirrors + ``test_manifest_golden``). + """ + assert type(got) is type(ref), f"type drift at {path or ''}: {type(got)} != {type(ref)}" + if isinstance(ref, dict): + assert isinstance(got, dict) + assert list(got) == list(ref), f"key set/order drift at {path or ''}" + for k in ref: + _same_shape(got[k], ref[k], f"{path}.{k}") + elif isinstance(ref, list): + assert isinstance(got, list) + assert len(got) == len(ref), f"length drift at {path or ''}: {len(got)} != {len(ref)}" + for i, (g, r) in enumerate(zip(got, ref, strict=True)): + _same_shape(g, r, f"{path}[{i}]") + elif isinstance(ref, float): + assert isinstance(got, float) + assert math.isclose(got, ref, rel_tol=1e-9, abs_tol=1e-12), ( + f"float drift at {path}: {got} != {ref}" + ) + else: + assert got == ref, f"value drift at {path}: {got!r} != {ref!r}" + + +def _labeled_scene(*, label: bool = True) -> Scene: + """Two moving entities; with ``label=True``, ``operator`` is standing over + [0, 4), crouching over [4, 7), reaching over [7, 10); ``walker`` is + standing over [2, 5). + """ + builder = ( + SceneBuilder(fps=10.0, num_frames=10) + .cameras( + CameraRig.ring( + n=2, + radius=5.0, + height=1.0, + look_at=(0.0, 0.0, 0.0), + focal=800.0, + width=640, + height_px=480, + ) + ) + .entity("operator", MotionPath.linear((0.0, 0.0, 0.0), (9.0, 0.0, 0.0))) + .entity("walker", MotionPath.linear((0.0, 5.0, 0.0), (9.0, 5.0, 0.0))) + ) + if label: + builder.activity("operator", ActivityState.standing, 0, 4) + builder.activity("operator", ActivityState.crouching, 4, 7) + builder.activity("operator", ActivityState.reaching, 7, 10) + builder.activity("walker", ActivityState.standing, 2, 5) + return builder.build() + + +def test_activity_state_serialises_as_plain_string() -> None: + """The label is str-backed: its JSON value is the plain state string, so + adding a state later is a new enum member and a new string — additive, + no schema fork.""" + assert ActivityState.standing == "standing" + assert ( + json.loads( + ActivitySegment( + entity_id="e", state=ActivityState.reaching, start_frame=0, end_frame=3 + ).model_dump_json() + )["state"] + == "reaching" + ) + + +def test_segment_rejects_invalid_windows() -> None: + """Frame indices must be >= 0 and end strictly greater than start.""" + with pytest.raises(ValueError, match=">= 0"): + ActivitySegment(entity_id="e", state=ActivityState.standing, start_frame=-1, end_frame=3) + with pytest.raises(ValueError, match="strictly greater"): + ActivitySegment(entity_id="e", state=ActivityState.standing, start_frame=3, end_frame=3) + with pytest.raises(ValueError, match="strictly greater"): + ActivitySegment(entity_id="e", state=ActivityState.standing, start_frame=5, end_frame=3) + + +def test_timeline_sorts_segments_and_rejects_overlap() -> None: + """Segments come out sorted by (entity_id, start_frame); overlapping + intervals for the SAME entity are rejected, disjoint entities may overlap + in time freely.""" + timeline = ActivityTimeline( + segments=[ + ActivitySegment( + entity_id="b", state=ActivityState.standing, start_frame=0, end_frame=5 + ), + ActivitySegment( + entity_id="a", state=ActivityState.crouching, start_frame=6, end_frame=9 + ), + ActivitySegment( + entity_id="a", state=ActivityState.standing, start_frame=0, end_frame=3 + ), + ] + ) + assert [(s.entity_id, s.start_frame) for s in timeline.segments] == [ + ("a", 0), + ("a", 6), + ("b", 0), + ] + + with pytest.raises(ValueError, match="overlapping activity segments"): + ActivityTimeline( + segments=[ + ActivitySegment( + entity_id="a", state=ActivityState.standing, start_frame=0, end_frame=5 + ), + ActivitySegment( + entity_id="a", state=ActivityState.crouching, start_frame=4, end_frame=8 + ), + ] + ) + + +def test_state_at_frame_half_open_boundaries() -> None: + """The timeline answers per-entity per-frame queries: the state holds over + [start, end), frames outside all segments are unlabeled (None), and an + unknown entity is always unlabeled.""" + scene = _labeled_scene(label=True) + assert scene.activity is not None + timeline = scene.activity + + # Contiguous intervals hand over exactly at the boundary frame. + for f in range(0, 4): + assert timeline.state_at_frame("operator", f) == ActivityState.standing + for f in range(4, 7): + assert timeline.state_at_frame("operator", f) == ActivityState.crouching + for f in range(7, 10): + assert timeline.state_at_frame("operator", f) == ActivityState.reaching + + # walker is labeled only on [2, 5); unlabeled before and from end_frame on. + assert timeline.state_at_frame("walker", 1) is None + assert timeline.state_at_frame("walker", 2) == ActivityState.standing + assert timeline.state_at_frame("walker", 4) == ActivityState.standing + assert timeline.state_at_frame("walker", 5) is None + assert timeline.state_at_frame("walker", 9) is None + + # Unknown entity is always unlabeled. + assert timeline.state_at_frame("unknown", 3) is None + + +def test_builder_validates_activity_window_and_entity() -> None: + """The opt-in hook rejects out-of-range windows immediately and unknown + entity ids at build time (mirroring ``attach``).""" + builder = ( + SceneBuilder(fps=10.0, num_frames=10) + .cameras( + CameraRig.ring( + n=2, + radius=5.0, + height=1.0, + look_at=(0.0, 0.0, 0.0), + focal=800.0, + width=640, + height_px=480, + ) + ) + .entity("operator", MotionPath.linear((0.0, 0.0, 0.0), (9.0, 0.0, 0.0))) + ) + with pytest.raises(ValueError, match="invalid activity window"): + builder.activity("operator", ActivityState.standing, -1, 5) + with pytest.raises(ValueError, match="invalid activity window"): + builder.activity("operator", ActivityState.standing, 5, 11) + with pytest.raises(ValueError, match="invalid activity window"): + builder.activity("operator", ActivityState.standing, 5, 5) + + builder.activity("ghost", ActivityState.standing, 0, 5) + with pytest.raises(ValueError, match="unknown entity 'ghost'"): + builder.build() + + +def test_scene_roundtrips_with_and_without_activity_sidecar() -> None: + """A Scene serialises and deserialises identically, both with and without + the optional activity sidecar.""" + labeled = _labeled_scene(label=True) + plain = _labeled_scene(label=False) + assert plain.activity is None + + for scene in (labeled, plain): + json_text = scene.model_dump_json(indent=2) + restored = Scene.model_validate_json(json_text) + assert restored.fps == scene.fps + assert restored.num_frames == scene.num_frames + assert [e.id for e in restored.entities] == [e.id for e in scene.entities] + assert restored.activity == scene.activity + + # The sidecar, when present, round-trips through its own to_json. + if scene.activity is not None: + sidecar_restored = ActivityTimeline.model_validate_json(scene.activity.to_json()) + assert sidecar_restored == scene.activity + + +def test_write_activity_json(tmp_path: Path) -> None: + """The sidecar writes to ``activity.json`` and returns the dumped dict.""" + timeline = _labeled_scene(label=True).activity + assert timeline is not None + path = tmp_path / "activity.json" + data = write_activity_json(timeline, path) + assert json.loads(path.read_text()) == data + assert data["segments"][0] == { + "entity_id": "operator", + "state": "standing", + "start_frame": 0, + "end_frame": 4, + } + + +def test_manifest_unchanged_without_opt_in() -> None: + """Scenes that do not opt into the activity channel keep the byte-golden + analytic manifest unchanged (compared to the smoke golden fixture). + """ + got = build_manifest(build_smoke_scene()).to_json() + ref = (_GOLDEN / "smoke.json").read_text() + _same_shape(json.loads(got), json.loads(ref)) + + +def test_manifest_byte_identical_with_and_without_opt_in() -> None: + """Opting into the activity channel changes NOTHING in the manifest: the + serialized manifest bytes of the labeled and unlabeled scene are exactly + equal.""" + labeled = build_manifest(_labeled_scene(label=True)).to_json().encode() + plain = build_manifest(_labeled_scene(label=False)).to_json().encode() + assert labeled == plain + + +def test_manifest_excludes_activity_sidecar() -> None: + """The analytic manifest never contains activity GT, even when the scene + carries the sidecar.""" + scene = _labeled_scene(label=True) + manifest_json = build_manifest(scene).to_json() + assert "activity" not in manifest_json + assert "ActivitySegment" not in manifest_json