orisys-sdk/tests/test_sensor_context.py

"""Read-only SensorContext (T04): SN/config/assets/cal_path bound once per instance.

The context is the stage-1 answer to "which physical sensor is this, which
configuration and assets does it run, and which force calibration was selected".
These tests protect the two properties the task names: the binding happens once
from explicit inputs (no SN guessed from enumeration order), and two Sensor
instances do not share or overwrite each other's context. The identity itself is
built through the frozen T03 ``FrameIdentity``, not a second shape.
"""

from __future__ import annotations

import dataclasses
import json
from pathlib import Path

import pytest

from orisys import Sensor, configs
from orisys.sensors import SensorContext, build_sensor_context
from orisys.stream_contract import FrameIdentity

# The shipped box calibration has no declared identity; the device tests add one,
# exactly like an operator-managed <product>/<sn>.json file.
BOX_CALIBRATION = Path(configs.resolve_sensor_bundle("box").force_calibration_json)
BOX_CONFIG = Path(configs.resolve_sensor_bundle("box").sensor_json)


def _device_calibration(root: Path, sn: str, *, family: str = "box") -> Path:
    """Write a device calibration whose declared SN is ``sn``."""
    document = json.loads(BOX_CALIBRATION.read_text(encoding="utf-8-sig"))
    document["identity"] = {"sensor_type": family, "sn": sn}
    path = root / family / f"{sn}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(document), encoding="utf-8")
    return path


def _external_sensor(**kwargs) -> Sensor:
    options = {"verbose": False, "backend": "cpu"}
    options.update(kwargs)
    return Sensor("external", **options)


# ------------------------------------------------------------ explicit binding
def test_context_binds_the_explicit_sn_config_and_calibration_path(
    tmp_path: Path,
) -> None:
    root = tmp_path / "calibrations"
    calibration = _device_calibration(root, "BOX-001")
    sensor = _external_sensor(
        config_name="box", calibration_dir=root, device_id="BOX-001"
    )
    try:
        context = sensor.context
        assert isinstance(context, SensorContext)
        assert context.sensor_id == "BOX-001"
        assert sensor.device_id == "BOX-001"
        assert context.product_family == "box"
        assert context.config_name == "box"
        assert context.config_source == str(BOX_CONFIG.resolve())
        assert len(context.config_fingerprint) == 64
        assert context.bundle_root == str(BOX_CONFIG.parent.resolve())
        assert context.calibration_path == str(calibration.resolve())
        assert context.calibration_fingerprint == sensor.calibration.fingerprint
        assert context.diagnostics == ()
        # The published attributes still say what they said before T04.
        assert sensor.calibration.source_path == calibration.resolve()
        assert sensor.force_unit == "N"
    finally:
        sensor.disconnect()


def test_context_never_guesses_an_sn_from_enumeration_order(tmp_path: Path) -> None:
    """No explicit SN means no identity; the stream header refuses to invent one."""
    first = _external_sensor(config_name="box")
    second = _external_sensor(config_name="box")
    try:
        assert first.context.sensor_id is None
        assert second.context.sensor_id is None
        # Same configuration and bundle, so the non-identity binding is stable.
        assert first.context.config_fingerprint == second.context.config_fingerprint
        assert first.context.bundle_root == second.context.bundle_root
        with pytest.raises(ValueError, match="explicit sensor_id"):
            first.context.frame_identity(
                session_id="run-1", frame_id=0, source_frame_id=None
            )
    finally:
        first.disconnect()
        second.disconnect()


def test_two_instances_keep_separate_contexts(tmp_path: Path) -> None:
    root_one = tmp_path / "one"
    root_two = tmp_path / "two"
    _device_calibration(root_one, "BOX-001")
    _device_calibration(root_two, "BOX-002")
    first = _external_sensor(
        config_name="box", calibration_dir=root_one, device_id="BOX-001"
    )
    second = _external_sensor(
        config_name="box", calibration_dir=root_two, device_id="BOX-002"
    )
    try:
        first_context = first.context
        first_identity = first_context.frame_identity(
            session_id="run-1", frame_id=0, source_frame_id=None
        )
        second_context = second.context
        second_identity = second_context.frame_identity(
            session_id="run-1", frame_id=0, source_frame_id=None
        )
        assert first_context != second_context
        assert first_identity != second_identity
        assert first_identity.sensor_id == "BOX-001"
        assert second_identity.sensor_id == "BOX-002"
        # Building the second Sensor did not mutate the first context.
        assert first.context == first_context
        assert first.context.calibration_path != second_context.calibration_path
        assert isinstance(first_identity, FrameIdentity)
    finally:
        first.disconnect()
        second.disconnect()


def test_context_is_frozen_read_only_value_data(tmp_path: Path) -> None:
    root = tmp_path / "calibrations"
    _device_calibration(root, "BOX-001")
    sensor = _external_sensor(
        config_name="box", calibration_dir=root, device_id="BOX-001"
    )
    try:
        context = sensor.context
        with pytest.raises(dataclasses.FrozenInstanceError):
            context.sensor_id = "BOX-999"  # type: ignore[misc]
        payload = context.as_dict()
        assert payload["sensor_id"] == "BOX-001"
        assert isinstance(payload["assets"], list)
    finally:
        sensor.disconnect()


def test_calibration_swap_replaces_the_context_atomically(tmp_path: Path) -> None:
    root = tmp_path / "calibrations"
    first_cal = _device_calibration(root, "BOX-001")
    # The same device SN, a different calibration document: a live save always
    # keeps the device identity, which is what a real recalibration does.
    variant = root / "box" / "BOX-001-variant.json"
    document = json.loads(BOX_CALIBRATION.read_text(encoding="utf-8-sig"))
    document["identity"] = {"sensor_type": "box", "sn": "BOX-001"}
    # A real recalibration is a different document; changing one point keeps the
    # curve valid while making the content fingerprint different.
    document["channels"]["normal"]["points"][1][1] = 0.3771
    variant.write_text(json.dumps(document), encoding="utf-8")
    sensor = _external_sensor(
        config_name="box", calibration_dir=root, device_id="BOX-001"
    )
    try:
        before = sensor.context
        assert before.calibration_path == str(first_cal.resolve())

        # The published live-swap entry stays available; the calibration identity
        # must still match the bound SN.
        sensor.set_calibration(variant)
        after = sensor.context
        assert after is not before
        assert after.sensor_id == before.sensor_id == "BOX-001"
        assert after.config_fingerprint == before.config_fingerprint
        assert after.bundle_root == before.bundle_root
        assert after.calibration_path == str(variant.resolve())
        assert after.calibration_fingerprint == sensor.calibration.fingerprint
        assert after.calibration_fingerprint != before.calibration_fingerprint

        sensor.set_calibration(None)
        assert sensor.context.calibration_path is None
        assert sensor.context.calibration_fingerprint == "none"
        assert sensor.force_unit == "raw"
    finally:
        sensor.disconnect()


def test_clearing_a_calibration_loaded_from_a_path_reports_no_path() -> None:
    """The context describes the calibration in effect, not the path ever requested.

    Constructing with an explicit ``cal_path`` then clearing the calibration must
    leave no calibration path behind: the old path would otherwise be recorded as
    if that calibration still shaped the measurements.
    """
    sensor = _external_sensor(config_name="box", cal_path=str(BOX_CALIBRATION))
    try:
        assert sensor.context.calibration_path == str(BOX_CALIBRATION.resolve())
        assert sensor.force_unit == "N"

        sensor.set_calibration(None)
        assert sensor.calibration is None
        assert sensor.context.calibration_path is None
        assert sensor.context.calibration_fingerprint == "none"
        assert sensor.force_unit == "raw"
    finally:
        sensor.disconnect()


def test_binding_failure_is_reported_without_breaking_construction(
    tmp_path: Path, monkeypatch
) -> None:
    """A context problem must not turn a previously constructible Sensor into an error."""
    import orisys.sensors.context as context_module

    def broken(**kwargs):
        raise RuntimeError("bundle exploded")

    monkeypatch.setattr(context_module, "build_sensor_context", broken)
    sensor = _external_sensor(config_name="box")
    try:
        assert sensor.context.sensor_id is None
        assert sensor.context.config_name == "box"
        assert sensor.context.diagnostics == (
            "context binding failed: RuntimeError: bundle exploded",
        )
        assert sensor.context.calibration_fingerprint == "none"
    finally:
        sensor.disconnect()


# --------------------------------------------------- configuration by path/name
def test_config_binding_follows_the_actual_loaded_file(tmp_path: Path) -> None:
    custom = tmp_path / "custom-box.json"
    custom.write_text(BOX_CONFIG.read_text(encoding="utf-8"), encoding="utf-8")
    sensor = _external_sensor(config_name=str(custom))
    try:
        context = sensor.context
        assert context.config_source == str(custom.resolve())
        assert context.config_name == str(custom)
        assert context.product_family is None
        assert context.bundle_root is None
        assert context.assets == ()
        assert any("no additive sensor bundle" in line for line in context.diagnostics)
    finally:
        sensor.disconnect()

    # The recorded source is the file load_config() actually reads.
    assert configs.resolve_config_source("box") == BOX_CONFIG.resolve()
    assert configs.resolve_config_source(custom) == custom.resolve()
    assert configs.resolve_config_source("no-such-config-name") is None


def test_same_configuration_has_the_same_config_fingerprint(tmp_path: Path) -> None:
    root = tmp_path / "calibrations"
    _device_calibration(root, "BOX-001")
    first = _external_sensor(config_name="box")
    second = _external_sensor(
        config_name="box", calibration_dir=root, device_id="BOX-001"
    )
    try:
        assert first.context.config_fingerprint == second.context.config_fingerprint
        assert (
            first.context.processing_backend
            == second.context.processing_backend
            == "compound_eye"
        )
    finally:
        first.disconnect()
        second.disconnect()


# --------------------------------------------------------------- asset binding
def test_bundle_assets_are_bound_as_relative_paths_once() -> None:
    finger_config = configs.load_config("finger")
    context = build_sensor_context(
        config_name="finger",
        config=finger_config,
        device_id="FINGER-001",
        bundle_candidates=["finger"],
    )
    assert context.sensor_id == "FINGER-001"
    assert context.product_family == "finger"
    assert context.bundle_root is not None
    assert context.manifest_digest is not None and len(context.manifest_digest) == 64
    roles = {asset.role for asset in context.assets}
    assert "assets_dir" in roles
    root = Path(context.bundle_root)
    for asset in context.assets:
        assert not Path(asset.path).is_absolute()
        assert (root / asset.path).parent.is_dir()

    # Two bindings of the same bundle describe the same immutable assets.
    again = build_sensor_context(
        config_name="finger", config=finger_config, bundle_candidates=["finger"]
    )
    assert again.assets == context.assets
    assert again.manifest_digest == context.manifest_digest
    assert again.config_fingerprint == context.config_fingerprint