diff --git a/tests/coordinators/test_detector_setup_coordinator.py b/tests/coordinators/test_detector_setup_coordinator.py index 25773aca..8252fbfb 100644 --- a/tests/coordinators/test_detector_setup_coordinator.py +++ b/tests/coordinators/test_detector_setup_coordinator.py @@ -1031,3 +1031,60 @@ def test_parallel_coordinator_instances( # Both coordinators should see the same state assert coordinator1.is_detector_initialized() is True assert coordinator2.is_detector_initialized() is True + + +class TestConfigureZonesAndParamUpdates: + def test_configure_zones_missing_deps_raises(self, detector_setup_coordinator): + detector_setup_coordinator.detector_service = None + with pytest.raises(CoordinatorValidationError, match="DetectorService is required"): + detector_setup_coordinator.configure_zones([{"type": "arena"}]) + + def test_configure_zones_legacy_dict_list(self, detector_setup_coordinator): + legacy_zones = [ + {"type": "arena", "polygon": [[0, 0], [10, 0], [10, 10], [0, 10]]}, + { + "type": "roi", + "polygon": [[2, 2], [4, 2], [4, 4], [2, 4]], + "name": "ROI1", + "color": "#ff0000", + }, + ] + success = detector_setup_coordinator.configure_zones( + legacy_zones, video_width=640, video_height=480 + ) + assert success is True + + def test_update_detector_parameters_invalid_scope(self, detector_setup_coordinator): + from zebtrack.core.exceptions import ValidationError + + with pytest.raises(ValidationError, match="Invalid scope"): + detector_setup_coordinator.update_detector_parameters( + {"conf_threshold": 0.5}, scope="invalid_scope" + ) + + def test_update_detector_parameters_value_error_raises_validation_error( + self, detector_setup_coordinator + ): + from zebtrack.core.exceptions import ValidationError + + detector_setup_coordinator.detector_service.update_tracking_parameters.side_effect = ( + ValueError("Out of range") + ) + with pytest.raises(ValidationError, match="Invalid detector parameter"): + detector_setup_coordinator.update_detector_parameters({"conf_threshold": 99.0}) + + def test_update_detector_parameters_generic_error_raises_coordinator_error( + self, detector_setup_coordinator + ): + detector_setup_coordinator.detector_service.update_tracking_parameters.side_effect = ( + RuntimeError("Crash") + ) + with pytest.raises( + DetectorSetupCoordinatorError, match="Failed to update detector parameters" + ): + detector_setup_coordinator.update_detector_parameters({"conf_threshold": 0.5}) + + def test_update_detector_parameters_maps_keys(self, detector_setup_coordinator): + params = {"confidence_threshold": 0.6, "track_buffer": 40} + success = detector_setup_coordinator.update_detector_parameters(params) + assert success is True diff --git a/tests/coordinators/test_live_batch_coordinator.py b/tests/coordinators/test_live_batch_coordinator.py index 47a8340d..7086d8e5 100644 --- a/tests/coordinators/test_live_batch_coordinator.py +++ b/tests/coordinators/test_live_batch_coordinator.py @@ -267,3 +267,60 @@ def test_persist_session_to_project_data_existing(coordinator): # Status should be updated, and metadata merged assert batch["videos"][0]["status"] == "recorded" assert batch["videos"][0]["metadata"]["group"] == "G1" + + +class TestUnifiedReportGeneration: + def test_generate_unified_report_no_videos(self, coordinator): + batch = BatchMetadata(batch_id="b1", group="G", day="1", subject_id="S", session_paths=[]) + assert coordinator._generate_unified_report(batch) is False + + def test_generate_unified_report_no_project_root(self, coordinator): + coordinator.project_manager.project_root = None + coordinator.project_manager.project_path = None + batch = BatchMetadata( + batch_id="b1", group="G", day="1", subject_id="S", session_paths=[Path("v1.mp4")] + ) + assert coordinator._generate_unified_report(batch) is False + + def test_generate_unified_report_no_summaries(self, coordinator, tmp_path): + coordinator.project_manager.project_root = tmp_path + coordinator.project_manager.find_video_entry.return_value = None + batch = BatchMetadata( + batch_id="b1", group="G", day="1", subject_id="S", session_paths=[Path("v1.mp4")] + ) + assert coordinator._generate_unified_report(batch) is False + + def test_generate_unified_report_success(self, coordinator, tmp_path): + coordinator.project_manager.project_root = tmp_path + coordinator.project_manager.find_video_entry.return_value = { + "summary_excel": str(tmp_path / "summary.xlsx") + } + batch = BatchMetadata( + batch_id="b1", session_paths=[Path("v1.mp4")], group="G1", day="1", subject_id="S1" + ) + + with patch.object( + coordinator, "_resolve_summary_excel_path", return_value=tmp_path / "summary.xlsx" + ): + ok = coordinator._generate_unified_report(batch) + assert ok is True + coordinator.analysis_service.aggregate_session_summaries.assert_called_once() + coordinator.project_manager.register_batch_outputs.assert_called_once() + + +class TestCollectMultiAquariumOutputs: + def test_collect_none_or_missing_dir(self): + assert LiveBatchCoordinator._collect_multi_aquarium_outputs(None, {}) == {} + assert LiveBatchCoordinator._collect_multi_aquarium_outputs(Path("/nonexistent"), {}) == {} + + def test_collect_with_aquarium_subdirectories(self, tmp_path): + aq1 = tmp_path / "aquarium_1" + aq1.mkdir() + (aq1 / "1_ProcessingArea_test.parquet").write_text("data") + + outputs = LiveBatchCoordinator._collect_multi_aquarium_outputs( + tmp_path, {"group": "G1", "subject_id": "S1", "day": 1} + ) + assert 0 in outputs + assert outputs[0]["group"] == "G1" + assert outputs[0]["subject_id"] == "S1" diff --git a/tests/coordinators/test_live_calibration_coordinator.py b/tests/coordinators/test_live_calibration_coordinator.py index ca7e1bf7..5b87dbd7 100644 --- a/tests/coordinators/test_live_calibration_coordinator.py +++ b/tests/coordinators/test_live_calibration_coordinator.py @@ -25,6 +25,7 @@ from zebtrack.coordinators.live_calibration_coordinator import ( LiveCalibrationCoordinator, ) +from zebtrack.ui.event_bus_v2 import UIEvents # --------------------------------------------------------------------------- # Helpers @@ -1218,3 +1219,101 @@ def _make_camera(settings_obj=None, **kwargs): assert 3 in captured_indices, "Camera deve ser criada com o índice ad-hoc (3)" # settings global restaurado após a construção. assert coordinator.settings.camera.index == 0 + + +class TestPrepareZonesForLiveSession: + def test_ensure_zones_auto_detect_success(self): + coordinator: Any = _make_coordinator() + coordinator.root = MagicMock() + coordinator.project_manager.get_project_type.return_value = "live" + coordinator.project_manager.get_zone_data.return_value = None + coordinator.run_live_calibration = MagicMock(return_value=True) + coordinator._wait_for_zone_confirmation = MagicMock(return_value=True) + + mock_dialog = MagicMock() + mock_dialog.show.return_value = {"method": "auto"} + + with patch( + "zebtrack.ui.dialogs.zone_calibration_dialog.ZoneCalibrationDialog", + return_value=mock_dialog, + ): + res = coordinator.ensure_zones_before_recording(camera_index=0) + + assert res is True + coordinator.run_live_calibration.assert_called_once() + coordinator._wait_for_zone_confirmation.assert_called_once() + coordinator.event_bus.publish.assert_called() + + def test_ensure_zones_auto_detect_user_cancelled(self): + coordinator: Any = _make_coordinator() + coordinator.root = MagicMock() + coordinator.project_manager.get_project_type.return_value = "live" + coordinator.project_manager.get_zone_data.return_value = None + coordinator.run_live_calibration = MagicMock(return_value=False) + coordinator._last_calibration_cancelled = True + + mock_dialog = MagicMock() + mock_dialog.show.return_value = {"method": "auto"} + + with patch( + "zebtrack.ui.dialogs.zone_calibration_dialog.ZoneCalibrationDialog", + return_value=mock_dialog, + ): + res = coordinator.ensure_zones_before_recording(camera_index=0) + + assert res is False + + def test_ensure_zones_auto_detect_failed_fallback_to_manual(self): + coordinator: Any = _make_coordinator() + coordinator.root = MagicMock() + coordinator.project_manager.get_project_type.return_value = "live" + coordinator.project_manager.get_zone_data.return_value = None + coordinator.run_live_calibration = MagicMock(return_value=False) + coordinator._last_calibration_cancelled = False + coordinator._capture_reference_frame_for_zones = MagicMock(return_value=True) + coordinator._wait_for_zone_confirmation = MagicMock(return_value=True) + + mock_dialog = MagicMock() + mock_dialog.show.return_value = {"method": "auto"} + + with patch( + "zebtrack.ui.dialogs.zone_calibration_dialog.ZoneCalibrationDialog", + return_value=mock_dialog, + ): + res = coordinator.ensure_zones_before_recording(camera_index=0) + + assert res is True + coordinator._capture_reference_frame_for_zones.assert_called_once() + coordinator._wait_for_zone_confirmation.assert_called_once() + event_types = [c[0][0].type for c in coordinator.event_bus.publish.call_args_list] + assert UIEvents.UI_SELECT_TAB in event_types + assert UIEvents.UI_UPDATE_ZONE_LIST in event_types + + def test_ensure_zones_calibration_dialog_cancelled(self): + coordinator: Any = _make_coordinator() + coordinator.root = MagicMock() + coordinator.project_manager.get_project_type.return_value = "live" + coordinator.project_manager.get_zone_data.return_value = None + + mock_dialog = MagicMock() + mock_dialog.show.return_value = None # user cancelled dialog + + with patch( + "zebtrack.ui.dialogs.zone_calibration_dialog.ZoneCalibrationDialog", + return_value=mock_dialog, + ): + res = coordinator.ensure_zones_before_recording(camera_index=0) + + assert res is False + + +class TestPolygonSourceManagement: + def test_source_lifecycle(self): + coordinator = _make_coordinator() + assert coordinator.last_polygon_source is None + + coordinator._set_last_polygon_source("auto") + assert coordinator.last_polygon_source == "auto" + + coordinator.clear_last_polygon_source() + assert coordinator.last_polygon_source is None diff --git a/tests/coordinators/test_live_camera_session_coordinator.py b/tests/coordinators/test_live_camera_session_coordinator.py index 053b8a46..7d4dc0bc 100644 --- a/tests/coordinators/test_live_camera_session_coordinator.py +++ b/tests/coordinators/test_live_camera_session_coordinator.py @@ -434,3 +434,211 @@ def test_active_finalizes(self, coordinator): coordinator._finalize_live_session_ui.assert_called_once_with( cancelled=False, publish_refresh=True, service_success=True ) + + +class TestPendingAndResumeFlows: + def test_publish_pending_with_calibration_source( + self, coordinator, mock_live_calibration_coordinator, mock_event_bus + ): + mock_live_calibration_coordinator.last_polygon_source = "auto" + ctx = {"experiment_id": "exp_1", "group": "G1", "day": "D1", "subject": "S1"} + + coordinator._publish_pending(ctx) + + mock_event_bus.publish.assert_called_once() + event_obj = mock_event_bus.publish.call_args[0][0] + assert event_obj.type == UIEvents.LIVE_RECORDING_PENDING + assert event_obj.data.polygon_source == "auto" + + def test_publish_pending_no_event_bus(self, coordinator): + coordinator.event_bus = None + coordinator._publish_pending({}) # Should not raise + + def test_on_resume_requested_no_pending_context(self, coordinator): + coordinator._pending_live_context = None + coordinator._pending_live_kind = None + coordinator._on_resume_requested() # Should return safely + + def test_on_resume_requested_project_kind(self, coordinator): + coordinator._pending_live_context = { + "day_int": 1, + "group": "G1", + "subject": "S1", + "duration_s": 60.0, + "camera_index_override": 0, + "camera_friendly_name_override": "Cam1", + } + coordinator._pending_live_kind = "project" + coordinator.start_live_project_session = Mock() + + coordinator._on_resume_requested() + + coordinator.start_live_project_session.assert_called_once_with( + day=1, + group="G1", + subject="S1", + duration_s=60.0, + camera_index_override=0, + camera_friendly_name_override="Cam1", + zones_validated=True, + ) + + def test_on_resume_requested_config_kind(self, coordinator): + coordinator._pending_live_context = {"config": {"key": "val"}} + coordinator._pending_live_kind = "config" + coordinator.start_session_from_config = Mock() + + coordinator._on_resume_requested() + + coordinator.start_session_from_config.assert_called_once_with( + config={"key": "val"}, + zones_validated=True, + ) + + def test_on_resume_requested_exception_shows_error(self, coordinator, mock_event_bus): + coordinator._pending_live_context = {"day_int": 1, "group": "G", "subject": "S"} + coordinator._pending_live_kind = "project" + coordinator.start_live_project_session = Mock(side_effect=RuntimeError("Hardware issue")) + + coordinator._on_resume_requested() + + mock_event_bus.publish.assert_called_once() + event_obj = mock_event_bus.publish.call_args[0][0] + assert event_obj.type == UIEvents.UI_SHOW_ERROR + + +class TestSessionInfo: + def test_get_live_session_info_not_active(self, coordinator): + coordinator._active_live_session_id = None + assert coordinator.get_live_session_info() is None + + def test_get_live_session_info_active(self, coordinator, mock_state_manager): + coordinator._active_live_session_id = "sess_123" + mock_proc = Mock(camera_index=0, experiment_id="exp_1", duration_s=100.0) + mock_state_manager.get_processing_state.return_value = mock_proc + + info = coordinator.get_live_session_info() + assert info is not None + assert info["session_id"] == "sess_123" + assert info["is_active"] is True + assert info["camera_index"] == 0 + + +class TestBatchSessionRegistration: + def test_register_batch_session_no_wizard_data(self, coordinator): + coordinator._active_wizard_data = None + coordinator._register_batch_session() # Should not raise + + def test_register_batch_session_incomplete_fields(self, coordinator): + coordinator._active_wizard_data = {"experimental_group": "G1"} # missing day and subject + coordinator._register_batch_session() + assert coordinator._active_wizard_data is None + + def test_register_batch_session_with_batch_coordinator(self, coordinator): + coordinator._active_wizard_data = { + "experimental_group": "G1", + "experiment_day": "D1", + "subject_id": "S1", + "recording_duration_s": 60.0, + "camera_index": 0, + "is_batch_last_session": True, + } + coordinator._active_live_session_id = "exp_batch" + coordinator._find_video_in_live_session = Mock(return_value=Path("/tmp/live.mp4")) + + mock_batch_coord = Mock() + mock_batch_coord.register_session.return_value = "batch_001" + coordinator.live_batch_coordinator = mock_batch_coord + + coordinator._register_batch_session() + + mock_batch_coord.register_session.assert_called_once() + mock_batch_coord.mark_batch_complete.assert_called_once_with("batch_001") + assert coordinator._active_wizard_data is None + + +class TestArduinoExternalTriggerFlows: + def test_on_arduino_event_start_with_pending_context(self, coordinator, mock_event_bus): + coordinator._pending_trigger_context = { + "experiment_id": "exp_arduino_1", + "day_int": 1, + "group": "Control", + "subject": "Fish1", + "duration_s": 120.0, + "camera_index_override": 0, + "camera_friendly_name_override": "Cam 0", + } + coordinator.start_live_project_session = Mock() + + coordinator.on_arduino_event(1) + + assert coordinator._pending_trigger_context is None + coordinator.start_live_project_session.assert_called_once_with( + day=1, + group="Control", + subject="Fish1", + duration_s=120.0, + camera_index_override=0, + camera_friendly_name_override="Cam 0", + zones_validated=True, + external_trigger_armed=True, + ) + + def test_on_arduino_event_start_without_context(self, coordinator): + coordinator._pending_trigger_context = None + coordinator.start_live_project_session = Mock() + + coordinator.on_arduino_event(1) + + coordinator.start_live_project_session.assert_not_called() + + def test_on_arduino_event_stop_with_pending_context(self, coordinator): + coordinator._pending_trigger_context = {"dummy": True} + coordinator.clear_pending_external_trigger = Mock() + + coordinator.on_arduino_event(0) + + coordinator.clear_pending_external_trigger.assert_called_once() + + def test_on_arduino_event_stop_active_session(self, coordinator): + coordinator._pending_trigger_context = None + coordinator.is_live_session_active = Mock(return_value=True) + coordinator.stop_live_session = Mock() + + coordinator.on_arduino_event(0) + + coordinator.stop_live_session.assert_called_once() + + def test_on_arduino_event_other_code_ignored(self, coordinator): + coordinator.start_live_project_session = Mock() + coordinator.stop_live_session = Mock() + + coordinator.on_arduino_event(42) + + coordinator.start_live_project_session.assert_not_called() + coordinator.stop_live_session.assert_not_called() + + def test_clear_pending_external_trigger(self, coordinator, mock_event_bus): + coordinator._pending_trigger_context = {"dummy": True} + coordinator.clear_pending_external_trigger() + + assert coordinator._pending_trigger_context is None + mock_event_bus.publish.assert_called() + + +class TestFindVideoInLiveSession: + def test_find_video_when_present(self, coordinator, tmp_path): + session_dir = tmp_path / "live_session" + session_dir.mkdir() + video_file = session_dir / "recorded.mp4" + video_file.write_text("dummy") + + coordinator.live_camera_service.current_output_dir = str(session_dir) + found = coordinator._find_video_in_live_session() + + assert found is not None + assert found.name == "recorded.mp4" + + def test_find_video_when_dir_missing(self, coordinator): + coordinator.live_camera_service.current_output_dir = None + assert coordinator._find_video_in_live_session() is None diff --git a/tests/coordinators/test_model_diagnostics_coordinator.py b/tests/coordinators/test_model_diagnostics_coordinator.py index 69f61060..36afc495 100644 --- a/tests/coordinators/test_model_diagnostics_coordinator.py +++ b/tests/coordinators/test_model_diagnostics_coordinator.py @@ -1,4 +1,5 @@ import threading +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -493,3 +494,87 @@ def test_weight_name_from_manager_method(self, mock_thread, mock_dialog, coordin } coordinator.run_model_diagnostic(config) mock_thread.assert_called_once() + + +class TestOpenVINODiagnosticsAndFormatting: + def test_load_openvino_model_plugin_missing(self, coordinator): + with patch.dict( + "zebtrack.coordinators.model_diagnostics_coordinator.DETECTOR_PLUGINS", {}, clear=True + ): + with patch( + "zebtrack.coordinators.model_diagnostics_coordinator._is_valid_openvino_directory", + return_value=True, + ): + with pytest.raises(DiagnosticAbortError): + coordinator._initialize_diagnostic_openvino_model( + "OpenVINO", {"openvino_path": "/valid/path"}, {}, MagicMock() + ) + + def test_load_openvino_model_missing_predict(self, coordinator): + dummy_cls = MagicMock(return_value=object()) # instance has no predict + with patch.dict( + "zebtrack.coordinators.model_diagnostics_coordinator.DETECTOR_PLUGINS", + {"OpenVINO": dummy_cls}, + clear=True, + ): + with patch( + "zebtrack.coordinators.model_diagnostics_coordinator._is_valid_openvino_directory", + return_value=True, + ): + with pytest.raises(DiagnosticAbortError): + coordinator._initialize_diagnostic_openvino_model( + "OpenVINO", {"openvino_path": "/valid/path"}, {}, MagicMock() + ) + + def test_load_openvino_model_success(self, coordinator): + mock_model = MagicMock() + mock_model.predict = MagicMock() + mock_model.set_context = MagicMock() + dummy_cls = MagicMock(return_value=mock_model) + results: dict[str, Any] = {} + + with patch.dict( + "zebtrack.coordinators.model_diagnostics_coordinator.DETECTOR_PLUGINS", + {"OpenVINO": dummy_cls}, + clear=True, + ): + with patch( + "zebtrack.coordinators.model_diagnostics_coordinator._is_valid_openvino_directory", + return_value=True, + ): + loaded = coordinator._initialize_diagnostic_openvino_model( + "OpenVINO", {"openvino_path": "/valid/path"}, results, MagicMock() + ) + assert loaded is mock_model + assert "OpenVINO" in results + mock_model.set_context.assert_called_with("diagnostic") + + def test_format_single_model_results_with_masks_and_orphans(self, coordinator): + import numpy as np + + mock_box = MagicMock() + mock_box.cls = 0 + mock_box.conf = 0.95 + mock_box.xyxy = [[10, 20, 30, 40]] + + mock_preds = MagicMock() + mock_preds.boxes = [mock_box] + mock_preds.names = {0: "fish"} + mock_preds.masks = MagicMock() + mock_preds.masks.xy = [ + np.array([[10, 20], [30, 20], [30, 40], [10, 40]]), + np.array([[50, 50], [100, 50], [100, 100], [50, 100]]), # orphan mask + ] + + config = { + "video_path": "video.mp4", + "frames_to_analyze": 1, + "confidence_threshold": 0.5, + } + results = {"YOLO (PyTorch)": [mock_preds]} + + text = coordinator._format_diagnostic_report(config, results) + + assert "Class 0 ('fish')" in text + assert "[MASK WITHOUT BOX]" in text + assert "YOLO (PYTORCH) RESULTS" in text diff --git a/tests/coordinators/test_multi_aquarium_coordinator.py b/tests/coordinators/test_multi_aquarium_coordinator.py index 4d6c55ed..b10f06be 100644 --- a/tests/coordinators/test_multi_aquarium_coordinator.py +++ b/tests/coordinators/test_multi_aquarium_coordinator.py @@ -307,3 +307,58 @@ def test_temporary_single_animal_mode(coordinator): assert coordinator.settings.video_processing.single_animal_per_aquarium is False assert coordinator.settings.tracking.use_single_subject_tracker is False assert mock_publish.call_count == 2 + + +class TestRelocateResultsFolders: + def test_relocate_no_outputs(self, coordinator): + coordinator._relocate_multi_aquarium_folders("v.mp4", {}, []) # Should not raise + + def test_relocate_with_files(self, coordinator, tmp_path): + video_path = str(tmp_path / "video.mp4") + old_dir = tmp_path / "old_results" + old_dir.mkdir() + dummy_file = old_dir / "3_CoordMovimento_video_aq1.parquet" + dummy_file.write_text("data") + + new_dir = tmp_path / "new_results" + + coordinator.project_manager.project_path = str(tmp_path) + coordinator.project_manager.resolve_results_directory.return_value = new_dir + + entry = { + "multi_aquarium_outputs": { + "0": { + "results_dir": str(old_dir), + "parquet_files": {"trajectory": str(dummy_file)}, + } + } + } + configs = [{"aquarium_id": 0, "group": "G_new", "subject_id": "S_new", "day": "1"}] + + coordinator._relocate_multi_aquarium_folders(video_path, entry, configs) + + assert not old_dir.exists() + assert new_dir.exists() + assert (new_dir / "3_CoordMovimento_video_aq1.parquet").exists() + assert entry["multi_aquarium_outputs"]["0"]["results_dir"] == str(new_dir) + + +class TestResolveSingleAnimalMode: + def test_from_top_level_flag(self, coordinator): + coordinator.project_manager.project_data = {"single_animal_per_aquarium": True} + assert coordinator._resolve_single_animal_mode() is True + + coordinator.project_manager.project_data = {"single_animal_per_aquarium": False} + assert coordinator._resolve_single_animal_mode() is False + + def test_from_config(self, coordinator): + assert coordinator._resolve_single_animal_mode({"single_animal_per_aquarium": True}) is True + assert coordinator._resolve_single_animal_mode({"animals_per_aquarium": 1}) is True + assert coordinator._resolve_single_animal_mode({"animals_per_aquarium": 3}) is False + + def test_from_calibration_animals_per_aquarium(self, coordinator): + coordinator.project_manager.project_data = {"calibration": {"animals_per_aquarium": 1}} + assert coordinator._resolve_single_animal_mode() is True + + coordinator.project_manager.project_data = {"calibration": {"animals_per_aquarium": 4}} + assert coordinator._resolve_single_animal_mode() is False diff --git a/tests/coordinators/test_project_lifecycle_coordinator_unit.py b/tests/coordinators/test_project_lifecycle_coordinator_unit.py index 424cdf94..ad941e32 100644 --- a/tests/coordinators/test_project_lifecycle_coordinator_unit.py +++ b/tests/coordinators/test_project_lifecycle_coordinator_unit.py @@ -446,3 +446,51 @@ def test_copy_global_with_apply_runtime_false_skips_defaults(): assert call_kwargs["active_weight_setter"] is None assert call_kwargs["use_openvino_setter"] is None assert call_kwargs["apply_runtime_callback"] is None + + +class TestCalibrationSessionsAndDefaults: + def test_global_calibration_session_with_and_without_coordinator(self, coordinator): + # Without calibration coordinator + with coordinator.global_calibration_session(lambda: "w", lambda: True): + pass + + # With calibration coordinator + mock_calib = MagicMock() + mock_cm = MagicMock() + mock_calib.global_calibration_session.return_value = mock_cm + coordinator._calibration_coordinator = mock_calib + + with coordinator.global_calibration_session(lambda: "w", lambda: True): + pass + mock_calib.global_calibration_session.assert_called_once() + + def test_project_calibration_session_with_and_without_coordinator(self, coordinator): + # Without calibration coordinator + with coordinator.project_calibration_session(): + pass + + # With calibration coordinator + mock_calib = MagicMock() + mock_cm = MagicMock() + mock_calib.project_calibration_session.return_value = mock_cm + coordinator._calibration_coordinator = mock_calib + + with coordinator.project_calibration_session(): + pass + mock_calib.project_calibration_session.assert_called_once() + + def test_setup_zones_from_project_no_callback(self, coordinator): + coordinator._setup_zones_from_project(None) # Should log warning and return + + def test_restore_global_model_defaults_fallback(self, coordinator): + mock_state = MagicMock() + mock_state.active_weight_name = "test_weight.pt" + mock_state.use_openvino = True + coordinator.state_manager.get_detector_state.return_value = mock_state + coordinator._model_override_service = None + + coordinator._restore_global_model_defaults() + + assert coordinator._global_model_defaults["active_weight"] == "test_weight.pt" + assert coordinator._global_model_defaults["use_openvino"] is True + assert coordinator._using_project_overrides is False diff --git a/tests/coordinators/test_recording_session_coordinator.py b/tests/coordinators/test_recording_session_coordinator.py index 92d9421f..ba5b553e 100644 --- a/tests/coordinators/test_recording_session_coordinator.py +++ b/tests/coordinators/test_recording_session_coordinator.py @@ -276,3 +276,53 @@ def test_on_zone_saved(coordinator): coordinator._on_zone_saved() coordinator.recording_service.schedule_recording.assert_called_once() + + +class TestStartRecordingDetailsAndExplicitPath: + def test_start_recording_no_params_no_view_raises(self, coordinator): + from zebtrack.coordinators.recording_session_coordinator import ( + RecordingSessionCoordinatorError, + ) + + coordinator.view = None + with pytest.raises( + RecordingSessionCoordinatorError, match="Cannot request recording details" + ): + coordinator.start_recording() + + def test_start_recording_dialog_cancelled(self, coordinator): + coordinator.view = MagicMock() + coordinator.view.ask_recording_details_unified.return_value = None + + res = coordinator.start_recording() + assert res is False + + def test_start_recording_dialog_confirmed(self, coordinator, tmp_path): + coordinator.view = MagicMock() + coordinator.view.ask_recording_details_unified.return_value = { + "day": 1, + "group": "G1", + "cobaia": "S1", + } + coordinator.project_manager.project_path = str(tmp_path) + coordinator._handle_external_trigger = MagicMock(return_value=False) + coordinator._schedule_recording = MagicMock() + + res = coordinator.start_recording() + assert res is True + coordinator.project_manager.save_last_session_details.assert_called_once_with(1, "G1") + coordinator._schedule_recording.assert_called_once() + + def test_start_recording_explicit_path_missing_params_raises(self, coordinator): + with pytest.raises(ValueError, match="experiment_id is required"): + coordinator.start_recording( + context=None, day=None, output_path="out.mp4", experiment_id=None + ) + + def test_on_arduino_event_disarm_on_zero(self, coordinator): + coordinator._pending_external_trigger = {"dummy": True} + recording_state = MagicMock(is_recording=False) + coordinator.state_manager.get_recording_state.return_value = recording_state + + coordinator.on_arduino_event(0) + assert coordinator._pending_external_trigger is None diff --git a/tests/coordinators/test_report_generation_coordinator.py b/tests/coordinators/test_report_generation_coordinator.py index def25727..b367935a 100644 --- a/tests/coordinators/test_report_generation_coordinator.py +++ b/tests/coordinators/test_report_generation_coordinator.py @@ -385,3 +385,92 @@ def test_process_multi_summary_video_exception(self, coordinator, mock_project_m ) assert status == "failed" assert "DB error" in msg + + +class TestProcessOneAquariumSummary: + def test_trajectory_file_missing_returns_none(self, coordinator): + res = coordinator._process_one_aquarium_summary( + {}, "exp1", "v.mp4", 0, {"parquet_files": {}}, MagicMock(), coordinator.settings, [] + ) + assert res is None + + def test_aquarium_not_in_multi_zone(self, coordinator, tmp_path): + traj_p = tmp_path / "traj.parquet" + traj_p.write_text("data") + multi_zone = MagicMock(aquariums=[]) + + res = coordinator._process_one_aquarium_summary( + {}, + "exp1", + "v.mp4", + 0, + {"parquet_files": {"trajectory": str(traj_p)}}, + multi_zone, + coordinator.settings, + [], + ) + assert res is None + + def test_empty_dataframe_returns_none(self, coordinator, tmp_path): + traj_p = tmp_path / "traj.parquet" + traj_p.write_text("data") + mock_aq = MagicMock(id=0) + multi_zone = MagicMock(aquariums=[mock_aq]) + coordinator._read_trajectory = MagicMock(return_value=pd.DataFrame()) + + res = coordinator._process_one_aquarium_summary( + {}, + "exp1", + "v.mp4", + 0, + {"parquet_files": {"trajectory": str(traj_p)}}, + multi_zone, + coordinator.settings, + [], + ) + assert res is None + + def test_process_one_aquarium_summary_success(self, coordinator, tmp_path): + traj_p = tmp_path / "traj.parquet" + traj_p.write_text("data") + mock_aq = MagicMock( + id=0, + polygon=[[0, 0], [10, 0], [10, 10], [0, 10]], + roi_polygons=[], + roi_names=[], + roi_colors=[], + ) + multi_zone = MagicMock(aquariums=[mock_aq]) + df = pd.DataFrame({"x": [1, 2], "y": [3, 4], "time": [0.1, 0.2]}) + coordinator.project_manager.project_data = {"calibration": {}} + coordinator._read_trajectory = MagicMock(return_value=df) + coordinator._prepare_summary_geometry = MagicMock( + return_value=(10.0, 10.0, [[0, 0], [10, 10]], 480, [], [], {}) + ) + + mock_reporter = MagicMock() + mock_reporter.generate_reports.return_value = { + "summary_parquet": str(tmp_path / "sum.parquet") + } + + with ( + patch("zebtrack.analysis.reporters.ReporterContext", return_value=MagicMock()), + patch("zebtrack.analysis.reporters.ParquetSummaryReporter", return_value=mock_reporter), + ): + out_info = { + "results_dir": str(tmp_path), + "parquet_files": {"trajectory": str(traj_p)}, + "group": "G1", + "subject_id": "S1", + "day": "1", + } + video_entry: dict[str, Any] = {"multi_aquarium_outputs": {"0": {"parquet_files": {}}}} + res = coordinator._process_one_aquarium_summary( + video_entry, "exp1", "v.mp4", 0, out_info, multi_zone, coordinator.settings, [] + ) + expected_summary_path = str(tmp_path / "exp1_aq1_summary.parquet") + assert res == expected_summary_path + assert ( + video_entry["multi_aquarium_outputs"]["0"]["parquet_files"]["summary"] + == expected_summary_path + ) diff --git a/tests/coordinators/test_ui_state_coordinator.py b/tests/coordinators/test_ui_state_coordinator.py index 7a9552e3..f817d5b6 100644 --- a/tests/coordinators/test_ui_state_coordinator.py +++ b/tests/coordinators/test_ui_state_coordinator.py @@ -376,3 +376,73 @@ def test_show_cancel_feedback(self, controller): controller.view, "stop_analysis_view_mode" ) controller.ui_coordinator.set_status.assert_called_once() + + +class TestProcessingUIAndDiagnostics: + def test_activate_analysis_view_mode(self, controller): + controller.activate_analysis_view_mode() + event_obj = controller.ui_event_bus.publish.call_args[0][0] + assert event_obj.type == UIEvents.UI_NAVIGATE_TO_ANALYSIS_VIEW + + def test_activate_analysis_view_mode_no_event_bus(self, controller): + controller.ui_event_bus = None + controller.view = Mock() + controller.activate_analysis_view_mode() + controller.ui_coordinator.update_view.assert_called_once_with( + controller.view, "start_analysis_view_mode" + ) + + def test_prepare_processing_ui(self, controller): + controller.view = Mock() + controller._prepare_processing_ui(5) + controller.ui_coordinator.show_progress_bar.assert_called_once_with(controller.view) + controller.project_manager.set_active_zone_video.assert_called_once_with(None) + + def test_finalize_processing_success_and_cancelled(self, controller): + controller.view = Mock() + + # Cancelled + controller._finalize_processing( + was_cancelled=True, videos_to_process=[{}], final_output_dir="/out" + ) + controller.ui_coordinator.hide_progress_bar.assert_called_with(controller.view) + + # Success + controller._finalize_processing( + was_cancelled=False, videos_to_process=[{}], final_output_dir="/out" + ) + assert controller.ui_coordinator.show_info.call_count == 2 + + def test_diagnostic_progress_helpers(self, controller): + mock_dialog = Mock() + + # Update without current/total + controller._update_diagnostic_progress(mock_dialog, "working") + controller.root.after.assert_called_with(0, mock_dialog.update_progress, "working") + + # Update with current/total + controller._update_diagnostic_progress(mock_dialog, "working", 1, 10) + controller.root.after.assert_called_with(0, mock_dialog.update_progress, "working", 1, 10) + + # Finish dialog + controller._finish_progress_dialog(mock_dialog) + controller.root.after.assert_called_with(0, mock_dialog.finish) + + +class TestSetActiveWeightExtended: + def test_set_active_weight_with_openvino_conversion(self, controller): + controller.main_view_model.get_all_weight_names.return_value = ["model.pt"] + controller.main_view_model.use_openvino = True + controller.convert_active_weight_to_openvino = Mock() + + controller.set_active_weight("model.pt") + + assert controller.main_view_model.active_weight_name == "model.pt" + controller.convert_active_weight_to_openvino.assert_called_once() + + def test_set_active_weight_not_found(self, controller): + controller.main_view_model.get_all_weight_names.return_value = ["model.pt"] + + controller.set_active_weight("unknown.pt") + + assert controller.main_view_model.active_weight_name == "" diff --git a/tests/coordinators/test_video_selection_and_completion_mixins.py b/tests/coordinators/test_video_selection_and_completion_mixins.py new file mode 100644 index 00000000..a4c83215 --- /dev/null +++ b/tests/coordinators/test_video_selection_and_completion_mixins.py @@ -0,0 +1,307 @@ +"""Unit tests for _VideoSelectionMixin and _VideoCompletionMixin in VideoProcessingCoordinator.""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import MagicMock, patch + +import pytest + +from zebtrack.coordinators.video_processing_coordinator import VideoProcessingCoordinator +from zebtrack.core.detection import AquariumData, MultiAquariumZoneData, ZoneData +from zebtrack.ui.event_bus_v2 import UIEvents + + +@pytest.fixture +def coordinator(): + state_manager = MagicMock() + event_bus = MagicMock() + settings = MagicMock() + settings.analysis_config = MagicMock() + settings.tracking = MagicMock() + settings.video_processing = MagicMock() + ui_coordinator = MagicMock() + project_manager = MagicMock() + project_manager.project_path = "/fake/project" + project_manager.project_data = {"calibration": {}} + weight_manager = MagicMock() + detector_service = MagicMock() + cancel_event = MagicMock() + ui_state_controller = MagicMock() + video_selection_service = MagicMock() + video_validation_service = MagicMock() + video_classification_service = MagicMock() + dialog_coordinator = MagicMock() + + coord = VideoProcessingCoordinator( + state_manager=state_manager, + settings_obj=settings, + ui_coordinator=ui_coordinator, + project_manager=project_manager, + weight_manager=weight_manager, + detector_service=detector_service, + cancel_event=cancel_event, + event_bus=event_bus, + ui_state_controller=ui_state_controller, + video_selection_service=video_selection_service, + video_validation_service=video_validation_service, + video_classification_service=video_classification_service, + dialog_coordinator=dialog_coordinator, + ) + coord.view = MagicMock() + coord.root = MagicMock() + coord.detector = MagicMock() + return coord + + +class TestLiveCameraSessionActiveCheck: + def test_live_session_active_true(self, coordinator): + mock_live = MagicMock() + mock_live.is_live_session_active.return_value = True + coordinator.view.controller.live_camera_session_coordinator = mock_live + processing_state = SimpleNamespace(is_live_session_active=True) + assert coordinator._is_live_session_currently_active(processing_state) is True + + def test_live_session_active_false_fallback_camera(self, coordinator): + mock_live = MagicMock() + mock_live.is_live_session_active.return_value = False + mock_live.live_camera_service.camera = MagicMock() + coordinator.view.controller.live_camera_session_coordinator = mock_live + processing_state = SimpleNamespace(is_live_session_active=True) + assert coordinator._is_live_session_currently_active(processing_state) is False + + def test_live_session_exception_fallback(self, coordinator): + mock_live = MagicMock() + mock_live.is_live_session_active.side_effect = RuntimeError("Disconnected") + mock_live.live_camera_service.camera = MagicMock() + coordinator.view.controller.live_camera_session_coordinator = mock_live + processing_state = SimpleNamespace(is_live_session_active=True) + assert coordinator._is_live_session_currently_active(processing_state) is True + + +class TestSelectionErrorHandlers: + def test_show_validation_error(self, coordinator): + val = SimpleNamespace(error_message="Zone missing") + coordinator._show_validation_error(val) + coordinator.event_bus.publish.assert_called_once() + assert coordinator.event_bus.publish.call_args[0][0].type == UIEvents.UI_SHOW_WARNING + + def test_handle_targeted_selection_no_paths(self, coordinator): + res = coordinator._handle_targeted_selection_errors(MagicMock(), []) + assert res is False + assert coordinator.event_bus.publish.call_args[0][0].type == UIEvents.UI_SHOW_INFO + + def test_handle_targeted_selection_has_missing_and_zero_candidates(self, coordinator): + selection_result = SimpleNamespace( + has_missing=True, + missing_targets=["v1.mp4", "v2.mp4", "v3.mp4", "v4.mp4", "v5.mp4", "v6.mp4"], + candidate_count=0, + ) + res = coordinator._handle_targeted_selection_errors(selection_result, ["v1.mp4"]) + assert res is False + assert coordinator.event_bus.publish.call_count == 2 + + def test_handle_targeted_selection_success(self, coordinator): + selection_result = SimpleNamespace( + has_missing=False, + missing_targets=[], + candidate_count=2, + ) + res = coordinator._handle_targeted_selection_errors(selection_result, ["v1.mp4", "v2.mp4"]) + assert res is True + + def test_handle_pending_selection_errors_zero_count(self, coordinator): + selection_result = SimpleNamespace(candidate_count=0) + res = coordinator._handle_pending_selection_errors(selection_result) + assert res is False + assert coordinator.event_bus.publish.call_args[0][0].type == UIEvents.UI_SHOW_INFO + + def test_handle_pending_selection_errors_success(self, coordinator): + selection_result = SimpleNamespace(candidate_count=3) + res = coordinator._handle_pending_selection_errors(selection_result) + assert res is True + + def test_extract_and_validate_candidate_paths_empty(self, coordinator): + paths = coordinator._extract_and_validate_candidate_paths([{"path": ""}, {"no_path": 1}]) + assert paths is None + assert coordinator.event_bus.publish.call_args[0][0].type == UIEvents.UI_SHOW_ERROR + + def test_extract_and_validate_candidate_paths_valid(self, coordinator): + paths = coordinator._extract_and_validate_candidate_paths( + [{"path": "1.mp4"}, {"path": "2.mp4"}] + ) + assert paths == ["1.mp4", "2.mp4"] + + def test_handle_missing_files_warning(self, coordinator): + scan_result = SimpleNamespace( + has_missing=True, + missing_files=["m1.mp4", "m2.mp4", "m3.mp4", "m4.mp4", "m5.mp4", "m6.mp4"], + ) + coordinator._handle_missing_files_warning(scan_result) + assert coordinator.event_bus.publish.call_args[0][0].type == UIEvents.UI_SHOW_WARNING + + +class TestLoadZonesForEligibleVideos: + def test_load_zones_multi_aquarium_filtered(self, coordinator): + aq1 = AquariumData(id=0, polygon=[[0, 0], [10, 0], [10, 10], [0, 10]]) + aq2 = AquariumData(id=1, polygon=[[20, 0], [30, 0], [30, 10], [20, 10]]) + multi_data = MultiAquariumZoneData( + aquariums=[aq1, aq2], + video_width=640, + video_height=480, + sequential_processing=True, + ) + coordinator.project_manager.get_multi_aquarium_zone_data.return_value = multi_data + coordinator.project_manager.resolve_results_directory.return_value = Path("/results") + + eligible: list[dict[str, Any]] = [{"path": "video1.mp4", "metadata": {}}] + coordinator._load_zones_for_eligible_videos(eligible, aquarium_filter={"video1.mp4": [0]}) + + assert "zone_data" in eligible[0] + zd = cast(dict[str, Any], eligible[0]["zone_data"]) + assert len(zd["aquariums"]) == 1 + assert zd["aquariums"][0]["id"] == 0 + + def test_load_zones_standard_video_from_parquet(self, coordinator): + coordinator.project_manager.get_multi_aquarium_zone_data.return_value = None + coordinator.project_manager.resolve_results_directory.return_value = Path("/results") + mock_zone = ZoneData( + polygon=[[0, 0], [100, 100]], roi_polygons=[], roi_names=[], roi_colors=[] + ) + + with patch( + "zebtrack.core.project.project_manager.ProjectManager.load_zones_from_parquet", + return_value=mock_zone, + ): + eligible: list[dict[str, Any]] = [{"path": "video1.mp4", "has_arena": True}] + coordinator._load_zones_for_eligible_videos(eligible) + + zd = cast(dict[str, Any], eligible[0]["zone_data"]) + assert zd["polygon"] == [[0, 0], [100, 100]] + coordinator.project_manager.save_zone_data.assert_called_once() + + +class TestExplodeSequentialTasks: + def test_explode_single_aquarium(self, coordinator): + videos = [{"path": "v1.mp4", "zone_data": {"polygon": [[0, 0]]}}] + tasks = coordinator._explode_sequential_tasks(videos) + assert len(tasks) == 1 + assert tasks[0]["path"] == "v1.mp4" + + def test_explode_sequential_multi_aquarium(self, coordinator): + aq1 = AquariumData( + id=0, polygon=[[0, 0], [10, 0], [10, 10], [0, 10]], group="G1", subject_id="S1" + ) + aq2 = AquariumData( + id=1, polygon=[[20, 0], [30, 0], [30, 10], [20, 10]], group="G2", subject_id="S2" + ) + multi_data = MultiAquariumZoneData( + aquariums=[aq1, aq2], + video_width=640, + video_height=480, + sequential_processing=True, + ) + from zebtrack.core.project.zone_manager import ZoneManager + + zone_dict = ZoneManager.multi_aquarium_zone_data_to_dict(multi_data) + + coordinator.project_manager.resolve_multi_aquarium_results_directories.return_value = { + 0: Path("/results/aq1"), + 1: Path("/results/aq2"), + } + + videos = [ + { + "path": "v_multi.mp4", + "zone_data": zone_dict, + } + ] + tasks = coordinator._explode_sequential_tasks(videos) + assert len(tasks) == 2 + assert tasks[0]["aquarium_id"] == 0 + assert tasks[1]["aquarium_id"] == 1 + + +class TestSelectEligibleVideos: + def test_skip_dialog_returns_combined_ready_and_arena(self, coordinator): + ready_traj = [{"path": "t1.mp4"}] + ready_zones = [{"path": "z1.mp4"}] + arena_only = [{"path": "a1.mp4"}] + without_arena = [{"path": "w1.mp4"}] + coordinator.view.dialog_manager.ask_ok_cancel.return_value = True + + res = coordinator.select_eligible_videos( + True, + ready_traj, + ready_zones, + arena_only, + without_arena, + ) + assert res == [{"path": "t1.mp4"}, {"path": "z1.mp4"}, {"path": "a1.mp4"}] + + def test_with_dialog_user_cancelled(self, coordinator): + coordinator.view.dialog_manager.show_pending_videos_dialog.return_value = None + res = coordinator.select_eligible_videos( + False, + [], + [{"path": "z1.mp4"}], + [], + [], + ) + assert res is None + + +class TestOnVideoCompleted: + def test_on_video_completed_unsuccessful(self, coordinator): + coordinator._on_video_completed([], 0, 1, "exp1", False) # Should return early + + def test_on_video_completed_index_out_of_bounds(self, coordinator): + coordinator._on_video_completed([], 5, 1, "exp1", True) # Should log warning and return + + def test_on_video_completed_single_video(self, coordinator, tmp_path): + video_file = tmp_path / "vid.mp4" + video_file.write_text("dummy") + results_dir = tmp_path / "vid_results" + results_dir.mkdir() + traj_file = results_dir / "3_CoordMovimento_vid.parquet" + traj_file.write_text("data") + + videos = [ + {"path": str(video_file), "results_dir": str(results_dir), "experiment_id": "vid"} + ] + + coordinator._on_video_completed(videos, 0, 1, "vid", True) + + coordinator.project_manager.update_video_status.assert_called_once_with( + str(video_file), "complete" + ) + coordinator.project_manager.register_processing_outputs.assert_called_once_with( + video_path=str(video_file), + results_dir=str(results_dir), + trajectory_path=str(traj_file), + ) + + def test_on_video_completed_exploded_task(self, coordinator, tmp_path): + video_file = tmp_path / "vid_multi.mp4" + video_file.write_text("dummy") + results_dir = tmp_path / "vid_multi_aq1_results" + results_dir.mkdir() + traj_file = results_dir / "3_CoordMovimento_vid_multi.parquet" + traj_file.write_text("data") + + videos = [ + { + "path": str(video_file), + "results_dir": str(results_dir), + "experiment_id": "vid_multi", + "aquarium_id": 0, + "group": "G1", + "subject": "S1", + } + ] + + coordinator._on_video_completed(videos, 0, 1, "vid_multi", True) + + coordinator.project_manager.register_multi_aquarium_outputs.assert_called_once()