diff --git a/KNOWN_ISSUES.md b/KNOWN_ISSUES.md index fac3009..87e6489 100644 --- a/KNOWN_ISSUES.md +++ b/KNOWN_ISSUES.md @@ -64,3 +64,53 @@ in `configure_scan_trigger` is not enough for FastFrame to re-arm after an AVERAGE-mode sequence, and the row-packing warning ("N frames acquired, M expected") will say so in the log. Raise the settle rather than the ramp buffer — the stage geometry is not what changed. + +## Auto-align: constants that are guesses until the rig confirms them + +`core/auto_align.py` closes a loop over hardware whose gain nobody has +measured. Three numbers in `AlignSettings`/`TAxisSettings` are reasoned +defaults, not readings: + +- `probe_steps = 200` — the first move of every search, made only to learn how + many millivolts a microstep is worth. Too small and each search wastes + iterations doubling it (the status line says so: "moved N microsteps and the + DC difference did not change"); too large and the first move overshoots by + more than the platform should be asked to travel in one go. +- `max_excursion_steps = 20000` — the per-axis safety limit, measured from + wherever the axis started. It exists to stop a runaway before the actuator + reaches its end stop, so it has to be smaller than the real travel. +- `hold_current_ma = 300` — the run current (600 mA) and microstepping (32) + are specified; the standstill current is half the run current by analogy + with the GR axis, and has not been checked against the platform's weight. + +**Bench check:** run one auto-align and read the log. The first search's +iteration count is the probe verdict — 3 or 4 steps means the probe is about +right, and a "did not change by 2 mV" line means it is too small. Convert the +applied corrections into actuator travel and compare against the T-axis +travel to set the excursion limit. Watch whether the platform holds its tilt +between the two phases; if it sags, raise the hold current. + +## Auto-align: does the X phase's piston matter, and where is the pivot? + +The X phase moves T1 alone, as specified. T1 is the only axis lying along X, +so it does tilt the platform along X — but moving one leg of three also lifts +the platform by a third of the move (`tilt_response(X_TILT)` returns a piston +of 1/3 alongside the 2/3 tilt). The search nulls the split-detector +difference, which a piston should not move, so the assumption is that the +piston is harmless. The piston-free alternative is T1 +1 with T0 and T2 at +−0.5 each. + +Separately, the procedure assumes the tilt pivot is under the beam: if it is +not, applying the correction shifts the DC levels at the reference point +itself, and the Y phase then chases levels that no longer describe the rig. +The code reports this rather than compensating for it — `AxisResult`'s +"back at the reference" reading after the X phase is exactly that +measurement. + +**Bench check:** during an X search, watch DC1 + DC2 (the sum, not the +difference) on the scope. If the sum moves as T1 moves, the piston is +changing the amount of collected light and `X_TILT` should become the +piston-free triple. Then read the X phase's reference residual out of the +log: more than a few millivolts means the pivot is not under the beam, and +the Y phase's reference should be re-measured after the X correction instead +of reusing the operator's original numbers. diff --git a/README.md b/README.md index c492f84..86410c2 100755 --- a/README.md +++ b/README.md @@ -19,6 +19,9 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio - **SAW Quality Check**: Acquire one row per angle — the row-wise middle of the ROI — as a v11 `.sras`, then compare every angle's SAW frequency on one graph to judge the alignment before a full run +- **Auto-Align**: Level the sample from the camera window — step the stage + 1.5 mm either side on X and then Y, tilt the T-axes until the DC bias levels + read what they read at the reference point, and leave the correction applied - **Real-time Monitoring**: Live status updates and progress tracking ## Hardware Components @@ -63,8 +66,9 @@ scanengine-3/ │ ├── scan_resume.py # Resume planning (frontier rule) │ ├── 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 +│ ├── scope_inspect.py # Scope setup for inspection + bias read-back │ ├── angle_inspect.py # AngleInspector — park on a point per angle +│ ├── auto_align.py # AutoAligner — tilt the sample level on the DC levels │ ├── saw_check.py # Middle-row SAW check: plan + alignment read-out │ ├── rotation.py # GR rotation axis settings + moves │ ├── sras_format.py # v7/v11 .sras writer, v6/v10 reader (mmap) @@ -84,6 +88,7 @@ scanengine-3/ ├── gui/ # Shared PyQt6 layer │ ├── scan_bridge.py # QtScanController over core.scan_engine │ ├── inspect_bridge.py # QtAngleInspector over core.angle_inspect +│ ├── align_bridge.py # QtAutoAligner over core.auto_align │ ├── qt_t3r.py # Qt adapter over the T3R driver │ ├── qt_workers.py # QueueWorker / PollingQueueWorker bases │ ├── jog_panel.py # T3R + BBD202 jog controls (camera window) @@ -246,6 +251,32 @@ with SrasFile("/data/SRAS/demo-sawcheck.sras") as sras: `saw_check_viewer.py` is the same read-out with the curves drawn. +### Levelling the sample (auto-align) + +Two phases, because the operator sits between them: `prepare()` configures +the rig and reads the DC levels where the stage stands, and `run()` only +starts once those levels have been confirmed as the ones to hold. + +```python +from core.auto_align import AlignCallbacks, AutoAligner + +aligner = AutoAligner(stage, scope, t3r, + callbacks=AlignCallbacks(on_status=print)) +reference = aligner.prepare() # scope + T-axes configured, one reading +print(reference.describe()) # "is the image correct?" happens here +result = aligner.run() # X on T1, then Y on T0/T2 +print(result.describe()) +aligner.stop() # stage parked; the tilt stays applied +``` + +The scope has to be cabled CH1 SAW / CH2 trigger / CH3 DC 1 / CH4 DC 2 — the +same channels a scan uses, except that CH3 carries the DC monitor here rather +than the max-velocity gate. Nothing rewires it; the app asks the operator to +confirm the cabling, and refuses to servo on a scope that is not triggering. + +In the main app the button is in the camera window, because judging the image +is the first step of the procedure. + ### Reading a scan file `SrasFile` memory-maps the data block, so opening a multi-gigabyte scan diff --git a/core/auto_align.py b/core/auto_align.py new file mode 100644 index 0000000..bd148ea --- /dev/null +++ b/core/auto_align.py @@ -0,0 +1,708 @@ +"""Auto-align: level the sample against the stage's travel plane. + +The operator frames a good spot by eye and confirms the DC bias levels the +detector reads there. Those two numbers — DC 1 on CH3, DC 2 on CH4 — are the +definition of "aligned" for this rig, and they are the only thing this module +optimises. + +Why moving the stage tells you about tilt: the detection beam is fixed in +space and the XY stage carries the sample under it, so the height of the +surface under the beam is ``h(x) = h0 + theta * x`` when the sample sits at an +angle ``theta`` to the travel plane. Step 1.5 mm along X and the bias levels +move by ``theta * 1.5``; tilt the platform until they read what they read at +the reference point and you have measured ``theta`` directly, because a +platform tilt changes the height under the beam in proportion to x as well. +The correction that fixes the offset point is therefore the same correction +that levels the whole travel — which is why this procedure ends by applying +it and leaving it applied. + +Both directions are measured, from the same starting tilt, and the two answers +are averaged. On a flat sample they agree; a disagreement is the read-out +saying the surface is not a plane (or that the platform has backlash), and it +is reported rather than averaged away silently. + +The three T-axes form a tip/tilt platform. Their azimuths on the platform +(see T_AXIS_AZIMUTH_DEG) decide which axis corrects which stage direction: +T1 lies along +X, so it alone tilts the platform along X; T0 and T2 sit at ++/-120 degrees from it and have to move as an equal-and-opposite pair to tilt +along Y without also tilting along X. ``tilt_response`` derives that from the +azimuths, so a re-plumbed platform is a one-line change to the azimuth map and +not a re-derivation of the whole procedure. + +Qt-free, like ScanEngine and AngleInspector: gui/align_bridge.py wraps it. +""" +from __future__ import annotations + +import logging +import math +import time +from dataclasses import dataclass, field +from typing import Callable + +from core import scope_inspect +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, StageLimits + +logger = logging.getLogger(__name__) + +# Where each T-axis sits on the tilt platform, in degrees from the stage's +X +# axis. T1 is co-linear with +X; T0 and T2 are the other two legs of the +# kinematic triangle. This map is the whole geometry — everything else about +# which axis moves when is derived from it. +T_AXIS_AZIMUTH_DEG = {0: 120.0, 1: 0.0, 2: 240.0} +T_AXES = tuple(sorted(T_AXIS_AZIMUTH_DEG)) +T_AXIS_LABELS = {ch: f"T{ch}" for ch in T_AXES} + +# Positioning moves only, so there is no reason to cross the tray at full scan +# velocity — the same halving the angle inspector uses. +ALIGN_VELOCITY_MM_S = SCAN_VELOCITY_MM_S / 2.0 + + +class AutoAlignError(RuntimeError): + """The procedure cannot continue: bad rig state, or nothing responding.""" + + +class AutoAlignAborted(RuntimeError): + """The operator stopped the procedure part-way through.""" + + +# ── Platform geometry ──────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class TiltGroup: + """The T-axis move that tilts the platform along one stage axis. + + ``weights`` maps a T-axis channel to the microsteps it contributes per + unit of correction, so a correction of ``c`` moves channel ``ch`` by + ``c * weights[ch]``. + """ + stage_axis: int # APT axis address of the stage axis this corrects + label: str # "X" or "Y", for the operator + weights: dict[int, float] + + def describe(self) -> str: + if len(self.weights) == 1: + return T_AXIS_LABELS[next(iter(self.weights))] + return " / ".join(f"{T_AXIS_LABELS[ch]} {w:+.0f}" + for ch, w in sorted(self.weights.items())) + + +# X is corrected by the one axis that lies along it, and Y by the other two +# driven equal and opposite — that pairing is what makes the Y move a pure +# tilt along Y (tilt_response(Y_TILT) has no X term), so the two phases of the +# procedure do not fight each other. Moving T1 alone does raise the platform +# as well as tilt it, which the search absorbs: it nulls a measured level, not +# a model of the platform. +X_TILT = TiltGroup(AXIS_X, "X", {1: +1.0}) +Y_TILT = TiltGroup(AXIS_Y, "Y", {0: +1.0, 2: -1.0}) +TILT_GROUPS = (X_TILT, Y_TILT) + + +def tilt_response(group: TiltGroup) -> tuple[float, float, float]: + """What one unit of ``group`` does to the platform: (piston, x_tilt, y_tilt). + + The three actuators define a plane, so their heights fix it exactly: + fitting ``z = piston + x_tilt * x + y_tilt * y`` through the three + (azimuth, weight) points is a closed-form solution on a symmetric triangle + — the mean is the piston and the projections onto x and y are the tilts, + scaled by 2/3 because each actuator sits one unit radius out. + + Used to check the groups above are the moves they claim to be, and to say + in one place what "moving T0 and T2 as a pair" actually produces. + """ + heights = {ch: group.weights.get(ch, 0.0) for ch in T_AXES} + piston = sum(heights.values()) / len(T_AXES) + x_tilt = y_tilt = 0.0 + for ch, h in heights.items(): + theta = math.radians(T_AXIS_AZIMUTH_DEG[ch]) + x_tilt += h * math.cos(theta) + y_tilt += h * math.sin(theta) + scale = 2.0 / len(T_AXES) + return piston, x_tilt * scale, y_tilt * scale + + +# ── Settings ───────────────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class TAxisSettings: + """Drive settings for the three T-axes during the procedure. + + 32 microsteps and 600 mA are the operating point this procedure is + specified at; they are applied to all three axes at the start rather than + trusted from whatever the T3R panel last left behind, because the search + reports its corrections in microsteps and a different microstep setting + would silently change what a step means. + """ + microsteps: int = 32 + run_current_ma: int = 600 + hold_current_ma: int = 300 # half of run: holds the platform, runs cool + ihold_delay: int = 6 + velocity: int = 4000 # steps/s — small moves, so ramps dominate + accel: int = 2000 # steps/s^2 + + +@dataclass(frozen=True) +class AlignSettings: + """How far to step, how close to get, and how hard to try.""" + offset_mm: float = 1.5 # stage step either side of the reference + tolerance_mv: float = 5.0 # "same DC values" means within this + probe_steps: int = 200 # first move of a search: gain is unknown + max_step_steps: int = 2000 # per-iteration clamp on a correction + max_excursion_steps: int = 20000 # per-axis limit from the starting tilt + max_iterations: int = 25 + max_probe_doublings: int = 4 # escalation when a probe reads as no response + min_response_mv: float = 2.0 # below this a probe has told us nothing + settle_s: float = 0.3 # after a move, before believing a reading + acquisition_retries: int = 2 # re-reads before calling the scope stalled + reads_per_measurement: int = scope_inspect.BIAS_READS + move_timeout_margin_s: float = 5.0 + stage_timeout_s: float = 60.0 + + +DEFAULT_T_AXIS = TAxisSettings() +DEFAULT_ALIGN = AlignSettings() + + +# ── Read-out ───────────────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class Reading: + """One measurement of the two DC bias levels, in millivolts.""" + dc1_mv: float + dc2_mv: float + + @property + def difference_mv(self) -> float: + """DC 1 - DC 2. + + The split-detector difference is what a tilt actually steers, so it is + the signal the search drives to zero; the sum is set by the laser and + the surface reflectivity, which no amount of tilting will change. + """ + return self.dc1_mv - self.dc2_mv + + def error_vs(self, ref: "Reading") -> tuple[float, float]: + return self.dc1_mv - ref.dc1_mv, self.dc2_mv - ref.dc2_mv + + def difference_error_vs(self, ref: "Reading") -> float: + return self.difference_mv - ref.difference_mv + + def matches(self, ref: "Reading", tolerance_mv: float) -> bool: + return all(abs(e) <= tolerance_mv for e in self.error_vs(ref)) + + def describe(self) -> str: + return f"DC1 {self.dc1_mv:+.1f} mV, DC2 {self.dc2_mv:+.1f} mV" + + +@dataclass +class OffsetResult: + """What one search — one stage offset, one tilt group — ended up doing.""" + axis_label: str + offset_mm: float + correction_steps: float + iterations: int + nulled: bool # difference back within tolerance: the tilt loop worked + converged: bool # both levels back within tolerance: the operator's test + reason: str + final: Reading + + def describe(self) -> str: + return (f"{self.axis_label}{self.offset_mm:+.2f} mm: " + f"{self.correction_steps:+.0f} usteps in {self.iterations} steps " + f"→ {self.final.describe()} ({self.reason})") + + +@dataclass +class AxisResult: + """Both offsets for one stage axis, and the tilt they agreed on.""" + axis_label: str + group: TiltGroup + offsets: list[OffsetResult] + applied_steps: float + disagreement_steps: float + applied: bool + reference_residual: Reading | None # measured back at the reference point + + @property + def ok(self) -> bool: + return self.applied and all(o.nulled for o in self.offsets) + + def describe(self) -> str: + if not self.applied: + return (f"{self.axis_label}: no correction applied — " + + "; ".join(o.reason for o in self.offsets)) + residual = (f", back at the reference {self.reference_residual.describe()}" + if self.reference_residual else "") + return (f"{self.axis_label}: applied {self.applied_steps:+.0f} usteps on " + f"{self.group.describe()} (the two directions disagreed by " + f"{self.disagreement_steps:.0f} usteps){residual}") + + +@dataclass +class AlignResult: + """The whole procedure, as the summary the operator is shown.""" + reference: Reading + axes: list[AxisResult] = field(default_factory=list) + final: Reading | None = None + tolerance_mv: float = DEFAULT_ALIGN.tolerance_mv + + @property + def ok(self) -> bool: + return bool(self.axes) and all(a.ok for a in self.axes) and ( + self.final is not None + and self.final.matches(self.reference, self.tolerance_mv)) + + def verdict(self) -> str: + """The closing lines: where it ended up, and whether that is aligned.""" + lines = [] + if self.final is not None: + d1, d2 = self.final.error_vs(self.reference) + lines.append(f"Final at the reference point: {self.final.describe()} " + f"({d1:+.1f} / {d2:+.1f} mV from the good values)") + lines.append("Aligned." if self.ok else + "Finished without meeting the tolerance — see the log.") + return "\n".join(lines) + + def describe(self) -> str: + lines = [f"Reference: {self.reference.describe()}"] + for axis in self.axes: + lines.append(axis.describe()) + lines.extend(f" {o.describe()}" for o in axis.offsets) + lines.append(self.verdict()) + return "\n".join(lines) + + +@dataclass +class AlignCallbacks: + """Progress reporting. Defaults are no-ops so the core needs no front end.""" + on_status: Callable[[str], None] = lambda msg: None + on_reading: Callable[[Reading], None] = lambda r: None + on_busy: Callable[[bool], None] = lambda busy: None + on_offset_done: Callable[[OffsetResult], None] = lambda r: None + on_axis_done: Callable[[AxisResult], None] = lambda r: None + + +# ── The tilt platform ──────────────────────────────────────────────────────── + +class _TiltPlatform: + """The three T-axes, driven in microsteps relative to where they started. + + Positions are tracked as floats and commanded as integers, with the + rounding residue carried forward, so a long run of fractional corrections + cannot drift the platform away from what the procedure thinks it applied. + """ + + def __init__(self, driver, settings: TAxisSettings, max_excursion_steps: int, + timeout_margin_s: float = 5.0): + self._driver = driver + self._s = settings + self._max = max_excursion_steps + self._timeout_margin_s = timeout_margin_s + self._target = {ch: 0.0 for ch in T_AXES} # wanted, fractional + self._actual = {ch: 0 for ch in T_AXES} # commanded, integral + + @property + def positions(self) -> dict[int, int]: + return dict(self._actual) + + def configure(self) -> None: + s = self._s + for ch in T_AXES: + self._driver.set_microstep(ch, s.microsteps) + self._driver.set_current(ch, s.run_current_ma, s.hold_current_ma, + s.ihold_delay) + self._driver.enable(ch) + + def snapshot(self) -> dict[int, float]: + return dict(self._target) + + def apply(self, group: TiltGroup, amount: float) -> None: + """Move the group by ``amount`` units of its weights.""" + self._goto({ch: self._target[ch] + amount * w + for ch, w in group.weights.items()}) + + def restore(self, snapshot: dict[int, float]) -> None: + self._goto(snapshot) + + def _goto(self, targets: dict[int, float]) -> None: + for ch, target in targets.items(): + if abs(target) > self._max: + raise AutoAlignError( + f"{T_AXIS_LABELS[ch]} would travel {target:+.0f} microsteps " + f"from where it started, past the {self._max} microstep " + f"safety limit. Stopping before the actuator runs out of " + f"travel — align the rig by hand and start again.") + self._target[ch] = target + delta = round(target) - self._actual[ch] + if delta: + self._move(ch, delta) + self._actual[ch] += delta + + def _move(self, ch: int, steps: int) -> None: + s = self._s + self._driver.move(ch, steps, s.velocity, s.accel) + timeout = (abs(steps) / s.velocity + s.velocity / s.accel + + self._timeout_margin_s) + if not self._driver.wait_motion_done(ch, timeout): + logger.warning("%s did not report MOTION_DONE within %.1f s for a " + "%+d microstep move; continuing", + T_AXIS_LABELS[ch], timeout, steps) + + +# ── The procedure ──────────────────────────────────────────────────────────── + +class AutoAligner: + """Drives the stage and the tilt platform to level the sample. + + Usage is two-phase because the operator sits in the middle of it: + ``prepare()`` puts the rig in a known state and reads the bias levels at + the reference point, the operator confirms the image and those levels are + the ones to hold, then ``run()`` measures and applies the tilt. + """ + + def __init__(self, stage, scope, t3r, + settings: AlignSettings = DEFAULT_ALIGN, + t_axis: TAxisSettings = DEFAULT_T_AXIS, + limits: StageLimits = DEFAULT_STAGE_LIMITS, + callbacks: AlignCallbacks | None = None, + should_abort: Callable[[], bool] = lambda: False): + self._stage = stage + self._scope = scope + self._t3r = t3r + self._s = settings + self._cb = callbacks if callbacks is not None else AlignCallbacks() + self._limits = limits + self._should_abort = should_abort + + self._platform = _TiltPlatform(t3r, t_axis, settings.max_excursion_steps, + settings.move_timeout_margin_s) + self._reference: Reading | None = None + self._ref_mm: tuple[float, float] | None = None + self._last_acq: int | None = None + self._started = False + + # ── Lifecycle ───────────────────────────────────────────────────────────── + + @property + def reference(self) -> Reading | None: + return self._reference + + @property + def reference_position_mm(self) -> tuple[float, float] | None: + return self._ref_mm + + def prepare(self) -> Reading: + """Configure the rig and read the bias levels where it stands. + + The returned reading is a candidate, not yet the reference: the + operator has to confirm the camera image is the one to align on before + anything moves. + """ + self._require_hardware() + self._cb.on_busy(True) + try: + x_mm, y_mm = self._stage_position() + self._check_travel(x_mm, y_mm) + self._ref_mm = (x_mm, y_mm) + + self._cb.on_status("Configuring stage …") + for axis in (AXIS_X, AXIS_Y): + self._stage.set_velocity_params(axis, + max_velocity=ALIGN_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. + self._stage.set_trigger_gate_off(AXIS_X) + + self._cb.on_status("Configuring oscilloscope …") + scope_inspect.configure_inspection(self._scope) + + self._cb.on_status( + f"Configuring T-axes ({DEFAULT_T_AXIS.microsteps} usteps, " + f"{DEFAULT_T_AXIS.run_current_ma} mA) …") + self._platform.configure() + + self._started = True + self._cb.on_status("Reading the DC levels at the reference point …") + reading = self._measure() + self._reference = reading + return reading + finally: + self._cb.on_busy(False) + + def run(self) -> AlignResult: + """Measure and apply the tilt, X first and then Y. + + Y follows X because the two corrections are independent moves (see + ``tilt_response``) but not independent measurements: the X phase is + the one that can reveal a platform whose pivot is not under the beam, + and its reference residual is reported before Y adds to it. + """ + if not self._started or self._reference is None: + raise AutoAlignError("prepare() must run, and the operator must " + "confirm the image, before run()") + result = AlignResult(reference=self._reference, + tolerance_mv=self._s.tolerance_mv) + self._cb.on_busy(True) + try: + for group in TILT_GROUPS: + axis_result = self._align_axis(group) + result.axes.append(axis_result) + self._cb.on_axis_done(axis_result) + if not axis_result.applied: + break + result.final = self._measure() + return result + finally: + self._cb.on_busy(False) + + def stop(self) -> None: + """Park the rig: stage back at the reference point, scope idle. + + The tilt correction stays applied — it is the result. Safe to call + twice, and safe to call after a failure part-way through. + """ + if not self._started: + return + self._started = False + self._cb.on_busy(True) + try: + if self._ref_mm is not None: + try: + self._cb.on_status("Returning to the reference point …") + # Deliberately not abort-checked: this *is* the response to + # an abort, and a stop that left the stage 1.5 mm off the + # operator's point would be worse than no stop at all. + self._goto_reference(check_abort=False) + except Exception: + logger.exception("Could not return the stage to the " + "reference point") + try: + scope_inspect.stop_inspection(self._scope) + except Exception: + logger.exception("Could not stop the inspection acquisition") + self._cb.on_status("Auto-align finished.") + finally: + self._cb.on_busy(False) + + # ── One stage axis ──────────────────────────────────────────────────────── + + def _align_axis(self, group: TiltGroup) -> AxisResult: + name = group.label + start_tilt = self._platform.snapshot() + offsets: list[OffsetResult] = [] + + for sign in (+1.0, -1.0): + offset_mm = sign * self._s.offset_mm + self._cb.on_status( + f"{name}: stepping {offset_mm:+.2f} mm and re-tilting on " + f"{group.describe()} …") + self._goto_offset(group, offset_mm) + offsets.append(self._null(group, name, offset_mm)) + self._cb.on_offset_done(offsets[-1]) + # Both directions are measured from the same tilt, so they are two + # independent estimates of the same angle rather than one estimate + # and one correction to it. + self._platform.restore(start_tilt) + + applied = 0.5 * sum(o.correction_steps for o in offsets) + disagreement = abs(offsets[0].correction_steps - offsets[1].correction_steps) + can_apply = all(o.nulled for o in offsets) + + if can_apply: + self._cb.on_status(f"{name}: applying {applied:+.0f} microsteps …") + self._platform.apply(group, applied) + else: + self._cb.on_status( + f"{name}: no correction applied — a search did not null the " + f"DC difference, so the tilt it found means nothing.") + + residual = None + self._cb.on_status(f"{name}: back to the reference point …") + self._goto_reference() + if can_apply: + residual = self._measure() + + return AxisResult(axis_label=name, group=group, offsets=offsets, + applied_steps=applied if can_apply else 0.0, + disagreement_steps=disagreement, + applied=can_apply, reference_residual=residual) + + # ── One search ──────────────────────────────────────────────────────────── + + def _null(self, group: TiltGroup, name: str, offset_mm: float) -> OffsetResult: + """Tilt until the bias levels read what they read at the reference. + + A secant search on the split-detector difference: probe once to learn + how many millivolts a microstep is worth (the sign included — which + way is "up" is a wiring question this refuses to assume), then step + straight at the null and re-estimate the slope from each pair of + readings. + + Ends on one of three outcomes, which mean different things: + *converged* — both levels back inside the tolerance, the result the + operator asked for; *difference nulled* — the beam is back on the + centre of the detector but both levels sit at the wrong height, which + no tilt can fix and which therefore still leaves a usable tilt answer; + anything else means the search failed and its answer must not be used. + """ + s = self._s + reading = self._measure() + applied = 0.0 + gain: float | None = None # mV of difference error per microstep + probe = float(s.probe_steps) + doublings = 0 + + for iteration in range(1, s.max_iterations + 1): + err = reading.difference_error_vs(self._reference) + d1, d2 = reading.error_vs(self._reference) + self._cb.on_status( + f"{name}{offset_mm:+.2f} mm, step {iteration}: " + f"DC1 {d1:+.1f} / DC2 {d2:+.1f} mV from the good values") + + if reading.matches(self._reference, s.tolerance_mv): + return self._offset_result(name, offset_mm, applied, iteration, + True, True, "within tolerance", reading) + if abs(err) <= s.tolerance_mv: + return self._offset_result( + name, offset_mm, applied, iteration, True, False, + f"difference nulled, but both levels are off by " + f"{0.5 * (d1 + d2):+.1f} mV — not something tilt can fix", + reading) + + if gain is None: + step = probe + else: + step = max(-s.max_step_steps, min(s.max_step_steps, -err / gain)) + if abs(step) < 1.0: + step = math.copysign(1.0, step) + + self._platform.apply(group, step) + applied += step + previous, reading = reading, self._measure() + response = reading.difference_error_vs(self._reference) - err + + if abs(response) >= s.min_response_mv: + gain = response / step + elif gain is None: + # The probe moved nothing measurable. Usually the probe is + # simply too small for this actuator's pitch, so escalate + # before concluding the axis is dead. + self._platform.apply(group, -step) + applied -= step + reading = previous + doublings += 1 + if doublings > s.max_probe_doublings: + raise AutoAlignError( + f"{group.describe()} moved {probe:.0f} microsteps and " + f"the DC difference did not change by " + f"{s.min_response_mv:.0f} mV. Check that the laser is " + f"pulsing, that CH3/CH4 are the DC monitors, and that " + f"the T-axes are energised.") + probe *= 2.0 + + return self._offset_result( + name, offset_mm, applied, s.max_iterations, False, False, + f"gave up after {s.max_iterations} steps", reading) + + def _offset_result(self, name, offset_mm, applied, iterations, nulled, + converged, reason, reading) -> OffsetResult: + return OffsetResult(axis_label=name, offset_mm=offset_mm, + correction_steps=applied, iterations=iterations, + nulled=nulled, converged=converged, reason=reason, + final=reading) + + # ── Hardware ────────────────────────────────────────────────────────────── + + def _require_hardware(self): + if self._stage is None: + raise AutoAlignError("BBD202 not connected") + if self._scope is None: + raise AutoAlignError("Oscilloscope not connected") + if self._t3r is None or not self._t3r.is_open: + raise AutoAlignError( + "The T3R is not connected, so the T-axes cannot be moved. " + "Connect it from the T3R panel and try again.") + + def _stage_position(self) -> tuple[float, float]: + try: + return float(self._stage.positions[0]), float(self._stage.positions[1]) + except Exception as exc: + raise AutoAlignError(f"Cannot read the stage position: {exc}") from exc + + def _check_travel(self, x_mm: float, y_mm: float) -> None: + """Both offsets on both axes have to be reachable before anything moves.""" + off = self._s.offset_mm + lim = self._limits + for name, value, lo, hi in (("X", x_mm, lim.x_min, lim.x_max), + ("Y", y_mm, lim.y_min, lim.y_max)): + if value - off < lo or value + off > hi: + raise AutoAlignError( + f"{name} is at {value:.3f} mm, and this procedure needs " + f"{off:.2f} mm either side of it — that leaves the " + f"{lo:g}–{hi:g} mm travel. Move to a point further from " + f"the end of travel and try again.") + + def _goto_offset(self, group: TiltGroup, offset_mm: float) -> None: + x_mm, y_mm = self._ref_mm + target = (x_mm if group.stage_axis == AXIS_X else y_mm) + offset_mm + self._move_stage(group.stage_axis, target) + + def _goto_reference(self, check_abort: bool = True) -> None: + x_mm, y_mm = self._ref_mm + self._move_stage(AXIS_X, x_mm, check_abort=check_abort) + self._move_stage(AXIS_Y, y_mm, check_abort=check_abort) + + def _move_stage(self, stage_axis: int, position_mm: float, + check_abort: bool = True) -> None: + if check_abort: + self._check_abort() + self._stage.move_axis_absolute(stage_axis, position_mm, + timeout=self._s.stage_timeout_s) + + def _measure(self) -> Reading: + """Settle, check the scope is still acquiring, then read both levels.""" + self._check_abort() + time.sleep(self._s.settle_s) + self._require_fresh_acquisition() + dc1, dc2 = scope_inspect.read_bias_mv(self._scope, + self._s.reads_per_measurement) + reading = Reading(dc1_mv=dc1, dc2_mv=dc2) + self._cb.on_reading(reading) + return reading + + def _require_fresh_acquisition(self) -> None: + """Refuse to servo on a record the scope has not re-taken. + + A stale record reads as a perfectly stable measurement, which is the + one failure this loop cannot see for itself: it would keep stepping + the actuators against a number that never moves. + + The very first reading has no previous count to compare against, so it + makes one — waiting for the count to move rather than assuming it + does. That reading becomes the reference the operator confirms and + the whole procedure then chases, which makes it the worst one to take + off a scope that is not triggering. + """ + baseline = self._last_acq + count = self._scope.get_acquisition_count() + if baseline is None: + baseline = count + for _ in range(self._s.acquisition_retries): + if count != baseline: + self._last_acq = count + return + time.sleep(self._s.settle_s) + count = self._scope.get_acquisition_count() + if count == baseline: + raise AutoAlignError( + "The oscilloscope has not triggered since the last reading, " + "so its DC levels are stale. Check that the laser is pulsing " + "and that CH2 carries the trigger.") + self._last_acq = count + + def _check_abort(self) -> None: + if self._should_abort(): + raise AutoAlignAborted("Auto-align stopped by the operator") diff --git a/core/scope_inspect.py b/core/scope_inspect.py index 3e305cc..9837f02 100644 --- a/core/scope_inspect.py +++ b/core/scope_inspect.py @@ -14,6 +14,11 @@ 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). + +``read_bias_mv`` is the one exception to "nothing is transferred": it reads +the two bias levels back as scalars, not waveforms, because the auto-align +procedure (core.auto_align) has to close a loop on them. The operator still +watches the same screen this configures. """ from __future__ import annotations @@ -46,6 +51,12 @@ BIAS_POSITION_DIV = -3.5 BIAS_LABELS = {3: "Bias - A", 4: "Bias - B"} +# One MEAN measurement carries the shot-to-shot noise of a single record, and +# the alignment loop has to resolve 5 mV. The median of a handful of reads +# rejects the odd outlier without the averaging acquisition mode, which would +# hide exactly the intermittent response the operator is watching CH1 for. +BIAS_READS = 5 + def inspect_channel_profiles() -> dict: """Channel front-end config for inspection. @@ -102,3 +113,24 @@ def stop_inspection(scope) -> None: stop the sweep — it does not try to restore the acquisition profile. """ scope.write("ACQuire:STATE STOP") + + +def read_bias_mv(scope, reads: int = BIAS_READS) -> tuple[float, float]: + """Read the two DC bias levels in millivolts. + + Returns ``(ch3_mv, ch4_mv)`` — DC 1 and DC 2 in the auto-align channel + map. Each channel is read ``reads`` times and reduced by the median. + + The two channels are read in separate batches rather than interleaved: + switching the immediate-measurement source costs a round trip, and these + are DC levels, so the few milliseconds between the batches are not a + source of error the way they would be for a transient. + """ + if reads < 1: + raise ValueError("read_bias_mv needs at least one read per channel") + + levels = [] + for ch in BIAS_CHANNELS: + samples = sorted(scope.measure_immediate(ch, "MEAN") for _ in range(reads)) + levels.append(samples[len(samples) // 2] * 1000.0) + return levels[0], levels[1] diff --git a/gui/align_bridge.py b/gui/align_bridge.py new file mode 100644 index 0000000..515fb34 --- /dev/null +++ b/gui/align_bridge.py @@ -0,0 +1,124 @@ +"""Qt bridge over the headless AutoAligner. + +The procedure is two long blocking runs with an operator decision between +them — ``prepare()`` puts the rig in a known state and reads the reference +levels, the operator confirms the camera image, then ``run()`` spends a minute +or two moving the stage and the tilt platform. Both belong on a worker +thread; the window only enqueues and reacts to signals. + +Stopping cannot go through the command queue: while ``run()`` is executing, +the worker is inside a handler and will not look at the queue until it +returns. The stop request is therefore a threading.Event the core polls +between moves (``should_abort``), and the queued "stop" command only handles +the tidy-up afterwards. +""" +from __future__ import annotations + +import threading +import traceback + +from PyQt6.QtCore import pyqtSignal + +from core.auto_align import AlignCallbacks, AutoAligner, AutoAlignAborted +from gui.qt_workers import QueueWorker + + +class QtAutoAligner(QueueWorker): + """Runs an AutoAligner on its own QThread and republishes its events.""" + + prepared = pyqtSignal(object) # Reading — the candidate reference + prepare_failed = pyqtSignal(str) + status_msg = pyqtSignal(str) + reading_taken = pyqtSignal(object) # Reading + busy_changed = pyqtSignal(bool) + offset_done = pyqtSignal(object) # OffsetResult + axis_done = pyqtSignal(object) # AxisResult + finished = pyqtSignal(object) # AlignResult + failed = pyqtSignal(str) + aborted = pyqtSignal() + stopped = pyqtSignal() + + def __init__(self, stage, scope, t3r, settings=None, on_align_active=None): + super().__init__() + self._on_align_active = on_align_active + self._abort = threading.Event() + + callbacks = AlignCallbacks( + on_status=self.status_msg.emit, + on_reading=self.reading_taken.emit, + on_busy=self.busy_changed.emit, + on_offset_done=self.offset_done.emit, + on_axis_done=self.axis_done.emit, + ) + kwargs = {"settings": settings} if settings is not None else {} + self._aligner = AutoAligner(stage, scope, t3r, callbacks=callbacks, + should_abort=self._abort.is_set, **kwargs) + self._handlers = { + "prepare": self._do_prepare, + "run": self._do_run, + "stop": self._do_stop, + } + + # ── Command submission (GUI thread) ─────────────────────────────────────── + + def request_prepare(self): + self._abort.clear() + self._enqueue("prepare") + + def request_run(self): + self._enqueue("run") + + def request_abort(self): + """Stop the procedure at the next move, wherever it has got to.""" + self._abort.set() + + def request_stop(self): + self._abort.set() + self._enqueue("stop") + + # ── Handlers (worker thread) ────────────────────────────────────────────── + + def _do_prepare(self): + if self._on_align_active is not None: + self._on_align_active(True) + try: + reading = self._aligner.prepare() + except Exception as exc: + traceback.print_exc() + if self._on_align_active is not None: + self._on_align_active(False) + self.prepare_failed.emit(str(exc)) + return + self.prepared.emit(reading) + + def _do_run(self): + try: + result = self._aligner.run() + except AutoAlignAborted: + self.status_msg.emit("Auto-align stopped.") + self.aborted.emit() + return + except Exception as exc: + traceback.print_exc() + self.failed.emit(str(exc)) + return + self.finished.emit(result) + + def _do_stop(self): + try: + self._aligner.stop() + finally: + if self._on_align_active is not None: + self._on_align_active(False) + self.stopped.emit() + + def _on_stop(self): + """Worker loop exiting — leave the rig parked even if the window went + away without a clean stop command reaching the queue.""" + try: + self._aligner.stop() + except Exception: + traceback.print_exc() + finally: + if self._on_align_active is not None: + self._on_align_active(False) diff --git a/gui/jog_panel.py b/gui/jog_panel.py index 1e9b703..342c091 100644 --- a/gui/jog_panel.py +++ b/gui/jog_panel.py @@ -15,13 +15,13 @@ already does it. from __future__ import annotations from PyQt6.QtCore import Qt, QTimer -from PyQt6.QtGui import QFont from PyQt6.QtWidgets import ( QCheckBox, QComboBox, QDoubleSpinBox, QFrame, QGridLayout, QGroupBox, QLabel, QPushButton, QSpinBox, ) import hardware.t3r_protocol as proto +from gui.widgets import mono_font from hardware.t3r_driver import T3RDriver # BBD202 jog defaults, shared with the main window's worker. @@ -44,13 +44,6 @@ BBD_JOG_REPEAT_MS = 200 BBD_VELOCITY_DEBOUNCE_MS = 300 -def _mono(size: int = 11) -> QFont: - f = QFont("Menlo") - f.setStyleHint(QFont.StyleHint.Monospace) - f.setPointSize(size) - return f - - def _hline() -> QFrame: line = QFrame() line.setFrameShape(QFrame.Shape.HLine) @@ -157,7 +150,7 @@ class T3RJogPanel(QGroupBox): self._micro_combos[ch] = combo pos_lbl = QLabel("—") - pos_lbl.setFont(_mono(11)) + pos_lbl.setFont(mono_font(11)) pos_lbl.setMinimumWidth(76) pos_lbl.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) grid.addWidget(pos_lbl, row, 4) @@ -280,11 +273,11 @@ class BBDJogPanel(QGroupBox): grid.addWidget(QLabel("X"), row, 0) self.x_pos_lbl = QLabel("---.---") - self.x_pos_lbl.setFont(_mono(11)) + self.x_pos_lbl.setFont(mono_font(11)) grid.addWidget(self.x_pos_lbl, row, 1) grid.addWidget(QLabel("Y"), row, 2) self.y_pos_lbl = QLabel("---.---") - self.y_pos_lbl.setFont(_mono(11)) + self.y_pos_lbl.setFont(mono_font(11)) grid.addWidget(self.y_pos_lbl, row, 3) row += 1 diff --git a/gui/widgets.py b/gui/widgets.py index aaace8a..825973d 100644 --- a/gui/widgets.py +++ b/gui/widgets.py @@ -17,6 +17,14 @@ from PyQt6.QtWidgets import ( from hardware.serial_util import scored_ports +def mono_font(size: int = 11) -> QFont: + """Monospace font for numeric read-outs, so columns of digits line up.""" + font = QFont("Menlo") + font.setStyleHint(QFont.StyleHint.Monospace) + font.setPointSize(size) + return font + + def set_toggle(btn, checked: bool, text: str, enabled: bool = True): """Update a checkable button without re-triggering its toggled signal.""" btn.blockSignals(True) @@ -138,10 +146,7 @@ class LogConsole(QWidget): self.view = QPlainTextEdit() self.view.setReadOnly(True) self.view.setMaximumBlockCount(max_blocks) - font = QFont("Menlo") - font.setStyleHint(QFont.StyleHint.Monospace) - font.setPointSize(11) - self.view.setFont(font) + self.view.setFont(mono_font()) layout.addWidget(self.view) row = QHBoxLayout() diff --git a/hardware/tektronix_base.py b/hardware/tektronix_base.py index b1184c8..54f8f3a 100755 --- a/hardware/tektronix_base.py +++ b/hardware/tektronix_base.py @@ -344,6 +344,72 @@ class TektronixOscilloscopeBase: self.write(f"CH{channel}:TERmination {termination}") + # ========== Measurement Methods ========== + + # Immediate measurements the alignment/inspection code asks for. The + # instrument accepts many more; this list is what has been exercised here, + # and an unlisted type is far more likely to be a typo than a deliberate + # choice. + MEASUREMENT_TYPES = { + 'MEAN': ['MEAN'], + 'AMPLITUDE': ['AMPlitude', 'AMPLITUDE'], + 'MAXIMUM': ['MAXimum', 'MAXIMUM'], + 'MINIMUM': ['MINImum', 'MINIMUM'], + 'PK2PK': ['PK2pk', 'PK2PK'], + 'RMS': ['RMS'], + } + + # Tektronix returns this sentinel when a measurement cannot be made (no + # acquisition yet, source off, signal outside the graticule). It is a + # valid float, so it has to be caught explicitly or it reads as a + # 1e38 V measurement. + MEASUREMENT_INVALID = 9.9e37 + + def measure_immediate(self, channel, measurement_type='MEAN'): + """Take an immediate measurement on one channel and return it in volts. + + "Immediate" measurements are computed on demand and are not added to + the scope's on-screen measurement badges, so this leaves whatever the + operator has set up on the front panel untouched. + + Raises ValueError if the instrument reports the measurement as + unavailable, which on a triggered-acquisition scope usually means it + has not acquired anything yet. + """ + channel = self._normalize_channel(channel) + + if measurement_type.upper() not in self.MEASUREMENT_TYPES: + raise ValueError( + f"Invalid measurement type: {measurement_type}. " + f"Valid options: {', '.join(self.MEASUREMENT_TYPES)}") + + self.write(f"MEASUrement:IMMed:SOUrce1 CH{channel}") + self.write(f"MEASUrement:IMMed:TYPe {measurement_type}") + response = self.query("MEASUrement:IMMed:VALue?") + try: + value = float(response) + except ValueError as exc: + raise ValueError( + f"Unparseable {measurement_type} measurement on CH{channel}: " + f"{response!r}") from exc + + if abs(value) >= self.MEASUREMENT_INVALID: + raise ValueError( + f"CH{channel} {measurement_type} is unavailable (the scope " + f"returned its no-measurement sentinel). Check that the " + f"channel is on and that the acquisition is triggering.") + return value + + def get_acquisition_count(self): + """Number of acquisitions since the acquisition was last started. + + A caller polling a free-running scope uses this to tell a fresh + reading from a stale one: if the count has not moved, the record has + not changed and every measurement taken off it is the previous + answer. + """ + return int(float(self.query("ACQuire:NUMACq?"))) + # ========== Waveform Transfer Methods ========== def set_data_source(self, source): diff --git a/sc3-aui-camera.ui b/sc3-aui-camera.ui index f7404d4..aba9b5e 100755 --- a/sc3-aui-camera.ui +++ b/sc3-aui-camera.ui @@ -78,6 +78,16 @@ + + + + Auto-Align… + + + Level the sample: step the stage 1.5 mm each way and re-tilt the T-axes until the DC bias levels read what they read here. + + + diff --git a/sc3_aui_app.py b/sc3_aui_app.py index 787911c..2a811c9 100755 --- a/sc3_aui_app.py +++ b/sc3_aui_app.py @@ -18,8 +18,8 @@ from PyQt6.QtCore import QThread, QTimer, Qt, pyqtSignal from PyQt6.QtGui import QImage, QPixmap from PyQt6.QtWidgets import ( QApplication, QDialog, QDialogButtonBox, QFileDialog, QHBoxLayout, QLabel, - QListWidget, QListWidgetItem, QMainWindow, QMessageBox, QPushButton, - QSizePolicy, QVBoxLayout, QWidget, + QListWidget, QListWidgetItem, QMainWindow, QMessageBox, QPlainTextEdit, + QPushButton, QSizePolicy, QVBoxLayout, QWidget, ) from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from matplotlib.figure import Figure @@ -27,6 +27,7 @@ from matplotlib.figure import Figure ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) +from core.auto_align import DEFAULT_ALIGN, DEFAULT_T_AXIS from core.config import ScanDefaults from core.rotation import DEFAULT_ROTATION, RotationAxis from core.scan_engine import ( @@ -36,11 +37,13 @@ from core.scan_engine import ( from core.saw_check import middle_row_plan from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta from core.scan_resume import is_compatible, plan_resume +from core.scope_inspect import BIAS_LABELS from core.scope_sras import SAMPLE_RATE_HZ, configure_channels from core.sras_format import ( SCAN_CHANNELS, VERSION, VERSION_SAW_CHECK, WRITABLE_VERSIONS, SrasFile, plan_from_header, ) +from gui.align_bridge import QtAutoAligner from gui.jog_panel import ( BBD_JOG_ACCEL_MM_S2, BBD_JOG_SPEED_MM_S, BBD_JOG_STEP_MM, BBDJogPanel, T3RJogPanel, @@ -49,6 +52,7 @@ from gui.qt_t3r import QtT3RAdapter from gui.qt_workers import PollingQueueWorker, QueueWorker from gui.inspect_bridge import QtAngleInspector from gui.scan_bridge import QtScanController +from gui.widgets import mono_font from hardware.helios_laser import HeliosLaser from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver from hardware.tektronix_base import TektronixOscilloscopeBase @@ -439,18 +443,62 @@ class HeliosWorker(PollingQueueWorker): # ── Camera window popup ─────────────────────────────────────────────────────── +# The channel map auto-align assumes. Nothing reconfigures the scope inputs — +# CH3 is the max-velocity gate during a scan and the DC 1 monitor here — so the +# operator is asked to confirm the cabling instead of being trusted to +# remember it. +ALIGN_SCOPE_CHANNELS = ((1, "SAW"), (2, "Trigger"), (3, "DC 1"), (4, "DC 2")) + + +def _align_scope_prompt() -> str: + channels = "\n".join(f" CH{ch} — {role}" for ch, role in ALIGN_SCOPE_CHANNELS) + # The channels arrive on screen carrying the inspection labels, which are + # the names for the same two monitors — say so, or the operator is left + # comparing "DC 1" here against "Bias - A" on the instrument. + labels = " and ".join(BIAS_LABELS[ch] for ch in sorted(BIAS_LABELS)) + return ( + "Auto-align reads the DC levels off CH3 and CH4. Check the scope is " + "cabled this way before starting:\n\n" + f"{channels}\n\n" + f"The scope will label the last two {labels}.\n\n" + "The laser must be pulsing and the detection beam on the sample — the " + "procedure stops rather than servo on a scope that is not triggering.\n\n" + "Start auto-align?" + ) + + +def _align_reference_prompt(reading) -> str: + return ( + "The detector reads, where the stage is standing:\n\n" + f" DC 1 {reading.dc1_mv:8.1f} mV\n" + f" DC 2 {reading.dc2_mv:8.1f} mV\n\n" + "Is the image correct?\n\n" + f"Yes takes these as the good values and holds them to within " + f"{DEFAULT_ALIGN.tolerance_mv:.0f} mV at " + f"{DEFAULT_ALIGN.offset_mm:.2f} mm either side of here, first on X " + f"(T1) and then on Y (T0/T2 as a pair)." + ) + + class CameraWindow(QWidget): """Camera display popup — auto-connects on show, auto-disconnects on close. Focusing and framing are done by eye, so the T3R and BBD202 jog controls live beside the image. Both take the objects the main window already owns; passing neither leaves the window as a plain viewer. + + Auto-align lives here for the same reason: the operator judges the image + to decide the rig is on a good spot, and that judgement is the first step + of the procedure. It needs the oscilloscope as well, so the button is + only offered when all three are on hand. """ closed = pyqtSignal() + align_active = pyqtSignal(bool) # lock the scan panel while aligning def __init__(self, t3r_driver: QtT3RAdapter | None = None, bbd_worker: BBD202Worker | None = None, + oscope_worker: "OscopeWorker | None" = None, parent: QWidget | None = None): super().__init__(parent, Qt.WindowType.Window) uic.loadUi(ROOT / "sc3-aui-camera.ui", self) @@ -459,6 +507,14 @@ class CameraWindow(QWidget): self._camera: UC480Camera | None = None self._stream: CameraStreamThread | None = None + self._t3r_driver = t3r_driver + self._bbd_worker = bbd_worker + self._oscope_worker = oscope_worker + self._align_thread: QThread | None = None + self._align_worker: QtAutoAligner | None = None + self._align_window: "AutoAlignWindow | None" = None + self._align_available = True + # Banner warning when the camera shares a USB controller with serial # adapters — opening any of those ports (T3R, BBD202, …) collapses # the delivered frame rate to <2 fps (USB split-transaction @@ -489,6 +545,13 @@ class CameraWindow(QWidget): self.uc480_stop_btn.clicked.connect(self._stop_stream) self.uc480_close_window_btn.clicked.connect(self.close) + # Every one of the three is load-bearing: the T3R moves the tilt + # platform, the BBD202 makes the 1.5 mm steps, and the scope is the + # only thing that can say whether either helped. + self.uc480_auto_align_btn.clicked.connect(self._on_auto_align) + self.uc480_auto_align_btn.setVisible( + None not in (t3r_driver, bbd_worker, oscope_worker)) + self.t3r_jog_panel = self.bbd_jog_panel = None self._build_jog_column(t3r_driver, bbd_worker) @@ -515,6 +578,8 @@ class CameraWindow(QWidget): self._start_stream() def closeEvent(self, event): + # An alignment outlives this window otherwise, and it owns the stage. + self._teardown_align() # A jog whose button-release lands after the window is gone would # otherwise leave an axis running. for panel in (self.t3r_jog_panel, self.bbd_jog_panel): @@ -609,6 +674,298 @@ class CameraWindow(QWidget): self.uc480_exposure_slider.setEnabled(camera_ok) self.uc480_gain_slider.setEnabled(camera_ok) + # ── Auto-align ──────────────────────────────────────────────────────────── + + def _on_auto_align(self): + if self._align_thread is not None and self._align_thread.isRunning(): + if self._align_window is not None: + self._align_window.raise_() + self._align_window.activateWindow() + return + + problem = self._align_prerequisite_problem() + if problem: + QMessageBox.warning(self, "Cannot Auto-Align", problem) + return + + if QMessageBox.question(self, "Check the Oscilloscope", + _align_scope_prompt()) \ + != QMessageBox.StandardButton.Yes: + return + + self._align_thread = QThread(self) + self._align_worker = QtAutoAligner( + stage=self._bbd_worker.controller, + scope=self._oscope_worker.scope, + t3r=self._t3r_driver.driver, + # Same reason the scan and the inspector do it: the procedure + # drives the stage from its own thread, and the position poll + # shares the BBD TX queue. + on_align_active=lambda active: setattr( + self._bbd_worker, "scanning_active", active), + ) + self._align_worker.moveToThread(self._align_thread) + + window = AutoAlignWindow(self) + self._align_window = window + worker = self._align_worker + worker.status_msg.connect(window.on_status) + worker.reading_taken.connect(window.on_reading) + worker.offset_done.connect(window.on_offset_done) + worker.axis_done.connect(window.on_axis_done) + worker.prepared.connect(self._on_align_prepared) + worker.prepare_failed.connect(self._on_align_prepare_failed) + worker.finished.connect(self._on_align_finished) + worker.failed.connect(self._on_align_failed) + worker.aborted.connect(self._on_align_aborted) + worker.stopped.connect(self._release_align_thread) + worker.error_occurred.connect(window.on_failed) + window.abort_requested.connect(worker.request_abort) + window.close_requested.connect(self._teardown_align) + + self._align_thread.started.connect(worker.run) + self._set_align_ui_active(True) + window.show() + self._align_thread.start() + worker.request_prepare() + + def _align_prerequisite_problem(self) -> str: + """Why auto-align cannot start, or "" if it can.""" + if self._oscope_worker is None or not self._oscope_worker.is_connected: + return ("The oscilloscope is not connected, and it is the only " + "thing that can read the DC levels. Connect it from the " + "main window and try again.") + if self._bbd_worker is None or not self._bbd_worker.is_connected: + return ("The BBD202 stage is not connected, so the 1.5 mm steps " + "cannot be made. Connect it from the main window and try " + "again.") + if self._t3r_driver is None or not self._t3r_driver.is_open: + return ("The T3R is not connected, so the T-axes cannot be moved. " + "Connect it from the T3R panel and try again.") + return "" + + def _on_align_prepared(self, reading): + """The rig is configured and standing on the point — ask the operator.""" + if self._align_window is None: + return + confirmed = QMessageBox.question( + self, "Is the Image Correct?", _align_reference_prompt(reading) + ) == QMessageBox.StandardButton.Yes + + if not confirmed: + self._align_window.on_status( + "Cancelled: the image was not confirmed. Nothing has moved — " + "re-align by hand and start again.") + self._finish_align() + return + + self._align_window.set_reference(reading) + self._align_worker.request_run() + + def _on_align_prepare_failed(self, message: str): + if self._align_window is not None: + self._align_window.on_failed(message) + QMessageBox.warning(self, "Cannot Auto-Align", message) + self._finish_align() + + def _on_align_finished(self, result): + if self._align_window is not None: + self._align_window.on_finished(result) + self._finish_align() + + def _on_align_failed(self, message: str): + if self._align_window is not None: + self._align_window.on_failed(message) + QMessageBox.warning(self, "Auto-Align Failed", message) + self._finish_align() + + def _on_align_aborted(self): + if self._align_window is not None: + self._align_window.on_aborted() + self._finish_align() + + def _finish_align(self): + """Park the rig and release the thread, leaving the summary on screen.""" + if self._align_worker is not None: + self._align_worker.request_stop() + + def _release_align_thread(self): + """Worker idle — take its thread down. Safe to call more than once.""" + if self._align_worker is not None: + self._align_worker.stop_worker() + if self._align_thread is not None: + self._align_thread.quit() + self._align_thread.wait(30000) + self._align_thread = None + self._align_worker = None + self._set_align_ui_active(False) + + def _teardown_align(self): + """Stop an alignment and close its window — the window or camera is + going away, so the procedure cannot be left running.""" + window, self._align_window = self._align_window, None + if self._align_worker is not None: + self._align_worker.request_stop() + self._release_align_thread() + if window is not None: + window.close() + + def set_align_available(self, available: bool): + """The main window handing the rig over, or taking it back. + + A scan or an angle inspection owns the same stage, so auto-align has + to be off the table while either runs. + """ + self._align_available = available + self._refresh_align_button() + + def _refresh_align_button(self): + self.uc480_auto_align_btn.setEnabled( + self._align_available and self._align_thread is None) + + def _set_align_ui_active(self, active: bool): + """A running alignment owns the stage and the T-axes; nothing else + may drive them, here or in the main window.""" + self._refresh_align_button() + for panel in (self.t3r_jog_panel, self.bbd_jog_panel): + if panel is not None: + panel.stop_jogs() + panel.setEnabled(not active) + self.align_active.emit(active) + + +class AutoAlignWindow(QWidget): + """Live progress for one auto-align run, and the summary it ends with. + + Shows the deviation from the good values rather than the raw levels: the + procedure is a null search, so how far off it is says more than what it + reads, and 5 mV out of ~400 mV does not show up in the raw number. + """ + + abort_requested = pyqtSignal() + close_requested = pyqtSignal() + + def __init__(self, parent: QWidget | None = None): + super().__init__(parent, Qt.WindowType.Window) + self.setWindowTitle("Auto-Align") + self.resize(560, 480) + self._reference = None + self._done = False + + layout = QVBoxLayout(self) + + self.header_label = QLabel( + f"Stepping {DEFAULT_ALIGN.offset_mm:.2f} mm either side of this " + f"point on X (T1), then on Y (T0/T2 as a pair), re-tilting until " + f"the DC levels come back to within " + f"{DEFAULT_ALIGN.tolerance_mv:.0f} mV.\n" + f"T-axes: {DEFAULT_T_AXIS.microsteps} µsteps, " + f"{DEFAULT_T_AXIS.run_current_ma} mA.") + self.header_label.setWordWrap(True) + layout.addWidget(self.header_label) + + self.reading_label = QLabel("—") + self.reading_label.setFont(mono_font(13)) + self.reading_label.setStyleSheet("font-weight: bold;") + layout.addWidget(self.reading_label) + + self.status_label = QLabel("Configuring the rig …") + self.status_label.setWordWrap(True) + layout.addWidget(self.status_label) + + # The verdict outlives the status line, which keeps reporting the + # parking moves after the answer is known. + self.verdict_label = QLabel() + self.verdict_label.setWordWrap(True) + self.verdict_label.setVisible(False) + layout.addWidget(self.verdict_label) + + self.log = QPlainTextEdit(self) + self.log.setReadOnly(True) + self.log.setFont(mono_font(11)) + layout.addWidget(self.log, 1) + + buttons = QHBoxLayout() + self.stop_btn = QPushButton("Stop") + self.stop_btn.setToolTip( + "Stop at the next move and put the stage back on the reference " + "point. Any tilt already applied stays applied.") + self.stop_btn.clicked.connect(self._on_stop_clicked) + self.close_btn = QPushButton("Close") + self.close_btn.clicked.connect(self.close) + self.close_btn.setEnabled(False) + buttons.addWidget(self.stop_btn) + buttons.addWidget(self.close_btn) + layout.addLayout(buttons) + + # ── Worker → window ─────────────────────────────────────────────────────── + + def set_reference(self, reading): + self._reference = reading + self._append(f"Reference: {reading.describe()}") + + def on_status(self, msg: str): + self.status_label.setText(msg) + + def on_reading(self, reading): + if self._reference is None: + self.reading_label.setText(reading.describe()) + return + d1, d2 = reading.error_vs(self._reference) + self.reading_label.setText( + f"DC1 {reading.dc1_mv:8.1f} mV ({d1:+6.1f}) " + f"DC2 {reading.dc2_mv:8.1f} mV ({d2:+6.1f})") + + def on_offset_done(self, result): + self._append(f" {result.describe()}") + + def on_axis_done(self, result): + self._append(result.describe()) + + def on_finished(self, result): + # The per-axis lines are already in the log, put there as they + # happened; only the closing verdict is new. + self._append("") + self._append(result.verdict()) + self._set_verdict(result.verdict(), ok=result.ok) + + def on_failed(self, message: str): + self._append(f"FAILED: {message}") + self._set_verdict(message, ok=False) + + def on_aborted(self): + message = ("Stopped by the operator. Any tilt already applied stays " + "applied.") + self._append(message) + self._set_verdict(message, ok=False) + + # ── Window → worker ─────────────────────────────────────────────────────── + + def _on_stop_clicked(self): + self.stop_btn.setEnabled(False) + self.on_status("Stopping at the next move …") + self.abort_requested.emit() + + def _set_verdict(self, text: str, ok: bool): + self.verdict_label.setText(text) + self.verdict_label.setStyleSheet( + f"font-weight: bold; color: {'green' if ok else '#b8860b'};") + self.verdict_label.setVisible(True) + self._done = True + self.stop_btn.setEnabled(False) + self.close_btn.setEnabled(True) + + def _append(self, line: str): + self.log.appendPlainText(line) + + def closeEvent(self, event): + # Closing part-way through is a stop: the procedure owns the stage, + # and nothing else can call it off once this window is gone. + if not self._done: + self.abort_requested.emit() + self.close_requested.emit() + super().closeEvent(event) + # ── Helios laser panel ──────────────────────────────────────────────────────── @@ -1024,7 +1381,8 @@ class MainWindow(QMainWindow): self._helios_thread.started.connect(self._helios_worker.run) # ── Popup windows ───────────────────────────────────────────────────── - self._camera_win = CameraWindow(self._t3r_driver, self._bbd_worker) + self._camera_win = CameraWindow(self._t3r_driver, self._bbd_worker, + self._oscope_worker) self._helios_win = HeliosWindow(self._helios_worker) self._scan_progress = ScanProgressWindow() @@ -1146,6 +1504,7 @@ class MainWindow(QMainWindow): self._camera_win.closed.connect( lambda: self._set_toggle(self.show_camera_toggle, False, "Show Camera Window") ) + self._camera_win.align_active.connect(self._on_camera_align_active) # Helios self.show_helios_toggle.toggled.connect(self._on_helios_toggle) @@ -1331,6 +1690,13 @@ class MainWindow(QMainWindow): """Both entry points drive the same rig, so they lock and unlock together.""" self.start_scan_btn.setEnabled(enabled) self.saw_check_btn.setEnabled(enabled) + # Auto-align lives in the camera window but drives this same stage. + self._camera_win.set_align_available(enabled) + + def _on_camera_align_active(self, active: bool): + """An alignment started or finished in the camera window.""" + self._set_scan_buttons_enabled(not active) + self.inspect_angles_btn.setEnabled(not active) def _on_browse_save_dir(self): d = QFileDialog.getExistingDirectory( diff --git a/tests/fakes.py b/tests/fakes.py index 62c4512..c0f6988 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -258,6 +258,9 @@ class FakeT3R: self._t = trace self.is_open = is_open self._motion_completes = motion_completes + # Microsteps commanded per channel, so a test can read the tilt the + # platform ended up at rather than replaying the move trace. + self.positions = {ch: 0 for ch in range(4)} def set_microstep(self, ch, micro): self._t.record("t3r_set_microstep", ch, micro) @@ -273,9 +276,113 @@ class FakeT3R: return round(self.MOTOR_FULL_STEPS_PER_REV * microsteps * ratio * angle_deg / 360.0) + def move(self, ch, steps, velocity, accel): + self._t.record("t3r_move", ch, steps) + self.positions[ch] += steps + def rotate_stage(self, angle_deg, microsteps, velocity, accel): self._t.record("t3r_rotate", round(angle_deg, 6)) def wait_motion_done(self, ch, timeout): self._t.record("t3r_wait_motion_done", ch) return self._motion_completes + + +class FakeAlignRig: + """A tilted sample on the tilt platform, as the DC levels would read it. + + The detection beam is fixed and the stage carries the sample under it, so + the height error under the beam is the sample's slope times how far the + stage has moved off the reference point. The T-axes tilt the sample the + other way: their three heights define a plane, and its slope adds to the + sample's. Nulling the split-detector difference therefore means cancelling + the sample slope — which is exactly what an aligner has to work out. + + The plane is fitted by least squares here, rather than reusing + core.auto_align's closed form, so the two are independent statements of + the same geometry. + + ``curvature_mv_per_mm2`` bends the surface: a curved sample needs opposite + corrections at +1.5 mm and -1.5 mm, which is the disagreement the + procedure is supposed to report instead of averaging away. + """ + + # Actuator azimuths on the platform, in degrees from stage +X. + AZIMUTH_DEG = {0: 120.0, 1: 0.0, 2: 240.0} + + def __init__(self, stage, t3r, ref_mm=(50.0, 40.0), + x_slope_mv_per_mm=40.0, y_slope_mv_per_mm=-25.0, + tilt_gain_mv_per_mm=0.2, base_mv=400.0, + curvature_mv_per_mm2=0.0, jitter_mv=0.0): + self._stage = stage + self._t3r = t3r + self.ref_mm = ref_mm + self.x_slope_mv_per_mm = x_slope_mv_per_mm + self.y_slope_mv_per_mm = y_slope_mv_per_mm + self.tilt_gain_mv_per_mm = tilt_gain_mv_per_mm + self.base_mv = base_mv + self.curvature_mv_per_mm2 = curvature_mv_per_mm2 + self.jitter_mv = jitter_mv + self._reads = 0 + + # -- geometry ----------------------------------------------------------- + def platform_tilt(self): + """(x_tilt, y_tilt) of the plane through the three actuator heights.""" + import numpy as np + rows, heights = [], [] + for ch, azimuth in self.AZIMUTH_DEG.items(): + theta = np.radians(azimuth) + rows.append([1.0, np.cos(theta), np.sin(theta)]) + heights.append(float(self._t3r.positions[ch])) + _, x_tilt, y_tilt = np.linalg.lstsq(np.array(rows), np.array(heights), + rcond=None)[0] + return float(x_tilt), float(y_tilt) + + def slopes_mv_per_mm(self): + """The residual sample slope the beam sees, after the platform tilt.""" + x_tilt, y_tilt = self.platform_tilt() + return (self.x_slope_mv_per_mm + self.tilt_gain_mv_per_mm * x_tilt, + self.y_slope_mv_per_mm + self.tilt_gain_mv_per_mm * y_tilt) + + def difference_mv(self): + x_off = self._stage.positions[0] - self.ref_mm[0] + y_off = self._stage.positions[1] - self.ref_mm[1] + slope_x, slope_y = self.slopes_mv_per_mm() + return (slope_x * x_off + slope_y * y_off + + self.curvature_mv_per_mm2 * (x_off ** 2 + y_off ** 2)) + + # -- what the scope reports --------------------------------------------- + def level_v(self, channel): + """CH3 and CH4 as volts: the difference straddling a constant sum. + + The sum is fixed because tilt steers the beam across the detector + rather than changing how much light comes back — so a nulled + difference does put both levels back where they were. + """ + self._reads += 1 + # A deterministic alternating wobble, so a test can check the median + # of several reads is what keeps the loop stable. + jitter = self.jitter_mv * (1 if self._reads % 2 else -1) + half = 0.5 * self.difference_mv() + mv = self.base_mv + (half if channel == 3 else -half) + jitter + return mv / 1000.0 + + +class FakeAlignScope(FakeScope): + """FakeScope that also answers the DC measurements auto-align reads.""" + + def __init__(self, trace: Trace, rig: FakeAlignRig, samples_per_frame=8, + acquisitions_advance=True): + super().__init__(trace, samples_per_frame=samples_per_frame) + self._rig = rig + self._acq = 0 + self._advance = acquisitions_advance + + def measure_immediate(self, channel, measurement_type="MEAN"): + self._t.record("measure_immediate", channel, measurement_type) + return self._rig.level_v(channel) + + def get_acquisition_count(self): + if self._advance: + self._acq += 1 + return self._acq diff --git a/tests/test_auto_align.py b/tests/test_auto_align.py new file mode 100644 index 0000000..f14a18f --- /dev/null +++ b/tests/test_auto_align.py @@ -0,0 +1,312 @@ +"""Auto-align, driven entirely by fake hardware. + +The feature is a closed loop over hardware, so the tests are built round a +model of the rig (fakes.FakeAlignRig): a sample at a known tilt, a platform +whose three actuators tilt it, and DC levels that follow from both. A test +therefore asks the question the operator does — is the sample level now? — +rather than replaying a command sequence. + +The other half is geometry. Which axis moves for which stage direction is +the one thing here that cannot be discovered at run time, and getting it +wrong would still converge (on the wrong axis, at the wrong point), so it is +pinned separately and explicitly. +""" +from dataclasses import replace + +import pytest + +from core.auto_align import ( + AutoAligner, AutoAlignAborted, AutoAlignError, DEFAULT_ALIGN, + T_AXIS_AZIMUTH_DEG, X_TILT, Y_TILT, tilt_response, +) +from core.scan_engine import AXIS_X, AXIS_Y +from core.scope_inspect import BIAS_CHANNELS, BIAS_SCALE_V_DIV, INSPECT_TRIG_LEVEL_V +from fakes import FakeAlignRig, FakeAlignScope, FakeStage, FakeT3R, Trace + +REF_MM = (50.0, 40.0) + +# No settle: the sleeps are there for the instrument, and every test here +# takes a few dozen readings. +FAST = replace(DEFAULT_ALIGN, settle_s=0.0) + + +def build(*, settings=FAST, acquisitions_advance=True, ref_mm=REF_MM, + should_abort=lambda: False, **rig_kwargs): + trace = Trace() + stage = FakeStage(trace) + stage.positions = list(ref_mm) + t3r = FakeT3R(trace) + rig = FakeAlignRig(stage, t3r, ref_mm=ref_mm, **rig_kwargs) + scope = FakeAlignScope(trace, rig, acquisitions_advance=acquisitions_advance) + aligner = AutoAligner(stage, scope, t3r, settings=settings, + should_abort=should_abort) + return aligner, rig, trace, stage, t3r + + +def moves(trace, ch=None): + return [c for c in trace.of("t3r_move") if ch is None or c[1] == ch] + + +# ── Geometry: the half that cannot be discovered at run time ───────────────── + +def test_tilt_groups_are_the_moves_they_claim_to_be(): + """X tilts along X only, Y along Y only — otherwise the phases interfere. + + If this fails, the azimuth map and the groups have drifted apart and the + Y phase would be undoing the X phase's correction. + """ + x_piston, x_x, x_y = tilt_response(X_TILT) + y_piston, y_x, y_y = tilt_response(Y_TILT) + + assert x_x != 0 and x_y == pytest.approx(0.0, abs=1e-9) + assert y_y != 0 and y_x == pytest.approx(0.0, abs=1e-9) + # The Y pair is equal and opposite, so it lifts nothing on average; the + # single X axis unavoidably lifts the platform as well as tilting it. + assert y_piston == pytest.approx(0.0, abs=1e-9) + assert x_piston != 0 + + +def test_x_is_corrected_by_the_axis_lying_along_x(): + """T1 sits at 0°, so it is the one that tilts the platform along X.""" + assert T_AXIS_AZIMUTH_DEG[1] == 0.0 + assert set(X_TILT.weights) == {1} + + +def test_y_is_corrected_by_the_other_two_as_an_opposed_pair(): + assert set(Y_TILT.weights) == {0, 2} + assert Y_TILT.weights[0] == -Y_TILT.weights[2] + + +# ── The loop does what it is for ───────────────────────────────────────────── + +def test_alignment_cancels_the_sample_slope_on_both_axes(): + """The point of the whole procedure: a level sample when it finishes.""" + aligner, rig, _, _, _ = build() + aligner.prepare() + result = aligner.run() + + slope_x, slope_y = rig.slopes_mv_per_mm() + # Residual slope over the +/-1.5 mm the scan cares about, in mV. + assert abs(slope_x * DEFAULT_ALIGN.offset_mm) <= DEFAULT_ALIGN.tolerance_mv + assert abs(slope_y * DEFAULT_ALIGN.offset_mm) <= DEFAULT_ALIGN.tolerance_mv + assert result.ok + + +def test_every_search_ends_inside_the_tolerance(): + aligner, _, _, _, _ = build() + reference = aligner.prepare() + result = aligner.run() + + for axis in result.axes: + for offset in axis.offsets: + assert offset.converged, offset.describe() + assert offset.final.matches(reference, DEFAULT_ALIGN.tolerance_mv) + assert result.final.matches(reference, DEFAULT_ALIGN.tolerance_mv) + + +def test_a_flat_sample_gives_the_same_answer_in_both_directions(): + """Both offsets measure one angle, so on a plane they must agree. + + The agreement is what licenses averaging them; see the curved case below + for what happens when it does not hold. + """ + aligner, _, _, _, _ = build() + aligner.prepare() + result = aligner.run() + + for axis in result.axes: + plus, minus = axis.offsets + assert plus.correction_steps == pytest.approx(minus.correction_steps, + rel=0.02, abs=5.0) + assert axis.disagreement_steps < 10.0 + assert axis.applied + + +def test_a_curved_sample_is_reported_rather_than_averaged_away(): + """Curvature needs opposite corrections either side, and says so.""" + # Big enough that the near side is still outside the tolerance once the + # far side's error has been curved past it — otherwise one search has + # nothing to do and the disagreement never shows up. + aligner, _, _, _, _ = build(curvature_mv_per_mm2=60.0) + aligner.prepare() + result = aligner.run() + + x_axis = result.axes[0] + plus, minus = x_axis.offsets + assert plus.correction_steps * minus.correction_steps < 0 # opposite signs + assert x_axis.disagreement_steps > 100.0 + + +def test_the_search_survives_a_noisy_detector(): + """Five reads and a median, so a wobbling level still converges.""" + aligner, _, _, _, _ = build(jitter_mv=1.5) + reference = aligner.prepare() + result = aligner.run() + assert result.final.matches(reference, DEFAULT_ALIGN.tolerance_mv) + + +# ── Which hardware moves, and how ─────────────────────────────────────────── + +def test_the_x_phase_moves_t1_and_the_y_phase_moves_t0_and_t2(): + """The phases stay on their own axes, in the order X then Y.""" + aligner, _, trace, _, t3r = build() + aligner.prepare() + aligner.run() + + channels = [c[1] for c in moves(trace)] + first_y = next(i for i, ch in enumerate(channels) if ch in (0, 2)) + assert set(channels[:first_y]) == {1}, "the X phase moved something else" + assert set(channels[first_y:]) == {0, 2}, "the Y phase moved something else" + + # The Y pair ends equal and opposite: anything else is a tilt along X the + # Y phase had no business applying. + assert t3r.positions[0] == -t3r.positions[2] + assert t3r.positions[1] != 0 + + +def test_the_rotation_axis_is_never_touched(): + """GR carries the scan's angle; an alignment that moved it would silently + re-datum every subsequent scan.""" + aligner, _, trace, _, t3r = build() + aligner.prepare() + aligner.run() + + assert moves(trace, ch=3) == [] + assert t3r.positions[3] == 0 + assert [c for c in trace.of("t3r_enable") if c[1] == 3] == [] + + +def test_the_t_axes_are_configured_before_they_are_moved(): + """32 microsteps and 600 mA, applied rather than assumed — a correction is + reported in microsteps, so what a microstep means has to be pinned.""" + aligner, _, trace, _, _ = build() + aligner.prepare() + + for ch in (0, 1, 2): + assert ("t3r_set_microstep", ch, 32) in trace.calls + run_ma = [c for c in trace.of("t3r_set_current") if c[1] == ch] + assert run_ma and run_ma[0][2] == 600 + assert ("t3r_enable", ch) in trace.calls + + names = trace.names() + assert "t3r_move" not in names[:names.index("t3r_enable")] + + +def test_the_stage_steps_either_side_and_comes_back(): + aligner, _, trace, stage, _ = build() + aligner.prepare() + aligner.run() + aligner.stop() + + x_targets = [c[2] for c in trace.of("move_axis_absolute") if c[1] == AXIS_X] + y_targets = [c[2] for c in trace.of("move_axis_absolute") if c[1] == AXIS_Y] + off = DEFAULT_ALIGN.offset_mm + assert REF_MM[0] + off in x_targets and REF_MM[0] - off in x_targets + assert REF_MM[1] + off in y_targets and REF_MM[1] - off in y_targets + assert stage.positions == list(REF_MM) + + +def test_the_scope_is_put_into_the_bias_reading_state(): + """The same free-running, edge-triggered state the angle inspector uses: + the operator has to be able to read CH1 while this runs.""" + aligner, _, trace, _, _ = build() + aligner.prepare() + + scales = {c[1]: c[2] for c in trace.of("set_channel_scale")} + for ch in BIAS_CHANNELS: + assert scales[ch] == BIAS_SCALE_V_DIV + assert ("set_trigger_level", 2, INSPECT_TRIG_LEVEL_V) in trace.calls + assert ("set_fastframe_state", False) in trace.calls + assert "ACQuire:STATE RUN" in [c[1] for c in trace.of("write")] + + # Only the two bias channels are ever measured. + assert {c[1] for c in trace.of("measure_immediate")} == set(BIAS_CHANNELS) + + +def test_the_gate_is_dropped_before_anything_moves(): + """An armed TRIGOUT would drive the scan gate on every positioning move.""" + aligner, _, trace, _, _ = build() + aligner.prepare() + assert ("set_trigger_gate_off", AXIS_X) in trace.calls + + +# ── Refusals ──────────────────────────────────────────────────────────────── + +def test_a_dead_axis_stops_the_procedure(): + """No response to a probe, however large: something is wrong upstream of + the tilt platform, and stepping the actuators further will not find it.""" + aligner, _, _, _, _ = build(tilt_gain_mv_per_mm=0.0) + aligner.prepare() + with pytest.raises(AutoAlignError, match="laser"): + aligner.run() + + +def test_a_scope_that_never_retriggers_stops_the_procedure(): + """A stale record reads as a rock-steady measurement — the one failure the + loop cannot see for itself.""" + aligner, _, _, _, _ = build(acquisitions_advance=False) + with pytest.raises(AutoAlignError, match="not triggered"): + aligner.prepare() + + +def test_an_axis_that_would_run_out_of_travel_stops_the_procedure(): + aligner, _, _, _, _ = build(x_slope_mv_per_mm=200.0, + tilt_gain_mv_per_mm=0.01) + aligner.prepare() + with pytest.raises(AutoAlignError, match="safety limit"): + aligner.run() + + +def test_there_has_to_be_room_either_side_of_the_reference_point(): + aligner, _, _, _, _ = build(ref_mm=(0.5, 40.0)) + with pytest.raises(AutoAlignError, match="either side"): + aligner.prepare() + + +def test_run_before_the_operator_confirms_is_refused(): + aligner, _, _, _, _ = build() + with pytest.raises(AutoAlignError, match="prepare"): + aligner.run() + + +def test_missing_hardware_is_named(): + trace = Trace() + stage = FakeStage(trace) + t3r = FakeT3R(trace, is_open=False) + rig = FakeAlignRig(stage, t3r) + scope = FakeAlignScope(trace, rig) + + with pytest.raises(AutoAlignError, match="T3R"): + AutoAligner(stage, scope, t3r, settings=FAST).prepare() + with pytest.raises(AutoAlignError, match="Oscilloscope"): + AutoAligner(stage, None, t3r, settings=FAST).prepare() + with pytest.raises(AutoAlignError, match="BBD202"): + AutoAligner(None, scope, t3r, settings=FAST).prepare() + + +def test_an_abort_stops_the_run_and_still_parks_the_stage(): + """Stopping is the operator's, so it must not leave the stage 1.5 mm off + the point they were looking at.""" + calls = {"n": 0} + + def abort_after_a_few_moves(): + calls["n"] += 1 + return calls["n"] > 12 + + aligner, _, _, stage, _ = build(should_abort=abort_after_a_few_moves) + aligner.prepare() + with pytest.raises(AutoAlignAborted): + aligner.run() + + aligner.stop() + assert stage.positions == list(REF_MM) + + +def test_stop_leaves_the_correction_applied(): + """The tilt is the result — a stop parks the stage, not the platform.""" + aligner, _, _, _, t3r = build() + aligner.prepare() + aligner.run() + applied = dict(t3r.positions) + aligner.stop() + assert t3r.positions == applied diff --git a/tests/test_smoke_apps.py b/tests/test_smoke_apps.py index b995501..dab4cb4 100644 --- a/tests/test_smoke_apps.py +++ b/tests/test_smoke_apps.py @@ -38,6 +38,51 @@ def test_sc3_aui_main_window(qapp): _pump(qapp) +def test_camera_window_without_hardware_is_a_plain_viewer(qapp): + """No stage, no scope, no auto-align button to press.""" + import sc3_aui_app + win = sc3_aui_app.CameraWindow() + _pump(qapp) + try: + assert not win.uc480_auto_align_btn.isVisible() + finally: + win.deleteLater() + _pump(qapp) + + +def test_camera_window_auto_align_needs_every_device(qapp): + """The button appears once the three devices exist, and says which one is + missing rather than starting and failing on the rig.""" + import sc3_aui_app + from gui.qt_t3r import QtT3RAdapter + win = sc3_aui_app.CameraWindow(QtT3RAdapter(), sc3_aui_app.BBD202Worker(), + sc3_aui_app.OscopeWorker()) + _pump(qapp) + try: + assert win.uc480_auto_align_btn.isVisibleTo(win) + assert "oscilloscope" in win._align_prerequisite_problem() + finally: + win.deleteLater() + _pump(qapp) + + +def test_auto_align_window_logs_a_result(qapp): + import sc3_aui_app + from core.auto_align import AlignResult, Reading + win = sc3_aui_app.AutoAlignWindow() + _pump(qapp) + try: + reference = Reading(400.0, 400.0) + win.set_reference(reference) + win.on_reading(Reading(403.0, 397.0)) + win.on_finished(AlignResult(reference=reference, final=reference)) + assert "Reference" in win.log.toPlainText() + assert win.close_btn.isEnabled() + finally: + win.deleteLater() + _pump(qapp) + + def test_sras_viewer_window(qapp): import sras_viewer win = sras_viewer.SrasViewerWindow()