"""Optional selection, live sample input and real target-phase diagnostics.""" from __future__ import annotations import copy import os import queue import threading import time import unittest from types import SimpleNamespace from unittest import mock import numpy as np os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") from PyQt6.QtCore import QTimer # noqa: E402 from PyQt6.QtWidgets import QApplication, QPushButton, QWidget # noqa: E402 from python_app.gui.controllers.object_calibration_controller import ObjectCalibrationController, _calculate # noqa: E402 from python_app.gui.object_calibration_dialog import ObjectCalibrationDialog # noqa: E402 from python_app.gui.preprocess_dialog import PreprocessDialog # noqa: E402 from python_app.tests import test_mimo_calibration as numeric_fixture # noqa: E402 def setUpModule() -> None: global app app = QApplication.instance() or QApplication([]) class ObjectCalibrationSelectorTest(unittest.TestCase): def setUp(self) -> None: self.dialog = PreprocessDialog() self.addCleanup(self.dialog.close) def test_optional_off_is_distinct_from_a_saved_set_named_off(self) -> None: self.dialog.set_available_sets({"s21_object_calibration": ["Off", "sphere"]}) changes = mock.Mock() self.dialog.selection_changed.connect(changes) combo = self.dialog._set_combos["s21_object_calibration"] self.assertEqual(combo.currentText(), "Off") self.assertEqual(self.dialog.selection_snapshot()["s21_object_calibration"], "") combo.setCurrentIndex(combo.findData("Off")) self.assertEqual(self.dialog.selection_snapshot()["s21_object_calibration"], "Off") changes.assert_called_once() self.dialog.set_available_sets({"s21_object_calibration": ["Off", "sphere"]}) self.assertEqual(self.dialog.selection_snapshot()["s21_object_calibration"], "Off") self.dialog.set_selected_sets({"s21_object_calibration": ""}, emit_signal=False) self.assertEqual(combo.currentText(), "Off") self.assertEqual(combo.currentData(), "") def test_object_actions_do_not_start_a_sequential_hardware_capture(self) -> None: capture, imported, created = mock.Mock(), mock.Mock(), mock.Mock() self.dialog.start_sequence_requested.connect(capture) self.dialog.import_object_calibration_requested.connect(imported) self.dialog.create_object_calibration_requested.connect(created) self.dialog._import_object_calibration_button.click() self.dialog._create_object_calibration_button.click() imported.assert_called_once() created.assert_called_once() capture.assert_not_called() self.assertEqual([button.text() for button in self.dialog.findChildren(QPushButton) if button.text().startswith("Start ")], ["Start S21 Calibration", "Start S21 Reference"]) class ObjectCalibrationDialogTest(unittest.TestCase): def setUp(self) -> None: self.fixture = numeric_fixture.MimoCalibrationTest() self.fixture.setUp() self.dialog = ObjectCalibrationDialog("sphere") self.addCleanup(self.dialog.close) def _capture_with_target(self) -> None: self.dialog.add_capture([self.fixture.collection()], self.fixture.config) for column, coordinate in enumerate(self.fixture.target, start=2): self.dialog._table.item(0, column).setText(str(coordinate)) def test_capture_is_frozen_and_xyz_must_be_entered_after_measurement(self) -> None: collection = self.fixture.collection() self.dialog.add_capture([collection], self.fixture.config) expected = collection.traces[0].s21.copy() collection.traces[0].s21.fill(0) with self.assertRaisesRegex(ValueError, "X, Y и Z"): self.dialog.calculation_inputs() for column, coordinate in enumerate(self.fixture.target, start=2): self.dialog._table.item(0, column).setText(str(coordinate)) groups, config, settings = self.dialog.calculation_inputs() np.testing.assert_array_equal(groups[0]["collections"][0].traces[0].s21, expected) self.assertEqual(groups[0]["target_scene_m"], self.fixture.target) self.assertEqual(config, self.fixture.config) self.assertEqual(settings["input_idx"], (2, 7)) self.assertEqual(settings["output_idx"], (1, 5)) self.dialog._table.item(0, 2).setText("nan") with self.assertRaisesRegex(ValueError, "конечными"): self.dialog.calculation_inputs() def test_changed_geometry_is_rejected_without_appending_a_capture(self) -> None: self._capture_with_target() changed = copy.deepcopy(self.fixture.config) changed["gpr"]["tx_geometry"][0]["x_m"] += 1 with self.assertRaisesRegex(ValueError, "изменились"): self.dialog.add_capture([self.fixture.collection()], changed) self.assertEqual(self.dialog._table.rowCount(), 1) def test_actual_numeric_output_draws_both_target_phase_curves(self) -> None: calibration = self.fixture.calibration() self.dialog.show_calibration(calibration) curves = self.dialog._phase_plot.listDataItems() self.assertEqual(len(curves), 2) diagnostic = calibration["meta"]["diagnostics"][0] phases = np.array([diagnostic["phase_before_deg"], diagnostic["phase_after_deg"]]).reshape(2, -1) expected = (phases - phases[0, 0] + 180) % 360 - 180 for curve, phases in zip(curves, expected): np.testing.assert_allclose(curve.getData()[1], phases) self.assertLess(np.ptp(curves[1].getData()[1]), .5) self.assertIn(f"{diagnostic['coherence_before']:.3f}", self.dialog._quality_label.text()) self.assertIn(f"{diagnostic['coherence_after']:.3f}", self.dialog._quality_label.text()) def test_phase_wrap_uses_the_same_first_pair_baseline_for_both_curves(self) -> None: calibration = self.fixture.calibration() diagnostic = calibration["meta"]["diagnostics"][0] diagnostic["phase_before_deg"] = [[179, -179], [178, -178]] diagnostic["phase_after_deg"] = [[-179, -179], [-179, -179]] self.dialog.show_calibration(calibration) before, after = self.dialog._phase_plot.listDataItems() np.testing.assert_allclose(before.getData()[1], [0, 2, -1, 3]) np.testing.assert_allclose(after.getData()[1], [2, 2, 2, 2]) def test_capture_and_calculation_disable_edits_but_keep_cancel_available(self) -> None: self._capture_with_target() for setter in (self.dialog.set_capture_pending, self.dialog.set_calculating): setter(True) self.assertFalse(self.dialog._capture_button.isEnabled()) self.assertFalse(self.dialog._table.isEnabled()) self.assertTrue(self.dialog._cancel_button.isEnabled()) setter(False) self.assertTrue(self.dialog._capture_button.isEnabled()) self.assertTrue(self.dialog._table.isEnabled()) class ObjectCalibrationWorkerTest(unittest.TestCase): def _controller_with_data(self): fixture = numeric_fixture.MimoCalibrationTest() fixture.setUp() owner = QWidget() self.addCleanup(owner.close) panel = SimpleNamespace(set_name=lambda: "sphere", set_set_name=mock.Mock(), set_status=mock.Mock()) current_config = SimpleNamespace(to_dict=lambda: copy.deepcopy(fixture.config)) owner._ensure_preprocess_dialog = mock.Mock(return_value=panel) owner._build_config = mock.Mock(return_value=current_config) owner._radar_key = mock.Mock(return_value="current-config-key") owner._store = mock.Mock() owner._selected_preprocess_sets = {} for name in ("_show_exception", "_show_error", "_refresh_sets", "_log"): setattr(owner, name, mock.Mock()) controller = ObjectCalibrationController(owner) controller.open_creation() self.addCleanup(controller.shutdown) dialog = controller._dialog dialog.add_capture([fixture.collection()], fixture.config) for column, coordinate in enumerate(fixture.target, start=2): dialog._table.item(0, column).setText(str(coordinate)) for key in ("sphere_radius_m", "antenna_height_m", "pitch_deg", "coarse_range_search_m"): dialog._settings[key].setValue(getattr(fixture.settings, key)) for field in dialog._refine_axes.values(): field.setChecked(False) return owner, controller, fixture, current_config def _wait_for_completion(self, controller) -> None: deadline = time.monotonic() + 5 while controller._worker is not None and time.monotonic() < deadline: app.processEvents() time.sleep(.005) self.assertIsNone(controller._worker, "Background calculation did not complete") def test_actual_calculation_runs_off_gui_thread_and_saves_under_current_config(self) -> None: owner, controller, _, current_config = self._controller_with_data() from python_app.workflows.mimo_calibration import compute_from_collections worker_threads, gui_ticks = [], [] def compute(*args, **kwargs): worker_threads.append(threading.get_ident()) return compute_from_collections(*args, **kwargs) timer = QTimer() timer.timeout.connect(lambda: gui_ticks.append(threading.get_ident())) timer.start(1) self.addCleanup(timer.stop) with mock.patch("python_app.workflows.mimo_calibration.compute_from_collections", side_effect=compute): controller._start_calculation() self.assertTrue(controller._dialog._calculating) self._wait_for_completion(controller) owner._show_exception.assert_not_called() self.assertEqual(len(worker_threads), 1) self.assertNotEqual(worker_threads[0], threading.get_ident()) self.assertTrue(gui_ticks) self.assertEqual(set(gui_ticks), {threading.get_ident()}) owner._radar_key.assert_called_once_with(current_config) key, name, calibration = owner._store.save_object_calibration.call_args.args self.assertEqual((key, name), ("current-config-key", "sphere")) self.assertEqual(owner._selected_preprocess_sets["s21_object_calibration"], name) self.assertEqual(len(controller._dialog._phase_plot.listDataItems()), 2) self.assertGreater(calibration["meta"]["diagnostics"][0]["coherence_after"], .999) def test_cancelled_background_calculation_keeps_samples_and_creates_no_saved_set(self) -> None: owner, controller, _, _ = self._controller_with_data() entered = threading.Event() def compute(*_args, cancelled, progress, **_kwargs): entered.set() progress("Safe cancellation boundary") while not cancelled(): time.sleep(.005) raise InterruptedError with mock.patch("python_app.workflows.mimo_calibration.compute_from_collections", side_effect=compute): controller._start_calculation() self.assertTrue(entered.wait(1)) controller.cancel() self._wait_for_completion(controller) owner._store.save_object_calibration.assert_not_called() owner._show_exception.assert_not_called() self.assertEqual(controller._dialog._table.rowCount(), 1) self.assertTrue(controller._dialog._capture_button.isEnabled()) self.assertIn("отменён", controller._dialog._status_label.text()) def test_thread_start_failure_restores_editor_without_saving(self) -> None: owner, controller, _, _ = self._controller_with_data() with mock.patch("python_app.gui.controllers.object_calibration_controller.threading.Thread.start", side_effect=RuntimeError("thread limit")): controller._start_calculation() self.assertIsNone(controller._worker) self.assertFalse(controller._dialog._calculating) self.assertTrue(controller._dialog._capture_button.isEnabled()) owner._store.save_object_calibration.assert_not_called() owner._show_exception.assert_called_once() self.assertIn("thread limit", controller._dialog._status_label.text()) def test_capture_registration_failure_is_visible_and_does_not_leave_busy_dialog(self) -> None: owner, controller, _, _ = self._controller_with_data() with mock.patch("python_app.gui.controllers.object_calibration_capture.capture_object_sweeps", side_effect=RuntimeError("Capture already running")): controller._capture_data() self.assertIsNone(controller._capture) self.assertTrue(controller._dialog._capture_button.isEnabled()) self.assertIn("Capture already running", controller._dialog._status_label.text()) owner._show_exception.assert_called_once() def test_import_waits_for_capture_or_calculation_before_opening_file_picker(self) -> None: owner, controller, _, _ = self._controller_with_data() for busy in ("capture", "calculation"): with self.subTest(busy=busy): owner._object_capture_job = object() if busy == "capture" else None controller._worker = mock.Mock() if busy == "calculation" else None with mock.patch("python_app.gui.controllers.object_calibration_controller.QFileDialog.getExistingDirectory") as picker: controller.import_calibration() picker.assert_not_called() owner._show_error.assert_called() controller._worker = None owner._object_capture_job = None def test_worker_uses_only_data_and_honours_cancellation_without_subtracting_another_background(self) -> None: fixture = numeric_fixture.MimoCalibrationTest() fixture.setUp() cancelled, messages, result = threading.Event(), queue.Queue(), queue.Queue() cancelled.set() with mock.patch("python_app.workflows.mimo_calibration.compute_from_collections", side_effect=InterruptedError) as compute: _calculate([{"collections": [fixture.collection()], "target_scene_m": fixture.target}], fixture.config, {}, cancelled, messages, result) self.assertEqual(result.get_nowait(), ("cancelled", None)) self.assertNotIn("empty_collections", compute.call_args.kwargs) def test_calculate_rejects_invalid_xyz_before_starting_a_worker(self) -> None: owner = QWidget() self.addCleanup(owner.close) owner._ensure_preprocess_dialog = mock.Mock(return_value=SimpleNamespace(set_name=lambda: "sphere")) owner._show_exception = mock.Mock() controller = ObjectCalibrationController(owner) controller.open_creation() self.addCleanup(controller.shutdown) fixture = numeric_fixture.MimoCalibrationTest() fixture.setUp() controller._dialog.add_capture([fixture.collection()], fixture.config) with mock.patch("python_app.gui.controllers.object_calibration_controller.threading.Thread") as thread: controller._start_calculation() thread.assert_not_called() owner._show_exception.assert_called_once() self.assertFalse(controller._dialog._calculating) def test_closing_cancels_capture_and_worker_without_waiting_for_a_thread(self) -> None: owner = QWidget() self.addCleanup(owner.close) controller = ObjectCalibrationController(owner) controller._dialog = ObjectCalibrationDialog("sphere", owner) controller._capture = mock.Mock() controller._worker = mock.Mock() controller.shutdown() self.assertTrue(controller._cancel.is_set()) controller._capture.cancel.assert_called_once() controller._worker.join.assert_not_called() self.assertFalse(controller._poll.isActive())