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.
392 lines
20 KiB
Python
392 lines
20 KiB
Python
"""Independent 3D SAR views of the volume published by the sar3d processor.
|
||
|
||
Scene axes: X across, Y height above the ground (Y = 0 is the ground), Z along
|
||
the ground away from the radar. Payloads share normalization by the maximum
|
||
away from the Y boundaries:
|
||
|
||
- ``sar3d_front``, ``sar3d_side``, ``sar3d_top``: maximum projections onto X-Y
|
||
(looking ahead from the radar), Z-Y (from the side) and X-Z (from above);
|
||
- ``sar3d_oblique``: orthographic projection along the camera of the notebook's
|
||
3D scene; ``sar3d_oblique_basis`` (2x3) gives its image axes u and v as scene
|
||
directions, which places the scene box, the ground and the maximum on it;
|
||
- ``sar3d_peak``: [level, X, Y, Z] of the marked maximum; level is the normalization.
|
||
- ``sar3d_target_width``: axial FWHM [X, Y, Z] in metres through that maximum,
|
||
without CLEAN objects.
|
||
- ``sar3d_background``: [peak over background, BG_GATE_3D_SHOW, BG_GATE_3D_FULL]
|
||
while BG_GATE_ENABLE is on; the frame is scaled by ``bg_gate_gain``.
|
||
- ``sar3d_objects``: CLEAN objects, strongest first, rows [X, Y, Z, level in
|
||
backgrounds]. The projections then show the residual with every object in its
|
||
own 0-1 scale, and the marked maximum is the first object.
|
||
- ``sar3d_marks``: boxes around groups of the objects, rows [class, X0, X1, Y0,
|
||
Y1, Z0, Z1], classes as in ``SAR3D_MARKS``.
|
||
- ``sar3d_status``: [code, UTF-8 bytes...] with ready CUDA (0), no signal (1),
|
||
processing error (2), or CUDA unavailable (3), including the explanation.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Callable
|
||
import itertools
|
||
|
||
import numpy as np
|
||
from numpy.typing import ArrayLike
|
||
import pyqtgraph as pg
|
||
from PyQt6.QtCore import QRectF, Qt
|
||
from PyQt6.QtWidgets import QGridLayout, QLabel, QVBoxLayout, QWidget
|
||
|
||
from python_app.gui.controllers.renderers.common import (
|
||
ANTENNA_STYLES, OVERLAY_FILL, build_lut, create_plot, image_rect, selected_antenna_coordinates,
|
||
)
|
||
from python_app.gui.result_payloads import collection_payload_by_name
|
||
from python_app.gui.theme import PLOT_FOREGROUND
|
||
from python_app.models.dataset_model import ResultCollection, ResultPayload
|
||
from python_app.models.gui_profile_schema import GuiSar3dStateModel
|
||
from python_app.models.run_config_schema import GprModel
|
||
|
||
# Status codes published by the sar3d processor in its "sar3d_status" table.
|
||
SAR3D_STATES = {0: "ready", 1: "no signal", 2: "error", 3: "warning"}
|
||
|
||
# Box classes of "sar3d_marks" (notebook _SAR_MARK_STYLE): name, color, dashed. A box at the
|
||
# boundary is not yet decidable: part of its response may lie outside the scene.
|
||
SAR3D_MARKS: dict[int, tuple[str, tuple[int, int, int], bool]] = {
|
||
0: ("very low", (229, 38, 43), False),
|
||
1: ("on the ground", (255, 154, 31), False),
|
||
2: ("above the ground", (47, 111, 222), False),
|
||
3: ("at the boundary", (138, 138, 138), True),
|
||
}
|
||
|
||
# Under the views and inside the captured grid, so the web view shows it too.
|
||
_BOX_LEGEND = "Boxes: " + " ".join(
|
||
f"<span style='color: rgb{color}'>■</span> {name}" for name, color, _dashed in SAR3D_MARKS.values())
|
||
|
||
_GROUND_COLOR = (115, 105, 88)
|
||
_SCENE_BOX_COLOR = "#898781"
|
||
|
||
# payload -> (title, x label, y label, scene axes (0 = X, 1 = Y, 2 = Z) along the plot axes)
|
||
SAR3D_VIEWS: dict[str, tuple[str, str, str, tuple[int, int] | None]] = {
|
||
"sar3d_front": ("Front (X-Y)", "X", "Y", (0, 1)),
|
||
"sar3d_side": ("Side (Z-Y)", "Z", "Y", (2, 1)),
|
||
"sar3d_top": ("Top (X-Z)", "X", "Z", (0, 2)),
|
||
"sar3d_oblique": ("3D Projection", "", "", None),
|
||
}
|
||
|
||
# Corners of a box indexed by bits (x, y, z); an edge joins corners one bit apart.
|
||
_BOX_EDGES = [(a, b) for a, b in itertools.combinations(range(8), 2) if bin(a ^ b).count("1") == 1]
|
||
# Axis names go on the edges meeting at the bottom corner nearest the camera (X max, Y min, Z min).
|
||
_AXIS_EDGES = {"X": (0, 1), "Y": (1, 3), "Z": (1, 5)}
|
||
|
||
|
||
def sar3d_status(collection: ResultCollection) -> tuple[int, str] | None:
|
||
"""Status code (a SAR3D_STATES key) and message, one UTF-8 byte per table column."""
|
||
payload = collection_payload_by_name(collection, "sar3d_status", kind=4)
|
||
if payload is None:
|
||
return None
|
||
if payload.table.ndim != 2 or not payload.table.size or not np.isfinite(payload.table).all():
|
||
return 2, "Invalid SAR3D status table"
|
||
row = payload.table[0]
|
||
code = int(row[0])
|
||
if code not in SAR3D_STATES or code != row[0]:
|
||
return 2, f"Unknown SAR3D status code: {row[0]}"
|
||
return code, bytes(row[1:].astype(np.uint8)).decode("utf-8", errors="replace")
|
||
|
||
|
||
def bg_gate_gain(ratio: float, show: float, full: float, floor: float) -> float:
|
||
"""Scale of a frame by its peak over background (notebook bg_gate_gain).
|
||
|
||
Below ``show`` the frame is empty; at ``show`` its maximum lands on ``floor``, the
|
||
cloud threshold, so an object appears as a point; from ``full`` the scale is full.
|
||
"""
|
||
if ratio < show:
|
||
return 0.0
|
||
if full <= show:
|
||
return 1.0
|
||
return float(np.clip(floor + (1.0 - floor) * (ratio - show) / (full - show), floor, 1.0))
|
||
|
||
|
||
Projection = Callable[[ArrayLike], np.ndarray]
|
||
|
||
|
||
def _projection(axes: tuple[int, int] | None, basis: np.ndarray) -> Projection:
|
||
"""Scene points -> plot coordinates of a view: two scene axes, or the oblique image axes."""
|
||
if axes is None:
|
||
return lambda points: np.asarray(points, dtype=float) @ basis.T
|
||
return lambda points: np.asarray(points, dtype=float)[..., list(axes)]
|
||
|
||
|
||
def _box_corners(bounds: np.ndarray) -> np.ndarray:
|
||
"""Eight corners of the box [X0, X1, Y0, Y1, Z0, Z1], indexed by bits (x, y, z)."""
|
||
return np.array([[bounds[2*axis + ((index >> axis) & 1)] for axis in range(3)] for index in range(8)])
|
||
|
||
|
||
def _table(collection: ResultCollection, name: str, columns: int) -> np.ndarray:
|
||
"""Rows of a table payload; none when it is absent."""
|
||
payload = collection_payload_by_name(collection, name, kind=4)
|
||
if payload is None:
|
||
return np.empty((0, columns))
|
||
table = np.asarray(payload.table, dtype=float)
|
||
if table.ndim != 2 or table.shape[1] != columns:
|
||
raise ValueError(f"{name} must have {columns} columns")
|
||
return table
|
||
|
||
|
||
class Sar3dRenderer:
|
||
"""Draws the 3D SAR views."""
|
||
|
||
def __init__(self) -> None:
|
||
self.widget = QWidget()
|
||
layout = QVBoxLayout(self.widget)
|
||
layout.setContentsMargins(0, 0, 0, 0)
|
||
layout.setSpacing(6)
|
||
self.status_label = QLabel()
|
||
self.status_label.setTextFormat(Qt.TextFormat.PlainText)
|
||
self.status_label.setWordWrap(True)
|
||
layout.addWidget(self.status_label)
|
||
self.grid = QWidget()
|
||
grid_layout = QGridLayout(self.grid)
|
||
grid_layout.setContentsMargins(0, 0, 0, 0)
|
||
grid_layout.setSpacing(6)
|
||
layout.addWidget(self.grid, stretch=1)
|
||
lut = build_lut(["#081c15", "#1b4332", "#ffd166", "#f94144"]) # the GPR palette
|
||
lut = np.column_stack([lut, np.full(len(lut), 255, dtype=np.uint8)])
|
||
lut[0, 3] = 0 # below SAR_CLOUD_THRESHOLD
|
||
self.plots: dict[str, pg.PlotWidget] = {}
|
||
self.antennas: dict[str, tuple[pg.ScatterPlotItem, pg.ScatterPlotItem]] = {}
|
||
self.images: dict[str, pg.ImageItem] = {}
|
||
self.targets: dict[str, pg.PlotCurveItem] = {}
|
||
self.marks: dict[str, dict[int, pg.PlotCurveItem]] = {}
|
||
self.objects: dict[str, pg.ScatterPlotItem] = {}
|
||
self.object_labels: dict[str, list[pg.TextItem]] = {}
|
||
for index, (name, (title, x_label, y_label, axes)) in enumerate(SAR3D_VIEWS.items()):
|
||
plot = create_plot(title)
|
||
grid_layout.addWidget(plot, index // 2, index % 2)
|
||
self.plots[name] = plot
|
||
plot.setAspectLocked(True)
|
||
if axes is not None:
|
||
markers = (pg.ScatterPlotItem(), pg.ScatterPlotItem())
|
||
for marker in markers:
|
||
marker.setZValue(4)
|
||
plot.addItem(marker)
|
||
self.antennas[name] = markers
|
||
item = plot.getPlotItem()
|
||
assert item is not None
|
||
view = item.getViewBox()
|
||
view.setMouseEnabled(x=False, y=False)
|
||
view.enableAutoRange(x=False, y=False)
|
||
item.hideButtons()
|
||
image = pg.ImageItem(axisOrder="row-major", lut=lut)
|
||
plot.addItem(image)
|
||
# Hollow, so the response at the maximum stays visible.
|
||
peak = pg.PlotCurveItem(pen=pg.mkPen(PLOT_FOREGROUND, width=1.5))
|
||
peak.setZValue(3)
|
||
plot.addItem(peak)
|
||
self.images[name] = image
|
||
self.targets[name] = peak
|
||
self.marks[name] = {}
|
||
for kind, (_name, color, dashed) in SAR3D_MARKS.items():
|
||
style = Qt.PenStyle.DashLine if dashed else Qt.PenStyle.SolidLine
|
||
# A dark outline keeps the box visible over pixels of its own colour.
|
||
curve = pg.PlotCurveItem(pen=pg.mkPen(color, width=2, style=style), connect="pairs",
|
||
shadowPen=pg.mkPen(PLOT_FOREGROUND, width=4, style=style))
|
||
curve.setZValue(3)
|
||
plot.addItem(curve)
|
||
self.marks[name][kind] = curve
|
||
objects = pg.ScatterPlotItem(symbol="d", size=11, brush=pg.mkBrush("#ffffff"),
|
||
pen=pg.mkPen("#111111", width=1.5))
|
||
objects.setZValue(5)
|
||
plot.addItem(objects)
|
||
self.objects[name] = objects
|
||
self.object_labels[name] = []
|
||
if x_label:
|
||
plot.setLabel("bottom", x_label, units="m")
|
||
plot.setLabel("left", y_label, units="m")
|
||
if y_label == "Y":
|
||
ground = pg.InfiniteLine(pos=0.0, angle=0,
|
||
pen=pg.mkPen(_GROUND_COLOR, width=1, style=Qt.PenStyle.DashLine))
|
||
ground.setZValue(2)
|
||
plot.addItem(ground)
|
||
|
||
oblique = self.plots["sar3d_oblique"]
|
||
oblique.hideAxis("bottom")
|
||
oblique.hideAxis("left")
|
||
# The notebook scene's ground, under the image so it shows through weak voxels.
|
||
self.ground = pg.PlotCurveItem(
|
||
pen=pg.mkPen(_GROUND_COLOR), brush=pg.mkBrush(*_GROUND_COLOR, 110), fillLevel="enclosed"
|
||
)
|
||
self.ground.setZValue(-1)
|
||
oblique.addItem(self.ground)
|
||
self.box = pg.PlotCurveItem(pen=pg.mkPen(_SCENE_BOX_COLOR), connect="pairs")
|
||
self.box.setZValue(2)
|
||
oblique.addItem(self.box)
|
||
self.axis_labels: dict[str, pg.TextItem] = {}
|
||
for axis_name in _AXIS_EDGES:
|
||
label = pg.TextItem(axis_name, color=PLOT_FOREGROUND, anchor=(0.5, 0.5))
|
||
oblique.addItem(label)
|
||
self.axis_labels[axis_name] = label
|
||
self.box_legend = QLabel(_BOX_LEGEND)
|
||
self.box_legend.setTextFormat(Qt.TextFormat.RichText)
|
||
self.box_legend.setWordWrap(True)
|
||
self.box_legend.hide()
|
||
grid_layout.addWidget(self.box_legend, 2, 0, 1, 2)
|
||
self.clear()
|
||
|
||
def clear(self, message: str = "SAR3D: waiting for measurements") -> None:
|
||
"""Clear all four 3D SAR views."""
|
||
for name, (title, *_rest) in SAR3D_VIEWS.items():
|
||
self.images[name].hide()
|
||
self.targets[name].setData([], [])
|
||
self.plots[name].setTitle(title)
|
||
for markers in self.antennas.values():
|
||
for marker in markers:
|
||
marker.setData([], [])
|
||
self._draw_marks(np.empty((0, 7)), {})
|
||
self._draw_objects(np.empty((0, 4)), {})
|
||
self.box.setData([], [])
|
||
self.ground.setData([], [])
|
||
for label in self.axis_labels.values():
|
||
label.hide()
|
||
self.status_label.setText(message)
|
||
|
||
def _draw_marks(self, boxes: np.ndarray, projections: dict[str, Projection]) -> None:
|
||
"""Boxes around the CLEAN object groups, one curve per class and view."""
|
||
for view, curves in self.marks.items():
|
||
for kind, curve in curves.items():
|
||
rows = boxes[boxes[:, 0] == kind, 1:]
|
||
if not len(rows) or view not in projections:
|
||
curve.setData([], [])
|
||
continue
|
||
points = projections[view](np.concatenate([_box_corners(row)[np.ravel(_BOX_EDGES)] for row in rows]))
|
||
curve.setData(points[:, 0], points[:, 1])
|
||
|
||
def _draw_objects(self, objects: np.ndarray, projections: dict[str, Projection]) -> None:
|
||
"""Numbered CLEAN objects with their level in frame backgrounds."""
|
||
for name, markers in self.objects.items():
|
||
plot = self.plots[name]
|
||
for label in self.object_labels[name]:
|
||
plot.removeItem(label)
|
||
self.object_labels[name].clear()
|
||
if not len(objects) or name not in projections:
|
||
markers.setData([], [])
|
||
continue
|
||
points = projections[name](objects[:, :3])
|
||
markers.setData(points[:, 0], points[:, 1])
|
||
for number, ((x, y), level) in enumerate(zip(points, objects[:, 3]), start=1):
|
||
label = pg.TextItem(f"{number}: {level:.0f} bg", color=PLOT_FOREGROUND,
|
||
fill=pg.mkBrush(*OVERLAY_FILL), anchor=(0.5, 1.2))
|
||
label.setZValue(6)
|
||
label.setPos(float(x), float(y))
|
||
plot.addItem(label)
|
||
self.object_labels[name].append(label)
|
||
|
||
def draw(self, collection: ResultCollection, settings: GuiSar3dStateModel, geometry: GprModel) -> bool:
|
||
"""Draw four maximum projections of the volume; return success."""
|
||
status = sar3d_status(collection)
|
||
if status is not None and status[0] != 0:
|
||
self.clear(f"SAR3D: {SAR3D_STATES[status[0]]} — {status[1]}")
|
||
return False
|
||
payloads: dict[str, ResultPayload] = {}
|
||
for name in SAR3D_VIEWS:
|
||
payload = collection_payload_by_name(collection, name, kind=3)
|
||
if payload is None or not payload.image.size:
|
||
self.clear("SAR3D: waiting for projection data")
|
||
return False
|
||
payloads[name] = payload
|
||
basis = collection_payload_by_name(collection, "sar3d_oblique_basis", kind=4)
|
||
if basis is None:
|
||
self.clear("SAR3D: waiting for projection data")
|
||
return False
|
||
peak = collection_payload_by_name(collection, "sar3d_peak", kind=4)
|
||
level, *peak_xyz = peak.table[0] if peak is not None else (None,)
|
||
objects = _table(collection, "sar3d_objects", 4)
|
||
marks = _table(collection, "sar3d_marks", 7)
|
||
# Frame relative to its background; CLEAN objects are each in their own scale already.
|
||
background = _table(collection, "sar3d_background", 3)
|
||
gain = 1.0
|
||
background_note = ""
|
||
if len(background):
|
||
ratio, appear, full = (float(value) for value in background[0])
|
||
background_note = f"peak {ratio:.1f} backgrounds"
|
||
if not len(objects):
|
||
gain = bg_gate_gain(ratio, appear, full, settings.cloud_threshold)
|
||
state = ("below the appearance level, frame empty" if gain <= 0.0
|
||
else "full scale" if gain >= 1.0 else f"scale ×{gain:.2f}")
|
||
background_note += f" (appears from {appear:g}, full scale from {full:g}): {state}"
|
||
# Objects replace the peak marker, and so does a frame below the appearance level.
|
||
show_peak = not len(objects) and gain > 0.0
|
||
|
||
antennas = selected_antenna_coordinates(geometry, settings.input_positions, settings.output_positions)
|
||
pitch = np.radians(settings.look_angle_deg)
|
||
for coordinates in antennas:
|
||
y, z = coordinates[:, 1].copy(), coordinates[:, 2].copy()
|
||
coordinates[:, 1] = settings.bp_ground_depth_m + y*np.cos(pitch) - z*np.sin(pitch)
|
||
coordinates[:, 2] = y*np.sin(pitch) + z*np.cos(pitch)
|
||
width_payload = collection_payload_by_name(collection, "sar3d_target_width", kind=4)
|
||
steps = np.array([settings.voxel_x_m, settings.voxel_y_m, settings.voxel_z_m])
|
||
widths = width_payload.table[0] if width_payload is not None else steps
|
||
zoom = settings.zoom_percent / 100.0
|
||
projections = {name: _projection(axes, basis.table) for name, (*_labels, axes) in SAR3D_VIEWS.items()}
|
||
for name, payload in payloads.items():
|
||
title, _x_label, _y_label, axes = SAR3D_VIEWS[name]
|
||
rect = image_rect(payload.image_x_axis, payload.image_y_axis)
|
||
image = self.images[name]
|
||
shown = payload.image * gain
|
||
image.setImage(np.where(shown >= settings.cloud_threshold, shown, 0), autoLevels=False, levels=(0, 1))
|
||
image.setRect(rect)
|
||
image.show()
|
||
plot = self.plots[name]
|
||
position = projections[name](peak_xyz) if peak_xyz else None
|
||
if axes is not None:
|
||
for marker, coordinates, (symbol, brush, pen) in zip(self.antennas[name], antennas, ANTENNA_STYLES):
|
||
xx, yy = coordinates[:, axes[0]], coordinates[:, axes[1]]
|
||
marker.setData(x=xx, y=yy, symbol=symbol, size=13, brush=pg.mkBrush(brush), pen=pg.mkPen(pen))
|
||
if len(xx):
|
||
pad = steps.max()
|
||
rect = rect.united(QRectF(xx.min()-pad, yy.min()-pad, np.ptp(xx)+2*pad, np.ptp(yy)+2*pad))
|
||
if position is not None and show_peak:
|
||
extent = np.abs(basis.table) @ widths if axes is None else widths[list(axes)]
|
||
half = .5 * max(extent)
|
||
self.targets[name].setData(position[0]+np.array([-half, half, half, -half, -half]),
|
||
position[1]+np.array([-half, -half, half, half, -half]))
|
||
else:
|
||
self.targets[name].setData([], [])
|
||
if position is not None and zoom > 1.0:
|
||
center = position + (np.array([rect.center().x(), rect.center().y()])-position) / zoom**2
|
||
rect = QRectF(center[0]-rect.width()/(2*zoom), center[1]-rect.height()/(2*zoom),
|
||
rect.width()/zoom, rect.height()/zoom)
|
||
plot.setRange(rect, padding=0.04)
|
||
plot.setTitle(title)
|
||
boxes = marks if settings.mark_enabled else np.empty((0, 7))
|
||
self._draw_marks(boxes, projections)
|
||
# The legend follows the settings rather than each frame, so the views keep their size.
|
||
self.box_legend.setVisible(settings.mark_enabled and settings.clean_enabled)
|
||
self._draw_objects(np.empty((0, 4)) if settings.mark_enabled else objects, projections)
|
||
|
||
# Scene box and ground on the oblique view, from the extents of the other views.
|
||
front, top = payloads["sar3d_front"], payloads["sar3d_top"]
|
||
bounds = np.array([front.image_x_axis[0], front.image_x_axis[-1], front.image_y_axis[0],
|
||
front.image_y_axis[-1], top.image_y_axis[0], top.image_y_axis[-1]])
|
||
projected = _box_corners(bounds) @ basis.table.T
|
||
box = projected[np.ravel(_BOX_EDGES)]
|
||
self.box.setData(box[:, 0], box[:, 1])
|
||
centre = projected.mean(axis=0)
|
||
for axis_name, (start, end) in _AXIS_EDGES.items():
|
||
middle = (projected[start] + projected[end]) / 2
|
||
self.axis_labels[axis_name].setPos(*(middle + 0.08 * (middle - centre))) # just outside
|
||
self.axis_labels[axis_name].show()
|
||
x0, x1, _y0, _y1, z0, z1 = bounds
|
||
ground = (np.array([[x0, 0.0, z0], [x1, 0.0, z0], [x1, 0.0, z1], [x0, 0.0, z1], [x0, 0.0, z0]]) @ basis.table.T).T
|
||
self.ground.setData(ground[0], ground[1])
|
||
|
||
notes = [f"SAR3D: ready — {status[1]}" if status else "SAR3D: ready"]
|
||
if background_note:
|
||
notes.append(background_note)
|
||
if len(boxes):
|
||
counts = {kind: int(np.sum(boxes[:, 0] == kind)) for kind in SAR3D_MARKS}
|
||
notes.append("boxes: " + ", ".join(f"{counts[kind]} {name}" for kind, (name, *_style)
|
||
in SAR3D_MARKS.items() if kind != 3 or counts[kind]))
|
||
if level is not None:
|
||
x, y, z = peak_xyz
|
||
notes.append(f"max X {x:+.2f}, Y {y:+.2f}, Z {z:.2f} m | level {level:.3g}")
|
||
notes.append(f"zoom {settings.zoom_percent}%")
|
||
self.status_label.setText(" | ".join(notes))
|
||
return True
|