Per-observation appearance noise for a graded re-id dose-response

src/multicam_sim/appearance.py

@@ -19,6 +19,15 @@
 a shared mean, indistinguishable — the impossible case). Sweeping it traces how
 re-id accuracy degrades as identities become harder to separate.
 
+**Per-observation noise.** :meth:`AppearanceTable.observe` draws a noisy unit
+observation around an entity's anchor — one camera's *look* at the entity — via an
+isotropic Gaussian jitter of stddev ``sigma``. With a single clean anchor per
+entity a separation sweep is a degenerate cliff (perfect until total collapse,
+zero variance); an intra-entity ``sigma`` gives each observation real spread, so
+the sweep becomes a graded dose-response. ``separation`` is the inter-entity gap,
+``sigma`` the intra-entity spread; ``sigma == 0.0`` reduces exactly to the clean
+anchor, so the anchor-only behaviour is the special case.
+
 Like :mod:`multicam_sim.activity` and :mod:`multicam_sim.possession`, this is a
 pure typed-model sidecar. The table rides in a JSON sidecar and attaches to
 :class:`~multicam_sim.scene.Scene` via an optional field that the manifest builder
@@ -142,6 +151,56 @@ def sample(
         ]
         return cls(dim=dim, separation=separation, seed=seed, entries=entries)
 
+    def observe(self, entity_id: str, *, obs_seed: int, sigma: float) -> list[float]:
+        """Draw one noisy, unit-norm *observation* of an entity around its anchor.
+
+        Where :meth:`sample` fixes one clean anchor per entity, this models a
+        single camera's *observation* of that entity: an isotropic-Gaussian jitter
+        around the anchor, re-projected to the unit sphere::
+
+            observed = normalize(anchor(entity_id) + sigma * N(0, I_dim))
+
+        The noise is drawn from ``numpy.random.default_rng(obs_seed)`` with a fixed
+        draw order (a single ``standard_normal(dim)`` call), so a given
+        ``(entity_id, obs_seed, sigma)`` is fully deterministic; different
+        ``obs_seed`` values give different valid unit observations of the *same*
+        anchor.
+
+        ``sigma`` (>= 0) is the observation-noise stddev: the intra-entity spread,
+        modelling sensor / viewpoint variation across cameras. It is orthogonal to
+        the table's :attr:`separation`, which is the inter-entity gap. Together they
+        set the confusability of a single observation: ``separation`` sets how far
+        apart two entities' anchors sit, ``sigma`` sets how far a single look drifts
+        from its own anchor, and re-id gets hard once the intra-entity spread rivals
+        the inter-entity gap. ``sigma == 0.0`` reduces EXACTLY to the clean anchor
+        (a no-op), so the pre-noise behaviour is the special case. Like the anchor,
+        an observation does NOT encode or leak the ground-truth ``entity_id``.
+
+        Raises ``ValueError`` if ``sigma`` is negative or ``entity_id`` is unknown.
+        """
+        if sigma < 0.0:
+            raise ValueError("sigma must be >= 0")
+        anchor = self._anchor(entity_id)
+        # sigma == 0 is an exact no-op: return the stored anchor verbatim. (Adding
+        # zero noise then renormalising would perturb the anchor's float-rounded
+        # last bits, so short-circuit to keep the reduction bitwise-exact.)
+        if sigma == 0.0:
+            return list(anchor)
+        rng = np.random.default_rng(obs_seed)
+        vec = np.asarray(anchor, dtype=np.float64) + sigma * rng.standard_normal(self.dim)
+        norm = float(np.linalg.norm(vec))
+        # Guard the astronomically unlikely zero vector by falling back to the
+        # anchor so the observation stays unit-norm and finite.
+        if norm <= 0.0:
+            return list(anchor)
+        return [float(x) for x in vec / norm]
+
+    def _anchor(self, entity_id: str) -> list[float]:
+        for entry in self.entries:
+            if entry.entity_id == entity_id:
+                return entry.descriptor
+        raise ValueError(f"unknown entity_id: {entity_id!r}")
+
     def to_json(self, *, indent: int | None = 2) -> str:
         """Serialise to a JSON string (the ``appearance.json`` sidecar payload)."""
         return self.model_dump_json(indent=indent)

tests/test_appearance.py

@@ -137,6 +137,71 @@ def test_roundtrip_json() -> None:
     assert restored == table
 
 
+def test_observe_sigma_zero_is_exact_anchor() -> None:
+    """sigma == 0 is a bitwise no-op: observe() returns the stored anchor verbatim."""
+    table = AppearanceTable.sample(_ENTITY_IDS, dim=256, separation=0.8, seed=7)
+    for entry in table.entries:
+        obs = table.observe(entry.entity_id, obs_seed=99, sigma=0.0)
+        # Pinned to the ACTUAL stored anchor, not a recomputed sample().
+        assert obs == entry.descriptor
+
+
+def test_observe_deterministic_and_seed_sensitive() -> None:
+    """Same (entity_id, obs_seed, sigma) -> identical vector; different obs_seed ->
+    a different but still valid unit observation of the same anchor."""
+    table = AppearanceTable.sample(_ENTITY_IDS, dim=256, separation=0.8, seed=7)
+    a = table.observe("entity_a", obs_seed=3, sigma=0.5)
+    b = table.observe("entity_a", obs_seed=3, sigma=0.5)
+    assert a == b
+
+    c = table.observe("entity_a", obs_seed=4, sigma=0.5)
+    assert c != a
+    assert _is_unit(c)
+    assert all(math.isfinite(x) for x in c)
+
+
+def test_observe_unit_norm_finite_and_dim() -> None:
+    """Every noisy observation is unit-norm, finite, and exactly `dim` long."""
+    for dim in (32, 256):
+        table = AppearanceTable.sample(_ENTITY_IDS, dim=dim, separation=0.7, seed=5)
+        for sigma in (0.0, 0.25, 1.0, 2.0):
+            for entry in table.entries:
+                obs = table.observe(entry.entity_id, obs_seed=11, sigma=sigma)
+                assert len(obs) == dim
+                assert _is_unit(obs)
+                assert all(math.isfinite(x) for x in obs)
+
+
+def test_observe_noise_monotonic_drift_from_anchor() -> None:
+    """Mean cosine(observation, anchor) DECREASES as sigma increases: more noise
+    means observations drift further from the anchor. Averaged over >=3 seeds and
+    >=3 sigma values so the assertion clears sampling jitter."""
+    sigmas = [0.25, 0.5, 1.0, 2.0]
+    seeds = [0, 1, 2, 3, 4]
+    table = AppearanceTable.sample(_ENTITY_IDS, dim=512, separation=0.8, seed=7)
+    anchors = {e.entity_id: np.asarray(e.descriptor) for e in table.entries}
+
+    mean_cos = []
+    for sigma in sigmas:
+        cosines = [
+            float(np.asarray(table.observe(eid, obs_seed=s, sigma=sigma)) @ anchors[eid])
+            for eid in anchors
+            for s in seeds
+        ]
+        mean_cos.append(sum(cosines) / len(cosines))
+
+    for lower_sigma, higher_sigma in zip(mean_cos, mean_cos[1:], strict=False):
+        assert higher_sigma < lower_sigma, f"non-monotone drift: {mean_cos}"
+
+
+def test_observe_rejects_negative_sigma_and_unknown_id() -> None:
+    table = AppearanceTable.sample(_ENTITY_IDS, dim=32, separation=0.8, seed=7)
+    with pytest.raises(ValueError):
+        table.observe("entity_a", obs_seed=0, sigma=-0.1)
+    with pytest.raises(ValueError):
+        table.observe("nope", obs_seed=0, sigma=0.5)
+
+
 def _scene_with_appearance(*, attach: bool) -> Scene:
     scene = build_smoke_scene()
     if not attach: