tka/scanengine-3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added all the other shit

Browse Source
This commit is contained in:
Thomas Ales [M S E]
2026-06-14 18:54:03 -05:00
parent 002e9ace13
commit a465a3dd72
2 changed files with 142 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sc3_aui_app.py
+142 -12
View File
@@ -15,13 +15,17 @@ from pathlib import Path
import threading
from threading import Thread
import numpy as np
import serial
from PyQt6 import uic
from PyQt6.QtCore import QObject, QThread, QTimer, Qt, pyqtSignal, pyqtSlot
from PyQt6.QtGui import QImage, QPixmap
from PyQt6.QtGui import QImage, QPixmap, QPainter, QColor, QPen
from PyQt6.QtWidgets import (
QApplication, QFileDialog, QLabel, QMainWindow, QMessageBox, QWidget,
QApplication, QFileDialog, QLabel, QMainWindow, QMessageBox, QSizePolicy,
QWidget,
)
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
ROOT = Path(__file__).parent
sys.path.insert(0, str(ROOT))
@@ -95,6 +99,54 @@ T1T2T3_JOG_STEPS = 500 # steps per manual jog press for T1/T2/T3
_T3R_AXIS_NAMES = {1: "T1", 2: "T2", 3: "T3", 4: "GR"}
BBD_DEFAULT_JOG_MM = 0.5 # default jog step for BBD202
class DCBiasImageWidget(FigureCanvas):
"""Live 2D DC-bias image: rows × frames, matching sras_viewer's compute_dc_image view."""
def __init__(self, parent=None):
fig = Figure(tight_layout=True)
super().__init__(fig)
self.setParent(parent)
self.setMinimumSize(300, 150)
self.setSizePolicy(
QSizePolicy.Policy.Expanding,
QSizePolicy.Policy.Expanding,
)
self.ax = fig.add_subplot(111)
self._row_data: list[np.ndarray] = []
self._im = None
self.ax.set_title("DC Bias Preview (ADC counts)")
self.ax.set_xlabel("Frame")
self.ax.set_ylabel("Row")
self.ax.text(0.5, 0.5, "Waiting for data…",
transform=self.ax.transAxes, ha="center", va="center", color="gray")
self.draw()
def add_row(self, row_index: int, values):
"""Store per-frame DC mean values for row_index (1-based) and refresh the image."""
idx = row_index - 1
while len(self._row_data) <= idx:
self._row_data.append(np.empty(0, dtype=np.float32))
self._row_data[idx] = np.asarray(values, dtype=np.float32)
# Build rectangular image — pad shorter rows with NaN so they render distinctly
max_len = max(len(r) for r in self._row_data)
img = np.full((len(self._row_data), max_len), np.nan, dtype=np.float32)
for i, row in enumerate(self._row_data):
img[i, :len(row)] = row
self.ax.clear()
self.ax.set_title("DC Bias Preview (ADC counts)")
self.ax.set_xlabel("Frame")
self.ax.set_ylabel("Row")
self.ax.imshow(
img, aspect="auto", origin="upper",
cmap="viridis", interpolation="nearest",
)
self.draw_idle()
BBD_JOG_SPEED_MM_S = 10.0
BBD_JOG_ACCEL_MM_S2 = 50.0
@@ -108,6 +160,17 @@ SCAN_CHANNELS = [1, 3, 4] # oscilloscope channels recorded, in order
BLOB_HDR_FMT = ">4sBHHffffIIdBB"
def _format_eta(secs: float) -> str:
secs = max(0.0, secs)
m, s = divmod(int(secs), 60)
h, m = divmod(m, 60)
if h > 0:
return f"{h}h {m:02d}m"
if m > 0:
return f"{m}m {s:02d}s"
return f"{s}s"
def _open_scan_file(path: Path, n_angles: int, n_rows: int,
x_start: float, x_delta: float,
velocity: float, laser_freq: float,
@@ -994,15 +1057,63 @@ class ScanProgressWindow(QWidget):
uic.loadUi(ROOT / "sc3-aui-scanprogress.ui", self)
self.setWindowTitle("Scan Progress")
self.abort_btn.clicked.connect(self.abort_requested.emit)
self.bias_image_widget = DCBiasImageWidget()
insert_pos = self.verticalLayout.count() - 1
self.verticalLayout.insertWidget(insert_pos, QLabel("DC Bias Preview"))
self.verticalLayout.insertWidget(insert_pos + 1, self.bias_image_widget)
self.verticalLayout.setStretch(insert_pos + 1, 1)
self._row_start_time: float | None = None
self._row_durations: list[float] = []
self._last_angle_idx: int = -1
def on_row_started(self, row: int, n_rows: int, angle_idx: int, n_angles: int):
self._row_start_time = time.monotonic()
def update_progress(self, row: int, n_rows: int, angle_idx: int, n_angles: int):
n_rows = max(1, n_rows)
n_angles = max(1, n_angles)
self.current_scan_current_row_indicator.setText(f"Row {row} of {n_rows}")
pct_row = round(row / n_rows * 100) if n_rows > 0 else 0
self.current_scan_progbar.setValue(pct_row)
self.overall_scan_current_angle_indicator.setText(f"Angle {angle_idx} of {n_angles}")
pct_ang = round(angle_idx / n_angles * 100) if n_angles > 0 else 0
if row == 0:
self._row_durations = []
self._row_start_time = None
self._last_angle_idx = -1
self.current_scan_progbar.setMaximum(n_rows)
self.current_scan_progbar.setValue(0)
self.current_scan_progbar.setFormat("Row %v/%m")
else:
# Reset duration history when the angle changes
if angle_idx != self._last_angle_idx and self._last_angle_idx != -1:
self._row_durations = []
self._last_angle_idx = angle_idx
if self._row_start_time is not None:
self._row_durations.append(time.monotonic() - self._row_start_time)
self._row_start_time = None
self.current_scan_progbar.setMaximum(n_rows)
self.current_scan_progbar.setValue(row)
rows_left = n_rows - row
if self._row_durations and rows_left > 0:
recent = self._row_durations[-5:]
avg = sum(recent) / len(recent)
eta_str = _format_eta(avg * rows_left)
self.current_scan_progbar.setFormat(f"Row %v/%m — ETA: {eta_str}")
elif rows_left == 0:
self.current_scan_progbar.setFormat("Row %v/%m — Done")
else:
self.current_scan_progbar.setFormat("Row %v/%m")
pct_ang = round(angle_idx / n_angles * 100)
self.overall_scan_progbar.setValue(pct_ang)
def update_dc_bias_plot(self, row: int, values):
self.bias_image_widget.add_row(row, values)
# ── Scan worker ───────────────────────────────────────────────────────────────
@@ -1013,12 +1124,14 @@ class ScanWorker(QObject):
make blocking calls. BBD202Worker.scanning_active is set to True for the
duration to suppress position polling in the BBD worker thread.
"""
started = pyqtSignal()
completed = pyqtSignal()
failed = pyqtSignal(str)
row_done = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles
status_msg = pyqtSignal(str)
user_prompt = pyqtSignal(str, str) # title, message — ask user to acknowledge
started = pyqtSignal()
completed = pyqtSignal()
dc_bias_updated = pyqtSignal(int, object) # row index, list[float] per-frame DC means
failed = pyqtSignal(str)
row_started = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles
row_done = pyqtSignal(int, int, int, int) # row, n_rows, angle_idx, n_angles
status_msg = pyqtSignal(str)
user_prompt = pyqtSignal(str, str) # title, message
def __init__(self, bbd: BBD202Worker, t3r: T3RWorker,
oscope: OscopeWorker, params: dict):
@@ -1268,6 +1381,7 @@ class ScanWorker(QObject):
if self._abort:
break
self.row_started.emit(ri + 1, n_rows, ai + 1, n_angles)
self.status_msg.emit(
f"Angle {ai+1}/{n_angles} Row {ri+1}/{n_rows} "
f"(Y={y_pos:.3f} mm)"
@@ -1302,6 +1416,20 @@ class ScanWorker(QObject):
n_frames = int(scope.query("ACQuire:NUMFRAMESACQuired?"))
for _ in range(n_frames):
scan_file.write(zero_frame)
elif ch == 4:
self.status_msg.emit(f"Fetching CH{ch} data …")
scope.set_data_source(ch)
waveforms: list[bytes] = scope.transfer_fastframe(parse=False)
if waveforms:
frame_avgs = []
for w in waveforms:
n = len(w)
frame_avgs.append(
sum(struct.unpack(f"{n}b", w)) / n if n > 0 else 0.0
)
self.dc_bias_updated.emit(ri + 1, frame_avgs)
for w in waveforms:
scan_file.write(w)
else:
self.status_msg.emit(f"Fetching CH{ch} data …")
scope.set_data_source(ch)
@@ -1729,7 +1857,9 @@ class MainWindow(QMainWindow):
)
self._scan_worker.moveToThread(self._scan_thread)
self._scan_thread.started.connect(self._scan_worker.run)
self._scan_worker.row_started.connect(self._scan_progress.on_row_started)
self._scan_worker.row_done.connect(self._on_row_done)
self._scan_worker.dc_bias_updated.connect(self._scan_progress.update_dc_bias_plot)
self._scan_worker.completed.connect(self._on_scan_complete)
self._scan_worker.failed.connect(self._on_scan_failed)
self._scan_worker.status_msg.connect(lambda m: print(f"[SCAN] {m}"))
@@ -1757,7 +1887,7 @@ class MainWindow(QMainWindow):
x_delta = _f(self.x_delta_edit, "XD")
y_delta = _f(self.y_delta_edit, "YD")
num_angles = _i(self.num_angles_edit, "NumAngles")
row_spacing = _f(self.row_spacing_edit,"RowSpacing")
row_spacing = _f(self.row_spacing_edit, "RowSpacing")
prefix = self.scan_prefix_edit.text().strip() or "scan"
save_dir = self.scan_save_dir_edit.text().strip() or DEFAULT_SAVE_DIR