#!/usr/bin/env python3 """ SRAS Scan File Viewer PyQt6 application for visualizing channel data from v6 .sras scan files. Channel semantics (fixed by sc3_aui_app.py acquisition settings): CH1 — RF Acoustic Packet: FFT → peak frequency CH3 — Bias A (DC): waveform mean CH4 — Bias B (DC): waveform mean RF images are masked: pixels where CH4_dc < dc_threshold show 0. Scan data is memory-mapped (core.sras_format.SrasFile), so opening a multi-GB file costs only the pages actually touched by the current view. Incomplete (aborted/resumed) files are handled via the format's frontier analysis: only the rows actually on disk are shown. """ import sys from pathlib import Path import numpy as np from PyQt6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QPushButton, QComboBox, QSpinBox, QDoubleSpinBox, QFileDialog, QSizePolicy, QSplitter, QCheckBox, QFrame, QProgressDialog, ) from PyQt6.QtCore import Qt, QThread, QTimer, pyqtSignal, QObject from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg, NavigationToolbar2QT from matplotlib.figure import Figure sys.path.insert(0, str(Path(__file__).resolve().parent)) from core.sras_format import SrasFile from core.sras_analysis import ( CH1_IDX, CH3_IDX, CH4_IDX, ChannelCalibration, SawPipeline, compute_dc_image, compute_rf_image, compute_saw_image, power_spectrum, ) CH_LABELS = [ "CH1 — RF (FFT peak freq)", "CH3 — Bias A (DC mean)", "CH4 — Bias B (DC mean)", "CH1 — Velocity (SRAS)", "CH1 — SAW Amplitude (matched filter)", "CH1 — SAW Arrival time (matched filter)", ] CH_NAMES = ["CH1", "CH3", "CH4", "VEL", "SAW-AMP", "SAW-TOF"] # Combo indices for derived modes (all use CH1_IDX data) VELOCITY_MODE_IDX = 3 SAW_MODE_AMP_IDX = 4 SAW_MODE_TOF_IDX = 5 SAW_MODES = (SAW_MODE_AMP_IDX, SAW_MODE_TOF_IDX) CH1_DERIVED_MODES = (CH1_IDX, VELOCITY_MODE_IDX) + SAW_MODES # Per-mode display strings: (mode_str, unit, colorbar_label) MODE_INFO = { CH1_IDX: ("RF", "Peak frequency (MHz)", "MHz"), VELOCITY_MODE_IDX: ("Velocity", "Velocity (m/s)", "m/s"), SAW_MODE_AMP_IDX: ("SAW-AMP", "MF envelope peak (arb.)", "amplitude"), SAW_MODE_TOF_IDX: ("SAW-TOF", "SAW arrival time (ns)", "ns"), CH3_IDX: ("DC", "DC mean (mV)", "mV"), CH4_IDX: ("DC", "DC mean (mV)", "mV"), } CMAPS = ["gray", "viridis", "plasma", "inferno", "hot", "jet", "RdBu_r", "seismic"] RECOMPUTE_DEBOUNCE_MS = 250 class LoadedScan: """A parsed scan plus everything the viewer derives from it once.""" def __init__(self, path: str): self.sras = SrasFile(Path(path)) self.calib = ChannelCalibration.from_preambles(self.sras.preambles) bg = np.frombuffer(self.sras.background, dtype=np.int8) self.background = bg.astype(np.float32) if len(bg) else None # Rows actually on disk per angle (aborted/resumed scans) self.rows_available = [s.n_rows_available for s in self.sras.angle_status()] def angle_view(self, angle_idx: int) -> np.ndarray: """Available rows of one angle: (rows, n_ch, n_frames, spf) int8 view.""" return self.sras.load_angle(angle_idx, n_rows=self.rows_available[angle_idx]) def close(self): self.sras.close() # --------------------------------------------------------------------------- # Background workers # --------------------------------------------------------------------------- class FnWorker(QObject): """Runs a callable on a QThread; emits its return value or the error.""" finished = pyqtSignal(object) error = pyqtSignal(str) def __init__(self, fn): super().__init__() self._fn = fn def run(self): try: self.finished.emit(self._fn()) except Exception as exc: self.error.emit(str(exc)) def compute_image(scan: LoadedScan, angle_idx: int, ch_idx: int, dc_threshold_mv: float, apply_bg_sub: bool, gate_start_ns: float | None, gate_end_ns: float | None, saw_pipeline: SawPipeline | None) -> np.ndarray: """Channel-mode dispatch for the image computation worker. VELOCITY mode returns the plain RF image (MHz) — the grating scaling is a display-time scalar multiply, so changing the grating never triggers a recompute. """ view = scan.angle_view(angle_idx) if view.shape[0] == 0: return np.zeros((0, 0), dtype=np.float32) sras = scan.sras bg = scan.background if apply_bg_sub else None if ch_idx in (CH1_IDX, VELOCITY_MODE_IDX): return compute_rf_image( view, scan.calib, sras.freq_axis_mhz(sras.header.samples_per_frame), dc_threshold_mv, background=bg, gate_start_ns=gate_start_ns, gate_end_ns=gate_end_ns, time_axis_ns=sras.time_axis_ns(), ) if ch_idx in SAW_MODES: if saw_pipeline is None or saw_pipeline.template is None: raise RuntimeError( "SAW pipeline: no template built yet.\n" "Use \"Build Template\" in the SAW Pipeline panel first.") mode = "amplitude" if ch_idx == SAW_MODE_AMP_IDX else "tof" return compute_saw_image(view, scan.calib, dc_threshold_mv, saw_pipeline, mode, background=bg) # DC channels: ADC counts → mV return scan.calib.adc_to_mv(compute_dc_image(view, ch_idx), ch_idx) # --------------------------------------------------------------------------- # Matplotlib canvases # --------------------------------------------------------------------------- class ImageCanvas(FigureCanvasQTAgg): pixel_clicked = pyqtSignal(int, int) # row_idx, frame_idx def __init__(self, parent=None): fig = Figure(figsize=(7, 5), tight_layout=True) self.ax = fig.add_subplot(111) super().__init__(fig) self.setParent(parent) self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) self._extent = None self._img_shape = None self._im = None self._cbar = None self.mpl_connect("button_press_event", self._on_click) def show_image(self, img: np.ndarray, extent: list[float], cmap: str, vmin: float, vmax: float, xlabel: str, ylabel: str, title: str, colorbar_label: str = ""): # Rebuild the artist tree only when the geometry changes; colormap, # limits, and data updates reuse the existing AxesImage + colorbar. rebuild = (self._im is None or img.shape != self._img_shape or extent != self._extent) self._extent = extent self._img_shape = img.shape if rebuild: self.figure.clf() self.ax = self.figure.add_subplot(111) self._im = self.ax.imshow( img, aspect="auto", origin="upper", extent=extent, cmap=cmap, vmin=vmin, vmax=vmax, interpolation="nearest", ) self._cbar = self.figure.colorbar(self._im, ax=self.ax, fraction=0.046, pad=0.04) else: self._im.set_data(img) self._im.set_cmap(cmap) self._im.set_clim(vmin, vmax) self._cbar.set_label(colorbar_label or "") self.ax.set_xlabel(xlabel) self.ax.set_ylabel(ylabel) self.ax.set_title(title) self.draw_idle() def _on_click(self, event): if event.inaxes is not self.ax or self._extent is None: return x0, x1, y_bot, y_top = self._extent n_rows, n_frames = self._img_shape if n_rows == 0 or n_frames == 0: return col = int((event.xdata - x0) / (x1 - x0) * n_frames) row = int((event.ydata - y_top) / (y_bot - y_top) * n_rows) col = max(0, min(col, n_frames - 1)) row = max(0, min(row, n_rows - 1)) self.pixel_clicked.emit(row, col) class WaveformCanvas(FigureCanvasQTAgg): def __init__(self, parent=None): fig = Figure(figsize=(8, 3), tight_layout=True) self.ax_wave = fig.add_subplot(121) self.ax_right = fig.add_subplot(122) super().__init__(fig) self.setParent(parent) self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) def show_rf_waveform(self, scan: LoadedScan, angle_idx: int, row_idx: int, frame_idx: int, apply_bg_sub: bool = True, gate_start_ns: float | None = None, gate_end_ns: float | None = None): """CH1 RF: time-domain + FFT spectrum (background overlay if present).""" sras = scan.sras waveform = sras.load_row(angle_idx, row_idx, CH1_IDX)[frame_idx].astype(np.float32) t_ns = sras.time_axis_ns() f_mhz = sras.freq_axis_mhz(sras.header.samples_per_frame) dc3_val = float(sras.load_row(angle_idx, row_idx, CH3_IDX)[frame_idx] .mean(dtype=np.float32)) dc4_val = float(sras.load_row(angle_idx, row_idx, CH4_IDX)[frame_idx] .mean(dtype=np.float32)) bg = scan.background if apply_bg_sub else None waveform_plot = waveform - bg if bg is not None else waveform self.ax_wave.cla() self.ax_right.cla() if bg is not None: self.ax_wave.plot(t_ns, waveform, linewidth=0.5, color="#aaaaaa", label="raw", zorder=1) self.ax_wave.plot(t_ns, bg, linewidth=0.5, color="#e07030", linestyle="--", label="background", zorder=2) self.ax_wave.plot(t_ns, waveform_plot, linewidth=0.7, color="#4488cc", label="subtracted", zorder=3) self.ax_wave.legend(fontsize=7, loc="upper right") else: self.ax_wave.plot(t_ns, waveform, linewidth=0.7, color="#4488cc") if gate_start_ns is not None: self.ax_wave.axvline(gate_start_ns, color="#22cc44", linestyle="--", linewidth=1.0, label=f"gate start {gate_start_ns:.0f} ns") if gate_end_ns is not None: self.ax_wave.axvline(gate_end_ns, color="#cc4422", linestyle="--", linewidth=1.0, label=f"gate end {gate_end_ns:.0f} ns") if gate_start_ns is not None or gate_end_ns is not None: lo = gate_start_ns if gate_start_ns is not None else t_ns[0] hi = gate_end_ns if gate_end_ns is not None else t_ns[-1] self.ax_wave.axvspan(t_ns[0], lo, alpha=0.10, color="#cc4422") self.ax_wave.axvspan(hi, t_ns[-1], alpha=0.10, color="#cc4422") calib = scan.calib self.ax_wave.set_xlabel("Time (ns)") self.ax_wave.set_ylabel("ADC counts") bg_tag = " [bg sub]" if bg is not None else "" self.ax_wave.set_title( f"CH1 RF row={row_idx} frame={frame_idx}{bg_tag}\n" f"CH3={dc3_val:.1f} CH4={dc4_val:.1f} " f"({calib.adc_to_mv(dc3_val, CH3_IDX):.2f} / " f"{calib.adc_to_mv(dc4_val, CH4_IDX):.2f} mV)", fontsize=8, ) power_sub = power_spectrum(waveform_plot) peak_idx = int(np.argmax(power_sub)) peak_mhz = f_mhz[peak_idx] if bg is not None: power_raw = power_spectrum(waveform) self.ax_right.plot(f_mhz, power_raw, linewidth=0.5, color="#aaaaaa", label="raw FFT", zorder=1) self.ax_right.plot(f_mhz, power_sub, linewidth=0.7, color="#4488cc", label="subtracted FFT" if bg is not None else None, zorder=2) self.ax_right.axvline(peak_mhz, color="tomato", linestyle="--", linewidth=1.2, label=f"peak = {peak_mhz:.1f} MHz") self.ax_right.set_xlabel("Frequency (MHz)") self.ax_right.set_ylabel("Power (arb.)") self.ax_right.set_title("FFT Power Spectrum") self.ax_right.set_xlim(0, 500) self.ax_right.legend(fontsize=8) self.draw_idle() def show_dc_waveform(self, scan: LoadedScan, angle_idx: int, ch_idx: int, row_idx: int, frame_idx: int): """CH3 or CH4 DC: time-domain + mean annotation.""" sras = scan.sras waveform = sras.load_row(angle_idx, row_idx, ch_idx)[frame_idx].astype(np.float32) t_ns = sras.time_axis_ns() mean_val = float(waveform.mean()) mean_mv = scan.calib.adc_to_mv(mean_val, ch_idx) self.ax_wave.cla() self.ax_right.cla() self.ax_wave.plot(t_ns, waveform, linewidth=0.7, color="#4488cc") self.ax_wave.axhline(mean_val, color="tomato", linestyle="--", linewidth=1.2, label=f"mean = {mean_val:.2f} ADC") self.ax_wave.set_xlabel("Time (ns)") self.ax_wave.set_ylabel("ADC counts") self.ax_wave.set_title( f"{CH_NAMES[ch_idx]} DC row={row_idx} frame={frame_idx}" ) self.ax_wave.legend(fontsize=8) self.ax_right.text( 0.5, 0.5, f"DC mode\n\n" f"mean = {mean_val:.3f} ADC\n" f" = {mean_mv:.3f} mV", ha="center", va="center", transform=self.ax_right.transAxes, fontsize=11, ) self.ax_right.set_axis_off() self.draw_idle() # --------------------------------------------------------------------------- # SAW diagnostic window # --------------------------------------------------------------------------- class SawDiagnosticWindow(QMainWindow): """6-panel matplotlib window showing every SAW pipeline stage for one pixel. Panels: raw / EMI-gated / bandpassed / PSD / matched filter / shot-to-shot overlay. The pipeline runs (once for the pixel + up to 20 overlay shots) happen on a worker thread; only the plotting touches the GUI thread. """ _C_EMI = "#e05030" # red — EMI gate _C_NOISE = "#ccaa00" # amber — noise / inter-packet region _C_SAW = "#30c060" # green — SAW window def __init__(self, scan: LoadedScan, pipeline: SawPipeline, angle_idx: int, row_idx: int, frame_idx: int, parent=None): super().__init__(parent) self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose) self.setWindowTitle( f"SAW Diagnostics angle={angle_idx} row={row_idx} frame={frame_idx}") self.resize(1400, 940) self._scan = scan self._pipeline = pipeline self._angle_idx = angle_idx self._row_idx = row_idx self._frame_idx = frame_idx self._worker_thread: QThread | None = None central = QWidget() self.setCentralWidget(central) vl = QVBoxLayout(central) vl.setContentsMargins(4, 4, 4, 4) vl.setSpacing(4) toolbar_row = QHBoxLayout() fig = Figure(figsize=(14, 9), tight_layout=True) self._canvas = FigureCanvasQTAgg(fig) mpl_toolbar = NavigationToolbar2QT(self._canvas, central) toolbar_row.addWidget(mpl_toolbar, stretch=1) self._btn_refresh = QPushButton("Re-apply filter & refresh PSDs") self._btn_refresh.setToolTip( "Re-run the current pipeline on this pixel and redraw all panels.\n" "Use after rebuilding the template or changing pipeline parameters.") self._btn_refresh.clicked.connect(self._start_compute) toolbar_row.addWidget(self._btn_refresh) vl.addLayout(toolbar_row) vl.addWidget(self._canvas) self._start_compute() def _compute(self): """Worker-thread body: pixel result + overlay envelopes.""" sras = self._scan.sras row = sras.load_row(self._angle_idx, self._row_idx, CH1_IDX) raw_adc = row[self._frame_idx].astype(np.float32) result = self._pipeline.process_shot(raw_adc) n_frames = row.shape[0] n_overlay = min(20, n_frames) overlay_idxs = np.linspace(0, n_frames - 1, n_overlay, dtype=int) overlays = [ self._pipeline.process_shot(row[fi].astype(np.float32))["envelope"] for fi in overlay_idxs ] return raw_adc, result, overlays def _start_compute(self): if self._worker_thread is not None: return self._btn_refresh.setEnabled(False) worker = FnWorker(self._compute) thread = QThread() worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect(self._on_computed) worker.error.connect(lambda msg: self.statusBar().showMessage(f"Error: {msg}")) worker.finished.connect(thread.quit) worker.error.connect(thread.quit) thread.finished.connect(self._on_thread_finished) self._worker = worker self._worker_thread = thread thread.start() def _on_thread_finished(self): self._worker_thread = None self._worker = None self._btn_refresh.setEnabled(True) def _shade_time_zones(self, ax, t_full: np.ndarray, emi_end_ns: float, s0: float, s1: float, show_legend: bool = False): """Shade EMI gate, noise region, and SAW window on a time-domain axes.""" t0, t_end = float(t_full[0]), float(t_full[-1]) ax.axvspan(t0, emi_end_ns, alpha=0.15, color=self._C_EMI, label=f"EMI gate 0–{emi_end_ns:.0f} ns") if emi_end_ns < s0: ax.axvspan(emi_end_ns, s0, alpha=0.08, color=self._C_NOISE, label=f"noise {emi_end_ns:.0f}–{s0:.0f} ns") ax.axvspan(s0, min(s1, t_end), alpha=0.10, color=self._C_SAW, label=f"SAW {s0:.0f}–{s1:.0f} ns") if show_legend: ax.legend(fontsize=7, loc="upper right") def _on_computed(self, payload): raw_adc, result, overlays = payload sras = self._scan.sras pipeline = self._pipeline row_idx, frame_idx = self._row_idx, self._frame_idx t_full = sras.time_axis_ns() sr = result["sample_rate_hz"] t_proc = np.arange(len(result["envelope"])) / sr * 1e9 peak_amps = [float(e.max()) if len(e) > 0 else 0.0 for e in overlays] peak_var = float(np.var(peak_amps)) n_overlay = len(overlays) fig = self._canvas.figure fig.clf() axes = fig.subplots(3, 2) ax_raw, ax_gated = axes[0] ax_filt, ax_spec = axes[1] ax_mf, ax_over = axes[2] emi_end_ns = pipeline.emi_gate_ns s0, s1 = pipeline.saw_window_ns # --- 1. Raw signal --- ax_raw.plot(t_full, raw_adc, lw=0.6, color="#4488cc", zorder=3) self._shade_time_zones(ax_raw, t_full, emi_end_ns, s0, s1, show_legend=True) ax_raw.set_xlabel("Time (ns)") ax_raw.set_ylabel("ADC counts") ax_raw.set_title(f"1 — Raw signal (row={row_idx}, frame={frame_idx})") # --- 2. After EMI gating --- ax_gated.plot(t_full[:len(result["gated"])], result["gated"], lw=0.6, color="#cc8833", zorder=3) self._shade_time_zones(ax_gated, t_full, emi_end_ns, s0, s1) ax_gated.set_xlabel("Time (ns)") ax_gated.set_ylabel("Amplitude") ax_gated.set_title("2 — After EMI gating (cosine taper)") # --- 3. After bandpass --- ax_filt.plot(t_full[:len(result["filtered"])], result["filtered"], lw=0.6, color="#44aa44", zorder=3) self._shade_time_zones(ax_filt, t_full, emi_end_ns, s0, s1) ax_filt.set_xlabel("Time (ns)") ax_filt.set_ylabel("Amplitude") ax_filt.set_title( f"3 — After bandpass ({pipeline.bp_lo_mhz:.0f}–{pipeline.bp_hi_mhz:.0f} MHz, " f"6th-order Butterworth, zero-phase)") # --- 4. Frequency spectrum — raw PSD + post-bandpass PSD --- filt_sig = result["filtered"] f_mhz = np.fft.rfftfreq(len(filt_sig), d=1.0 / sras.header.sample_rate) / 1e6 spec_raw = np.abs(np.fft.rfft(raw_adc, n=len(filt_sig))) ** 2 spec_filt = np.abs(np.fft.rfft(filt_sig)) ** 2 spec_raw[0] = 0.0 spec_filt[0] = 0.0 ax_spec.plot(f_mhz, spec_raw, lw=0.5, color="#aaaaaa", alpha=0.7, label="raw PSD", zorder=1) ax_spec.plot(f_mhz, spec_filt, lw=0.8, color="#44aa44", label="bandpass PSD", zorder=2) ax_spec.axvspan(pipeline.bp_lo_mhz, pipeline.bp_hi_mhz, alpha=0.14, color=self._C_SAW, label="passband", zorder=0) ax_spec.set_xlabel("Frequency (MHz)") ax_spec.set_ylabel("Power (arb.)") ax_spec.set_title("4 — FFT PSD: raw vs. after bandpass") ax_spec.set_xlim(0, min(600.0, sras.header.sample_rate / 2e6)) ax_spec.legend(fontsize=7) # --- 5. Matched filter output + envelope --- ax_mf.plot(t_proc, result["mf_output"], lw=0.5, color="#8855cc", alpha=0.55, label="MF output", zorder=3) ax_mf.plot(t_proc, result["envelope"], lw=1.3, color="#cc4488", label="envelope (Hilbert)", zorder=4) if pipeline.template is not None: ax_mf.axvline(result["peak_time_ns"], color="#ffaa00", linestyle="--", lw=1.2, zorder=5, label=f"peak {result['peak_time_ns']:.1f} ns") self._shade_time_zones(ax_mf, t_proc, emi_end_ns, s0, s1) ax_mf.set_xlabel("Time (ns)") ax_mf.set_ylabel("Amplitude") ax_mf.set_title( f"5 — Matched filter | " f"A = {result['peak_amplitude']:.3f} | " f"SNR = {result['snr']:.1f} | " f"t = {result['peak_time_ns']:.1f} ns") ax_mf.legend(fontsize=7) # --- 6. Shot-to-shot overlay --- for env in overlays: t_ov = np.arange(len(env)) / sr * 1e9 ax_over.plot(t_ov, env, lw=0.5, alpha=0.45, color="#cc4488", zorder=3) self._shade_time_zones(ax_over, t_proc, emi_end_ns, s0, s1) ax_over.set_xlabel("Time (ns)") ax_over.set_ylabel("MF envelope amplitude") ax_over.set_title( f"6 — Shot-to-shot overlay (row {row_idx}, {n_overlay} frames) | " f"peak-amp variance = {peak_var:.4g}") fig.text( 0.5, 0.005, f"Peak amplitude: {result['peak_amplitude']:.4g} | " f"Arrival time: {result['peak_time_ns']:.2f} ns | " f"SNR: {result['snr']:.1f} | " f"Shot-to-shot peak-amp variance ({n_overlay} shots): {peak_var:.4g}", ha="center", va="bottom", fontsize=9, bbox=dict(boxstyle="round,pad=0.3", facecolor="#2a2a2a", alpha=0.85), color="#e8e8e8", ) self._canvas.draw_idle() def closeEvent(self, event): if self._worker_thread is not None: self._worker_thread.quit() self._worker_thread.wait(2000) super().closeEvent(event) # --------------------------------------------------------------------------- # Main window # --------------------------------------------------------------------------- class SrasViewerWindow(QMainWindow): def __init__(self, initial_path: str | None = None): super().__init__() self.setWindowTitle("SRAS Scan Viewer") self.resize(1560, 840) self.setAcceptDrops(True) self._scan: LoadedScan | None = None self._current_image: np.ndarray | None = None # unscaled (RF in MHz for VEL mode) self._current_angle: int = 0 self._current_ch: int = 0 self._load_thread: QThread | None = None self._compute_thread: QThread | None = None self._compute_pending = False self._progress_dlg: QProgressDialog | None = None # Debounce: rapid spinbox changes coalesce into one recompute self._recompute_timer = QTimer(self) self._recompute_timer.setSingleShot(True) self._recompute_timer.setInterval(RECOMPUTE_DEBOUNCE_MS) self._recompute_timer.timeout.connect(self._start_compute) # SAW pipeline state self._saw_pipeline: SawPipeline | None = None self._template_thread: QThread | None = None self._diag_window: SawDiagnosticWindow | None = None self._last_row: int | None = None self._last_frame: int | None = None self._build_ui() if initial_path: self._load_file(initial_path) # ------------------------------------------------------------------ # UI construction # ------------------------------------------------------------------ def _build_ui(self): central = QWidget() self.setCentralWidget(central) root = QHBoxLayout(central) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) # ---- Left control panel ---------------------------------------- panel = QWidget() panel.setFixedWidth(260) panel_layout = QVBoxLayout(panel) panel_layout.setContentsMargins(0, 0, 0, 0) panel_layout.setSpacing(6) root.addWidget(panel) # File grp_file = QGroupBox("File") fl = QVBoxLayout(grp_file) self.btn_open = QPushButton("Open .sras…") self.btn_open.clicked.connect(self._on_open) self.lbl_filename = QLabel("No file loaded") self.lbl_filename.setWordWrap(True) self.lbl_filename.setStyleSheet("color: #888; font-size: 11px;") fl.addWidget(self.btn_open) fl.addWidget(self.lbl_filename) panel_layout.addWidget(grp_file) # Scan info grp_info = QGroupBox("Scan Info") il = QVBoxLayout(grp_info) self._info = {} for key in ("Angles", "Rows", "Frames / row", "Samples / frame", "Sample rate", "X start", "Pixel Δx", "Laser freq"): lbl = QLabel(f"{key}: —") lbl.setWordWrap(True) lbl.setStyleSheet("font-size: 11px;") il.addWidget(lbl) self._info[key] = lbl # Truncation / background notes (hidden until needed) self.lbl_frame_warn = QLabel("") self.lbl_frame_warn.setWordWrap(True) self.lbl_frame_warn.setStyleSheet("color: #e07000; font-size: 11px;") il.addWidget(self.lbl_frame_warn) panel_layout.addWidget(grp_info) # View settings grp_view = QGroupBox("View Settings") vl = QVBoxLayout(grp_view) # Angle ar = QHBoxLayout() ar.addWidget(QLabel("Angle:")) self.spin_angle = QSpinBox() self.spin_angle.setRange(0, 0) self.spin_angle.setEnabled(False) self.spin_angle.valueChanged.connect(self._on_view_changed) self.lbl_angle_deg = QLabel("—") ar.addWidget(self.spin_angle) ar.addWidget(self.lbl_angle_deg) vl.addLayout(ar) # Channel cr = QHBoxLayout() cr.addWidget(QLabel("Channel:")) self.combo_channel = QComboBox() self.combo_channel.addItems(CH_LABELS) self.combo_channel.setEnabled(False) self.combo_channel.currentIndexChanged.connect(self._on_channel_changed) cr.addWidget(self.combo_channel) vl.addLayout(cr) # DC threshold (for RF / CH1 masking) sep = QFrame() sep.setFrameShape(QFrame.Shape.HLine) sep.setStyleSheet("color: #555;") vl.addWidget(sep) self.grp_threshold = QGroupBox("RF Mask Threshold (CH1 only)") tl = QVBoxLayout(self.grp_threshold) thr_row = QHBoxLayout() thr_row.addWidget(QLabel("DC threshold:")) self.spin_threshold_mv = QDoubleSpinBox() self.spin_threshold_mv.setRange(-500.0, 500.0) self.spin_threshold_mv.setDecimals(3) self.spin_threshold_mv.setSingleStep(0.025) self.spin_threshold_mv.setSuffix(" mV") self.spin_threshold_mv.setValue(50.0) self.spin_threshold_mv.setEnabled(False) self.spin_threshold_mv.valueChanged.connect(self._on_threshold_changed) thr_row.addWidget(self.spin_threshold_mv) tl.addLayout(thr_row) self.lbl_threshold_adc = QLabel("") self.lbl_threshold_adc.setStyleSheet("font-size: 11px; color: #888;") tl.addWidget(self.lbl_threshold_adc) vl.addWidget(self.grp_threshold) # Background subtraction self.chk_bg_sub = QCheckBox("Background subtraction (CH1 only)") self.chk_bg_sub.setChecked(True) self.chk_bg_sub.setEnabled(False) self.chk_bg_sub.setToolTip( "Subtract the stored background waveform from each CH1 frame\n" "before computing the FFT." ) self.chk_bg_sub.toggled.connect(self._on_bg_sub_toggled) vl.addWidget(self.chk_bg_sub) # Time gate (for FFT; CH1/velocity only) self.grp_gate = QGroupBox("Time Gate (CH1 only)") gl = QVBoxLayout(self.grp_gate) self.chk_gate = QCheckBox("Enable time gate") self.chk_gate.setChecked(False) self.chk_gate.setEnabled(False) self.chk_gate.setToolTip( "Zero-out samples outside the specified time window before\n" "computing the FFT (useful for isolating a specific acoustic packet)." ) self.chk_gate.toggled.connect(self._on_gate_toggled) gl.addWidget(self.chk_gate) gate_start_row = QHBoxLayout() gate_start_row.addWidget(QLabel("Start:")) self.spin_gate_start = QDoubleSpinBox() self.spin_gate_start.setRange(0.0, 100000.0) self.spin_gate_start.setDecimals(1) self.spin_gate_start.setSingleStep(10.0) self.spin_gate_start.setSuffix(" ns") self.spin_gate_start.setValue(50.0) self.spin_gate_start.setEnabled(False) self.spin_gate_start.valueChanged.connect(self._on_gate_changed) gate_start_row.addWidget(self.spin_gate_start) gl.addLayout(gate_start_row) gate_end_row = QHBoxLayout() gate_end_row.addWidget(QLabel("End:")) self.spin_gate_end = QDoubleSpinBox() self.spin_gate_end.setRange(0.0, 100000.0) self.spin_gate_end.setDecimals(1) self.spin_gate_end.setSingleStep(10.0) self.spin_gate_end.setSuffix(" ns") self.spin_gate_end.setValue(200.0) self.spin_gate_end.setEnabled(False) self.spin_gate_end.valueChanged.connect(self._on_gate_changed) gate_end_row.addWidget(self.spin_gate_end) gl.addLayout(gate_end_row) vl.addWidget(self.grp_gate) # Velocity settings (visible only in velocity mode) self.grp_velocity = QGroupBox("Velocity Settings (CH1 only)") vel_l = QVBoxLayout(self.grp_velocity) grat_row = QHBoxLayout() grat_row.addWidget(QLabel("Grating size:")) self.spin_grating_um = QDoubleSpinBox() self.spin_grating_um.setRange(0.1, 1000.0) self.spin_grating_um.setDecimals(2) self.spin_grating_um.setSingleStep(0.5) self.spin_grating_um.setSuffix(" µm") self.spin_grating_um.setValue(12.5) self.spin_grating_um.setEnabled(False) self.spin_grating_um.valueChanged.connect(self._on_grating_changed) grat_row.addWidget(self.spin_grating_um) vel_l.addLayout(grat_row) self.lbl_velocity_formula = QLabel("v (m/s) = freq (MHz) × grating (µm)") self.lbl_velocity_formula.setStyleSheet("font-size: 10px; color: #888;") vel_l.addWidget(self.lbl_velocity_formula) self.grp_velocity.setVisible(False) vl.addWidget(self.grp_velocity) # Export sep2 = QFrame() sep2.setFrameShape(QFrame.Shape.HLine) sep2.setStyleSheet("color: #555;") vl.addWidget(sep2) self.btn_export_csv = QPushButton("Export Image as CSV…") self.btn_export_csv.setEnabled(False) self.btn_export_csv.setToolTip( "Save the current CH1 image (one scan row per CSV line)." ) self.btn_export_csv.clicked.connect(self._on_export_csv) vl.addWidget(self.btn_export_csv) panel_layout.addWidget(grp_view) # ---- SAW Pipeline panel ---------------------------------------- grp_saw = QGroupBox("SAW Pipeline (CH1 only)") sl = QVBoxLayout(grp_saw) emi_row = QHBoxLayout() emi_row.addWidget(QLabel("EMI gate end:")) self.spin_saw_emi_ns = QDoubleSpinBox() self.spin_saw_emi_ns.setRange(1.0, 10000.0) self.spin_saw_emi_ns.setDecimals(1) self.spin_saw_emi_ns.setSingleStep(5.0) self.spin_saw_emi_ns.setSuffix(" ns") self.spin_saw_emi_ns.setValue(50.0) emi_row.addWidget(self.spin_saw_emi_ns) sl.addLayout(emi_row) sl.addWidget(QLabel("SAW window (ns):")) saw_win_row = QHBoxLayout() self.spin_saw_win_start = QDoubleSpinBox() self.spin_saw_win_start.setRange(0.0, 100000.0) self.spin_saw_win_start.setDecimals(1) self.spin_saw_win_start.setSuffix(" ns") self.spin_saw_win_start.setValue(80.0) saw_win_row.addWidget(self.spin_saw_win_start) saw_win_row.addWidget(QLabel("–")) self.spin_saw_win_end = QDoubleSpinBox() self.spin_saw_win_end.setRange(0.0, 100000.0) self.spin_saw_win_end.setDecimals(1) self.spin_saw_win_end.setSuffix(" ns") self.spin_saw_win_end.setValue(350.0) saw_win_row.addWidget(self.spin_saw_win_end) sl.addLayout(saw_win_row) sl.addWidget(QLabel("Bandpass (MHz):")) bp_row = QHBoxLayout() self.spin_saw_bp_lo = QDoubleSpinBox() self.spin_saw_bp_lo.setRange(1.0, 3000.0) self.spin_saw_bp_lo.setDecimals(1) self.spin_saw_bp_lo.setSuffix(" MHz") self.spin_saw_bp_lo.setValue(85.0) bp_row.addWidget(self.spin_saw_bp_lo) bp_row.addWidget(QLabel("–")) self.spin_saw_bp_hi = QDoubleSpinBox() self.spin_saw_bp_hi.setRange(1.0, 3000.0) self.spin_saw_bp_hi.setDecimals(1) self.spin_saw_bp_hi.setSuffix(" MHz") self.spin_saw_bp_hi.setValue(200.0) bp_row.addWidget(self.spin_saw_bp_hi) sl.addLayout(bp_row) tmpl_row = QHBoxLayout() tmpl_row.addWidget(QLabel("Template shots:")) self.spin_saw_n_shots = QSpinBox() self.spin_saw_n_shots.setRange(1, 10000) self.spin_saw_n_shots.setValue(50) tmpl_row.addWidget(self.spin_saw_n_shots) sl.addLayout(tmpl_row) self.btn_build_template = QPushButton("Build Template") self.btn_build_template.setEnabled(False) self.btn_build_template.setToolTip( "Average N shots (EMI-gated + bandpass-filtered) to form a\n" "Hann-windowed template for the matched filter.") self.btn_build_template.clicked.connect(self._on_build_template_clicked) sl.addWidget(self.btn_build_template) self.lbl_saw_status = QLabel("No template") self.lbl_saw_status.setStyleSheet("font-size: 11px; color: #888;") self.lbl_saw_status.setWordWrap(True) sl.addWidget(self.lbl_saw_status) sep_mf = QFrame() sep_mf.setFrameShape(QFrame.Shape.HLine) sep_mf.setStyleSheet("color: #555;") sl.addWidget(sep_mf) sl.addWidget(QLabel("Apply matched filter:")) mf_mode_row = QHBoxLayout() mf_mode_row.addWidget(QLabel("Output:")) self.combo_mf_mode = QComboBox() self.combo_mf_mode.addItems(["Amplitude", "Time-of-Flight"]) self.combo_mf_mode.setEnabled(False) mf_mode_row.addWidget(self.combo_mf_mode) sl.addLayout(mf_mode_row) self.btn_apply_mf = QPushButton("Apply Filter → Image") self.btn_apply_mf.setEnabled(False) self.btn_apply_mf.setToolTip( "Switch to the SAW matched-filter channel and compute the image.\n" "Requires a template to be built first.") self.btn_apply_mf.clicked.connect(self._on_apply_mf_clicked) sl.addWidget(self.btn_apply_mf) panel_layout.addStretch() # Colormap sep3 = QFrame() sep3.setFrameShape(QFrame.Shape.HLine) sep3.setStyleSheet("color: #555;") vl.addWidget(sep3) cmr = QHBoxLayout() cmr.addWidget(QLabel("Colormap:")) self.combo_cmap = QComboBox() self.combo_cmap.addItems(CMAPS) self.combo_cmap.setCurrentText("gray") self.combo_cmap.setEnabled(False) self.combo_cmap.currentIndexChanged.connect(self._on_display_changed) cmr.addWidget(self.combo_cmap) vl.addLayout(cmr) # Auto-scale self.chk_auto = QCheckBox("Auto-scale colormap") self.chk_auto.setChecked(True) self.chk_auto.toggled.connect(self._on_autoscale_toggled) vl.addWidget(self.chk_auto) for label, attr in (("min:", "spin_vmin"), ("max:", "spin_vmax")): row = QHBoxLayout() row.addWidget(QLabel(label)) spin = QDoubleSpinBox() spin.setRange(-1e9, 1e9) spin.setDecimals(4) spin.setEnabled(False) spin.valueChanged.connect(self._on_manual_range_changed) setattr(self, attr, spin) row.addWidget(spin) vl.addLayout(row) # ---- Right: image + waveform splitter -------------------------- splitter = QSplitter(Qt.Orientation.Vertical) root.addWidget(splitter, stretch=1) img_widget = QWidget() img_vl = QVBoxLayout(img_widget) img_vl.setContentsMargins(0, 0, 0, 0) self.image_canvas = ImageCanvas() self.image_canvas.pixel_clicked.connect(self._on_pixel_clicked) toolbar = NavigationToolbar2QT(self.image_canvas, img_widget) img_vl.addWidget(toolbar) img_vl.addWidget(self.image_canvas) splitter.addWidget(img_widget) wave_widget = QWidget() wave_vl = QVBoxLayout(wave_widget) wave_vl.setContentsMargins(0, 0, 0, 0) self.lbl_wave_hint = QLabel( "Click a pixel in the image above to inspect its waveform." ) self.lbl_wave_hint.setAlignment(Qt.AlignmentFlag.AlignCenter) self.lbl_wave_hint.setStyleSheet("color: #888; font-size: 11px;") self.wave_canvas = WaveformCanvas() self.btn_saw_diag = QPushButton("Open SAW Diagnostics…") self.btn_saw_diag.setEnabled(False) self.btn_saw_diag.setToolTip( "Show the 6-panel SAW pipeline diagnostic for the clicked pixel.\n" "Requires a SAW template to be built first.") self.btn_saw_diag.clicked.connect(self._on_open_diagnostics) wave_vl.addWidget(self.lbl_wave_hint) wave_vl.addWidget(self.btn_saw_diag) wave_vl.addWidget(self.wave_canvas) splitter.addWidget(wave_widget) splitter.setSizes([580, 250]) # ---- Right control panel (SAW pipeline) ------------------------ right_panel = QWidget() right_panel.setFixedWidth(260) right_panel_layout = QVBoxLayout(right_panel) right_panel_layout.setContentsMargins(0, 0, 0, 0) right_panel_layout.setSpacing(6) right_panel_layout.addWidget(grp_saw) right_panel_layout.addStretch() root.addWidget(right_panel) self.statusBar().showMessage("Open an .sras file to begin.") self._update_threshold_label() # ------------------------------------------------------------------ # Drag-and-drop # ------------------------------------------------------------------ def dragEnterEvent(self, event): urls = event.mimeData().urls() if urls and urls[0].toLocalFile().lower().endswith(".sras"): event.acceptProposedAction() def dropEvent(self, event): self._load_file(event.mimeData().urls()[0].toLocalFile()) # ------------------------------------------------------------------ # File loading # ------------------------------------------------------------------ def _on_open(self): path, _ = QFileDialog.getOpenFileName( self, "Open SRAS File", "", "SRAS Files (*.sras);;All Files (*)" ) if path: self._load_file(path) def _load_file(self, path: str): if self._load_thread is not None: self.statusBar().showMessage("Still loading previous file…") return self.btn_open.setEnabled(False) self.statusBar().showMessage(f"Loading {Path(path).name}…") self._show_progress(f"Loading {Path(path).name}…") worker = FnWorker(lambda: LoadedScan(path)) thread = QThread() worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect(self._on_load_done) worker.error.connect(self._on_load_error) worker.finished.connect(thread.quit) worker.error.connect(thread.quit) thread.finished.connect(lambda: setattr(self, "_load_thread", None)) self._load_worker = worker self._load_thread = thread thread.start() def _on_load_error(self, msg: str): self._close_progress() self.btn_open.setEnabled(True) self.statusBar().showMessage(f"Error: {msg}") def _on_load_done(self, scan: LoadedScan): self._close_progress() self.btn_open.setEnabled(True) # Drop every reference to the previous file so its mapping (and any # multi-GB views derived from it) can actually be freed. if self._diag_window is not None: self._diag_window.close() self._diag_window = None self._load_worker = None self._current_image = None self._last_row = None self._last_frame = None old = self._scan self._scan = scan if old is not None: old.close() sras = scan.sras h = sras.header self.lbl_filename.setText(sras.path.name) self._info["Angles"].setText(f"Angles: {h.n_angles}") self._info["Samples / frame"].setText(f"Samples / frame: {h.samples_per_frame}") self._info["Sample rate"].setText(f"Sample rate: {h.sample_rate/1e9:.4g} GS/s") self._info["Pixel Δx"].setText(f"Pixel Δx: {sras.pixel_pitch_x_mm()*1e3:.3g} µm") self._info["Laser freq"].setText(f"Laser freq: {h.laser_freq/1e3:.4g} kHz") self.spin_angle.blockSignals(True) self.spin_angle.setRange(0, max(0, h.n_angles - 1)) self.spin_angle.setValue(0) self.spin_angle.blockSignals(False) self._update_controls_enabled(True) self._update_threshold_label() self._on_view_changed() def _update_angle_info(self): """Per-angle info fields (v6 geometry is ragged across angles).""" scan = self._scan if scan is None: return ai = self.spin_angle.value() pa = scan.sras.per_angle[ai] avail = scan.rows_available[ai] rows_txt = f"Rows: {pa.n_rows}" if avail == pa.n_rows \ else f"Rows: {avail} of {pa.n_rows} on disk" self._info["Rows"].setText(rows_txt) self._info["Frames / row"].setText(f"Frames / row: {pa.n_frames}") self._info["X start"].setText(f"X start: {pa.x_start:.4g} mm") notes = [] if avail < pa.n_rows: notes.append(f"! Angle {ai}: only {avail}/{pa.n_rows} rows on disk " "(scan was aborted or is still running)") if scan.background is not None: notes.append(f"Background waveform: {len(scan.background)} samples") self.lbl_frame_warn.setText("\n".join(notes)) # ------------------------------------------------------------------ # Controls # ------------------------------------------------------------------ def _update_controls_enabled(self, enabled: bool): scan = self._scan has_file = enabled and scan is not None self.spin_angle.setEnabled(has_file and scan.sras.header.n_angles > 1) self.combo_channel.setEnabled(enabled) self.combo_cmap.setEnabled(enabled) self.chk_auto.setEnabled(enabled) manual = enabled and not self.chk_auto.isChecked() self.spin_vmin.setEnabled(manual) self.spin_vmax.setEnabled(manual) ch_idx = self.combo_channel.currentIndex() is_ch1 = enabled and ch_idx in CH1_DERIVED_MODES is_fft = enabled and ch_idx in (CH1_IDX, VELOCITY_MODE_IDX) self.spin_threshold_mv.setEnabled(is_ch1) has_bg = has_file and scan.background is not None self.chk_bg_sub.setEnabled(has_bg and is_ch1) # Time gate only for FFT modes (the SAW pipeline has its own gating) self.chk_gate.setEnabled(is_fft) gate_active = is_fft and self.chk_gate.isChecked() self.spin_gate_start.setEnabled(gate_active) self.spin_gate_end.setEnabled(gate_active) is_vel = enabled and ch_idx == VELOCITY_MODE_IDX self.spin_grating_um.setEnabled(is_vel) self.grp_velocity.setVisible(is_vel) self.btn_build_template.setEnabled(has_file) has_template = (self._saw_pipeline is not None and self._saw_pipeline.template is not None) has_pixel = self._last_row is not None self.btn_saw_diag.setEnabled(has_file and has_template and has_pixel) self.btn_apply_mf.setEnabled(has_file and has_template) self.combo_mf_mode.setEnabled(has_file and has_template) self.btn_export_csv.setEnabled(is_ch1 and self._current_image is not None) def _on_channel_changed(self): self._update_controls_enabled(True) self._request_compute() def _on_bg_sub_toggled(self): if self._scan is not None and self.combo_channel.currentIndex() in CH1_DERIVED_MODES: self._request_compute() def _on_gate_toggled(self, checked: bool): self.spin_gate_start.setEnabled(checked) self.spin_gate_end.setEnabled(checked) if self._scan is not None and \ self.combo_channel.currentIndex() in (CH1_IDX, VELOCITY_MODE_IDX): self._request_compute() def _on_gate_changed(self): if self._scan is not None and self.chk_gate.isChecked() and \ self.combo_channel.currentIndex() in (CH1_IDX, VELOCITY_MODE_IDX): self._request_compute() def _on_grating_changed(self): # Velocity = RF (MHz) × grating (µm): a display-time scalar multiply, # so no recompute — just redraw. if self._scan is not None and self._current_image is not None \ and self._current_ch == VELOCITY_MODE_IDX: self._redraw_image(self._current_image) def _on_export_csv(self): if self._current_image is None or self._scan is None: return ch_idx = self._current_ch angle = self._current_angle ch_name = CH_NAMES[ch_idx] stem = self._scan.sras.path.stem default_name = f"{stem}_angle{angle}_{ch_name}.csv" path, _ = QFileDialog.getSaveFileName( self, "Export Image as CSV", str(self._scan.sras.path.parent / default_name), "CSV files (*.csv);;All files (*)", ) if not path: return img = self._display_image(self._current_image) worker = FnWorker(lambda: np.savetxt(path, img, delimiter=",", fmt="%.6g")) thread = QThread() worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect( lambda _: self.statusBar().showMessage(f"Exported {Path(path).name}")) worker.error.connect( lambda msg: self.statusBar().showMessage(f"Export failed: {msg}")) worker.finished.connect(thread.quit) worker.error.connect(thread.quit) thread.finished.connect(lambda: setattr(self, "_csv_thread", None)) self._csv_worker = worker self._csv_thread = thread thread.start() def _update_threshold_label(self): mv = self.spin_threshold_mv.value() if self._scan is not None: adc = self._scan.calib.mv_to_adc(mv, CH4_IDX) else: adc = ChannelCalibration.from_preambles([""] * 3).mv_to_adc(mv, CH4_IDX) self.lbl_threshold_adc.setText(f"≈ {adc:.1f} ADC counts") def _on_threshold_changed(self, mv: float): self._update_threshold_label() if self._scan is not None and self.combo_channel.currentIndex() in CH1_DERIVED_MODES: self._request_compute() def _on_autoscale_toggled(self, checked: bool): manual = not checked self.spin_vmin.setEnabled(manual and self._scan is not None) self.spin_vmax.setEnabled(manual and self._scan is not None) self._on_display_changed() def _on_manual_range_changed(self): if not self.chk_auto.isChecked(): self._on_display_changed() def _on_display_changed(self): """Colormap/scale changes: redraw only, no recompute.""" if self._scan is not None and self._current_image is not None: self._redraw_image(self._current_image) def _on_view_changed(self): if self._scan is None: return idx = self.spin_angle.value() self.lbl_angle_deg.setText(f"({self._scan.sras.per_angle[idx].angle_deg:.1f}°)") self._update_angle_info() self._request_compute() # ------------------------------------------------------------------ # Computation # ------------------------------------------------------------------ def _request_compute(self): """Debounced entry point — rapid widget changes coalesce into one run.""" if self._scan is None: return self._recompute_timer.start() def _start_compute(self): if self._scan is None: return if self._compute_thread is not None: # A run is in flight; run again when it finishes. self._compute_pending = True return scan = self._scan angle_idx = self.spin_angle.value() ch_idx = self.combo_channel.currentIndex() threshold_mv = self.spin_threshold_mv.value() apply_bg_sub = self.chk_bg_sub.isChecked() gate_enabled = self.chk_gate.isChecked() gate_start = self.spin_gate_start.value() if gate_enabled else None gate_end = self.spin_gate_end.value() if gate_enabled else None pipeline = self._saw_pipeline self._computing_angle = angle_idx self._computing_ch = ch_idx self.statusBar().showMessage("Computing image…") self._show_progress("Computing image…") worker = FnWorker(lambda: compute_image( scan, angle_idx, ch_idx, threshold_mv, apply_bg_sub, gate_start, gate_end, pipeline)) thread = QThread() worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect(self._on_compute_done) worker.error.connect(self._on_compute_error) worker.finished.connect(thread.quit) worker.error.connect(thread.quit) thread.finished.connect(self._on_compute_thread_finished) self._compute_worker = worker self._compute_thread = thread thread.start() def _on_compute_thread_finished(self): self._compute_thread = None self._compute_worker = None if self._compute_pending: self._compute_pending = False self._start_compute() def _on_compute_error(self, msg: str): self._close_progress() self.statusBar().showMessage(f"Compute error: {msg}") def _on_compute_done(self, img: np.ndarray): self._close_progress() self._current_image = img self._current_angle = self._computing_angle self._current_ch = self._computing_ch self.btn_export_csv.setEnabled(self._current_ch in CH1_DERIVED_MODES) self._redraw_image(img) # ------------------------------------------------------------------ # Drawing # ------------------------------------------------------------------ def _display_image(self, img: np.ndarray) -> np.ndarray: """The image as displayed (velocity mode scales RF MHz by grating µm).""" if self._current_ch == VELOCITY_MODE_IDX: return img * self.spin_grating_um.value() return img def _redraw_image(self, img: np.ndarray): scan = self._scan sras = scan.sras ai = self._current_angle pa = sras.per_angle[ai] if img.size == 0: self.statusBar().showMessage( f"{sras.path.name} | angle {ai}: no data on disk") return img = self._display_image(img) x_axis = sras.x_axis_mm(ai) y_axis = np.asarray(pa.y_positions[:img.shape[0]]) dx = x_axis[1] - x_axis[0] if len(x_axis) > 1 else sras.pixel_pitch_x_mm() dy = float(y_axis[1] - y_axis[0]) if len(y_axis) > 1 else 1.0 extent = [ x_axis[0] - dx / 2, x_axis[-1] + dx / 2, y_axis[-1] + dy / 2, y_axis[0] - dy / 2, ] if self.chk_auto.isChecked(): vmin, vmax = float(img.min()), float(img.max()) for spin, val in ((self.spin_vmin, vmin), (self.spin_vmax, vmax)): spin.blockSignals(True) spin.setValue(val) spin.blockSignals(False) else: vmin = self.spin_vmin.value() vmax = self.spin_vmax.value() ch_idx = self._current_ch angle_deg = pa.angle_deg mode_str, unit, colorbar_label = MODE_INFO[ch_idx] ch_label = CH_LABELS[ch_idx] if ch_idx == VELOCITY_MODE_IDX: ch_label = f"Velocity [grating={self.spin_grating_um.value():.2f} µm]" title = f"{CH_NAMES[ch_idx]} | {mode_str} | {angle_deg:.1f}°" self.image_canvas.show_image( img, extent, cmap=self.combo_cmap.currentText(), vmin=vmin, vmax=vmax, xlabel="X (mm)", ylabel="Y (mm)", title=title, colorbar_label=colorbar_label, ) self.statusBar().showMessage( f"{sras.path.name} | {ch_label} @ {angle_deg:.1f}° " f"| {img.shape[1]} × {img.shape[0]} px | {unit}" ) # ------------------------------------------------------------------ # Pixel inspector # ------------------------------------------------------------------ def _on_pixel_clicked(self, row_idx: int, frame_idx: int): if self._scan is None or self._current_image is None: return self._last_row = row_idx self._last_frame = frame_idx self.lbl_wave_hint.hide() ch_idx = self._current_ch if ch_idx in CH1_DERIVED_MODES: gate_enabled = self.chk_gate.isChecked() and ch_idx in (CH1_IDX, VELOCITY_MODE_IDX) self.wave_canvas.show_rf_waveform( self._scan, self._current_angle, row_idx, frame_idx, apply_bg_sub=self.chk_bg_sub.isChecked(), gate_start_ns=self.spin_gate_start.value() if gate_enabled else None, gate_end_ns=self.spin_gate_end.value() if gate_enabled else None, ) else: self.wave_canvas.show_dc_waveform( self._scan, self._current_angle, ch_idx, row_idx, frame_idx ) has_template = (self._saw_pipeline is not None and self._saw_pipeline.template is not None) self.btn_saw_diag.setEnabled(has_template) # ------------------------------------------------------------------ # SAW pipeline management # ------------------------------------------------------------------ def _on_build_template_clicked(self): if self._scan is None or self._template_thread is not None: return scan = self._scan self._saw_pipeline = SawPipeline( sample_rate_hz=scan.sras.header.sample_rate, emi_gate_ns=self.spin_saw_emi_ns.value(), bp_lo_mhz=self.spin_saw_bp_lo.value(), bp_hi_mhz=self.spin_saw_bp_hi.value(), saw_window_ns=(self.spin_saw_win_start.value(), self.spin_saw_win_end.value()), ) pipeline = self._saw_pipeline angle_idx = self.spin_angle.value() n_shots_req = self.spin_saw_n_shots.value() row_sel, frame_sel = self._last_row, self._last_frame apply_bg = scan.background is not None and self.chk_bg_sub.isChecked() if row_sel is not None and frame_sel is not None: src_desc = f"selected pixel (row={row_sel}, frame={frame_sel})" else: src_desc = f"{n_shots_req} shots (full scan)" self._template_src_desc = src_desc def _build(): view = scan.angle_view(angle_idx) n_rows, _, n_frames, _ = view.shape if n_rows == 0: raise RuntimeError("This angle has no data on disk.") if row_sel is not None and frame_sel is not None: shots = view[row_sel, CH1_IDX, frame_sel:frame_sel + 1].astype(np.float32) else: # Fancy-index only the selected shots — never materialize the # whole angle in float32. n = min(n_shots_req, n_rows * n_frames) flat = np.linspace(0, n_rows * n_frames - 1, n, dtype=int) rows, frames = np.divmod(flat, n_frames) shots = view[rows, CH1_IDX, frames].astype(np.float32) if apply_bg: shots -= scan.background pipeline.build_template(shots) return None self.btn_build_template.setEnabled(False) self.lbl_saw_status.setText(f"Building template from {src_desc}…") self._show_progress("Building SAW template…") worker = FnWorker(_build) thread = QThread() worker.moveToThread(thread) thread.started.connect(worker.run) worker.finished.connect(self._on_template_built) worker.error.connect(self._on_template_error) worker.finished.connect(thread.quit) worker.error.connect(thread.quit) thread.finished.connect(self._on_template_thread_finished) self._template_worker = worker self._template_thread = thread thread.start() def _on_template_thread_finished(self): self._template_thread = None self._template_worker = None def _on_template_built(self, _result=None): self._close_progress() self.btn_build_template.setEnabled(True) self.lbl_saw_status.setText( f"Template ready ({self._template_src_desc})\n" f"EMI gate: {self.spin_saw_emi_ns.value():.0f} ns " f"BP: {self.spin_saw_bp_lo.value():.0f}–{self.spin_saw_bp_hi.value():.0f} MHz") self.lbl_saw_status.setStyleSheet("font-size: 11px; color: #44cc66;") self.btn_saw_diag.setEnabled(self._last_row is not None) self.btn_apply_mf.setEnabled(True) self.combo_mf_mode.setEnabled(True) def _on_template_error(self, msg: str): self._close_progress() self.btn_build_template.setEnabled(True) self.lbl_saw_status.setText(f"Error: {msg}") self.lbl_saw_status.setStyleSheet("font-size: 11px; color: #e05030;") def _on_open_diagnostics(self): if (self._scan is None or self._saw_pipeline is None or self._last_row is None): return if self._diag_window is not None: try: self._diag_window.close() except RuntimeError: pass # C++ object already deleted (user closed the window) self._diag_window = None self._diag_window = SawDiagnosticWindow( self._scan, self._saw_pipeline, self._current_angle, self._last_row, self._last_frame, parent=None, # free-floating window ) # Clear our reference when the user closes the window so we never # call into a deleted C++ object again. self._diag_window.destroyed.connect( lambda: setattr(self, "_diag_window", None)) self._diag_window.show() def _on_apply_mf_clicked(self): if self._scan is None or self._saw_pipeline is None: return if self._saw_pipeline.template is None: self.statusBar().showMessage( "No template built yet — use 'Build Template' first.") return target_ch = (SAW_MODE_AMP_IDX if self.combo_mf_mode.currentIndex() == 0 else SAW_MODE_TOF_IDX) self.combo_channel.blockSignals(True) self.combo_channel.setCurrentIndex(target_ch) self.combo_channel.blockSignals(False) self._on_channel_changed() # ------------------------------------------------------------------ # Progress dialog helpers # ------------------------------------------------------------------ def _show_progress(self, message: str): if self._progress_dlg is not None: self._progress_dlg.setLabelText(message) return dlg = QProgressDialog(message, "", 0, 0, self) dlg.setWindowTitle("Please wait…") dlg.setCancelButton(None) dlg.setWindowModality(Qt.WindowModality.WindowModal) dlg.setMinimumDuration(300) # only appears if operation takes > 300 ms dlg.show() self._progress_dlg = dlg def _close_progress(self): if self._progress_dlg is not None: self._progress_dlg.close() self._progress_dlg = None # ------------------------------------------------------------------ def closeEvent(self, event): for attr in ("_load_thread", "_compute_thread", "_template_thread", "_csv_thread"): t = getattr(self, attr, None) if t is not None: t.quit() t.wait(2000) if self._diag_window is not None: try: self._diag_window.close() except RuntimeError: pass super().closeEvent(event) # --------------------------------------------------------------------------- def main(): app = QApplication(sys.argv) initial = sys.argv[1] if len(sys.argv) > 1 else None window = SrasViewerWindow(initial_path=initial) window.show() sys.exit(app.exec()) if __name__ == "__main__": main()