Ayzen
b88d704126
- BG_GATE_ENABLE: frame peak over its median background (sar3d_background); the GUI scales the frame with bg_gate_gain and its cloud threshold. - SAR_CLEAN_ENABLE: CLEAN of a one-frame aperture on a 4x10x4 cm search grid over the whole scene; objects in their own scale (sar3d_objects). - SAR_MARK_ENABLE: boxes by class around object groups (sar3d_marks), with the legend under the views. - SAR_VOXEL_M as three steps X/Y/Z with the notebook defaults 0.025/0.02/0.02. - GPU backend: image/download/load/project API, per-plane block reduction for the maximum, depth gate without FMA; sar3d host code built with -ffp-contract=off. - References regenerated from notebook cells 1 and 5, with new CLEAN scenes and a synthetic box-rule case; docs updated.
355 lines
21 KiB
Python
355 lines
21 KiB
Python
"""SAR3D projection rendering and visible processor failures, without live resources."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
import numpy as np
|
||
import pyqtgraph as pg
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
from PyQt6.QtCore import QPointF, QRectF, Qt, QSignalBlocker # noqa: E402
|
||
from PyQt6.QtWidgets import QApplication, QCheckBox, QComboBox, QDoubleSpinBox, QSlider # noqa: E402
|
||
|
||
from python_app.gui.controllers.renderers.sar3d import SAR3D_MARKS, SAR3D_STATES, SAR3D_VIEWS, Sar3dRenderer # noqa: E402
|
||
from python_app.models.dataset_model import ComboKey, ResultCollection, ResultPayload, TraceData # noqa: E402
|
||
from python_app.models.gui_profile_schema import GuiSar3dStateModel # noqa: E402
|
||
from python_app.models.run_config_schema import GprModel, GprTxGeometryModel, GprRxGeometryModel # noqa: E402
|
||
from python_app.tests.gui_test_support import widget
|
||
|
||
|
||
def _status(code: int, message: str) -> ResultPayload:
|
||
return ResultPayload(processing_name="sar3d_status", kind=4,
|
||
table=np.array([[code, *message.encode("utf-8")]], dtype=np.float32))
|
||
|
||
|
||
def _collection(*payloads: ResultPayload) -> ResultCollection:
|
||
return ResultCollection(collection_id=5, monotonic_ns=1500, collection_payloads=list(payloads),
|
||
capture_start_ns=1200, capture_end_ns=1400)
|
||
|
||
|
||
def _points(item: pg.PlotCurveItem | pg.ScatterPlotItem) -> tuple[np.ndarray, np.ndarray]:
|
||
"""Drawn x and y of a curve or scatter item, empty when nothing is drawn."""
|
||
xx, yy = item.getData()
|
||
assert xx is not None and yy is not None
|
||
return np.asarray(xx), np.asarray(yy)
|
||
|
||
|
||
def _table(name: str, *rows: list[float]) -> ResultPayload:
|
||
return ResultPayload(processing_name=name, kind=4, table=np.array(rows, dtype=np.float32))
|
||
|
||
|
||
def _projections() -> ResultCollection:
|
||
x, y, z = np.array([-1, 0, 1]), np.array([-0.5, 0.5]), np.array([1, 2, 3])
|
||
axes = [(x, y), (z, y), (x, z), (x, y)]
|
||
images = [ResultPayload(processing_name=name, kind=3, image_x_axis=xx, image_y_axis=yy,
|
||
image=np.ones((len(yy), len(xx)), dtype=np.float32))
|
||
for name, (xx, yy) in zip(SAR3D_VIEWS, axes)]
|
||
return _collection(*images, _status(0, "CUDA"),
|
||
ResultPayload(processing_name="sar3d_oblique_basis", kind=4,
|
||
table=np.array([[1, 0, 0], [0, 1, 0]], dtype=np.float32)),
|
||
ResultPayload(processing_name="sar3d_peak", kind=4,
|
||
table=np.array([[2, 0.5, -0.2, 2]], dtype=np.float32)))
|
||
|
||
|
||
class Sar3dPlotTest(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.app = QApplication.instance() or QApplication([])
|
||
|
||
def setUp(self) -> None:
|
||
self.renderer = Sar3dRenderer()
|
||
self.settings, self.geometry = GuiSar3dStateModel(), GprModel()
|
||
self.addCleanup(self.renderer.widget.close)
|
||
|
||
def test_all_four_projections_have_visible_images_and_correct_peak_axes(self) -> None:
|
||
self.assertTrue(self.renderer.draw(_projections(), self.settings, self.geometry))
|
||
self.assertEqual(self.renderer.status_label.text(),
|
||
"SAR3D: ready — CUDA | max X +0.50, Y -0.20, Z 2.00 m | level 2 | zoom 100%")
|
||
positions = {"sar3d_front": (0.5, -0.2), "sar3d_side": (2, -0.2),
|
||
"sar3d_top": (0.5, 2), "sar3d_oblique": (0.5, -0.2)}
|
||
for name, position in positions.items():
|
||
image = self.renderer.images[name]
|
||
self.assertTrue(image.isVisible(), name)
|
||
self.assertGreater(image.boundingRect().width(), 0)
|
||
self.assertGreater(image.boundingRect().height(), 0)
|
||
xx, yy = self.renderer.targets[name].getData()
|
||
assert xx is not None and yy is not None
|
||
np.testing.assert_allclose([.5*(xx.min()+xx.max()), .5*(yy.min()+yy.max())], position)
|
||
self.assertAlmostEqual(float(np.ptp(xx)), float(np.ptp(yy)))
|
||
self.assertTrue(all(label.isVisible() for label in self.renderer.axis_labels.values()))
|
||
self.renderer.clear()
|
||
self.assertEqual(self.renderer.status_label.text(), "SAR3D: waiting for measurements")
|
||
|
||
def test_unavailable_warning_clears_previous_images_and_labels_and_shows_reason(self) -> None:
|
||
self.renderer.draw(_projections(), self.settings, self.geometry)
|
||
message = "CUDA unavailable: драйвер <libcuda>"
|
||
self.assertFalse(self.renderer.draw(_collection(_status(3, message)), self.settings, self.geometry))
|
||
self.assertEqual(self.renderer.status_label.text(), f"SAR3D: warning — {message}")
|
||
self.assertEqual(self.renderer.status_label.textFormat(), Qt.TextFormat.PlainText)
|
||
self.assertTrue(all(not image.isVisible() for image in self.renderer.images.values()))
|
||
self.assertTrue(all(not label.isVisible() for label in self.renderer.axis_labels.values()))
|
||
|
||
def test_no_signal_and_missing_projections_explain_empty_panels(self) -> None:
|
||
self.assertFalse(self.renderer.draw(_collection(_status(1, "No signal in selected range")), self.settings, self.geometry))
|
||
self.assertIn("no signal", self.renderer.status_label.text())
|
||
self.assertFalse(self.renderer.draw(_collection(), self.settings, self.geometry))
|
||
self.assertEqual(self.renderer.status_label.text(), "SAR3D: waiting for projection data")
|
||
|
||
def test_response_size_antenna_positions_and_zoom_use_scene_coordinates(self) -> None:
|
||
result = _projections()
|
||
result.collection_payloads.append(ResultPayload("sar3d_target_width", 4, table=np.array([[.4, .6, .8]])))
|
||
self.geometry = GprModel(tx_geometry=[GprTxGeometryModel(0, .7, .3, .4)],
|
||
rx_geometry=[GprRxGeometryModel(0, -.7, -.3, .2)])
|
||
self.settings.look_angle_deg = 30
|
||
self.settings.bp_ground_depth_m = 1.25
|
||
self.renderer.widget.resize(800, 600)
|
||
self.renderer.widget.show()
|
||
self.renderer.draw(result, self.settings, self.geometry)
|
||
self.app.processEvents()
|
||
baseline = {}
|
||
expected_antenna = [.7, 1.25 + .3*np.cos(np.pi/6) - .4*.5, .3*.5 + .4*np.cos(np.pi/6)]
|
||
for name, (_, _, _, axes) in SAR3D_VIEWS.items():
|
||
xx, yy = self.renderer.targets[name].getData()
|
||
assert xx is not None and yy is not None
|
||
side = .6 if name in {"sar3d_front", "sar3d_oblique"} else .8
|
||
self.assertAlmostEqual(float(np.ptp(xx)), side)
|
||
if axes is not None:
|
||
tx, _ = self.renderer.antennas[name]
|
||
ax, ay = tx.getData()
|
||
np.testing.assert_allclose([ax[0], ay[0]], np.asarray(expected_antenna)[list(axes)])
|
||
view = self.renderer.plots[name].getViewBox()
|
||
for marker in self.renderer.antennas[name]:
|
||
ax, ay = marker.getData()
|
||
for x, y in zip(ax, ay):
|
||
point = view.mapViewToScene(QPointF(float(x), float(y)))
|
||
self.assertTrue(view.sceneBoundingRect().contains(QRectF(point.x()-7, point.y()-7, 14, 14)), name)
|
||
rect = self.renderer.plots[name].getViewBox().targetRect()
|
||
baseline[name] = np.array([[rect.left(), rect.right()], [rect.top(), rect.bottom()]])
|
||
self.assertNotIn("sar3d_oblique", self.renderer.antennas)
|
||
self.settings.zoom_percent = 250
|
||
self.renderer.draw(result, self.settings, self.geometry)
|
||
self.app.processEvents()
|
||
for name, ranges in baseline.items():
|
||
rect = self.renderer.plots[name].getViewBox().targetRect()
|
||
zoomed = np.array([[rect.left(), rect.right()], [rect.top(), rect.bottom()]])
|
||
np.testing.assert_allclose(np.ptp(zoomed, axis=1), np.ptp(ranges, axis=1)/2.5)
|
||
xx, yy = self.renderer.targets[name].getData()
|
||
assert xx is not None and yy is not None
|
||
peak = np.array([.5*(xx.min()+xx.max()), .5*(yy.min()+yy.max())])
|
||
self.assertLess(np.linalg.norm(zoomed.mean(axis=1)-peak), np.linalg.norm(ranges.mean(axis=1)-peak))
|
||
|
||
def test_bg_gate_scales_the_frame_and_hides_it_below_appearance(self) -> None:
|
||
self.settings.cloud_threshold = .5
|
||
for ratio, gain, state in ((9., 1., "full scale"), (7.5, .8, "scale ×0.80"),
|
||
(5., 0., "below the appearance level, frame empty")):
|
||
with self.subTest(ratio=ratio):
|
||
result = _projections()
|
||
result.collection_payloads.append(_table("sar3d_background", [ratio, 6, 8.5]))
|
||
self.assertTrue(self.renderer.draw(result, self.settings, self.geometry))
|
||
for name, image in self.renderer.images.items():
|
||
np.testing.assert_allclose(np.asarray(image.image), gain, err_msg=name)
|
||
self.assertEqual(len(_points(self.renderer.targets[name])[0]) > 0, gain > 0, name)
|
||
self.assertIn(f"peak {ratio:.1f} backgrounds (appears from 6, full scale from 8.5): {state}",
|
||
self.renderer.status_label.text())
|
||
|
||
def test_clean_objects_are_boxed_by_class_or_numbered(self) -> None:
|
||
result = _projections()
|
||
result.collection_payloads += [
|
||
_table("sar3d_background", [7.5, 6, 8.5]),
|
||
_table("sar3d_objects", [.5, -.2, 2., 45.], [-.5, .3, 1.5, 12.]),
|
||
_table("sar3d_marks", [0, .25, .75, -.3, .1, 1.75, 2.25], [2, -.75, -.25, .1, .5, 1.25, 1.75]),
|
||
]
|
||
self.assertTrue(self.renderer.draw(result, self.settings, self.geometry))
|
||
text = self.renderer.status_label.text()
|
||
self.assertIn("peak 7.5 backgrounds | boxes: 1 very low, 0 on the ground, 1 above the ground |", text)
|
||
self.assertNotIn("at the boundary", text)
|
||
for name, (_title, _x, _y, axes) in SAR3D_VIEWS.items():
|
||
# Objects are in their own scale: no background gain and no peak marker.
|
||
np.testing.assert_allclose(np.asarray(self.renderer.images[name].image), 1)
|
||
self.assertFalse(len(_points(self.renderer.targets[name])[0]))
|
||
self.assertFalse(len(_points(self.renderer.objects[name])[0]))
|
||
curves = self.renderer.marks[name]
|
||
self.assertFalse(len(_points(curves[1])[0]) or len(_points(curves[3])[0]))
|
||
plane = axes or (0, 1) # the test basis shows the oblique view along X and Y
|
||
for kind, low, high in ((0, (.25, -.3, 1.75), (.75, .1, 2.25)), (2, (-.75, .1, 1.25), (-.25, .5, 1.75))):
|
||
xx, yy = _points(curves[kind])
|
||
self.assertEqual(len(xx), 24)
|
||
np.testing.assert_allclose([xx.min(), yy.min()], np.asarray(low)[list(plane)])
|
||
np.testing.assert_allclose([xx.max(), yy.max()], np.asarray(high)[list(plane)])
|
||
# The legend lists every class under the views while boxes are on.
|
||
self.assertFalse(self.renderer.box_legend.isHidden())
|
||
self.assertTrue(all(name in self.renderer.box_legend.text() for name, *_style in SAR3D_MARKS.values()))
|
||
|
||
self.settings.mark_enabled = False
|
||
self.renderer.draw(result, self.settings, self.geometry)
|
||
self.assertTrue(self.renderer.box_legend.isHidden())
|
||
self.assertNotIn("boxes:", self.renderer.status_label.text())
|
||
for name, (_title, _x, _y, axes) in SAR3D_VIEWS.items():
|
||
self.assertTrue(all(not len(_points(curve)[0]) for curve in self.renderer.marks[name].values()))
|
||
xx, yy = _points(self.renderer.objects[name])
|
||
plane = list(axes or (0, 1))
|
||
np.testing.assert_allclose(np.column_stack([xx, yy]), np.array([[.5, -.2, 2.], [-.5, .3, 1.5]])[:, plane])
|
||
self.assertEqual([label.textItem.toPlainText() for label in self.renderer.object_labels[name]],
|
||
["1: 45 bg", "2: 12 bg"])
|
||
self.renderer.clear()
|
||
self.assertFalse(any(self.renderer.object_labels.values()))
|
||
|
||
def test_table_with_another_width_is_an_error(self) -> None:
|
||
result = _projections()
|
||
result.collection_payloads.append(_table("sar3d_objects", [.5, -.2, 2.]))
|
||
with self.assertRaisesRegex(ValueError, "sar3d_objects must have 4 columns"):
|
||
self.renderer.draw(result, self.settings, self.geometry)
|
||
|
||
def test_cloud_threshold_masks_weak_response_without_changing_levels(self) -> None:
|
||
result = _projections()
|
||
for payload in result.collection_payloads:
|
||
if payload.kind == 3:
|
||
payload.image.flat[:2] = [.2, .8]
|
||
for threshold, expected in ((.65, [0, .8, 1]), (1., [0, 0, 1])):
|
||
self.settings.cloud_threshold = threshold
|
||
self.renderer.draw(result, self.settings, self.geometry)
|
||
for image in self.renderer.images.values():
|
||
np.testing.assert_allclose(np.asarray(image.image).flat[:3], expected)
|
||
np.testing.assert_allclose(np.asarray(image.levels), [0, 1])
|
||
|
||
|
||
class Sar3dIntegrationTest(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.app = QApplication.instance() or QApplication([])
|
||
|
||
def setUp(self) -> None:
|
||
from python_app.tests.gui_test_support import isolated_window
|
||
|
||
self.view = isolated_window(self.addCleanup)
|
||
self.view._pipeline.timer.stop()
|
||
self.mode = self.view._view.processing.mode
|
||
with QSignalBlocker(self.mode):
|
||
self.mode.setCurrentText("sar3d")
|
||
self.view._plots.set_mode("sar3d")
|
||
self.view._plots.required_revision = None
|
||
self.renderer = self.view._plots.sar3d
|
||
|
||
def test_display_controls_redraw_without_processing_revision_or_live_fields(self) -> None:
|
||
result = _projections()
|
||
self.view._history.record_result(result)
|
||
revision = self.view._live.revision
|
||
widget(self.view._view.processing, "_sar3d_zoom_percent", QSlider).setValue(247)
|
||
widget(self.view._view.processing, "_sar3d_cloud_threshold", QDoubleSpinBox).setValue(1.)
|
||
self.assertEqual(self.view._live.revision, revision)
|
||
self.assertIn("zoom 247%", self.renderer.status_label.text())
|
||
widget(self.view._view.processing, "_sar3d_mark_enabled", QCheckBox).setChecked(False)
|
||
self.assertEqual(self.view._live.revision, revision)
|
||
live = self.view._live.build().to_dict()
|
||
self.assertNotIn("sar3d_zoom_percent", live)
|
||
self.assertNotIn("sar3d_cloud_threshold", live)
|
||
self.assertNotIn("sar3d_mark_enabled", live)
|
||
schema = {entry["name"]: entry for entry in self.view._config.web_schema()}
|
||
self.assertEqual(schema["sar3d_zoom_percent"]["value"], 247)
|
||
self.assertEqual(schema["sar3d_zoom_percent"]["step"], 1)
|
||
self.assertEqual(schema["sar3d_cloud_threshold"]["value"], 1.)
|
||
self.assertIs(schema["sar3d_mark_enabled"]["value"], False)
|
||
|
||
def test_bg_gate_and_clean_publish_a_processing_revision(self) -> None:
|
||
self.view._pipeline.drain_results = mock.Mock()
|
||
live = self.view._live.build().to_dict()
|
||
self.assertIs(live["sar3d_bg_gate_enabled"], True)
|
||
self.assertIs(live["sar3d_clean_enabled"], True)
|
||
for name, key in (("_sar3d_bg_gate_enabled", "sar3d_bg_gate_enabled"),
|
||
("_sar3d_clean_enabled", "sar3d_clean_enabled")):
|
||
with self.subTest(key):
|
||
revision = self.view._live.revision
|
||
widget(self.view._view.processing, name, QCheckBox).setChecked(False)
|
||
self.assertNotEqual(self.view._live.revision, revision)
|
||
published = json.loads(self.view._live.writer.path.read_text(encoding="utf-8"))
|
||
self.assertIs(published[key], False)
|
||
self.assertEqual(published["settings_revision"], self.view._live.revision)
|
||
schema = {entry["name"]: entry for entry in self.view._config.web_schema()}
|
||
self.assertEqual(schema["sar3d_bg_gate_enabled"]["label"], "BG_GATE_ENABLE")
|
||
self.assertEqual(schema["sar3d_clean_enabled"]["label"], "SAR_CLEAN_ENABLE")
|
||
|
||
def test_inactive_sar_scene_is_validated_before_publishing_gpr_settings(self) -> None:
|
||
with QSignalBlocker(self.mode):
|
||
self.mode.setCurrentText("gpr")
|
||
with QSignalBlocker(widget(self.view._view.processing, "_sar3d_y_min_m", QDoubleSpinBox)):
|
||
widget(self.view._view.processing, "_sar3d_y_min_m", QDoubleSpinBox).setValue(3.)
|
||
with self.assertRaisesRegex(ValueError, "SAR_Y_RANGE_M"):
|
||
self.view._config.build_live_config()
|
||
|
||
def test_capture_preview_keeps_sar_page_and_uses_saved_trace_settings(self) -> None:
|
||
self.view._plots.set_mode("sar3d")
|
||
fields = self.view._view.processing
|
||
phase = widget(fields, "_show_phase_checkbox", QCheckBox)
|
||
unwrap = widget(fields, "_unwrap_phase_checkbox", QCheckBox)
|
||
fixed = widget(fields, "_pass_through_fixed_y_enabled", QCheckBox)
|
||
lower = widget(fields, "_pass_through_y_min_db", QDoubleSpinBox)
|
||
upper = widget(fields, "_pass_through_y_max_db", QDoubleSpinBox)
|
||
revision = self.view._live.revision
|
||
phase.setChecked(False)
|
||
unwrap.setChecked(True)
|
||
fixed.setChecked(True)
|
||
lower.setValue(-70)
|
||
upper.setValue(-5)
|
||
self.assertEqual(self.view._live.revision, revision)
|
||
trace = TraceData(ComboKey(0, 0), np.array([1., 2.]), np.array([0.1 + 0j, 0.2 + 0j]))
|
||
self.view._plots.draw_trace(trace, "Captured reference")
|
||
self.assertIs(self.view._plots.stack.currentWidget(), self.renderer.widget)
|
||
self.assertFalse(self.view._plots.trace.phase.curves)
|
||
self.assertTrue(self.view._config.processing_settings().pass_through.unwrap_phase)
|
||
np.testing.assert_allclose(self.view._plots.trace.magnitude.view.viewRange()[1], [-70, -5])
|
||
|
||
def test_single_capture_finishes_reported_failure_without_losing_its_result(self) -> None:
|
||
for code in (1, 2, 3):
|
||
with self.subTest(code=code):
|
||
result = _collection(_status(code, "processor explanation"))
|
||
self.view._history.results.clear()
|
||
self.view._history.record_result(result)
|
||
self.view._pipeline.single_capture_active = True
|
||
self.view._pipeline._single_capture_start_ns = 1000
|
||
self.view._pipeline._single_capture_target_key = (5, 1200)
|
||
with self.assertLogs("python_app.gui.controllers.pipeline_controller", level="WARNING") as logged:
|
||
self.assertTrue(self.view._pipeline._finish_single_capture_if_ready())
|
||
self.assertFalse(self.view._pipeline.single_capture_active)
|
||
self.assertEqual(list(self.view._history.results), [result])
|
||
self.assertIn("processor explanation", self.renderer.status_label.text())
|
||
state = SAR3D_STATES[code]
|
||
self.assertEqual(self.view._view.status.text(), f"Status: {state}")
|
||
self.assertEqual(logged.records[-1].levelname, "ERROR" if code == 2 else "WARNING")
|
||
self.assertEqual(logged.records[-1].getMessage(),
|
||
f"Single capture: SAR3D {state}: processor explanation")
|
||
|
||
def test_live_edit_retains_failure_reason_then_recovers_to_ready_images(self) -> None:
|
||
self.view._history.results.clear()
|
||
self.view._live.publish = mock.Mock()
|
||
self.view._log.show_exception = mock.Mock()
|
||
for result, expected in ((_collection(_status(2, "CUDA kernel failure")), "error"),
|
||
(_collection(_status(3, "Driver initialization unavailable")), "warning"),
|
||
(_collection(_status(2, "fewer than two frequency points")), "error"),
|
||
(_collection(_status(1, "No signal in selected depth range")), "no signal"),
|
||
(_projections(), "ready — CUDA")):
|
||
# Like the real drain, the recomputed result replaces its sweep in history.
|
||
self.view._pipeline.drain_results = mock.Mock(
|
||
side_effect=lambda result=result, **_timing: self.view._history.record_result(result))
|
||
self.view._live.changed()
|
||
self.assertIn(expected, self.renderer.status_label.text())
|
||
self.view._log.show_exception.assert_not_called()
|
||
self.assertTrue(all(image.isVisible() for image in self.renderer.images.values()))
|
||
|
||
def test_history_redraw_preserves_error_but_stale_revision_clears_old_images(self) -> None:
|
||
self.view._history.record_result(_collection(_status(2, "processor error")))
|
||
self.view._plots.redraw()
|
||
self.assertEqual(self.renderer.status_label.text(), "SAR3D: error — processor error")
|
||
self.view._plots.draw(_projections())
|
||
self.view._plots.required_revision = 10
|
||
self.view._plots.redraw()
|
||
self.assertEqual(self.renderer.status_label.text(), "SAR3D: waiting for measurements")
|
||
self.assertTrue(all(not image.isVisible() for image in self.renderer.images.values()))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|