diff --git a/ssms/basic_simulators/__init__.py b/ssms/basic_simulators/__init__.py index a33e7e19..20394ecd 100644 --- a/ssms/basic_simulators/__init__.py +++ b/ssms/basic_simulators/__init__.py @@ -9,6 +9,10 @@ normalize_simulator_result, validate_observation_result, ) +from .observation_metadata import ( + get_observation_metadata, + validate_observation_metadata, +) from .simulator import OMISSION_SENTINEL from .simulator_class import Simulator @@ -18,8 +22,10 @@ "Simulator", "boundary_functions", "drift_functions", + "get_observation_metadata", "modular_parameter_simulator_adapter", "normalize_simulator_result", "simulator", + "validate_observation_metadata", "validate_observation_result", ] diff --git a/ssms/basic_simulators/observation_metadata.py b/ssms/basic_simulators/observation_metadata.py new file mode 100644 index 00000000..b1f2239b --- /dev/null +++ b/ssms/basic_simulators/observation_metadata.py @@ -0,0 +1,156 @@ +"""Validate and inspect explicit producer observation metadata.""" + +from collections.abc import Callable, Mapping, Sequence +from numbers import Integral +from typing import Any + +import numpy as np + +from .observation_results import ( + OBSERVATION_SCHEMA_VERSION, + _validate_schema, + _validate_schema_version, +) + +_SCHEMA_ATTRIBUTE_NAMES = ( + "observation_schema_version", + "observation_schema", + "observation_schema_profile", +) +_LEGACY_RT_CHOICE_PROFILE = "legacy_rt_choice" + + +def validate_observation_metadata(metadata: Mapping[str, Any]) -> dict[str, Any]: + """Validate an explicit producer descriptor and return its canonical view. + + A descriptor must declare schema version 1 and exactly one of an ordered + ``observation_schema`` or the named ``legacy_rt_choice`` profile. The profile + additionally requires explicit ``choices``. An optional ``obs_dim`` is checked + for consistency, never used to infer a schema. + + The returned plain dictionary contains only the version, a freshly copied schema, + and the width derived from that schema. Input mappings are not mutated. + """ + if not isinstance(metadata, Mapping): + raise TypeError("observation metadata must be a mapping") + if "observation_schema_version" not in metadata: + raise ValueError("observation metadata is missing observation_schema_version") + + _validate_schema_version(metadata["observation_schema_version"]) + schema = _resolve_schema(metadata) + validated_schema = _validate_schema(schema) + copied_schema = tuple(_copy_schema_entry(entry) for entry in validated_schema) + obs_dim = len(copied_schema) + if "obs_dim" in metadata: + _validate_declared_obs_dim(metadata["obs_dim"], obs_dim) + + return { + "observation_schema_version": OBSERVATION_SCHEMA_VERSION, + "observation_schema": copied_schema, + "obs_dim": obs_dim, + } + + +def get_observation_metadata( + producer: Mapping[str, Any] | Callable[..., Any], +) -> dict[str, Any]: + """Return explicit observation metadata for a mapping or callable producer. + + Callable producers are inspected through the explicit schema attributes only and + are never executed. Their legacy ``obs_dim`` attribute is deliberately ignored: + semantic width always comes from the declared schema. + """ + if isinstance(producer, Mapping): + return validate_observation_metadata(producer) + if not callable(producer): + raise TypeError("producer must be an observation metadata mapping or callable") + + metadata = _callable_metadata(producer) + if not { + "observation_schema", + "observation_schema_profile", + }.intersection(metadata): + raise ValueError( + "callable must declare explicit observation metadata through " + "observation_schema or observation_schema_profile" + ) + return validate_observation_metadata(metadata) + + +def _resolve_schema(metadata: Mapping[str, Any]) -> object: + has_schema = "observation_schema" in metadata + has_profile = "observation_schema_profile" in metadata + if has_schema == has_profile: + raise ValueError( + "observation metadata must define exactly one of observation_schema " + "or observation_schema_profile" + ) + if has_schema: + return metadata["observation_schema"] + + profile = metadata["observation_schema_profile"] + if profile != _LEGACY_RT_CHOICE_PROFILE: + raise ValueError( + "observation_schema_profile must be the supported profile " + f"{_LEGACY_RT_CHOICE_PROFILE!r}; got {profile!r}" + ) + if "choices" not in metadata: + raise ValueError("legacy_rt_choice profile requires explicit choices") + + choices = _profile_choices(metadata["choices"]) + return ( + { + "name": "rt", + "kind": "continuous", + "lower": 0.0, + "lower_inclusive": False, + }, + {"name": "response", "kind": "categorical", "values": choices}, + ) + + +def _profile_choices(choices: object) -> tuple[object, ...]: + if isinstance(choices, np.ndarray): + if choices.ndim != 1: + raise TypeError( + "legacy_rt_choice choices must be a one-dimensional sequence" + ) + return tuple(choices.tolist()) + if isinstance(choices, (str, bytes)) or not isinstance(choices, Sequence): + raise TypeError("legacy_rt_choice choices must be a non-empty sequence") + return tuple(choices) + + +def _copy_schema_entry(entry: Mapping[str, Any]) -> dict[str, Any]: + copied = dict(entry) + if entry["kind"] == "categorical": + copied["values"] = tuple(entry["values"]) + return copied + + +def _validate_declared_obs_dim(value: object, derived_obs_dim: int) -> None: + if isinstance(value, (bool, np.bool_)) or not isinstance(value, Integral): + raise TypeError("obs_dim must be an integer when supplied") + if int(value) != derived_obs_dim: + raise ValueError( + "obs_dim must equal the declared observation schema length: " + f"expected {derived_obs_dim}, got {value!r}" + ) + + +def _callable_metadata(producer: object) -> dict[str, Any]: + metadata: dict[str, Any] = {} + for name in _SCHEMA_ATTRIBUTE_NAMES: + try: + metadata[name] = getattr(producer, name) + except AttributeError: + continue + if "observation_schema_profile" in metadata: + try: + metadata["choices"] = getattr(producer, "choices") + except AttributeError: + pass + return metadata + + +__all__ = ["get_observation_metadata", "validate_observation_metadata"] diff --git a/ssms/basic_simulators/observation_results.py b/ssms/basic_simulators/observation_results.py index 9b080515..56ee1259 100644 --- a/ssms/basic_simulators/observation_results.py +++ b/ssms/basic_simulators/observation_results.py @@ -227,12 +227,7 @@ def validate_observation_result(result: Mapping[str, Any]) -> dict[str, Any]: f"{_format_keys(missing_metadata_keys)}" ) - version = metadata["observation_schema_version"] - if type(version) is not int or version != OBSERVATION_SCHEMA_VERSION: - raise ValueError( - "observation_schema_version must be the supported integer version " - f"{OBSERVATION_SCHEMA_VERSION}; got {version!r}" - ) + _validate_schema_version(metadata["observation_schema_version"]) schema = _validate_schema(metadata["observation_schema"], observations.dtype) if observations.shape[-1] != len(schema): @@ -250,7 +245,7 @@ def validate_observation_result(result: Mapping[str, Any]) -> dict[str, Any]: def _validate_schema( - schema: object, observation_dtype: np.dtype[Any] + schema: object, observation_dtype: np.dtype[Any] | None = None ) -> tuple[Mapping[str, Any], ...]: if not isinstance(schema, tuple): raise TypeError("observation_schema must be an ordered tuple of mappings") @@ -305,7 +300,9 @@ def _validate_schema( def _validate_categorical_schema( - entry: Mapping[str, Any], name: str, observation_dtype: np.dtype[Any] + entry: Mapping[str, Any], + name: str, + observation_dtype: np.dtype[Any] | None, ) -> None: if "values" not in entry: raise ValueError(f"categorical field {name!r} requires values") @@ -326,7 +323,9 @@ def _validate_categorical_schema( f"categorical field {name!r} values must be finite, " "integer-valued numeric labels" ) - if not _integer_is_exactly_representable(integer_value, observation_dtype): + if observation_dtype is not None and not _integer_is_exactly_representable( + integer_value, observation_dtype + ): raise ValueError( f"categorical field {name!r} value {value!r} is not exactly " f"representable in observations dtype {observation_dtype.name}" @@ -337,6 +336,14 @@ def _validate_categorical_schema( raise ValueError(f"categorical field {name!r} values must be unique") +def _validate_schema_version(version: object) -> None: + if type(version) is not int or version != OBSERVATION_SCHEMA_VERSION: + raise ValueError( + "observation_schema_version must be the supported integer version " + f"{OBSERVATION_SCHEMA_VERSION}; got {version!r}" + ) + + def _validate_continuous_schema(entry: Mapping[str, Any], name: str) -> None: for endpoint in ("lower", "upper"): inclusive = f"{endpoint}_inclusive" diff --git a/tests/test_observation_metadata.py b/tests/test_observation_metadata.py new file mode 100644 index 00000000..cd6ab16e --- /dev/null +++ b/tests/test_observation_metadata.py @@ -0,0 +1,308 @@ +"""Tests for explicit producer observation metadata.""" + +from collections.abc import Mapping +from types import MappingProxyType +from typing import Any + +import numpy as np +import pytest + + +RT = { + "name": "rt", + "kind": "continuous", + "lower": 0.0, + "lower_inclusive": False, +} +RESPONSE = {"name": "response", "kind": "categorical", "values": (-1, 1)} +CONFIDENCE = { + "name": "confidence", + "kind": "continuous", + "lower": 0.0, + "upper": 1.0, +} +ANGLE = { + "name": "angle", + "kind": "circular", + "lower": -np.pi, + "upper": np.pi, +} + + +def _validate(metadata: object) -> dict[str, Any]: + from ssms.basic_simulators import validate_observation_metadata + + return validate_observation_metadata(metadata) + + +def _get(producer: object) -> dict[str, Any]: + from ssms.basic_simulators import get_observation_metadata + + return get_observation_metadata(producer) + + +def _explicit( + schema: tuple[Mapping[str, Any], ...], *, obs_dim: object | None = None +) -> dict[str, Any]: + metadata: dict[str, Any] = { + "observation_schema_version": 1, + "observation_schema": schema, + } + if obs_dim is not None: + metadata["obs_dim"] = obs_dim + return metadata + + +def test_validate_observation_metadata_expands_explicit_legacy_profile() -> None: + source = MappingProxyType( + { + "observation_schema_version": 1, + "observation_schema_profile": "legacy_rt_choice", + "choices": [-1, 1], + "obs_dim": 2, + "producer_extension": object(), + } + ) + + descriptor = _validate(source) + + assert descriptor == { + "observation_schema_version": 1, + "observation_schema": ( + RT, + {"name": "response", "kind": "categorical", "values": (-1, 1)}, + ), + "obs_dim": 2, + } + assert tuple(descriptor) == ( + "observation_schema_version", + "observation_schema", + "obs_dim", + ) + + +@pytest.mark.parametrize( + "schema", + [ + (RESPONSE,), + ({"name": "latency", "kind": "continuous"},), + (RT, CONFIDENCE, RESPONSE), + (RT, CONFIDENCE, ANGLE, RESPONSE), + ], + ids=("response-only", "continuous-only", "mixed-three", "mixed-four"), +) +def test_validate_observation_metadata_accepts_explicit_generic_schemas( + schema: tuple[Mapping[str, Any], ...], +) -> None: + descriptor = _validate(_explicit(schema, obs_dim=len(schema))) + + assert descriptor["obs_dim"] == len(schema) + assert tuple(field["name"] for field in descriptor["observation_schema"]) == tuple( + field["name"] for field in schema + ) + assert all(type(field) is dict for field in descriptor["observation_schema"]) + + +@pytest.mark.parametrize("version", [0, 2, True, "1"]) +def test_validate_observation_metadata_rejects_unsupported_versions( + version: object, +) -> None: + metadata = _explicit((RT,)) + metadata["observation_schema_version"] = version + + with pytest.raises(ValueError, match="observation_schema_version"): + _validate(metadata) + + +@pytest.mark.parametrize( + ("metadata", "error", "message"), + [ + ([], TypeError, "mapping"), + ({"observation_schema": (RT,)}, ValueError, "schema_version"), + ( + { + "observation_schema_version": 1, + "observation_schema": (RT,), + "observation_schema_profile": "legacy_rt_choice", + "choices": (-1, 1), + }, + ValueError, + "exactly one", + ), + ({"observation_schema_version": 1}, ValueError, "exactly one"), + ( + {"observation_schema_version": 1, "observation_schema": [RT]}, + TypeError, + "ordered tuple", + ), + (_explicit((RT,), obs_dim=True), TypeError, "obs_dim"), + (_explicit((RT,), obs_dim=0), ValueError, "obs_dim"), + (_explicit((RT,), obs_dim=2), ValueError, "obs_dim"), + ], +) +def test_validate_observation_metadata_rejects_malformed_descriptors( + metadata: object, + error: type[Exception], + message: str, +) -> None: + with pytest.raises(error, match=message): + _validate(metadata) + + +@pytest.mark.parametrize( + ("profile", "choices", "error", "message"), + [ + ("unknown", (-1, 1), ValueError, "profile"), + ("legacy_rt_choice", None, ValueError, "choices"), + ("legacy_rt_choice", (), ValueError, "values"), + ("legacy_rt_choice", (0.5, 1), ValueError, "values"), + ("legacy_rt_choice", (1, 1), ValueError, "unique"), + ("legacy_rt_choice", "01", TypeError, "sequence"), + ( + "legacy_rt_choice", + np.zeros((1, 2), dtype=int), + TypeError, + "one-dimensional", + ), + ], +) +def test_validate_observation_metadata_rejects_malformed_profiles( + profile: str, + choices: object, + error: type[Exception], + message: str, +) -> None: + metadata: dict[str, Any] = { + "observation_schema_version": 1, + "observation_schema_profile": profile, + } + if choices is not None: + metadata["choices"] = choices + + with pytest.raises(error, match=message): + _validate(metadata) + + +@pytest.mark.parametrize( + "metadata", + [ + {"observation_schema_version": 1, "choices": (-1, 1), "obs_dim": 2}, + {"observation_schema_version": 1, "nchoices": 2, "obs_dim": 2}, + ], +) +def test_validate_observation_metadata_never_infers_schema( + metadata: Mapping[str, Any], +) -> None: + with pytest.raises(ValueError, match="exactly one"): + _validate(metadata) + + +def test_get_observation_metadata_reads_callable_attributes_without_execution() -> None: + class Producer: + observation_schema_version = 1 + observation_schema_profile = "legacy_rt_choice" + choices = np.asarray([0, 1, 2]) + obs_dim = 99 + calls = 0 + + def __call__(self) -> None: + self.calls += 1 + raise AssertionError("metadata inspection must not execute the producer") + + producer = Producer() + + descriptor = _get(producer) + + assert producer.calls == 0 + assert descriptor["obs_dim"] == 2 + assert descriptor["observation_schema"][1]["values"] == (0, 1, 2) + + +def test_get_observation_metadata_reads_explicit_schema_callable() -> None: + calls: list[bool] = [] + + def producer() -> None: + calls.append(True) + raise AssertionError("metadata inspection must not execute the producer") + + setattr(producer, "observation_schema_version", 1) + setattr(producer, "observation_schema", (RESPONSE,)) + + descriptor = _get(producer) + + assert not calls + assert descriptor["observation_schema"] == (RESPONSE,) + assert descriptor["obs_dim"] == 1 + + +def test_get_observation_metadata_requires_explicit_callable_declaration() -> None: + class LegacyProducer: + choices = (-1, 1) + nchoices = 2 + obs_dim = 2 + calls = 0 + + def __call__(self) -> None: + self.calls += 1 + + producer = LegacyProducer() + + with pytest.raises(ValueError, match="explicit observation metadata"): + _get(producer) + assert producer.calls == 0 + + +def test_get_observation_metadata_requires_profile_choices_on_callable() -> None: + class Producer: + observation_schema_version = 1 + observation_schema_profile = "legacy_rt_choice" + + def __call__(self) -> None: + raise AssertionError("metadata inspection must not execute the producer") + + with pytest.raises(ValueError, match="requires explicit choices"): + _get(Producer()) + + +def test_get_observation_metadata_rejects_nonproducer_objects() -> None: + with pytest.raises(TypeError, match="mapping or callable"): + _get(object()) + + +def test_observation_metadata_access_is_pure_and_returns_fresh_plain_values() -> None: + values = [-1, 1] + field = {"name": "response", "kind": "categorical", "values": values} + source = _explicit((field,)) + + first = _get(source) + second = _validate(source) + + assert type(first) is dict + assert first is not second + assert first["observation_schema"] is not source["observation_schema"] + assert first["observation_schema"] is not second["observation_schema"] + assert first["observation_schema"][0] is not field + assert first["observation_schema"][0] is not second["observation_schema"][0] + assert first["observation_schema"][0]["values"] is not values + first["observation_schema"][0]["name"] = "changed" + assert field["name"] == "response" + assert second["observation_schema"][0]["name"] == "response" + + +def test_result_validation_keeps_dtype_specific_categorical_precision_check() -> None: + from ssms.basic_simulators import validate_observation_result + + label = 2**24 + 1 + result = { + "observations": np.asarray([[[0.0]]], dtype=np.float32), + "omission_mask": np.zeros((1, 1), dtype=bool), + "metadata": _explicit( + ({"name": "response", "kind": "categorical", "values": (label,)},) + ), + } + descriptor = _validate(result["metadata"]) + + assert descriptor["observation_schema"][0]["values"] == (label,) + + with pytest.raises(ValueError, match="not exactly representable.*float32"): + validate_observation_result(result)