Thomas Ales
6e8c1cb7a2
The operator frames a good spot, confirms the two DC levels the detector reads there, and the rig then measures its own tilt: step 1.5 mm either side on X and then on Y, and tilt the platform until those levels come back. The correction that fixes an offset point is the correction that levels the whole travel — height error and tilt effect are both proportional to the offset — so the procedure ends by applying it and leaving it applied. Both directions are measured from the same starting tilt and averaged, which makes their disagreement a flatness read-out rather than something averaged away silently. core/auto_align.py holds the geometry and the search, Qt-free. The three T-axes' azimuths are the whole geometry: T1 lies along +X so it alone tilts along X, and T0/T2 move as an equal-and-opposite pair to tilt along Y without touching X (tilt_response derives that, and the tests pin it — an axis map that drifts would still converge, on the wrong axis). The search is a secant null on the split-detector difference: probe once to learn what a microstep is worth, sign included, then step at the null. It refuses to servo on a scope that has not re-triggered, escalates a probe that reads as no response before calling an axis dead, and stops at a per-axis travel limit. gui/align_bridge.py runs it on a worker thread; stopping is a threading.Event rather than a queued command, because the worker is inside a long handler for the whole run. The camera window carries the button and the progress window, and locks the scan panel and the jog pads while a run owns the stage. Adds immediate MEAN measurements and an acquisition count to the scope driver, and read_bias_mv to core/scope_inspect — the one scalar the inspection state was missing. KNOWN_ISSUES.md records what only the rig can settle: the probe step, the travel limit, the hold current, and whether the piston the X phase applies alongside its tilt matters. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
709 lines
31 KiB
Python
709 lines
31 KiB
Python
"""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")
|