orisys-app/apps/studio/tests/test_session_metadata.py

"""Acceptance for the facts a recording states about its own session.

``session.json`` is the only witness a reader has later, so these are the rules that
keep it honest: which frame rate a container may declare, that the filters are read back
from the worker that ran them, that no metadata read can fail a recording, and that the
3D pane's own description is passed through untouched.
"""

from __future__ import annotations

import hashlib
from pathlib import Path
from types import SimpleNamespace

import numpy as np
import pytest

from orisys_app_shared.force.filters import FilterSettings
from orisys_app_shared.record import RecordConfig
from orisys_studio import session_metadata as sm

CONFIG = (
    Path(__file__).resolve().parents[1]
    / "config"
    / "sensors"
    / "finger"
    / "sensor.json"
)


# --------------------------------------------------------------------- config hash
def test_the_config_hash_is_the_file_content_or_nothing(tmp_path):
    path = tmp_path / "sensor.json"
    path.write_text('{"camera": {}}', encoding="utf-8")
    assert sm.config_sha256(path) == hashlib.sha256(path.read_bytes()).hexdigest()
    assert sm.config_sha256(tmp_path / "gone.json") == ""
    assert sm.config_sha256("") == ""


def test_the_sdk_version_is_a_string_even_with_no_sdk():
    value = sm.sdk_version()
    assert isinstance(value, str) and value


# ------------------------------------------------------------------- device serial
def test_the_device_serial_comes_from_whichever_field_the_sdk_uses():
    assert sm.device_serial(SimpleNamespace(sn="A1")) == "A1"
    assert sm.device_serial(SimpleNamespace(serial="B2")) == "B2"
    assert sm.device_serial(SimpleNamespace(serial_number="C3")) == "C3"
    assert sm.device_serial(SimpleNamespace(info=SimpleNamespace(SN="D4"))) == "D4"
    assert sm.device_serial(SimpleNamespace()) == ""
    assert sm.device_serial(None) == ""


# -------------------------------------------------------------- capture / filters
def test_capture_metadata_copies_what_the_worker_says_and_survives_failure():
    assert sm.capture_metadata(None) == {}
    assert sm.capture_metadata(SimpleNamespace()) == {}
    worker = SimpleNamespace(capture_status=lambda: {"requested_fps": 30.0})
    assert sm.capture_metadata(worker) == {"requested_fps": 30.0}

    def explode():
        raise RuntimeError("driver gone")

    # A recording must not fail because the camera cannot describe itself.
    assert sm.capture_metadata(SimpleNamespace(capture_status=explode)) == {}


def test_filter_metadata_names_what_the_sdk_could_not_apply():
    settings = FilterSettings()
    worker = SimpleNamespace(
        filter_state=lambda: {"flow_ema_alpha": 1.0},
        filter_settings=lambda: settings,
        filters_unsupported=lambda: ("force_lowpass_order",),
    )
    block = sm.filter_metadata(worker)
    assert block["unsupported_by_sdk"] == ["force_lowpass_order"]
    assert block  # the applied settings are stated too

    # No worker, and a worker that raises, both give a block rather than an exception.
    assert sm.filter_metadata(None)
    def explode():
        raise RuntimeError("worker is gone")
    assert sm.filter_metadata(SimpleNamespace(filter_state=explode))


def test_filter_settings_default_to_nothing_filters():
    """The shipped configs state "no filters"; an empty preference must agree."""
    empty = sm.filter_settings_from_prefs({})
    assert empty == FilterSettings()

    stored = sm.filter_settings_from_prefs(
        {"plot_force_median_window": 3, "plot_flow_ema_alpha": 0.5, "unrelated": 1}
    )
    assert stored != empty
    assert stored.median_window == 3
    assert stored.flow_ema_alpha == pytest.approx(0.5)


def test_a_broken_preferences_object_is_not_fatal():
    class Broken(dict):
        def get(self, key, default=None):  # noqa: D102 - deliberately broken
            raise RuntimeError("corrupt preferences")

    assert sm.filter_settings_from_prefs(Broken()) == FilterSettings()


# ------------------------------------------------------------------- frame rates
def test_a_camera_declares_its_configured_rate_and_a_file_its_measured_one():
    """The container's fps must describe the spacing of the pictures it holds."""
    if CONFIG.is_file():
        assert sm.video_fps(48.0, source_is_file=False, config_ref=CONFIG) == 30.0
    # A file source has no configured rate to honour.
    assert sm.video_fps(17.5, source_is_file=True, config_ref=CONFIG) == 17.5
    # An unreadable config falls back to the measured rate, never to zero.
    assert sm.video_fps(12.0, source_is_file=False, config_ref="missing.json") == 12.0
    assert sm.video_fps(0.0, source_is_file=True, config_ref=None) == float(
        RecordConfig().fps
    )


def test_the_filter_dialog_gets_a_rate_it_can_check_nyquist_against():
    assert sm.device_sample_rate(25.0, None, default=30.0) == 25.0
    assert sm.device_sample_rate(None, "15", default=30.0) == 15.0
    assert sm.device_sample_rate("garbage", "also garbage", default=30.0) == 30.0
    assert sm.device_sample_rate(None, 0, default=30.0) == 30.0


# ------------------------------------------------------------- the whole metadata
def test_the_session_block_states_the_device_the_filters_and_the_picture():
    info = SimpleNamespace(
        raw_frame=np.zeros((48, 64, 3), dtype=np.uint8), force_calibrated=True, sn="SN-9"
    )
    worker = SimpleNamespace(
        capture_status=lambda: {"requested_fps": 30.0, "measured_fps": 21.0},
        filter_state=lambda: {},
        filter_settings=lambda: FilterSettings(),
        filters_unsupported=lambda: (),
    )
    facts = sm.SessionFacts(
        sensor_type="finger",
        config_ref=str(CONFIG) if CONFIG.is_file() else "missing.json",
        worker=worker,
        info=info,
        video_fps=30.0,
        measured_fps=21.0,
        active_calibration_id="cal-7",
        operator_sn="  OPERATOR-1  ",
        recording_mode="video",
        visualization3d={"block_version": 1, "enabled": True},
    )
    config = RecordConfig()
    block = sm.build_session_metadata(config, facts)

    assert block["device"]["sensor_type"] == "finger"
    assert block["device"]["frame"] == {
        "width": 64,
        "height": 48,
        "fps": 30.0,
        "measured_fps": 21.0,
    }
    assert block["versions"]["sdk"]
    assert block["capture"] == {"requested_fps": 30.0, "measured_fps": 21.0}
    assert block["calibration"]["force_calibrated"] is True
    assert block["calibration"]["version"] == "cal-7"
    assert block["calibration"]["sn"] == "OPERATOR-1"
    assert block["calibration"]["device_sn"] == "SN-9"
    assert block["depth"]["stored"] is False
    assert block["depth"]["config_sha256"] == block["device"]["config_sha256"]
    assert block["filter"]
    assert block["recording_mode"] == "video"
    assert block["video_config"] == config.as_dict()
    assert block["visualization3d"] == {"block_version": 1, "enabled": True}
    assert block["conventions"]["axis_order"] == ["FN", "FX", "FY"]


def test_the_block_survives_a_worker_that_is_gone_and_a_packet_that_never_came():
    """Metadata that cannot be read is empty, never an exception (a record is running)."""
    block = sm.build_session_metadata(
        RecordConfig(),
        sm.SessionFacts(sensor_type="box", config_ref="", worker=None, info=None),
    )
    assert block["device"]["frame"]["width"] is None
    assert block["capture"] == {}
    assert block["filter"]
    assert block["device"]["config_sha256"] == ""


def test_the_3d_pane_description_is_passed_through_untouched():
    described = {"scene": {"asset_id": "x"}, "gl": {}, "display": {}, "stream": {}}
    block = sm.build_session_metadata(
        RecordConfig(),
        sm.SessionFacts(
            sensor_type="finger",
            config_ref="",
            visualization3d={"visualization3d": described, "block_version": 1},
        ),
    )
    assert block["visualization3d"]["visualization3d"] == described


@pytest.mark.parametrize("mode", ("video", "raw", "lossless"))
def test_every_recording_mode_is_recorded_as_stated(mode):
    block = sm.build_session_metadata(
        RecordConfig(),
        sm.SessionFacts(sensor_type="finger", config_ref="", recording_mode=mode),
    )
    assert block["recording_mode"] == mode