Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 23546a03f7 | |||
| 817da0160c |
@@ -12,6 +12,8 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio
|
|||||||
- **Laser Systems**: Helios pulsed laser and Genesis CW laser control
|
- **Laser Systems**: Helios pulsed laser and Genesis CW laser control
|
||||||
- **Data Acquisition**: Tektronix oscilloscope integration with fast-frame support
|
- **Data Acquisition**: Tektronix oscilloscope integration with fast-frame support
|
||||||
- **Scan Planning**: Automated raster scan generation and execution
|
- **Scan Planning**: Automated raster scan generation and execution
|
||||||
|
- **Angle Inspection**: Park the rig at random points across a plan's angles
|
||||||
|
to check the SAW response on the scope before committing to a long scan
|
||||||
- **Real-time Monitoring**: Live status updates and progress tracking
|
- **Real-time Monitoring**: Live status updates and progress tracking
|
||||||
|
|
||||||
## Hardware Components
|
## Hardware Components
|
||||||
@@ -55,6 +57,9 @@ scanengine-3/
|
|||||||
│ ├── scan_geometry.py # ScanPlan, rotated-bbox planning, limits
|
│ ├── scan_geometry.py # ScanPlan, rotated-bbox planning, limits
|
||||||
│ ├── scan_resume.py # Resume planning (frontier rule)
|
│ ├── scan_resume.py # Resume planning (frontier rule)
|
||||||
│ ├── scope_sras.py # Oscilloscope SCPI policy for SRAS
|
│ ├── scope_sras.py # Oscilloscope SCPI policy for SRAS
|
||||||
|
│ ├── scope_burst.py # Burst-mode FastFrame sizing + row splitting
|
||||||
|
│ ├── scope_inspect.py # Scope setup for pre-scan angle inspection
|
||||||
|
│ ├── angle_inspect.py # AngleInspector — park on a point per angle
|
||||||
│ ├── rotation.py # GR rotation axis settings + moves
|
│ ├── rotation.py # GR rotation axis settings + moves
|
||||||
│ ├── sras_format.py # v6 .sras writer/reader (memory-mapped)
|
│ ├── sras_format.py # v6 .sras writer/reader (memory-mapped)
|
||||||
│ ├── sras_analysis.py # Image reducers + SAW matched filter
|
│ ├── sras_analysis.py # Image reducers + SAW matched filter
|
||||||
@@ -72,6 +77,7 @@ scanengine-3/
|
|||||||
│
|
│
|
||||||
├── gui/ # Shared PyQt6 layer
|
├── gui/ # Shared PyQt6 layer
|
||||||
│ ├── scan_bridge.py # QtScanController over core.scan_engine
|
│ ├── scan_bridge.py # QtScanController over core.scan_engine
|
||||||
|
│ ├── inspect_bridge.py # QtAngleInspector over core.angle_inspect
|
||||||
│ ├── qt_t3r.py # Qt adapter over the T3R driver
|
│ ├── qt_t3r.py # Qt adapter over the T3R driver
|
||||||
│ ├── qt_workers.py # QueueWorker / PollingQueueWorker bases
|
│ ├── qt_workers.py # QueueWorker / PollingQueueWorker bases
|
||||||
│ └── widgets.py # ConnectionBar, LogConsole, PortSelector…
|
│ └── widgets.py # ConnectionBar, LogConsole, PortSelector…
|
||||||
|
|||||||
@@ -0,0 +1,251 @@
|
|||||||
|
"""Pre-scan angle inspection: park the rig on a point and let the operator look.
|
||||||
|
|
||||||
|
A multi-angle scan can take hours, and an angle that responds poorly produces
|
||||||
|
rows that look fine in the file but carry no usable SAW packet. This drives
|
||||||
|
the rig through the same angles the scan will use, parking at a random point
|
||||||
|
inside each angle's own bounding box so the response can be judged on the
|
||||||
|
oscilloscope before committing to the run.
|
||||||
|
|
||||||
|
Headless and Qt-free, like ScanEngine: gui/inspect_bridge.py wraps it.
|
||||||
|
|
||||||
|
No waveform ever crosses this boundary. The operator reads the scope screen
|
||||||
|
directly; this module's job is only to put the hardware in the right place and
|
||||||
|
the scope in a state worth looking at (see core.scope_inspect).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import random
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Callable
|
||||||
|
|
||||||
|
from core import scope_inspect
|
||||||
|
from core.rotation import RotationAxis
|
||||||
|
from core.scan_engine import (
|
||||||
|
AXIS_X, AXIS_Y, SCAN_ACCEL_MM_S2, SCAN_VELOCITY_MM_S,
|
||||||
|
)
|
||||||
|
from core.scan_geometry import DEFAULT_STAGE_LIMITS, ScanPlan, StageLimits
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Positioning moves only — no data is taken while moving, so there is no
|
||||||
|
# reason to cross the tray at full scan velocity.
|
||||||
|
INSPECT_VELOCITY_MM_S = SCAN_VELOCITY_MM_S / 2.0
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class InspectionPoint:
|
||||||
|
"""Where the rig is parked, and which angle it is parked for."""
|
||||||
|
angle_idx: int
|
||||||
|
angle_deg: float
|
||||||
|
x_mm: float
|
||||||
|
y_mm: float
|
||||||
|
|
||||||
|
def describe(self) -> str:
|
||||||
|
return (f"Angle {self.angle_idx + 1} ({self.angle_deg:.1f}°) "
|
||||||
|
f"X={self.x_mm:.3f} mm Y={self.y_mm:.3f} mm")
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class InspectCallbacks:
|
||||||
|
"""Progress reporting. Defaults are no-ops so the core needs no front end."""
|
||||||
|
on_status: Callable[[str], None] = lambda msg: None
|
||||||
|
on_point: Callable[[InspectionPoint], None] = lambda pt: None
|
||||||
|
on_busy: Callable[[bool], None] = lambda busy: None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class _State:
|
||||||
|
angle_idx: int = 0
|
||||||
|
point: InspectionPoint | None = None
|
||||||
|
started: bool = False
|
||||||
|
rotator_ready: bool = False
|
||||||
|
|
||||||
|
|
||||||
|
class AngleInspector:
|
||||||
|
"""Drives stage + rotator to inspection points across a plan's angles."""
|
||||||
|
|
||||||
|
def __init__(self, stage, scope, rotator: RotationAxis | None,
|
||||||
|
plan: ScanPlan,
|
||||||
|
callbacks: InspectCallbacks | None = None,
|
||||||
|
limits: StageLimits = DEFAULT_STAGE_LIMITS,
|
||||||
|
rng: random.Random | None = None):
|
||||||
|
self._stage = stage
|
||||||
|
self._scope = scope
|
||||||
|
self._rotator = rotator
|
||||||
|
self._plan = plan
|
||||||
|
self._cb = callbacks if callbacks is not None else InspectCallbacks()
|
||||||
|
self._limits = limits
|
||||||
|
# Injectable so tests can pin the point selection.
|
||||||
|
self._rng = rng if rng is not None else random.Random()
|
||||||
|
self._st = _State()
|
||||||
|
|
||||||
|
# ── Introspection ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
@property
|
||||||
|
def n_angles(self) -> int:
|
||||||
|
return self._plan.n_angles
|
||||||
|
|
||||||
|
@property
|
||||||
|
def angle_idx(self) -> int:
|
||||||
|
return self._st.angle_idx
|
||||||
|
|
||||||
|
@property
|
||||||
|
def current_point(self) -> InspectionPoint | None:
|
||||||
|
return self._st.point
|
||||||
|
|
||||||
|
def angle_labels(self) -> list[str]:
|
||||||
|
return [f"Angle {i + 1}/{self.n_angles} — {pa.angle_deg:.2f}°"
|
||||||
|
for i, pa in enumerate(self._plan.per_angle)]
|
||||||
|
|
||||||
|
# ── Lifecycle ─────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def start(self) -> InspectionPoint:
|
||||||
|
"""Configure the hardware and park on the first angle."""
|
||||||
|
if self._stage is None:
|
||||||
|
raise RuntimeError("BBD202 not connected")
|
||||||
|
if self._scope is None:
|
||||||
|
raise RuntimeError("Oscilloscope not connected")
|
||||||
|
|
||||||
|
self._st.rotator_ready = (self._rotator is not None
|
||||||
|
and self._rotator.is_available)
|
||||||
|
if self.n_angles > 1 and not self._st.rotator_ready:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Inspecting {self.n_angles} angles requires the T3R rotation "
|
||||||
|
"stage (GR-axis), but it is not connected. Connect T3R from "
|
||||||
|
"the T3R panel, or inspect a single-angle plan."
|
||||||
|
)
|
||||||
|
|
||||||
|
self._cb.on_busy(True)
|
||||||
|
try:
|
||||||
|
self._cb.on_status("Configuring stage for inspection …")
|
||||||
|
ctrl = self._stage
|
||||||
|
for axis in (AXIS_X, AXIS_Y):
|
||||||
|
ctrl.set_velocity_params(axis,
|
||||||
|
max_velocity=INSPECT_VELOCITY_MM_S,
|
||||||
|
acceleration=SCAN_ACCEL_MM_S2)
|
||||||
|
# Nothing here is gated, and an armed trigger output would keep
|
||||||
|
# driving the gate line on every positioning move.
|
||||||
|
ctrl.set_trigger_gate_off(AXIS_X)
|
||||||
|
|
||||||
|
if self._st.rotator_ready:
|
||||||
|
self._cb.on_status("Configuring GR axis …")
|
||||||
|
self._rotator.configure()
|
||||||
|
|
||||||
|
self._cb.on_status("Configuring oscilloscope for inspection …")
|
||||||
|
scope_inspect.configure_inspection(self._scope)
|
||||||
|
|
||||||
|
self._st.started = True
|
||||||
|
return self._goto(0, new_point=True)
|
||||||
|
finally:
|
||||||
|
self._cb.on_busy(False)
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
"""Stop the sweep and send the rotator home. Safe to call twice."""
|
||||||
|
if not self._st.started:
|
||||||
|
return
|
||||||
|
self._st.started = False
|
||||||
|
self._cb.on_busy(True)
|
||||||
|
try:
|
||||||
|
try:
|
||||||
|
scope_inspect.stop_inspection(self._scope)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Could not stop the inspection acquisition")
|
||||||
|
if self._st.rotator_ready and abs(self._rotator.current_deg) > 0.001:
|
||||||
|
self._cb.on_status("Returning GR to home …")
|
||||||
|
try:
|
||||||
|
self._rotator.return_to_zero()
|
||||||
|
except Exception:
|
||||||
|
logger.exception("GR return-to-home failed")
|
||||||
|
self._cb.on_status("Inspection finished.")
|
||||||
|
finally:
|
||||||
|
self._cb.on_busy(False)
|
||||||
|
|
||||||
|
# ── Navigation ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def goto_angle(self, angle_idx: int) -> InspectionPoint:
|
||||||
|
"""Rotate to `angle_idx` and park on a fresh random point there."""
|
||||||
|
self._require_started()
|
||||||
|
self._cb.on_busy(True)
|
||||||
|
try:
|
||||||
|
return self._goto(angle_idx, new_point=True)
|
||||||
|
finally:
|
||||||
|
self._cb.on_busy(False)
|
||||||
|
|
||||||
|
def next_angle(self) -> InspectionPoint:
|
||||||
|
"""Advance one angle, wrapping at the end."""
|
||||||
|
return self.goto_angle((self._st.angle_idx + 1) % self.n_angles)
|
||||||
|
|
||||||
|
def prev_angle(self) -> InspectionPoint:
|
||||||
|
return self.goto_angle((self._st.angle_idx - 1) % self.n_angles)
|
||||||
|
|
||||||
|
def new_point(self) -> InspectionPoint:
|
||||||
|
"""Re-roll the point within the current angle, without rotating.
|
||||||
|
|
||||||
|
One point can be unrepresentative — a bad spot on the sample looks the
|
||||||
|
same as a bad angle. Re-rolling a few times is how you tell them
|
||||||
|
apart, so this deliberately skips the rotation.
|
||||||
|
"""
|
||||||
|
self._require_started()
|
||||||
|
self._cb.on_busy(True)
|
||||||
|
try:
|
||||||
|
return self._goto(self._st.angle_idx, new_point=True, rotate=False)
|
||||||
|
finally:
|
||||||
|
self._cb.on_busy(False)
|
||||||
|
|
||||||
|
# ── Internals ─────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _require_started(self):
|
||||||
|
if not self._st.started:
|
||||||
|
raise RuntimeError("Inspection has not been started")
|
||||||
|
|
||||||
|
def _goto(self, angle_idx: int, new_point: bool,
|
||||||
|
rotate: bool = True) -> InspectionPoint:
|
||||||
|
if not 0 <= angle_idx < self.n_angles:
|
||||||
|
raise IndexError(
|
||||||
|
f"Angle {angle_idx} out of range (plan has {self.n_angles})")
|
||||||
|
|
||||||
|
pa = self._plan.per_angle[angle_idx]
|
||||||
|
self._st.angle_idx = angle_idx
|
||||||
|
|
||||||
|
if rotate and self._st.rotator_ready:
|
||||||
|
delta = pa.angle_deg - self._rotator.current_deg
|
||||||
|
if abs(delta) > 0.001:
|
||||||
|
self._cb.on_status(
|
||||||
|
f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …")
|
||||||
|
self._rotator.rotate_to(pa.angle_deg)
|
||||||
|
|
||||||
|
point = self._pick_point(angle_idx) if new_point else self._st.point
|
||||||
|
|
||||||
|
self._cb.on_status(f"Moving to {point.describe()} …")
|
||||||
|
# Y first, then X — the same order the scan uses to reach a row.
|
||||||
|
self._stage.move_axis_absolute(AXIS_Y, point.y_mm, timeout=60.0)
|
||||||
|
self._stage.move_axis_absolute(AXIS_X, point.x_mm, timeout=60.0)
|
||||||
|
|
||||||
|
self._st.point = point
|
||||||
|
self._cb.on_point(point)
|
||||||
|
self._cb.on_status(f"Parked at {point.describe()}")
|
||||||
|
return point
|
||||||
|
|
||||||
|
def _pick_point(self, angle_idx: int) -> InspectionPoint:
|
||||||
|
"""A random point on this angle's scan grid.
|
||||||
|
|
||||||
|
Y is drawn from the angle's actual row positions and X uniformly from
|
||||||
|
its data window, so the point is somewhere the scan would really
|
||||||
|
sample — not merely inside the bounding box.
|
||||||
|
"""
|
||||||
|
pa = self._plan.per_angle[angle_idx]
|
||||||
|
if not pa.y_positions:
|
||||||
|
raise ValueError(f"Angle {angle_idx + 1} has no rows to inspect")
|
||||||
|
|
||||||
|
y = self._rng.choice(pa.y_positions)
|
||||||
|
x = self._rng.uniform(pa.x_start, pa.x_start + pa.x_delta)
|
||||||
|
|
||||||
|
lim = self._limits
|
||||||
|
if not (lim.x_min <= x <= lim.x_max and lim.y_min <= y <= lim.y_max):
|
||||||
|
raise ValueError(
|
||||||
|
f"Inspection point X={x:.3f} Y={y:.3f} is outside the stage "
|
||||||
|
f"travel ({lim.x_min}–{lim.x_max} × {lim.y_min}–{lim.y_max} mm)"
|
||||||
|
)
|
||||||
|
return InspectionPoint(angle_idx=angle_idx, angle_deg=pa.angle_deg,
|
||||||
|
x_mm=x, y_mm=y)
|
||||||
@@ -0,0 +1,104 @@
|
|||||||
|
"""Oscilloscope configuration for pre-scan angle inspection.
|
||||||
|
|
||||||
|
Inspection is read-on-the-instrument: nothing in this module transfers or
|
||||||
|
plots waveform data. The app puts the scope into a free-running, edge-
|
||||||
|
triggered state and drives the stage to the point being inspected; the
|
||||||
|
operator judges the SAW response and the bias levels on the scope screen.
|
||||||
|
|
||||||
|
That split is deliberate. A scan's acquisition trigger is the logic AND of
|
||||||
|
the laser pulse and the stage's max-velocity gate, and its transfers are
|
||||||
|
FastFrame blocks — neither is useful for looking at one point by eye. Here
|
||||||
|
the trigger is a plain edge on the laser pulse, FastFrame is off, and the
|
||||||
|
acquisition free-runs, so the display updates continuously while the stage
|
||||||
|
sits still.
|
||||||
|
|
||||||
|
CH1 keeps the acquisition front-end so what is on screen is what a scan would
|
||||||
|
record. CH3 and CH4 are rescaled as DC bias monitors (see BIAS_* below).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from dataclasses import replace
|
||||||
|
|
||||||
|
from core.scope_sras import SAMPLE_RATE_HZ, SRAS_CHANNELS, configure_channels
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# CH2 carries the laser pulse. The scan triggers it at 0.5 V as one term of a
|
||||||
|
# logic AND; inspection triggers well above that so a slow edge or a noisy
|
||||||
|
# baseline cannot free-run the display.
|
||||||
|
INSPECT_TRIG_LEVEL_V = 2.0
|
||||||
|
|
||||||
|
# CH3/CH4 are the DC bias monitors during inspection. The signal never goes
|
||||||
|
# negative and spans roughly 0–700 mV, so both channels get the *same* scale
|
||||||
|
# and position — the point of inspecting them is comparing the two by eye, and
|
||||||
|
# that only works if a division means the same thing on each.
|
||||||
|
#
|
||||||
|
# Ground sits BIAS_POSITION_DIV divisions below centre, which puts the whole
|
||||||
|
# 0–700 mV range above the centre line with a little room underneath for
|
||||||
|
# undershoot. With 100 mV/div and ground 3.5 divisions low, the visible window
|
||||||
|
# runs from about -50 mV to +750 mV on an 8-division display and wider on a
|
||||||
|
# 10-division one, so 0–700 mV sits comfortably inside either.
|
||||||
|
BIAS_CHANNELS = (3, 4)
|
||||||
|
BIAS_WINDOW_V = 0.700
|
||||||
|
BIAS_SCALE_V_DIV = 0.100
|
||||||
|
BIAS_POSITION_DIV = -3.5
|
||||||
|
|
||||||
|
BIAS_LABELS = {3: "Bias - A", 4: "Bias - B"}
|
||||||
|
|
||||||
|
|
||||||
|
def inspect_channel_profiles() -> dict:
|
||||||
|
"""Channel front-end config for inspection.
|
||||||
|
|
||||||
|
CH1 and CH2 are the acquisition profiles verbatim. CH3 and CH4 differ
|
||||||
|
only in label, scale and position — termination, coupling and bandwidth
|
||||||
|
stay as the scan sets them, so the bias reading is the same measurement
|
||||||
|
the scan records, just displayed usefully.
|
||||||
|
"""
|
||||||
|
profiles = dict(SRAS_CHANNELS)
|
||||||
|
for ch in BIAS_CHANNELS:
|
||||||
|
profiles[ch] = replace(
|
||||||
|
SRAS_CHANNELS[ch],
|
||||||
|
label=BIAS_LABELS[ch],
|
||||||
|
scale_v_div=BIAS_SCALE_V_DIV,
|
||||||
|
position_div=BIAS_POSITION_DIV,
|
||||||
|
)
|
||||||
|
return profiles
|
||||||
|
|
||||||
|
|
||||||
|
def configure_inspection(scope) -> None:
|
||||||
|
"""Put the scope into free-running inspection mode.
|
||||||
|
|
||||||
|
Leaves the acquisition running, so the display stays live while the
|
||||||
|
operator moves between angles and points.
|
||||||
|
"""
|
||||||
|
configure_channels(scope, inspect_channel_profiles())
|
||||||
|
|
||||||
|
# Plain edge trigger on the laser pulse — no logic pattern, so the stage
|
||||||
|
# gate plays no part and a stationary stage still triggers.
|
||||||
|
scope.write("TRIGger:A:TYPe EDGE")
|
||||||
|
scope.set_trigger_source(2)
|
||||||
|
scope.set_trigger_slope("RISE")
|
||||||
|
scope.set_trigger_level(2, INSPECT_TRIG_LEVEL_V)
|
||||||
|
scope.set_trigger_mode("NORMAL")
|
||||||
|
|
||||||
|
# No averaging: a weak or intermittent SAW response is exactly what the
|
||||||
|
# operator is looking for, and averaging would hide it.
|
||||||
|
scope.set_acquire_mode("SAMPLE")
|
||||||
|
scope.set_fastframe_state(False)
|
||||||
|
|
||||||
|
scope.set_sample_rate(SAMPLE_RATE_HZ)
|
||||||
|
scope.write("HORizontal:POSition 30")
|
||||||
|
|
||||||
|
# Free-run rather than single-sequence, so the trace keeps updating.
|
||||||
|
scope.write("ACQuire:STOPAfter RUNSTop")
|
||||||
|
scope.write("ACQuire:STATE RUN")
|
||||||
|
|
||||||
|
|
||||||
|
def stop_inspection(scope) -> None:
|
||||||
|
"""Halt the free-running acquisition.
|
||||||
|
|
||||||
|
The next scan reconfigures the scope from scratch, so this only needs to
|
||||||
|
stop the sweep — it does not try to restore the acquisition profile.
|
||||||
|
"""
|
||||||
|
scope.write("ACQuire:STATE STOP")
|
||||||
@@ -0,0 +1,113 @@
|
|||||||
|
"""Qt bridge over the headless AngleInspector.
|
||||||
|
|
||||||
|
Inspection is command-driven rather than one long run: the operator clicks an
|
||||||
|
angle, waits for the stage to park, looks at the scope, clicks again. That is
|
||||||
|
exactly the shape QueueWorker exists for — it blocks on the queue between
|
||||||
|
commands instead of polling, so an inspection window left open costs nothing.
|
||||||
|
|
||||||
|
Every stage move and rotation blocks for seconds, so all of it runs on this
|
||||||
|
worker's thread; the window only ever enqueues and reacts to signals.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import traceback
|
||||||
|
|
||||||
|
from PyQt6.QtCore import pyqtSignal
|
||||||
|
|
||||||
|
from core.angle_inspect import AngleInspector, InspectCallbacks
|
||||||
|
from gui.qt_workers import QueueWorker
|
||||||
|
|
||||||
|
|
||||||
|
class QtAngleInspector(QueueWorker):
|
||||||
|
"""Runs an AngleInspector on its own QThread and republishes its events."""
|
||||||
|
|
||||||
|
ready = pyqtSignal(object) # InspectionPoint — start() succeeded
|
||||||
|
start_failed = pyqtSignal(str)
|
||||||
|
point_changed = pyqtSignal(object) # InspectionPoint
|
||||||
|
status_msg = pyqtSignal(str)
|
||||||
|
busy_changed = pyqtSignal(bool) # True while a move is in flight
|
||||||
|
stopped = pyqtSignal()
|
||||||
|
|
||||||
|
def __init__(self, stage, scope, rotator, plan, on_inspect_active=None):
|
||||||
|
super().__init__()
|
||||||
|
self._on_inspect_active = on_inspect_active
|
||||||
|
|
||||||
|
callbacks = InspectCallbacks(
|
||||||
|
on_status=self.status_msg.emit,
|
||||||
|
on_point=self.point_changed.emit,
|
||||||
|
on_busy=self.busy_changed.emit,
|
||||||
|
)
|
||||||
|
self._inspector = AngleInspector(stage, scope, rotator, plan,
|
||||||
|
callbacks=callbacks)
|
||||||
|
self._handlers = {
|
||||||
|
"start": self._do_start,
|
||||||
|
"goto": self._do_goto,
|
||||||
|
"new_point": self._do_new_point,
|
||||||
|
"stop": self._do_stop,
|
||||||
|
}
|
||||||
|
|
||||||
|
# ── Introspection (safe from the GUI thread: reads the plan, not the rig) ──
|
||||||
|
|
||||||
|
def angle_labels(self) -> list[str]:
|
||||||
|
return self._inspector.angle_labels()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def n_angles(self) -> int:
|
||||||
|
return self._inspector.n_angles
|
||||||
|
|
||||||
|
# ── Command submission (GUI thread) ───────────────────────────────────────
|
||||||
|
|
||||||
|
def request_start(self):
|
||||||
|
self._enqueue("start")
|
||||||
|
|
||||||
|
def request_goto(self, angle_idx: int):
|
||||||
|
self._enqueue("goto", angle_idx=angle_idx)
|
||||||
|
|
||||||
|
def request_new_point(self):
|
||||||
|
self._enqueue("new_point")
|
||||||
|
|
||||||
|
def request_stop(self):
|
||||||
|
self._enqueue("stop")
|
||||||
|
|
||||||
|
# ── Handlers (worker thread) ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _do_start(self):
|
||||||
|
if self._on_inspect_active is not None:
|
||||||
|
self._on_inspect_active(True)
|
||||||
|
try:
|
||||||
|
point = self._inspector.start()
|
||||||
|
except Exception as exc:
|
||||||
|
traceback.print_exc()
|
||||||
|
if self._on_inspect_active is not None:
|
||||||
|
self._on_inspect_active(False)
|
||||||
|
self.start_failed.emit(str(exc))
|
||||||
|
return
|
||||||
|
self.ready.emit(point)
|
||||||
|
|
||||||
|
def _do_goto(self, angle_idx: int):
|
||||||
|
self._inspector.goto_angle(angle_idx)
|
||||||
|
|
||||||
|
def _do_new_point(self):
|
||||||
|
self._inspector.new_point()
|
||||||
|
|
||||||
|
def _do_stop(self):
|
||||||
|
try:
|
||||||
|
self._inspector.stop()
|
||||||
|
finally:
|
||||||
|
if self._on_inspect_active is not None:
|
||||||
|
self._on_inspect_active(False)
|
||||||
|
self.stopped.emit()
|
||||||
|
|
||||||
|
def _on_stop(self):
|
||||||
|
"""Worker loop exiting — make sure the rig is left in a safe state.
|
||||||
|
|
||||||
|
Covers the case where the window is closed without a clean stop
|
||||||
|
command reaching the queue.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
self._inspector.stop()
|
||||||
|
except Exception:
|
||||||
|
traceback.print_exc()
|
||||||
|
finally:
|
||||||
|
if self._on_inspect_active is not None:
|
||||||
|
self._on_inspect_active(False)
|
||||||
@@ -1017,6 +1017,16 @@
|
|||||||
</property>
|
</property>
|
||||||
</widget>
|
</widget>
|
||||||
</item>
|
</item>
|
||||||
|
<item>
|
||||||
|
<widget class="QPushButton" name="inspect_angles_btn">
|
||||||
|
<property name="toolTip">
|
||||||
|
<string>Rotate through the angles of the scan currently entered, parking at a random point in each so the SAW response can be checked on the oscilloscope before committing to the run.</string>
|
||||||
|
</property>
|
||||||
|
<property name="text">
|
||||||
|
<string>Inspect Angles…</string>
|
||||||
|
</property>
|
||||||
|
</widget>
|
||||||
|
</item>
|
||||||
<item>
|
<item>
|
||||||
<widget class="QPushButton" name="start_scan_btn">
|
<widget class="QPushButton" name="start_scan_btn">
|
||||||
<property name="text">
|
<property name="text">
|
||||||
|
|||||||
+176
@@ -38,6 +38,7 @@ from core.scope_sras import SAMPLE_RATE_HZ, configure_channels
|
|||||||
from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header
|
from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header
|
||||||
from gui.qt_t3r import QtT3RAdapter
|
from gui.qt_t3r import QtT3RAdapter
|
||||||
from gui.qt_workers import PollingQueueWorker, QueueWorker
|
from gui.qt_workers import PollingQueueWorker, QueueWorker
|
||||||
|
from gui.inspect_bridge import QtAngleInspector
|
||||||
from gui.scan_bridge import QtScanController
|
from gui.scan_bridge import QtScanController
|
||||||
from hardware.helios_laser import HeliosLaser
|
from hardware.helios_laser import HeliosLaser
|
||||||
from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver
|
from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver
|
||||||
@@ -844,6 +845,104 @@ class ScanProgressWindow(QWidget):
|
|||||||
|
|
||||||
# ── Main window ───────────────────────────────────────────────────────────────
|
# ── Main window ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
class AngleInspectWindow(QWidget):
|
||||||
|
"""Click through a plan's angles, parking the rig at a point in each.
|
||||||
|
|
||||||
|
Deliberately shows no waveform. The operator reads the SAW response and
|
||||||
|
the bias levels off the oscilloscope itself — this window only says where
|
||||||
|
the rig is and lets them move it somewhere else.
|
||||||
|
"""
|
||||||
|
|
||||||
|
goto_requested = pyqtSignal(int)
|
||||||
|
new_point_requested = pyqtSignal()
|
||||||
|
stop_requested = pyqtSignal()
|
||||||
|
|
||||||
|
def __init__(self, angle_labels: list[str], parent: QWidget | None = None):
|
||||||
|
super().__init__(parent, Qt.WindowType.Window)
|
||||||
|
self.setWindowTitle("Inspect Angles")
|
||||||
|
self.resize(420, 460)
|
||||||
|
|
||||||
|
layout = QVBoxLayout(self)
|
||||||
|
layout.addWidget(QLabel(
|
||||||
|
"Select an angle to rotate to it and park at a random point in "
|
||||||
|
"its scan area.\nRead the SAW response on the oscilloscope."
|
||||||
|
))
|
||||||
|
|
||||||
|
self.angle_list = QListWidget(self)
|
||||||
|
for i, label in enumerate(angle_labels):
|
||||||
|
item = QListWidgetItem(label)
|
||||||
|
item.setData(Qt.ItemDataRole.UserRole, i)
|
||||||
|
self.angle_list.addItem(item)
|
||||||
|
self.angle_list.setCurrentRow(0)
|
||||||
|
# currentRowChanged would also fire when the code syncs the highlight
|
||||||
|
# back after a move, re-triggering the move it is reporting.
|
||||||
|
self.angle_list.itemClicked.connect(self._on_item_clicked)
|
||||||
|
layout.addWidget(self.angle_list)
|
||||||
|
|
||||||
|
nav_row = QHBoxLayout()
|
||||||
|
self.prev_btn = QPushButton("◀ Previous")
|
||||||
|
self.next_btn = QPushButton("Next ▶")
|
||||||
|
self.new_point_btn = QPushButton("New Point")
|
||||||
|
self.new_point_btn.setToolTip(
|
||||||
|
"Pick another random point at this angle without rotating — a bad "
|
||||||
|
"spot on the sample looks the same as a bad angle until you move."
|
||||||
|
)
|
||||||
|
self.prev_btn.clicked.connect(self._on_prev)
|
||||||
|
self.next_btn.clicked.connect(self._on_next)
|
||||||
|
self.new_point_btn.clicked.connect(self.new_point_requested.emit)
|
||||||
|
nav_row.addWidget(self.prev_btn)
|
||||||
|
nav_row.addWidget(self.next_btn)
|
||||||
|
nav_row.addWidget(self.new_point_btn)
|
||||||
|
layout.addLayout(nav_row)
|
||||||
|
|
||||||
|
self.point_label = QLabel("—")
|
||||||
|
self.point_label.setStyleSheet("font-weight: bold;")
|
||||||
|
layout.addWidget(self.point_label)
|
||||||
|
|
||||||
|
self.status_label = QLabel("Starting …")
|
||||||
|
self.status_label.setWordWrap(True)
|
||||||
|
layout.addWidget(self.status_label)
|
||||||
|
|
||||||
|
self.close_btn = QPushButton("Close")
|
||||||
|
self.close_btn.clicked.connect(self.close)
|
||||||
|
layout.addWidget(self.close_btn)
|
||||||
|
|
||||||
|
self._n_angles = len(angle_labels)
|
||||||
|
self._angle_idx = 0
|
||||||
|
self.set_busy(True)
|
||||||
|
|
||||||
|
# ── Worker → window ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def set_busy(self, busy: bool):
|
||||||
|
"""Lock navigation while the stage is moving; the rig is not re-entrant."""
|
||||||
|
for w in (self.angle_list, self.prev_btn, self.next_btn,
|
||||||
|
self.new_point_btn):
|
||||||
|
w.setEnabled(not busy)
|
||||||
|
|
||||||
|
def on_status(self, msg: str):
|
||||||
|
self.status_label.setText(msg)
|
||||||
|
|
||||||
|
def on_point(self, point):
|
||||||
|
self._angle_idx = point.angle_idx
|
||||||
|
self.angle_list.setCurrentRow(point.angle_idx)
|
||||||
|
self.point_label.setText(point.describe())
|
||||||
|
|
||||||
|
# ── Window → worker ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _on_item_clicked(self, item):
|
||||||
|
self.goto_requested.emit(item.data(Qt.ItemDataRole.UserRole))
|
||||||
|
|
||||||
|
def _on_prev(self):
|
||||||
|
self.goto_requested.emit((self._angle_idx - 1) % self._n_angles)
|
||||||
|
|
||||||
|
def _on_next(self):
|
||||||
|
self.goto_requested.emit((self._angle_idx + 1) % self._n_angles)
|
||||||
|
|
||||||
|
def closeEvent(self, event):
|
||||||
|
self.stop_requested.emit()
|
||||||
|
super().closeEvent(event)
|
||||||
|
|
||||||
|
|
||||||
class MainWindow(QMainWindow):
|
class MainWindow(QMainWindow):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
@@ -878,6 +977,9 @@ class MainWindow(QMainWindow):
|
|||||||
self._scan_progress = ScanProgressWindow()
|
self._scan_progress = ScanProgressWindow()
|
||||||
|
|
||||||
self._scan_worker: QtScanController | None = None
|
self._scan_worker: QtScanController | None = None
|
||||||
|
self._inspect_thread: QThread | None = None
|
||||||
|
self._inspect_worker: QtAngleInspector | None = None
|
||||||
|
self._inspect_window: AngleInspectWindow | None = None
|
||||||
self._scan_thread: QThread | None = None
|
self._scan_thread: QThread | None = None
|
||||||
|
|
||||||
self._bbd_active_jog: tuple[str, int] | None = None # (axis, direction)
|
self._bbd_active_jog: tuple[str, int] | None = None # (axis, direction)
|
||||||
@@ -999,6 +1101,7 @@ class MainWindow(QMainWindow):
|
|||||||
|
|
||||||
# Scan
|
# Scan
|
||||||
self.start_scan_btn.clicked.connect(self._on_start_scan)
|
self.start_scan_btn.clicked.connect(self._on_start_scan)
|
||||||
|
self.inspect_angles_btn.clicked.connect(self._on_inspect_angles)
|
||||||
self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir)
|
self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir)
|
||||||
self._scan_progress.abort_requested.connect(self._on_abort_scan)
|
self._scan_progress.abort_requested.connect(self._on_abort_scan)
|
||||||
self._scan_progress.pause_toggled.connect(self._on_pause_scan)
|
self._scan_progress.pause_toggled.connect(self._on_pause_scan)
|
||||||
@@ -1172,6 +1275,79 @@ class MainWindow(QMainWindow):
|
|||||||
self.scan_save_dir_edit.setText(d)
|
self.scan_save_dir_edit.setText(d)
|
||||||
self._persist_defaults()
|
self._persist_defaults()
|
||||||
|
|
||||||
|
def _on_inspect_angles(self):
|
||||||
|
"""Open the pre-scan angle inspector for the plan currently entered."""
|
||||||
|
if self._scan_thread is not None and self._scan_thread.isRunning():
|
||||||
|
QMessageBox.warning(
|
||||||
|
self, "Scan In Progress",
|
||||||
|
"A scan is running — abort it before inspecting angles."
|
||||||
|
)
|
||||||
|
return
|
||||||
|
if self._inspect_thread is not None and self._inspect_thread.isRunning():
|
||||||
|
self._inspect_window.raise_()
|
||||||
|
self._inspect_window.activateWindow()
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
plan, _, _ = self._build_scan_plan()
|
||||||
|
except ValueError as e:
|
||||||
|
QMessageBox.warning(self, "Invalid Scan Parameters", str(e))
|
||||||
|
return
|
||||||
|
|
||||||
|
rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION)
|
||||||
|
self._inspect_thread = QThread(self)
|
||||||
|
self._inspect_worker = QtAngleInspector(
|
||||||
|
stage=self._bbd_worker.controller,
|
||||||
|
scope=self._oscope_worker.scope,
|
||||||
|
rotator=rotator,
|
||||||
|
plan=plan,
|
||||||
|
# Same reason the scan does it: the inspector drives the stage from
|
||||||
|
# its own thread, and the position poll shares the BBD TX queue.
|
||||||
|
on_inspect_active=lambda active: setattr(
|
||||||
|
self._bbd_worker, "scanning_active", active),
|
||||||
|
)
|
||||||
|
self._inspect_worker.moveToThread(self._inspect_thread)
|
||||||
|
|
||||||
|
window = AngleInspectWindow(self._inspect_worker.angle_labels(), self)
|
||||||
|
self._inspect_window = window
|
||||||
|
|
||||||
|
window.goto_requested.connect(self._inspect_worker.request_goto)
|
||||||
|
window.new_point_requested.connect(self._inspect_worker.request_new_point)
|
||||||
|
window.stop_requested.connect(self._on_inspect_window_closed)
|
||||||
|
|
||||||
|
self._inspect_worker.status_msg.connect(window.on_status)
|
||||||
|
self._inspect_worker.point_changed.connect(window.on_point)
|
||||||
|
self._inspect_worker.busy_changed.connect(window.set_busy)
|
||||||
|
self._inspect_worker.start_failed.connect(self._on_inspect_failed)
|
||||||
|
self._inspect_worker.error_occurred.connect(
|
||||||
|
lambda m: QMessageBox.warning(self, "Inspection Error", m))
|
||||||
|
|
||||||
|
self._inspect_thread.started.connect(self._inspect_worker.run)
|
||||||
|
self.start_scan_btn.setEnabled(False)
|
||||||
|
self.inspect_angles_btn.setEnabled(False)
|
||||||
|
window.show()
|
||||||
|
self._inspect_thread.start()
|
||||||
|
self._inspect_worker.request_start()
|
||||||
|
|
||||||
|
def _on_inspect_failed(self, message: str):
|
||||||
|
QMessageBox.warning(self, "Cannot Inspect Angles", message)
|
||||||
|
if self._inspect_window is not None:
|
||||||
|
self._inspect_window.close()
|
||||||
|
|
||||||
|
def _on_inspect_window_closed(self):
|
||||||
|
"""Tear the worker down and hand the hardware back to the scan panel."""
|
||||||
|
if self._inspect_worker is not None:
|
||||||
|
self._inspect_worker.request_stop()
|
||||||
|
self._inspect_worker.stop_worker()
|
||||||
|
if self._inspect_thread is not None:
|
||||||
|
self._inspect_thread.quit()
|
||||||
|
self._inspect_thread.wait(10000)
|
||||||
|
self._inspect_thread = None
|
||||||
|
self._inspect_worker = None
|
||||||
|
self._inspect_window = None
|
||||||
|
self.start_scan_btn.setEnabled(True)
|
||||||
|
self.inspect_angles_btn.setEnabled(True)
|
||||||
|
|
||||||
def _on_start_scan(self):
|
def _on_start_scan(self):
|
||||||
try:
|
try:
|
||||||
plan, prefix, save_dir = self._build_scan_plan()
|
plan, prefix, save_dir = self._build_scan_plan()
|
||||||
|
|||||||
@@ -0,0 +1,292 @@
|
|||||||
|
"""Pre-scan angle inspection, driven entirely by fake hardware.
|
||||||
|
|
||||||
|
The feature's defining constraint is that it reads nothing back from the
|
||||||
|
scope — the operator looks at the instrument. These tests pin that, the scope
|
||||||
|
state the app is responsible for putting the instrument into, and the motion
|
||||||
|
sequence across angles.
|
||||||
|
"""
|
||||||
|
import random
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from core.angle_inspect import AngleInspector, InspectCallbacks
|
||||||
|
from core.rotation import RotationAxis, RotationSettings
|
||||||
|
from core.scan_engine import AXIS_X, AXIS_Y
|
||||||
|
from core.scan_geometry import build_plan
|
||||||
|
from core.scope_inspect import (
|
||||||
|
BIAS_CHANNELS, BIAS_POSITION_DIV, BIAS_SCALE_V_DIV, BIAS_WINDOW_V,
|
||||||
|
INSPECT_TRIG_LEVEL_V, inspect_channel_profiles,
|
||||||
|
)
|
||||||
|
from core.scope_sras import SRAS_CHANNELS
|
||||||
|
from fakes import FakeScope, FakeStage, FakeT3R, Trace
|
||||||
|
|
||||||
|
SPF = 8
|
||||||
|
|
||||||
|
|
||||||
|
def make_plan(num_angles=3):
|
||||||
|
return build_plan(40.0, 30.0, 2.0, 1.0, num_angles, 0.25,
|
||||||
|
laser_freq_hz=20000.0, velocity_mm_s=100.0)
|
||||||
|
|
||||||
|
|
||||||
|
def build(num_angles=3, seed=1234, callbacks=None, rotator_open=True):
|
||||||
|
trace = Trace()
|
||||||
|
scope = FakeScope(trace, samples_per_frame=SPF)
|
||||||
|
stage = FakeStage(trace, scope=scope)
|
||||||
|
t3r = FakeT3R(trace, is_open=rotator_open)
|
||||||
|
rotator = RotationAxis(t3r, RotationSettings())
|
||||||
|
plan = make_plan(num_angles)
|
||||||
|
insp = AngleInspector(stage, scope, rotator, plan,
|
||||||
|
callbacks=callbacks or InspectCallbacks(),
|
||||||
|
rng=random.Random(seed))
|
||||||
|
return insp, trace, plan
|
||||||
|
|
||||||
|
|
||||||
|
def writes(trace):
|
||||||
|
return [c[1] for c in trace.of("write")]
|
||||||
|
|
||||||
|
|
||||||
|
# ── The defining constraint ──────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def test_inspection_never_reads_a_waveform_back():
|
||||||
|
"""The operator reads the scope; the app must not pull data off it.
|
||||||
|
|
||||||
|
If this fails, someone has added a transfer path to a feature whose whole
|
||||||
|
premise is that there isn't one.
|
||||||
|
"""
|
||||||
|
insp, trace, plan = build()
|
||||||
|
insp.start()
|
||||||
|
for i in range(plan.n_angles):
|
||||||
|
insp.goto_angle(i)
|
||||||
|
insp.new_point()
|
||||||
|
insp.stop()
|
||||||
|
|
||||||
|
forbidden = {"transfer_fastframe", "transfer_fastframe_bulk",
|
||||||
|
"transfer_curve", "set_data_source", "query_wfmoutpre"}
|
||||||
|
assert forbidden.isdisjoint(set(trace.names()))
|
||||||
|
assert "CURVe?" not in writes(trace)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Scope configuration ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def test_start_sets_an_edge_trigger_on_ch2_above_the_scan_level():
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
assert "TRIGger:A:TYPe EDGE" in writes(trace)
|
||||||
|
assert trace.of("set_trigger_source")[-1][1] == 2
|
||||||
|
assert trace.of("set_trigger_slope")[-1][1] == "RISE"
|
||||||
|
ch, level = trace.of("set_trigger_level")[-1][1:3]
|
||||||
|
assert (ch, level) == (2, INSPECT_TRIG_LEVEL_V)
|
||||||
|
assert INSPECT_TRIG_LEVEL_V >= 2.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_start_disables_fastframe_averaging_and_the_logic_trigger():
|
||||||
|
"""Everything the scan needs and inspection must not inherit."""
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
assert trace.of("set_fastframe_state")[-1][1] is False
|
||||||
|
assert trace.of("set_acquire_mode")[-1][1] == "SAMPLE"
|
||||||
|
w = writes(trace)
|
||||||
|
assert not any("LOGIc" in cmd or "LOGICPattern" in cmd for cmd in w)
|
||||||
|
|
||||||
|
|
||||||
|
def test_start_leaves_the_acquisition_free_running():
|
||||||
|
"""The display has to keep updating while the operator looks at it."""
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
w = writes(trace)
|
||||||
|
assert "ACQuire:STOPAfter RUNSTop" in w
|
||||||
|
assert w.index("ACQuire:STOPAfter RUNSTop") < w.index("ACQuire:STATE RUN")
|
||||||
|
assert "ACQuire:STATE STOP" not in w
|
||||||
|
|
||||||
|
|
||||||
|
def test_bias_channels_are_directly_comparable():
|
||||||
|
"""CH3/CH4 must share scale and position or the eye comparison is a lie."""
|
||||||
|
profiles = inspect_channel_profiles()
|
||||||
|
a, b = (profiles[ch] for ch in BIAS_CHANNELS)
|
||||||
|
assert a.scale_v_div == b.scale_v_div
|
||||||
|
assert a.position_div == b.position_div
|
||||||
|
# Same front end as the scan records — only the display changes.
|
||||||
|
for ch in BIAS_CHANNELS:
|
||||||
|
assert profiles[ch].termination_ohm == SRAS_CHANNELS[ch].termination_ohm
|
||||||
|
assert profiles[ch].coupling == SRAS_CHANNELS[ch].coupling
|
||||||
|
assert profiles[ch].bandwidth_hz == SRAS_CHANNELS[ch].bandwidth_hz
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("n_divisions", [8, 10])
|
||||||
|
def test_bias_window_shows_zero_to_700mv_with_headroom(n_divisions):
|
||||||
|
"""0–700 mV must fit on screen, above ground, on either graticule size.
|
||||||
|
|
||||||
|
Ground sits BIAS_POSITION_DIV divisions below centre, so the visible
|
||||||
|
window runs from (-N/2 - pos)*scale to (+N/2 - pos)*scale.
|
||||||
|
"""
|
||||||
|
half = n_divisions / 2
|
||||||
|
bottom = (-half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
|
||||||
|
top = (half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
|
||||||
|
|
||||||
|
assert bottom < 0.0, "no room below ground for undershoot"
|
||||||
|
assert top > BIAS_WINDOW_V, "700 mV is clipped or sitting on the top edge"
|
||||||
|
# The point of moving the trace down: most of the screen is above ground.
|
||||||
|
assert abs(bottom) < top
|
||||||
|
|
||||||
|
|
||||||
|
def test_ch1_keeps_the_acquisition_front_end():
|
||||||
|
"""What you see at a point is what a scan would record there."""
|
||||||
|
assert inspect_channel_profiles()[1] == SRAS_CHANNELS[1]
|
||||||
|
|
||||||
|
|
||||||
|
# ── Stage and rotation ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def test_start_parks_on_the_first_angle():
|
||||||
|
insp, _, plan = build()
|
||||||
|
point = insp.start()
|
||||||
|
|
||||||
|
assert point.angle_idx == 0
|
||||||
|
assert point.angle_deg == plan.per_angle[0].angle_deg
|
||||||
|
assert insp.current_point == point
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_gate_is_off_for_the_whole_inspection():
|
||||||
|
"""Nothing here is gated, and an armed output keeps driving the line."""
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
insp.goto_angle(2)
|
||||||
|
insp.new_point()
|
||||||
|
|
||||||
|
assert trace.count("set_trigger_gate_off") >= 1
|
||||||
|
assert trace.count("set_trigger_trigout_maxv") == 0
|
||||||
|
assert [c[2] for c in trace.of("arm_scan_gate") if c[2]] == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_points_land_on_the_scan_grid():
|
||||||
|
"""A point the scan would never sample tells you nothing about the scan."""
|
||||||
|
insp, _, plan = build()
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
for i in range(plan.n_angles):
|
||||||
|
pa = plan.per_angle[i]
|
||||||
|
for _ in range(5):
|
||||||
|
pt = insp.new_point() if insp.angle_idx == i else insp.goto_angle(i)
|
||||||
|
assert pt.angle_idx == i
|
||||||
|
assert pt.y_mm in pa.y_positions
|
||||||
|
assert pa.x_start <= pt.x_mm <= pa.x_start + pa.x_delta
|
||||||
|
|
||||||
|
|
||||||
|
def test_goto_angle_rotates_then_moves():
|
||||||
|
insp, trace, plan = build()
|
||||||
|
insp.start()
|
||||||
|
trace.calls.clear()
|
||||||
|
|
||||||
|
insp.goto_angle(2)
|
||||||
|
|
||||||
|
# t3r_rotate carries the delta, so assert the resulting absolute angle.
|
||||||
|
assert trace.count("t3r_rotate") == 1, "expected exactly one rotation"
|
||||||
|
assert insp._rotator.current_deg == pytest.approx(plan.per_angle[2].angle_deg)
|
||||||
|
moves = trace.of("move_axis_absolute")
|
||||||
|
assert [m[1] for m in moves] == [AXIS_Y, AXIS_X], "Y then X, as the scan does"
|
||||||
|
|
||||||
|
|
||||||
|
def test_new_point_re_rolls_without_rotating():
|
||||||
|
"""Distinguishing a bad spot from a bad angle depends on not rotating."""
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
insp.goto_angle(1)
|
||||||
|
trace.calls.clear()
|
||||||
|
|
||||||
|
first = insp.current_point
|
||||||
|
second = insp.new_point()
|
||||||
|
|
||||||
|
assert second.angle_idx == first.angle_idx == 1
|
||||||
|
assert (second.x_mm, second.y_mm) != (first.x_mm, first.y_mm)
|
||||||
|
assert trace.count("t3r_rotate") == 0, "new_point must not rotate"
|
||||||
|
assert [m[1] for m in trace.of("move_axis_absolute")] == [AXIS_Y, AXIS_X]
|
||||||
|
|
||||||
|
|
||||||
|
def test_next_and_prev_wrap_around():
|
||||||
|
insp, _, plan = build(num_angles=3)
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
assert insp.next_angle().angle_idx == 1
|
||||||
|
assert insp.next_angle().angle_idx == 2
|
||||||
|
assert insp.next_angle().angle_idx == 0, "should wrap forward"
|
||||||
|
assert insp.prev_angle().angle_idx == plan.n_angles - 1, "should wrap back"
|
||||||
|
|
||||||
|
|
||||||
|
def test_angle_labels_cover_every_angle():
|
||||||
|
insp, _, plan = build(num_angles=9)
|
||||||
|
labels = insp.angle_labels()
|
||||||
|
assert len(labels) == 9
|
||||||
|
assert labels[0].startswith("Angle 1/9")
|
||||||
|
|
||||||
|
|
||||||
|
# ── Guards ───────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def test_multi_angle_inspection_requires_the_rotator():
|
||||||
|
insp, _, _ = build(num_angles=3, rotator_open=False)
|
||||||
|
with pytest.raises(RuntimeError, match="T3R rotation stage"):
|
||||||
|
insp.start()
|
||||||
|
|
||||||
|
|
||||||
|
def test_single_angle_inspection_works_without_the_rotator():
|
||||||
|
insp, _, _ = build(num_angles=1, rotator_open=False)
|
||||||
|
point = insp.start()
|
||||||
|
assert point.angle_idx == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_navigation_before_start_is_rejected():
|
||||||
|
insp, _, _ = build()
|
||||||
|
with pytest.raises(RuntimeError, match="not been started"):
|
||||||
|
insp.goto_angle(1)
|
||||||
|
with pytest.raises(RuntimeError, match="not been started"):
|
||||||
|
insp.new_point()
|
||||||
|
|
||||||
|
|
||||||
|
def test_out_of_range_angle_is_rejected():
|
||||||
|
insp, _, _ = build(num_angles=3)
|
||||||
|
insp.start()
|
||||||
|
with pytest.raises(IndexError):
|
||||||
|
insp.goto_angle(3)
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_halts_the_sweep_and_sends_the_rotator_home():
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
insp.goto_angle(2)
|
||||||
|
trace.calls.clear()
|
||||||
|
|
||||||
|
insp.stop()
|
||||||
|
|
||||||
|
assert "ACQuire:STATE STOP" in writes(trace)
|
||||||
|
assert trace.count("t3r_rotate") == 1, "GR not sent home"
|
||||||
|
assert insp._rotator.current_deg == pytest.approx(0.0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_is_idempotent():
|
||||||
|
insp, trace, _ = build()
|
||||||
|
insp.start()
|
||||||
|
insp.stop()
|
||||||
|
trace.calls.clear()
|
||||||
|
|
||||||
|
insp.stop() # must not re-issue anything or raise
|
||||||
|
|
||||||
|
assert trace.calls == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_busy_callback_brackets_every_move():
|
||||||
|
"""The window disables its controls on this, so it has to pair up."""
|
||||||
|
events = []
|
||||||
|
insp, _, _ = build(callbacks=InspectCallbacks(on_busy=events.append))
|
||||||
|
insp.start()
|
||||||
|
insp.goto_angle(1)
|
||||||
|
insp.new_point()
|
||||||
|
insp.stop()
|
||||||
|
|
||||||
|
assert events, "no busy events emitted"
|
||||||
|
assert events[0] is True and events[-1] is False
|
||||||
|
depth = 0
|
||||||
|
for e in events:
|
||||||
|
depth += 1 if e else -1
|
||||||
|
assert depth in (0, 1), f"unbalanced busy events: {events}"
|
||||||
|
assert depth == 0
|
||||||
Reference in New Issue
Block a user