diff --git a/examples/assembly_station.py b/examples/assembly_station.py index 5b72bd3..630afea 100644 --- a/examples/assembly_station.py +++ b/examples/assembly_station.py @@ -22,6 +22,18 @@ * ``manifest.json`` — the full scene manifest (projection + in_view/visible); * ``order.json`` — the verified order result (fulfilled / missing / …). + +Opt-in **placement-synced preset** (``--placement-synced``): the continuous +wrist reach is replaced by discrete hand dips synced to the placements (a +strict local minimum of the tracked wrist's height at ``placed_at - δ`` per +placed item), plus negatives that falsify temporal association: a distractor +dip that places nothing, and a distractor item (``part_d``) whose uncaused +move follows that dip inside the causal lag window — a naive causal-forward +associator pairs them, and the ground truth says otherwise. The true +``(actor, item, action_frame, change_frame)`` pairs are written to +``interactions.json`` so a causal-fusion consumer can score precision/recall. +Off by default: without the flag the scene and every emitted file are +byte-identical to before. """ from __future__ import annotations @@ -33,6 +45,12 @@ from typing import Any from multicam_sim import write_manifest +from multicam_sim.actions import ( + CausalTiming, + DipSchedule, + build_action_ground_truth, + write_actions_json, +) from multicam_sim.dsl.rig import CameraRig, StationView from multicam_sim.entities import Entity, EntityFrame from multicam_sim.order import ( @@ -62,6 +80,24 @@ } _PLACED_AT = {"part_a": 2, "part_b": 5, "part_c": 8} +# --- placement-synced preset (opt-in via --placement-synced) ---------------- # +# The causal half of the fusion story: the tracked hand dips (strict local +# height minimum) at ``placed_at - δ`` for each placed item. The negatives are +# positioned to *falsify* a naive causal-forward associator ("pair each dip +# with the next change inside the lag window"): the distractor dip places +# nothing, but the distractor item's move — which it did not cause — follows +# inside the window, so the naive rule pairs them and interactions.json says +# otherwise. δ and the lag window are typed parameters (CausalTiming). +_TRACKED_HAND = "right_wrist" +_SYNC_NUM_FRAMES = 13 +_SYNC_TIMING = CausalTiming(action_lag=1, lag_window=2) +_DIP_DEPTH = 0.30 +_DIP_HALF_WIDTH = 1 +_DISTRACTOR_ITEM = "part_d" +_DISTRACTOR_STAGING = (2.60, -0.30, 0.90) +_DISTRACTOR_DIP = 10 # places nothing; >= 2*half_width+1 from the dip at 7 +_DISTRACTOR_PLACED_AT = 11 # uncaused move inside the dip-10 lag window + # Standing COCO-17 offsets (dx, dy, dz) from the foot base; +y is the facing dir. _JOINT_OFFSETS: dict[str, tuple[float, float, float]] = { "nose": (0.0, 0.10, 1.60), @@ -84,37 +120,66 @@ } -def operator_pose() -> PoseTrajectory: - """A standing COCO-17 operator whose wrists make a small assembling motion.""" +def operator_pose(placement_synced: bool = False, num_frames: int = NUM_FRAMES) -> PoseTrajectory: + """A standing COCO-17 operator whose wrists make a small assembling motion. + + Default: a single continuous sinusoidal wrist reach (order-verification + scene). With ``placement_synced``: the wrists rest and the tracked hand + (:data:`_TRACKED_HAND`) dips — a strict local height minimum at + ``placed_at - δ`` for each placed item, plus one distractor dip that + assembles nothing — so every dip is recoverable off the manifest alone. + """ bx, by = _OPERATOR_BASE frames: list[PoseFrame] = [] - for f in range(NUM_FRAMES): - phase = math.sin(2.0 * math.pi * f / (NUM_FRAMES - 1)) # -1..1, smooth + for f in range(num_frames): + phase = ( + 0.0 if placement_synced else math.sin(2.0 * math.pi * f / (num_frames - 1)) + ) # -1..1, smooth joints: dict[str, list[float]] = {} for name, (dx, dy, dz) in _JOINT_OFFSETS.items(): reach = 0.06 * phase if name.endswith("wrist") else 0.0 # wrists reach in +y joints[name] = [bx + dx, by + dy + reach, dz] frames.append(PoseFrame(frame=f, joints=joints)) - return PoseTrajectory(id="operator", skeleton=Skeleton.coco17(), frames=frames) + trajectory = PoseTrajectory(id="operator", skeleton=Skeleton.coco17(), frames=frames) + if placement_synced: + dips = DipSchedule( + frames=[*synced_dip_frames(), _DISTRACTOR_DIP], + rest_height=_JOINT_OFFSETS[_TRACKED_HAND][2], + depth=_DIP_DEPTH, + half_width=_DIP_HALF_WIDTH, + ) + trajectory = dips.author(trajectory, _TRACKED_HAND) + return trajectory + +def synced_dip_frames() -> list[int]: + """The dip frame (``placed_at - δ``) for each causally-backed placement.""" + return sorted(frame - _SYNC_TIMING.action_lag for frame in _PLACED_AT.values()) -def item_entity(item_id: str) -> Entity: + +def item_entity( + item_id: str, + staging: tuple[float, float, float] | None = None, + placed_at: int | None = None, + num_frames: int = NUM_FRAMES, +) -> Entity: """An item that sits at its staging spot, then jumps into the container at its ``placed_at`` frame (and stays).""" - staging = _ITEM_STAGING[item_id] - placed_at = _PLACED_AT[item_id] + staging = _ITEM_STAGING[item_id] if staging is None else staging + placed_at = _PLACED_AT[item_id] if placed_at is None else placed_at frames = [ EntityFrame( frame=f, points={"center": list(_CONTAINER if f >= placed_at else staging)}, ) - for f in range(NUM_FRAMES) + for f in range(num_frames) ] return Entity(id=item_id, frames=frames) -def build_scene() -> Scene: +def build_scene(placement_synced: bool = False) -> Scene: """Assemble the two-camera scene: overview (operator) + worktop (items).""" + num_frames = _SYNC_NUM_FRAMES if placement_synced else NUM_FRAMES cameras = CameraRig.stations( [ # overview: wide-ish, high, to the north (+y), framing the operator. @@ -125,25 +190,44 @@ def build_scene() -> Scene: width=1280, height_px=720, ) - entities = [operator_pose().to_entity(), *(item_entity(i) for i in _ITEM_STAGING)] - return Scene(fps=FPS, num_frames=NUM_FRAMES, cameras=cameras, entities=entities) + entities = [ + operator_pose(placement_synced, num_frames).to_entity(), + *(item_entity(i, num_frames=num_frames) for i in _ITEM_STAGING), + ] + if placement_synced: + entities.append( + item_entity( + _DISTRACTOR_ITEM, + staging=_DISTRACTOR_STAGING, + placed_at=_DISTRACTOR_PLACED_AT, + num_frames=num_frames, + ) + ) + return Scene(fps=FPS, num_frames=num_frames, cameras=cameras, entities=entities) -def build_order() -> tuple[Order, list[ItemPlacement]]: +def build_order(placement_synced: bool = False) -> tuple[Order, list[ItemPlacement]]: """The pick-list (one of each part) and the placements as items land.""" - bom = BillOfMaterials.from_counts({item: 1 for item in _ITEM_STAGING}) + counts = {item: 1 for item in _ITEM_STAGING} + placed_at = dict(_PLACED_AT) + if placement_synced: + counts[_DISTRACTOR_ITEM] = 1 + placed_at[_DISTRACTOR_ITEM] = _DISTRACTOR_PLACED_AT + bom = BillOfMaterials.from_counts(counts) order = Order(order_id="ORD-1", bom=bom) placements = [ ItemPlacement(item=item, placed_at_frame=frame, entity_id=item) - for item, frame in _PLACED_AT.items() + for item, frame in placed_at.items() ] return order, placements -def build_actions(placements: list[ItemPlacement]) -> list[ActionEvent]: +def build_actions( + placements: list[ItemPlacement], trajectory: PoseTrajectory | None = None +) -> list[ActionEvent]: """One 'place' ActionEvent per placement, synced to its frame, carrying the operator's right-wrist world position at that frame (causal-fusion GT).""" - joints_by_frame = {f.frame: f.joints for f in operator_pose().frames} + joints_by_frame = {f.frame: f.joints for f in (trajectory or operator_pose()).frames} hand = "right_wrist" events: list[ActionEvent] = [] for p in placements: @@ -171,17 +255,18 @@ def entity_in_view(manifest: dict[str, Any], entity_id: str, cam_id: int) -> tup return seen, len(entity["frames"]) -def run(out_dir: Path) -> dict[str, Any]: +def run(out_dir: Path, placement_synced: bool = False) -> dict[str, Any]: """Build, verify, write sidecars, and return a summary dict.""" out_dir.mkdir(parents=True, exist_ok=True) - scene = build_scene() - order, placements = build_order() + scene = build_scene(placement_synced) + order, placements = build_order(placement_synced) write_manifest(scene, out_dir / "manifest.json") # read back the on-disk manifest as a plain dict — the genuine consumer path. manifest = json.loads((out_dir / "manifest.json").read_text()) - actions = build_actions(placements) + trajectory = operator_pose(placement_synced, num_frames=scene.num_frames) + actions = build_actions(placements, trajectory=trajectory) # order.json = the order GT sidecar: status + per-item deltas + the synced # ActionEvents (manifest stays byte-golden — actions never touch it). result: OrderResult = verify_order( @@ -190,8 +275,21 @@ def run(out_dir: Path) -> dict[str, Any]: write_order_json(result, out_dir / "order.json") write_order_json(order, out_dir / "pick_list.json") + truth = None + if placement_synced: + # interactions.json = the causal GT sidecar: only the true pairs. The + # dip-10 → part_d pairing a temporal associator will make is absent, so + # it scores as a false positive — the authored negative. + truth = build_action_ground_truth( + _SYNC_TIMING, + actor_id="operator", + tracked_joint=_TRACKED_HAND, + placements=[p for p in placements if p.item != _DISTRACTOR_ITEM], + ) + write_actions_json(truth, out_dir / "interactions.json") + OVERVIEW, WORKTOP = 0, 1 - item_ids = list(_ITEM_STAGING) + item_ids = [*_ITEM_STAGING, *([_DISTRACTOR_ITEM] if placement_synced else [])] visibility = { "operator": { "overview": entity_in_view(manifest, "operator", OVERVIEW), @@ -209,6 +307,7 @@ def run(out_dir: Path) -> dict[str, Any]: "manifest": manifest, "result": result, "actions": actions, + "interactions": truth, "visibility": visibility, "out_dir": out_dir, } @@ -219,15 +318,22 @@ def main() -> int: parser.add_argument( "--out", type=Path, default=Path(__file__).parent / "out", help="output directory" ) + parser.add_argument( + "--placement-synced", + action="store_true", + help="placement-synced preset: hand dips at placed_at - δ per item, a " + "distractor dip, a late distractor item, and an interactions.json " + "causal ground-truth sidecar", + ) args = parser.parse_args() - summary = run(args.out) + summary = run(args.out, placement_synced=args.placement_synced) vis = summary["visibility"] print(f"[assembly_station] wrote manifest.json + order.json to {summary['out_dir']}") print(" camera 0 = overview (operator) | camera 1 = worktop (items)") for name, cams in vis.items(): - ov, wt = cams["overview"][0], cams["worktop"][0] - print(f" {name:9s} overview in_view {ov:2d}/11 worktop in_view {wt:2d}/11") + (ov, n), (wt, _) = cams["overview"], cams["worktop"] + print(f" {name:9s} overview in_view {ov:2d}/{n} worktop in_view {wt:2d}/{n}") print(f" order {summary['result'].status.value}") for ev in summary["actions"]: hx, hy, hz = ev.hand_position @@ -235,6 +341,14 @@ def main() -> int: f" action {ev.action} {ev.item_id} @frame {ev.frame} " f"hand({ev.hand_joint})=({hx:.2f},{hy:.2f},{hz:.2f})" ) + truth = summary["interactions"] + if truth is not None: + print(f" interactions.json: {len(truth.pairs)} causal pairs") + for pair in truth.pairs: + print( + f" {pair.actor_id} dip @frame {pair.action_frame} " + f"-> {pair.item_id} placed @frame {pair.change_frame}" + ) return 0 diff --git a/src/multicam_sim/__init__.py b/src/multicam_sim/__init__.py index 13e68ad..24df151 100644 --- a/src/multicam_sim/__init__.py +++ b/src/multicam_sim/__init__.py @@ -6,6 +6,14 @@ from __future__ import annotations +from .actions import ( + ActionChange, + ActionGroundTruth, + CausalTiming, + DipSchedule, + build_action_ground_truth, + write_actions_json, +) from .activity import ActivitySegment, ActivityState, ActivityTimeline, write_activity_json from .annotations import ( CocoAnnotation, @@ -67,12 +75,15 @@ __all__ = [ "COCO17_EDGES", "COCO17_JOINTS", + "ActionChange", + "ActionGroundTruth", "ActivitySegment", "ActivityState", "ActivityTimeline", "AssumedCalibration", "Box", "CalibrationDrift", + "CausalTiming", "Cylinder", "Camera", "CameraManifest", @@ -81,6 +92,7 @@ "CocoCategory", "CocoDataset", "CocoImage", + "DipSchedule", "Entity", "EntityFrame", "EntityManifest", @@ -111,6 +123,7 @@ "TransitEdge", "YoloDataset", "YoloLabel", + "build_action_ground_truth", "build_group_formation_scene", "build_manifest", "build_mtmc_scene", @@ -123,6 +136,7 @@ "export_overlay", "export_yolo", "validate_manifest", + "write_actions_json", "write_activity_json", "write_coco", "write_group_json", diff --git a/src/multicam_sim/actions.py b/src/multicam_sim/actions.py new file mode 100644 index 0000000..f1fe79f --- /dev/null +++ b/src/multicam_sim/actions.py @@ -0,0 +1,273 @@ +"""δ-lagged hand dips and action→change ground truth for causal fusion. + +Placement-synced operator action events are not new here: +:class:`multicam_sim.order.ActionEvent` already records the operator's hand +position at each placement frame. What this module adds is the recoverable +*motion* and the *negatives*: a discrete **hand dip** — a strict local minimum +of a tracked keypoint's height — authored into the pose at ``action_lag`` (δ) +frames *before* each item's placement frame, so a consumer with only the +emitted manifest can recover one dip per placement and associate +action → change in time. + +Two authored negatives make the data able to *falsify* a weak association rule +rather than merely confirm a strong one: + +* a **distractor action** — a dip that places nothing, timed so that a change + it did *not* cause follows inside the causal lag window: a naive + causal-forward associator ("pair each dip with the next change within the + window") pairs them, and the ground truth says otherwise; +* a **distractor change** — an item whose move has no true causal action (it + is simply omitted from the ground-truth pairs). + +The ground truth rides in a JSON sidecar (e.g. ``interactions.json``) listing +only the *true* ``(actor, item, action_frame, change_frame)`` pairs, so a +consumer scores precision/recall without re-deriving the truth — and any +association it makes beyond this list is a false positive by construction. + +:class:`DipSchedule` is the motion producer (it rewrites one joint's height +channel of a :class:`~multicam_sim.pose.PoseTrajectory`); the rest is pure +typed models + logic in the :mod:`multicam_sim.order` / +:mod:`multicam_sim.possession` style. The sidecar never touches the +byte-golden manifest. +""" + +from __future__ import annotations + +import json +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict, field_validator, model_validator + +from .order import ItemPlacement +from .pose import PoseFrame, PoseTrajectory + + +class CausalTiming(BaseModel): + """The timing contract between an operator action and the change it causes. + + ``action_lag`` (δ) is the authored gap in frames between a hand dip and the + placement it causes: a dip at frame ``a`` causes the change at frame + ``a + action_lag``. ``lag_window`` is the consumer-facing association + window: an action at ``a`` and a change at ``c`` are causally associable + only when ``0 < c - a <= lag_window``. Both are explicit typed parameters — + the authored δ and the scoring window are never magic numbers buried in a + consumer. + """ + + model_config = ConfigDict(frozen=True) + + action_lag: int + lag_window: int + + @field_validator("action_lag") + @classmethod + def _lag_at_least_one(cls, value: int) -> int: + if value < 1: + raise ValueError("action_lag (δ) must be >= 1: the dip precedes the placement") + return value + + @model_validator(mode="after") + def _window_covers_lag(self) -> CausalTiming: + if self.lag_window < self.action_lag: + raise ValueError( + f"lag_window {self.lag_window} must be >= action_lag {self.action_lag}, " + "or the authored pairs fall outside their own causal window" + ) + return self + + +class DipSchedule(BaseModel): + """Hand dips: strict local minima of a tracked joint's height at given frames. + + Each dip is a reach-and-return triangle: the joint rests at ``rest_height``, + descends to ``rest_height - depth`` at the dip frame, and climbs back over + ``half_width`` frames either side. Dips must sit at least + ``2 * half_width + 1`` frames apart so their profiles never overlap; every + dip frame is then a *strict* local minimum — strictly lower than both + neighbouring frames — recoverable from the manifest alone. + """ + + model_config = ConfigDict(frozen=True) + + frames: list[int] + rest_height: float + depth: float + half_width: int = 1 + + @field_validator("frames") + @classmethod + def _frames_valid(cls, value: list[int]) -> list[int]: + if not value: + raise ValueError("dip schedule needs at least one dip frame") + for frame in value: + if frame < 1: + raise ValueError( + f"dip frame {frame} must be >= 1: " + "a strict local minimum needs a preceding frame" + ) + if len(set(value)) != len(value): + raise ValueError("dip frames must be unique") + return sorted(value) + + @field_validator("depth") + @classmethod + def _positive_depth(cls, value: float) -> float: + if value <= 0.0: + raise ValueError("dip depth must be > 0, or the minimum is not strict") + return value + + @field_validator("half_width") + @classmethod + def _positive_half_width(cls, value: int) -> int: + if value < 1: + raise ValueError("half_width must be >= 1 frame") + return value + + @model_validator(mode="after") + def _profiles_disjoint(self) -> DipSchedule: + for prev, cur in zip(self.frames, self.frames[1:], strict=False): + if cur - prev < 2 * self.half_width + 1: + raise ValueError( + f"dip frames {prev} and {cur} are closer than " + f"2*half_width+1 ({2 * self.half_width + 1}): overlapping " + "profiles would break the strict local minimum" + ) + return self + + def height_at(self, frame: int) -> float: + """Tracked-joint height at ``frame``: rest height minus the dip profile.""" + height = self.rest_height + for dip in self.frames: + dist = abs(frame - dip) + height -= self.depth * max(0.0, 1.0 - dist / (self.half_width + 1)) + return height + + def author(self, trajectory: PoseTrajectory, joint: str) -> PoseTrajectory: + """Return ``trajectory`` with ``joint``'s height channel replaced by this profile. + + The joint's x/y pass through untouched; its z becomes + :meth:`height_at` at every frame, so outside dips the hand sits exactly + at ``rest_height`` and the only strict local minima in the channel are + the authored dips. Every dip needs both neighbouring frames present in + the trajectory — a strict local minimum is defined by its neighbours. + """ + available = {f.frame for f in trajectory.frames} + for dip in self.frames: + if dip - 1 not in available or dip + 1 not in available: + raise ValueError( + f"dip at frame {dip} needs frames {dip - 1} and {dip + 1} in the trajectory" + ) + frames: list[PoseFrame] = [] + for pose_frame in trajectory.frames: + if joint not in pose_frame.joints: + raise ValueError(f"joint {joint!r} missing at frame {pose_frame.frame}") + joints = dict(pose_frame.joints) + x, y, _z = joints[joint] + joints[joint] = [x, y, self.height_at(pose_frame.frame)] + frames.append(PoseFrame(frame=pose_frame.frame, joints=joints)) + return PoseTrajectory(id=trajectory.id, skeleton=trajectory.skeleton, frames=frames) + + +class ActionChange(BaseModel): + """One ground-truth causal pair: ``actor_id``'s dip at ``action_frame`` + caused ``item_id``'s placement at ``change_frame``. + + This is the ``(actor, item, action_time, change_time)`` tuple of the + causal-fusion contract, expressed in frames (the repo's time unit; seconds + are ``frame / fps`` for a consumer that needs them). + """ + + model_config = ConfigDict(frozen=True) + + actor_id: str + item_id: str + action_frame: int + change_frame: int + + @model_validator(mode="after") + def _action_precedes_change(self) -> ActionChange: + if self.change_frame <= self.action_frame: + raise ValueError( + f"change_frame {self.change_frame} must be strictly after " + f"action_frame {self.action_frame}: causes precede effects" + ) + return self + + +class ActionGroundTruth(BaseModel): + """The causal-fusion sidecar payload: the true action→change pairs + contract. + + ``pairs`` holds only the *true* associations, kept sorted by + ``(change_frame, item_id)`` so the sidecar is deterministic. Distractors + are deliberately absent: they live in the manifest, and a consumer that + associates one scores a false positive against this list. ``timing`` and + ``tracked_joint`` document the channel the pairs were authored against. + """ + + model_config = ConfigDict(frozen=True) + + timing: CausalTiming + actor_id: str + tracked_joint: str + pairs: list[ActionChange] = [] + + @field_validator("pairs") + @classmethod + def _sorted_pairs(cls, value: list[ActionChange]) -> list[ActionChange]: + return sorted(value, key=lambda p: (p.change_frame, p.item_id)) + + def to_json(self, *, indent: int | None = 2) -> str: + """Serialise to a JSON string (the ``interactions.json`` sidecar payload).""" + return self.model_dump_json(indent=indent) + + +def build_action_ground_truth( + timing: CausalTiming, + actor_id: str, + tracked_joint: str, + placements: Sequence[ItemPlacement], +) -> ActionGroundTruth: + """Derive the true pairs from placements: one dip at ``placed_at - δ`` each. + + ``placements`` must list only causally-backed placements; a distractor + change (an item whose move has no true causal action) is simply omitted, + which is exactly what makes it a negative for the consumer. + + Raises if a placement sits at frame ``<= δ``: its dip would land on a frame + no :class:`DipSchedule` can author (a strict local minimum needs a + preceding frame), so the pair would reference an action no manifest can + contain. Better to fail here than emit unrecoverable ground truth. + """ + pairs: list[ActionChange] = [] + for p in placements: + action_frame = p.placed_at_frame - timing.action_lag + if action_frame < 1: + raise ValueError( + f"placement of {p.item!r} at frame {p.placed_at_frame} with " + f"action_lag {timing.action_lag} puts the causal dip at frame " + f"{action_frame}: no dip can be authored there " + "(a strict local minimum needs a preceding frame)" + ) + pairs.append( + ActionChange( + actor_id=actor_id, + item_id=p.item, + action_frame=action_frame, + change_frame=p.placed_at_frame, + ) + ) + return ActionGroundTruth( + timing=timing, actor_id=actor_id, tracked_joint=tracked_joint, pairs=pairs + ) + + +def write_actions_json(truth: ActionGroundTruth, path: str | Path) -> dict[str, Any]: + """Write the action ground truth to ``path`` as JSON (e.g. ``interactions.json``). + + Returns the dumped dict so a caller can assert on it without re-reading. + """ + data: dict[str, Any] = truth.model_dump(mode="json") + Path(path).write_text(json.dumps(data, indent=2)) + return data diff --git a/tests/test_actions.py b/tests/test_actions.py new file mode 100644 index 0000000..8beb3d8 --- /dev/null +++ b/tests/test_actions.py @@ -0,0 +1,174 @@ +"""Placement-synced operator action events (#34) — the library half. + +A dip is a reach-and-return *strict local minimum* of a tracked joint's height: +the tests pin the profile with pasted-in literals (derived from the triangle +geometry, not from the implementation), then assert the authored trajectory +recovers exactly the authored minima — the property a manifest-only consumer +relies on. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from multicam_sim.actions import ( + ActionChange, + CausalTiming, + DipSchedule, + build_action_ground_truth, + write_actions_json, +) +from multicam_sim.order import ItemPlacement +from multicam_sim.pose import PoseFrame, PoseTrajectory, Skeleton + + +def _trajectory(num_frames: int, z: float = 5.0) -> PoseTrajectory: + """A two-joint skeleton whose ``hand`` zigzags in z — authoring must replace + the whole height channel, not just add to it.""" + skeleton = Skeleton(name="test", joints=["hand", "elbow"], edges=[("hand", "elbow")]) + frames = [ + PoseFrame( + frame=f, + joints={"hand": [1.0, 2.0, z + (-1.0) ** f], "elbow": [0.0, 0.0, 1.0]}, + ) + for f in range(num_frames) + ] + return PoseTrajectory(id="actor", skeleton=skeleton, frames=frames) + + +def test_causal_timing_validation() -> None: + with pytest.raises(ValueError, match="action_lag"): + CausalTiming(action_lag=0, lag_window=2) + with pytest.raises(ValueError, match="lag_window"): + CausalTiming(action_lag=3, lag_window=2) + assert CausalTiming(action_lag=2, lag_window=2).action_lag == 2 + + +def test_dip_profile_height_literals() -> None: + """Triangle half_width=1: dip frame at rest-depth, neighbours at rest-depth/2.""" + dips = DipSchedule(frames=[2, 6], rest_height=1.0, depth=0.2, half_width=1) + assert [dips.height_at(f) for f in range(8)] == pytest.approx( + [1.0, 0.9, 0.8, 0.9, 1.0, 0.9, 0.8, 0.9] + ) + + +def test_dip_profile_wider_half_width_literals() -> None: + """Triangle half_width=2: linear return to rest over two frames either side.""" + dips = DipSchedule(frames=[3], rest_height=2.0, depth=0.6, half_width=2) + assert [dips.height_at(f) for f in range(7)] == pytest.approx( + [2.0, 1.8, 1.6, 1.4, 1.6, 1.8, 2.0] + ) + + +def test_dip_schedule_validation() -> None: + with pytest.raises(ValueError, match="at least one"): + DipSchedule(frames=[], rest_height=1.0, depth=0.2) + with pytest.raises(ValueError, match=">= 1"): + DipSchedule(frames=[0], rest_height=1.0, depth=0.2) # no preceding frame + with pytest.raises(ValueError, match="unique"): + DipSchedule(frames=[4, 4], rest_height=1.0, depth=0.2) + with pytest.raises(ValueError, match="depth"): + DipSchedule(frames=[4], rest_height=1.0, depth=0.0) # minimum not strict + with pytest.raises(ValueError, match="half_width"): + DipSchedule(frames=[4], rest_height=1.0, depth=0.2, half_width=0) + with pytest.raises(ValueError, match="overlap"): + # 2 apart with half_width=1: the profiles share a frame, breaking strictness. + DipSchedule(frames=[4, 6], rest_height=1.0, depth=0.2) + + +def test_author_makes_exactly_the_authored_strict_minima() -> None: + dips = DipSchedule(frames=[2, 6], rest_height=1.0, depth=0.2, half_width=1) + authored = dips.author(_trajectory(9), "hand") + z = [f.joints["hand"][2] for f in authored.frames] + assert z == pytest.approx([1.0, 0.9, 0.8, 0.9, 1.0, 0.9, 0.8, 0.9, 1.0]) + # The recovered strict local minima are exactly the authored dips — nothing + # left over from the zigzag channel the schedule replaced. + minima = [i for i in range(1, len(z) - 1) if z[i] < z[i - 1] and z[i] < z[i + 1]] + assert minima == [2, 6] + + +def test_author_preserves_other_channels() -> None: + base = _trajectory(5, z=5.0) + authored = DipSchedule(frames=[2], rest_height=1.0, depth=0.2).author(base, "hand") + for f_before, f_after in zip(base.frames, authored.frames, strict=True): + assert f_after.joints["elbow"] == f_before.joints["elbow"] + assert f_after.joints["hand"][:2] == f_before.joints["hand"][:2] # x, y pass through + assert authored.id == base.id and authored.skeleton == base.skeleton + + +def test_author_requires_neighbour_frames_and_known_joint() -> None: + dips = DipSchedule(frames=[4], rest_height=1.0, depth=0.2) + with pytest.raises(ValueError, match="frames 3 and 5"): + dips.author(_trajectory(5), "hand") # frame 5 missing: no right neighbour + with pytest.raises(ValueError, match="missing"): + dips.author(_trajectory(6), "ankle") + + +def test_action_change_requires_cause_before_effect() -> None: + with pytest.raises(ValueError, match="strictly after"): + ActionChange(actor_id="op", item_id="part_a", action_frame=3, change_frame=3) + + +def test_build_action_ground_truth_pairs_and_sorting() -> None: + timing = CausalTiming(action_lag=1, lag_window=2) + placements = [ + ItemPlacement(item="part_b", placed_at_frame=5, entity_id="part_b"), + ItemPlacement(item="part_a", placed_at_frame=2, entity_id="part_a"), + ] + truth = build_action_ground_truth(timing, "operator", "right_wrist", placements) + assert [(p.item_id, p.action_frame, p.change_frame) for p in truth.pairs] == [ + ("part_a", 1, 2), + ("part_b", 4, 5), + ] + assert truth.actor_id == "operator" and truth.tracked_joint == "right_wrist" + assert truth.timing == timing + + +def test_build_action_ground_truth_rejects_unauthorable_dips() -> None: + """A placement at frame <= δ would put its dip where no DipSchedule can + author one (a strict local minimum needs a preceding frame) — fail loudly + at construction, naming the item and frames.""" + timing = CausalTiming(action_lag=5, lag_window=5) + # the reviewer's case: placed at 3 with δ=5 → dip at frame -2 + with pytest.raises(ValueError, match=r"'part_a'.*frame 3.*frame -2"): + build_action_ground_truth( + timing, + "operator", + "right_wrist", + [ItemPlacement(item="part_a", placed_at_frame=3, entity_id="part_a")], + ) + # boundary: placed exactly at δ → dip at frame 0, still unauthorable + with pytest.raises(ValueError, match=r"'part_a'.*frame 5.*frame 0"): + build_action_ground_truth( + timing, + "operator", + "right_wrist", + [ItemPlacement(item="part_a", placed_at_frame=5, entity_id="part_a")], + ) + # just above the boundary: placed at δ+1 → dip at frame 1, authorable + truth = build_action_ground_truth( + timing, + "operator", + "right_wrist", + [ItemPlacement(item="part_a", placed_at_frame=6, entity_id="part_a")], + ) + assert [(p.action_frame, p.change_frame) for p in truth.pairs] == [(1, 6)] + + +def test_sidecar_round_trip(tmp_path: Path) -> None: + timing = CausalTiming(action_lag=1, lag_window=2) + placements = [ItemPlacement(item="part_a", placed_at_frame=2, entity_id="part_a")] + truth = build_action_ground_truth(timing, "operator", "right_wrist", placements) + written = write_actions_json(truth, tmp_path / "interactions.json") + assert json.loads((tmp_path / "interactions.json").read_text()) == written + assert written == { + "timing": {"action_lag": 1, "lag_window": 2}, + "actor_id": "operator", + "tracked_joint": "right_wrist", + "pairs": [ + {"actor_id": "operator", "item_id": "part_a", "action_frame": 1, "change_frame": 2} + ], + } diff --git a/tests/test_assembly_example.py b/tests/test_assembly_example.py index 1b55642..5a4f64e 100644 --- a/tests/test_assembly_example.py +++ b/tests/test_assembly_example.py @@ -93,3 +93,150 @@ def test_order_status_matches_placements(example: ModuleType) -> None: assert result.status.value == "fulfilled" assert result.missing == {} and result.extra == {} and result.wrong == {} assert {p.item for p in placements} == {"part_a", "part_b", "part_c"} + + +# --- placement-synced preset (opt-in) --------------------------------------- # +# Everything below reads ONLY the emitted files (manifest.json / +# interactions.json on disk) — the genuine consumer path. The authored scenario +# constants are pasted as literals so the tests pin the contract, not the code. + + +def _strict_local_minima(series: list[float]) -> list[int]: + """Frames whose value is strictly lower than both neighbours.""" + return [ + i + for i in range(1, len(series) - 1) + if series[i] < series[i - 1] and series[i] < series[i + 1] + ] + + +def _joint_height_series(manifest: dict, entity_id: str, joint: str) -> list[float]: + """A tracked joint's height per frame, read from the on-disk manifest only.""" + entity = next(e for e in manifest["entities"] if e["id"] == entity_id) + return [fr["points"][joint]["xyz_gt"][2] for fr in entity["frames"]] + + +def _item_change_frames(manifest: dict, entity_id: str) -> list[int]: + """Frames at which an item's center moved vs the previous frame (manifest only).""" + entity = next(e for e in manifest["entities"] if e["id"] == entity_id) + positions = [fr["points"]["center"]["xyz_gt"] for fr in entity["frames"]] + return [f for f in range(1, len(positions)) if positions[f] != positions[f - 1]] + + +def test_synced_dips_recoverable_from_manifest_alone(example: ModuleType, tmp_path: Path) -> None: + """Strict local minima of the wrist height == the synced dips + distractor. + + The placements land at frames 2/5/8 with δ=1, so dips sit at 1/4/7, plus + the distractor dip at 10. A consumer with only manifest.json recovers them. + """ + example.run(tmp_path, placement_synced=True) + manifest = json.loads((tmp_path / "manifest.json").read_text()) + + z = _joint_height_series(manifest, "operator", "right_wrist") + dips = _strict_local_minima(z) + assert dips == [1, 4, 7, 10] + # each true dip is exactly δ=1 frame before its placement, strictly dipped + for dip, placed in [(1, 2), (4, 5), (7, 8)]: + assert placed - dip == 1 + assert z[dip] < z[dip - 1] and z[dip] < z[dip + 1] + + +def _naive_causal_forward_associate( + dips: list[int], changes: list[tuple[str, int]], lag_window: int +) -> list[tuple[str, str, int, int]]: + """The associator a consumer would write first: pair each dip with the next + change inside the lag window. ``changes`` are ``(item_id, frame)``.""" + pairs = [] + for dip in sorted(dips): + later = [(item, c) for item, c in changes if 0 < c - dip <= lag_window] + if later: + item, change = min(later, key=lambda ic: ic[1]) + pairs.append(("operator", item, dip, change)) + return pairs + + +def test_synced_negatives_falsify_naive_causal_association( + example: ModuleType, tmp_path: Path +) -> None: + """The negatives must make the naive causal-forward rule score below perfect. + + The distractor dip at 10 places nothing, but part_d's uncaused move at 11 + follows inside the lag window — the naive rule pairs them, and + interactions.json says otherwise: exactly one false positive. + """ + example.run(tmp_path, placement_synced=True) + manifest = json.loads((tmp_path / "manifest.json").read_text()) + truth = json.loads((tmp_path / "interactions.json").read_text()) + lag_window = truth["timing"]["lag_window"] + + dips = _strict_local_minima(_joint_height_series(manifest, "operator", "right_wrist")) + changes = [ + (item, c) + for item in ("part_a", "part_b", "part_c", "part_d") + for c in _item_change_frames(manifest, item) + ] + predicted = _naive_causal_forward_associate(dips, changes, lag_window) + truth_pairs = { + (p["actor_id"], p["item_id"], p["action_frame"], p["change_frame"]) for p in truth["pairs"] + } + + tp = [p for p in predicted if p in truth_pairs] + fp = [p for p in predicted if p not in truth_pairs] + fn = [p for p in truth_pairs if p not in predicted] + precision = len(tp) / len(predicted) + recall = len(tp) / len(truth_pairs) + + # the distractor dip→part_d pairing is THE false positive, by name + assert fp == [("operator", "part_d", 10, 11)] + assert (len(tp), len(fp), len(fn)) == (3, 1, 0) + assert precision == 3 / 4 < 1.0 + assert recall == 1.0 + + # even a δ-informed rule trips on the confounder: a dip sits exactly δ=1 + # before part_d's move, and the ground truth still says it is no pair + assert truth["timing"]["action_lag"] == 11 - 10 + assert all(p["item_id"] != "part_d" for p in truth["pairs"]) + # part_d itself is placed (order stays fulfilled) and fully worktop-visible + assert _item_change_frames(manifest, "part_d") == [11] + + +def test_synced_interactions_sidecar_contents(example: ModuleType, tmp_path: Path) -> None: + """The sidecar carries the timing contract and only the true pairs.""" + summary = example.run(tmp_path, placement_synced=True) + truth = json.loads((tmp_path / "interactions.json").read_text()) + assert truth == { + "timing": {"action_lag": 1, "lag_window": 2}, + "actor_id": "operator", + "tracked_joint": "right_wrist", + "pairs": [ + {"actor_id": "operator", "item_id": "part_a", "action_frame": 1, "change_frame": 2}, + {"actor_id": "operator", "item_id": "part_b", "action_frame": 4, "change_frame": 5}, + {"actor_id": "operator", "item_id": "part_c", "action_frame": 7, "change_frame": 8}, + ], + } + # the late distractor item is placed (order still fulfilled) but is NOT a pair + assert summary["result"].status.value == "fulfilled" + assert "part_d" in summary["result"].placed + + +def test_synced_preset_keeps_complementary_in_view(example: ModuleType, tmp_path: Path) -> None: + """The distractor item joins the fusion story: worktop-only, every frame.""" + vis = example.run(tmp_path, placement_synced=True)["visibility"] + d_ov, _ = vis["part_d"]["overview"] + d_wt, total = vis["part_d"]["worktop"] + assert d_ov == 0 + assert d_wt == total == 13 + + +def test_default_run_has_no_interactions_sidecar(example: ModuleType, tmp_path: Path) -> None: + """Off by default: no sidecar, no extra entities, no extra frames.""" + summary = example.run(tmp_path) + assert not (tmp_path / "interactions.json").exists() + assert summary["interactions"] is None + assert summary["manifest"]["num_frames"] == 11 + assert {e["id"] for e in summary["manifest"]["entities"]} == { + "operator", + "part_a", + "part_b", + "part_c", + }