"""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")