diff --git a/src/multicam_sim/appearance.py b/src/multicam_sim/appearance.py index ac824c9..2eab391 100644 --- a/src/multicam_sim/appearance.py +++ b/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) diff --git a/tests/test_appearance.py b/tests/test_appearance.py index 500863a..2b93cad 100644 --- a/tests/test_appearance.py +++ b/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: