feat(gt): per-entity activity-state ground-truth channel (opt-in sidecar)

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",

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

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,11 +345,19 @@ 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,
             cameras=self._cameras,
             entities=entities,
             occluders=occluders,
             possession=possession,
+            activity=activity,
         )

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: