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>
125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""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)
|