"""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