bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Flomni commissioning 5 #256

Merged
holler merged 30 commits from flomni_commissioning_5 into main 2026-07-12 07:28:11 +02:00
Conversation 0 Commits 30 Files Changed 17
+1123 -196
17 changed files with 1123 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+541 -154
View File
@@ -14,7 +14,7 @@ from bec_lib.alarm_handler import AlarmBase
from bec_lib.endpoints import MessageEndpoints
from bec_lib.pdf_writer import PDFWriter
from bec_lib.scan_repeat import scan_repeat
from typeguard import typechecked
from typeguard import check_type, typechecked
from csaxs_bec.bec_ipython_client.plugins.cSAXS import cSAXSBeamlineChecks
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_optics_mixin import FlomniOpticsMixin
@@ -421,6 +421,23 @@ class FlomniSampleTransferMixin:
if not low:
raise FlomniError("Ftray is not at the 'IN' position. Aborting.")
def ftransfer_confirm_dialog(self, message: str, default: str = "none") -> bool:
"""
Yes/No confirmation for sample-transfer steps.
Uses the GUI console widget (ConsoleButtonsWidget, opened by
flomnigui_show_cameras()) when it's already available, so the
operator can confirm/abort right next to the live camera view.
Falls back to the regular console yesno() prompt if the
cameras/console haven't been opened yet -- e.g. if a transfer step
is called directly, outside the usual ftransfer_get_sample /
ftransfer_put_sample sequence that opens them first.
"""
if self.console is not None:
print("Using GUI console for confirmation dialog.")
return self.OMNYTools.gui_yesno(message, gui=self.console, default=default)
return self.OMNYTools.yesno(message, default)
def ftransfer_flomni_stage_in(self):
time.sleep(1)
sample_in_position = dev.flomni_samples.is_sample_slot_used(0)
@@ -440,14 +457,14 @@ class FlomniSampleTransferMixin:
print("Moving X-ray eye in.")
if self.OMNYTools.yesno(
"Please confirm that this is ok with the flight tube. This check is to be removed after commissioning",
"n",
):
print("OK. continue.")
else:
print("Stopping.")
raise FlomniError("Manual abort of x-ray eye in.")
# if self.OMNYTools.yesno(
# "Please confirm that this is ok with the flight tube. This check is to be removed after commissioning",
# "n",
# ):
# print("OK. continue.")
# else:
# print("Stopping.")
# raise FlomniError("Manual abort of x-ray eye in.")
self.feye_in()
print("Moving X-ray optics out.")
@@ -480,7 +497,7 @@ class FlomniSampleTransferMixin:
def laser_parameters_set_psd_intensity_threshold_tracking_low(self, val: float):
dev.rtx.controller.emitter_set_psd_intensity_threshold_tracking_low(val)
def laser_tweak(self):
def laser_parameters_tweak_tracking(self):
"""Interactive tweak of the two laser tracking target positions.
Arrow keys nudge the ConstEmitter tracking targets by a fixed step of
@@ -494,15 +511,27 @@ class FlomniSampleTransferMixin:
After every change the current y/z targets and the fzp x interferometer
signal strength (channel 1) are printed, so one can watch the signal
respond while tweaking.
Requires the laser tracker to be enabled: with the tracker off nothing
moves on the hardware side, so the tweak would be a no-op.
"""
import fcntl
import sys
import termios
import tty
step = 0.01
controller = dev.rtx.controller
if not controller.laser_tracker_check_enabled():
print(
"The laser tracker is disabled. Enable it with laser_tracker_on() "
"before tweaking the tracking targets - otherwise nothing moves on "
"the hardware side."
)
return
step = 0.02
def read_targets():
values = controller.emitter_get()
return (
@@ -742,15 +771,22 @@ class FlomniSampleTransferMixin:
self.transfer_step = 0
time.sleep(1)
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm(step_name="get")
try:
time.sleep(1)
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm(step_name="get")
time.sleep(1)
except KeyboardInterrupt:
self.ftransfer_abort()
raise FlomniError(
"Sample transfer aborted by user (Ctrl+C / GUI abort). Assess gripper and"
" sample state manually before continuing."
) from None
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
@@ -791,17 +827,24 @@ class FlomniSampleTransferMixin:
print("The mount process started.")
time.sleep(1)
self.transfer_step = 0
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm(step_name="put")
try:
time.sleep(1)
self.transfer_step = 0
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm(step_name="put")
time.sleep(1)
except KeyboardInterrupt:
self.ftransfer_abort()
raise FlomniError(
"Sample transfer aborted by user (Ctrl+C / GUI abort). Assess gripper and"
" sample state manually before continuing."
) from None
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
@@ -827,6 +870,47 @@ class FlomniSampleTransferMixin:
if sample_in_gripper:
raise FlomniError("There is already a sample in the gripper. Aborting.")
if new_sample_position == -1:
# Remove-only: stow the sample currently in the sample stage into a
# free tray slot and stop there -- no new sample is mounted.
sample_in_sample_stage = dev.flomni_samples.is_sample_slot_used(0)
if not sample_in_sample_stage:
raise FlomniError("There is no sample in the sample stage to remove. Aborting.")
empty_slots = []
for j in range(1, 21):
if not dev.flomni_samples.is_sample_slot_used(j):
empty_slots.append(j)
if not empty_slots:
raise FlomniError("There are no empty slots available. Aborting.")
print(f"The following slots are empty: {empty_slots}.")
while True:
user_input = input(
"Where shall I put the sample currently in the sample stage? "
f"Default: [{empty_slots[0]}] "
)
if user_input.strip() == "":
user_input = empty_slots[0]
break
try:
user_input = int(user_input)
if user_input not in empty_slots:
raise ValueError
break
except ValueError:
print("Please specify a valid number.")
continue
self.check_position_is_valid(user_input)
self.ftransfer_get_sample(0)
self.ftransfer_put_sample(user_input)
return
self.check_position_is_valid(new_sample_position)
if new_sample_position == 0:
@@ -853,7 +937,7 @@ class FlomniSampleTransferMixin:
# continue
# if val.get("value") == 0:
# empty_slots.append(int(name.split("flomni_samples_sample_placed_sample")[1]))
for j in range(1, 20):
for j in range(1, 21):
if not dev.flomni_samples.is_sample_slot_used(j):
empty_slots.append(j)
if not empty_slots:
@@ -928,6 +1012,38 @@ class FlomniSampleTransferMixin:
)
return in_mount_mode
def ftransfer_abort(self):
"""
Hard abort of a running sample transfer routine on the Galil controller.
Stops the controller via dev.ftransy.stop(), which publishes a stop
request that the device server turns into motor.stop() and thus
XQ#STOP on the controller. #STOP halts the transfer thread (3),
aborts all motion (AB1) and clears mntprgs/mntmod. Afterwards the
controller is put back into positioning mode.
Deliberately does NOT call ensure_gripper_up(): after a mid-transfer
abort the gripper may be closed around a partially inserted sample,
so any recovery motion must be assessed and performed manually.
"""
print("Aborting sample transfer: stopping the controller routine.")
dev.ftransy.stop()
# The stop request is asynchronous (Redis -> device server). Wait
# until the transfer thread is actually halted before switching mode:
# #POSMODE refuses while mntprgs=1 and disable_mount_mode would raise.
timeout = 5
start = time.time()
while dev.ftransy.controller.is_thread_active(3):
if time.time() - start > timeout:
raise FlomniError(
"Transfer abort requested but the controller transfer routine (thread 3)"
f" did not stop within {timeout} s. Check the controller."
)
time.sleep(0.1)
# Ensure the controller is back in positioning mode. #STOP already
# clears mntmod, so this is mostly a verification step.
self.ftransfer_controller_disable_mount_mode()
def ftransfer_confirm(self, step_name: str = ""):
confirm = int(float(dev.ftransy.controller.socket_put_and_receive("MG confirm").strip()))
@@ -935,7 +1051,7 @@ class FlomniSampleTransferMixin:
return
self.transfer_step += 1
if self.OMNYTools.yesno("All OK? Continue?", "y"):
if self.ftransfer_confirm_dialog("All OK? Continue?", "y"):
print("OK. continue.")
data = self.client.connector.get_last(
MessageEndpoints.device_preview("cam_flomni_gripper", "preview")
@@ -944,6 +1060,7 @@ class FlomniSampleTransferMixin:
dev.ftransy.controller.socket_put_confirmed("confirm=1")
else:
print("Stopping.")
self.ftransfer_abort()
raise FlomniError("User abort sample transfer.")
def save_reference_image(self, image: np.ndarray, file_suffix: str = ""):
@@ -979,16 +1096,19 @@ class FlomniSampleTransferMixin:
self.check_position_is_valid(position)
# this is not used for sample stage position!
self._ftransfer_shiftx = -0.15
self._ftransfer_shiftx = -0.1
self._ftransfer_shiftz = -0.5
fsamx_pos = dev.fsamx.readback.get()
if position == 0 and fsamx_pos > -160:
if self.OMNYTools.yesno(
"May the flomni stage be moved out for the sample change? Feedback will be disabled and alignment will be lost!",
"y",
):
print("Use GUI interface.")
if self.ftransfer_confirm_dialog("May the flomni stage be moved out for the sample change? Feedback will be disabled and alignment will be lost!", "y"):
# if self.OMNYTools.yesno(
# "May the flomni stage be moved out for the sample change? Feedback will be disabled and alignment will be lost!",
# "y",
# ):
print("OK. continue.")
self.ftransfer_flomni_stage_out()
else:
@@ -999,7 +1119,7 @@ class FlomniSampleTransferMixin:
self.check_tray_in()
if position == 0:
umv(dev.ftransx, 11, dev.ftransz, 3.5950)
umv(dev.ftransx, 11.02, dev.ftransz, 3.5950)
if position == 1:
umv(
dev.ftransx,
@@ -1357,10 +1477,10 @@ class FlomniAlignmentMixin:
correction_z = tomo_alignment_fit[0][0] * np.sin(
np.radians(angle + 90) + tomo_alignment_fit[0][1]
)
#Todo
print(
f"Alignment offset x {correction_x}, y {correction_y}, z {correction_z} for angle"
f" {angle}\n"
f" {angle}\nCurrently not applying z shift (HW damage)."
)
return (correction_x, correction_y, correction_z)
@@ -1553,6 +1673,50 @@ class _TomoQueueProxy:
return self._load()[index]
class _GlobalVarParam:
"""Descriptor for a Flomni parameter backed by a BEC global variable.
Consolidates the boilerplate getter/setter pairs that read from and write
to the BEC global-var store. The value persists across BEC client restarts
(instance attributes do not), and is readable from other client sessions
via ``client.get_global_var(<name>)``.
The global-var key is taken from the attribute name the descriptor is
assigned to (see :meth:`__set_name__`), so it always matches the property
name -- exactly the invariant the hand-written pairs relied on.
Parameters
----------
default
Value returned by the getter when the global var is unset (``None``).
type_
Optional expected type. When given, the setter validates ``val`` with
:func:`typeguard.check_type` before storing, reproducing the
``@typechecked`` guard that previously wrapped some setters (raising
``TypeCheckError`` on a mismatch).
"""
def __init__(self, default, type_=None):
self._default = default
self._type = type_
def __set_name__(self, owner, name):
self._key = name
def __get__(self, obj, objtype=None):
if obj is None:
return self
val = obj.client.get_global_var(self._key)
if val is None:
return self._default
return val
def __set__(self, obj, val):
if self._type is not None:
check_type(val, self._type)
obj.client.set_global_var(self._key, val)
class Flomni(
FlomniInitStagesMixin,
FlomniSampleTransferMixin,
@@ -1601,10 +1765,72 @@ class Flomni(
self.align = XrayEyeAlign(self.client, self)
self.set_client(client)
self._maybe_reset_params_on_account_change()
def set_web_password(self, password: str) -> None:
"""Set the web password for the current BEC account."""
self._webpage_gen.set_web_password(password)
def _maybe_reset_params_on_account_change(self) -> None:
"""Offer a tomo-parameter reset when the active account has changed.
Called at BEC session start. The account for which defaults were last
applied is stored in the global var ``defaults_applied_for_account`` so
it survives client restarts. Only a genuine change of account triggers
the (interactive) reset; the same account is a silent no-op, so
restarting a client within the same experiment never disturbs tuned
parameters.
"""
bec = builtins.__dict__.get("bec")
try:
account = bec.active_account
except Exception as exc:
print(f"account-change check skipped: cannot read active_account: {exc}")
return
if not account:
return
last_account = self.client.get_global_var("defaults_applied_for_account")
if account == last_account:
return
if self.OMNYTools.yesno(
f"New account '{account}' detected (previous: '{last_account}').\n"
"Reset tomo parameters to defaults for the new experiment?",
"y",
):
self._set_default_tomo_params()
print(f"Tomo parameters reset to defaults for account '{account}'.")
self.client.set_global_var("defaults_applied_for_account", account)
def _set_default_tomo_params(self) -> None:
"""Write all tomo scan parameters back to their default values.
These are the same baseline values used as the getter fallbacks. Per-
sample alignment state (corrections, alignment offsets) is deliberately
not reset here, as it is overwritten by the next alignment anyway.
"""
self.tomo_shellstep = 1
self.tomo_countingtime = 0.1
self.manual_shift_y = 0.0
self.single_point_random_shift_max = 0.0
self.fovx = 20
self.fovy = 20
self.tomo_type = 1
self.corridor_size = -1
self.stitch_x = 0
self.stitch_y = 0
self.ptycho_reconstruct_foldername = "ptycho_reconstruct"
self.tomo_angle_stepsize = 10.0
self.golden_max_number_of_projections = 1000.0
self.tomo_stitch_overlap = 0.2
self.tomo_angle_range = 180
self.golden_projections_at_0_deg_for_damage_estimation = 0
self.zero_deg_reference_at_each_subtomo = False
self.golden_ratio_bunch_size = 20
self.frames_per_trigger = 1
self.single_point_instead_of_fermat_scan = False
def start_x_ray_eye_alignment(self, keep_shutter_open=False):
if self.OMNYTools.yesno(
@@ -1622,7 +1848,45 @@ class Flomni(
umv(dev.fsamx, fsamx_in)
raise exc
def _check_beamline_states_valid(self):
"""Return (all_valid, invalid_labels).
Reads every configured beamline state (bec.beamline_states) and
collects the ones whose status is not "valid" (i.e. "invalid",
"warning" or "unknown"). Used to warn before taking a manual X-ray
eye frame: if e.g. the beam is down, the frame would come back empty
and the user is left confused about why they see nothing.
Returns all_valid=True (with an empty list) if the beamline-state
machinery isn't available at all, so this never blocks on a setup
that doesn't use beamline states.
"""
bec = builtins.__dict__.get("bec")
manager = getattr(bec, "beamline_states", None) if bec is not None else None
if manager is None:
return True, []
bad = []
for name in list(getattr(manager, "_states", {})):
try:
status = manager.get_status_by_name(name)
except Exception: # pylint: disable=broad-except
continue
if status is not None and status != "valid":
bad.append(f"{name}: {status}")
return (len(bad) == 0), bad
def xrayeye_update_frame(self, keep_shutter_open=False):
all_valid, bad_states = self._check_beamline_states_valid()
if not all_valid:
print(
"Attention: not all beamline states are valid:\n "
+ "\n ".join(bad_states)
+ "\nThe beam may be down, so the frame could come back empty."
)
if not self.OMNYTools.yesno("Take a frame anyway?", "n"):
print("Stopping.")
return
self.align.update_frame(keep_shutter_open)
def xrayeye_alignment_start(self, keep_shutter_open=False):
@@ -1690,38 +1954,11 @@ class Flomni(
raise TypeError(f"progress must be a dict, got {type(val).__name__!r}")
self._progress_proxy._save(val)
@property
def tomo_shellstep(self):
val = self.client.get_global_var("tomo_shellstep")
if val is None:
return 1
return val
tomo_shellstep = _GlobalVarParam(1)
@tomo_shellstep.setter
def tomo_shellstep(self, val: float):
self.client.set_global_var("tomo_shellstep", val)
tomo_countingtime = _GlobalVarParam(0.1)
@property
def tomo_countingtime(self):
val = self.client.get_global_var("tomo_countingtime")
if val is None:
return 0.1
return val
@tomo_countingtime.setter
def tomo_countingtime(self, val: float):
self.client.set_global_var("tomo_countingtime", val)
@property
def manual_shift_y(self):
val = self.client.get_global_var("manual_shift_y")
if val is None:
return 0.0
return val
@manual_shift_y.setter
def manual_shift_y(self, val: float):
self.client.set_global_var("manual_shift_y", val)
manual_shift_y = _GlobalVarParam(0.0)
@property
def single_point_random_shift_max(self):
@@ -1789,40 +2026,11 @@ class Flomni(
else:
raise ValueError("Unknown tomo_type.")
@property
def corridor_size(self):
val = self.client.get_global_var("corridor_size")
if val is None:
val = -1
return val
corridor_size = _GlobalVarParam(-1)
@corridor_size.setter
def corridor_size(self, val: float):
self.client.set_global_var("corridor_size", val)
stitch_x = _GlobalVarParam(0, type_=int)
@property
def stitch_x(self):
val = self.client.get_global_var("stitch_x")
if val is None:
return 0
return val
@stitch_x.setter
@typechecked
def stitch_x(self, val: int):
self.client.set_global_var("stitch_x", val)
@property
def stitch_y(self):
val = self.client.get_global_var("stitch_y")
if val is None:
return 0
return val
@stitch_y.setter
@typechecked
def stitch_y(self, val: int):
self.client.set_global_var("stitch_y", val)
stitch_y = _GlobalVarParam(0, type_=int)
@property
def ptycho_reconstruct_foldername(self):
@@ -1836,38 +2044,11 @@ class Flomni(
self.client.set_global_var("ptycho_reconstruct_foldername", val)
self.reconstructor.folder_name = val # keep reconstructor in sync
@property
def tomo_angle_stepsize(self):
val = self.client.get_global_var("tomo_angle_stepsize")
if val is None:
return 10.0
return val
tomo_angle_stepsize = _GlobalVarParam(10.0)
@tomo_angle_stepsize.setter
def tomo_angle_stepsize(self, val: float):
self.client.set_global_var("tomo_angle_stepsize", val)
golden_max_number_of_projections = _GlobalVarParam(1000.0)
@property
def golden_max_number_of_projections(self):
val = self.client.get_global_var("golden_max_number_of_projections")
if val is None:
return 1000.0
return val
@golden_max_number_of_projections.setter
def golden_max_number_of_projections(self, val: float):
self.client.set_global_var("golden_max_number_of_projections", val)
@property
def tomo_stitch_overlap(self):
val = self.client.get_global_var("tomo_stitch_overlap")
if val is None:
return 0.2
return val
@tomo_stitch_overlap.setter
def tomo_stitch_overlap(self, val: float):
self.client.set_global_var("tomo_stitch_overlap", val)
tomo_stitch_overlap = _GlobalVarParam(0.2)
@property
def tomo_angle_range(self):
@@ -1885,16 +2066,7 @@ class Flomni(
raise ValueError("tomo_angle_range must be 180 or 360 degrees.")
self.client.set_global_var("tomo_angle_range", val)
@property
def golden_projections_at_0_deg_for_damage_estimation(self):
val = self.client.get_global_var("golden_projections_at_0_deg_for_damage_estimation")
if val is None:
return 0
return val
@golden_projections_at_0_deg_for_damage_estimation.setter
def golden_projections_at_0_deg_for_damage_estimation(self, val: float):
self.client.set_global_var("golden_projections_at_0_deg_for_damage_estimation", val)
golden_projections_at_0_deg_for_damage_estimation = _GlobalVarParam(0)
@property
def zero_deg_reference_at_each_subtomo(self):
@@ -1917,16 +2089,7 @@ class Flomni(
def zero_deg_reference_at_each_subtomo(self, val: bool):
self.client.set_global_var("zero_deg_reference_at_each_subtomo", val)
@property
def golden_ratio_bunch_size(self):
val = self.client.get_global_var("golden_ratio_bunch_size")
if val is None:
return 20
return val
@golden_ratio_bunch_size.setter
def golden_ratio_bunch_size(self, val: float):
self.client.set_global_var("golden_ratio_bunch_size", val)
golden_ratio_bunch_size = _GlobalVarParam(20)
@property
def frames_per_trigger(self):
@@ -1977,7 +2140,22 @@ class Flomni(
dev = builtins.__dict__.get("dev")
bec = builtins.__dict__.get("bec")
self.feye_out()
# Only run the full eye-out / optics-in transition if we are not
# already in measurement condition with feedback running. That
# transition disables and re-enables-with-reset the rt feedback, which
# re-zeros the interferometers and moves us away from the current
# sample position. Skipping it when unnecessary lets us repeat
# alignment scans (e.g. after changing scan parameters) or interleave
# an alignment run into a tomogram while staying right where the last
# measurement was.
feedback_running = dev.rtx.controller.feedback_is_running()
if self._check_eye_out_and_optics_in() and feedback_running:
print(
"Setup already in measurement condition with feedback running; "
"skipping feye_out() to avoid an interferometer reset."
)
else:
self.feye_out()
tags = ["BEC_alignment_tomo", self.sample_name]
self.write_alignment_scan_numbers(bec.queue.next_scan_number)
start_angle = 0
@@ -2247,6 +2425,68 @@ class Flomni(
end_scan_number=end_scan_number,
)
def collect_empty_frames(self):
"""Acquire 10 empty-frame (flat-field) images at angle 0, with fsamx
shifted out of the beam by half the field of view (fovx/2), in the
direction opposite the normal alignment x-offset at angle 0.
Called once at the start of a new tomo_scan() (not on resume).
These frames are deliberately not passed through tomo_reconstruct()
-- they are flat fields, not ptycho projections, and must not be
added to the reconstruction queue. Logged to
tomography_scannumbers.txt with subtomo_number=0 so they remain
traceable but are clearly distinguished from the angular grid
(subtomo_number 1-8).
"""
scans = builtins.__dict__.get("scans")
angle = 0
# --- rotation ---
fsamroy_current_setpoint = dev.fsamroy.user_setpoint.get()
if angle != fsamroy_current_setpoint:
umv(dev.fsamroy, angle)
else:
print("No rotation required")
# --- alignment offset at angle 0, then push x out of the beam by
# fovx/2, in the direction opposite the normal offset ---
offsets = self.get_alignment_offset(angle)
normal_offset_x = offsets[0]
direction = -1 if normal_offset_x >= 0 else 1
sum_offset_x = normal_offset_x + direction * (self.fovx / 2) + direction * 20
sum_offset_y = (
offsets[1]
- self.compute_additional_correction_y(angle)
- self.compute_additional_correction_y_2(angle)
+ self.manual_shift_y
)
# sum_offset_z = offsets[2]
# Todo
sum_offset_z = 0
dev.rtx.controller.laser_tracker_on()
umv(dev.rtx, sum_offset_x, dev.rty, sum_offset_y, dev.rtz, sum_offset_z)
tracker_signal = dev.rtx.controller.laser_tracker_check_signalstrength()
# checks that the fsamx coarse stage is at a position that leaves
# sufficient piezo range on the fine (rtx) stage
dev.rtx.controller.move_samx_to_scan_region(sum_offset_x)
if tracker_signal == "low":
logger.warning("Signal strength of the laser tracker is low. Realignment recommended!")
elif tracker_signal == "toolow":
raise FlomniError(
"Signal strength of the laser tracker is too low for scanning. Realignment required!"
)
print("Acquiring 10 empty frames at angle 0, fsamx shifted out of the beam.")
start_scan_number = bec.queue.next_scan_number
scans.acquire(exp_time=self.tomo_countingtime, frames_per_trigger=10)
end_scan_number = bec.queue.next_scan_number
for scan_nr in range(start_scan_number, end_scan_number):
self._write_tomo_scan_number(scan_nr, angle, subtomo_number=0)
def tomo_scan(self, subtomo_start=1, start_angle=None, projection_number=None):
"""start a tomo scan"""
@@ -2299,6 +2539,7 @@ class Flomni(
self.progress["estimated_finish_time"] = None
self.progress["accumulated_idle_time"] = 0.0
self.progress["heartbeat"] = None
self.collect_empty_frames()
with scans.dataset_id_on_hold:
if self.tomo_type == 1:
@@ -2423,9 +2664,35 @@ class Flomni(
self._print_progress()
self._log_tomogram_timing()
self.OMNYTools.printgreenbold("Tomoscan finished")
print(
f"Total measurement time lost to detected gaps: {self._format_duration(self.progress.get('accumulated_idle_time', 0.0))}"
idle_s = self.progress.get("accumulated_idle_time", 0.0)
start_str = self.progress.get("tomo_start_time")
elapsed_s = None
if start_str is not None:
try:
elapsed_s = (
datetime.datetime.now() - datetime.datetime.fromisoformat(start_str)
).total_seconds()
except (ValueError, TypeError):
elapsed_s = None
timing_lines = [
"Tomoscan finished.",
f"Measurement ID: {self.tomo_id}",
f"Sample: {self.sample_name}",
]
if elapsed_s is not None:
timing_lines.append(f"Total measurement time: {self._format_duration(elapsed_s)}")
timing_lines.append(
f"Total measurement time excluding detected gaps: {self._format_duration(elapsed_s - idle_s)}"
)
timing_lines.append(
f"Total measurement time lost to detected gaps: {self._format_duration(idle_s)}"
)
for line in timing_lines[3:]:
print(line)
timing_content = "\n".join(timing_lines)
bec.messaging.scilog.new().add_text(timing_content.replace("\n", "<br>")).add_tags(
"tomoscan"
).send()
def tomo_scan_resume(self) -> None:
"""Resume a tomo_scan() that crashed or was interrupted, picking up
@@ -2606,6 +2873,7 @@ class Flomni(
):
"""write the tomo reconstruct file for the reconstruction queue"""
bec = builtins.__dict__.get("bec")
self.reconstructor.folder_name = self.ptycho_reconstruct_foldername
self.reconstructor.write(
scan_list=self._current_scan_list,
next_scan_number=bec.queue.next_scan_number,
@@ -2726,6 +2994,108 @@ class Flomni(
}
self._append_timing_record(self._TOMOGRAM_TIMING_LOG, record)
def _read_last_timing_records(self, number_of_scans: int) -> list[dict]:
"""Return the last ``number_of_scans`` projection timing records.
Reads back the append-only projection timing log written by
_log_projection_timing(). Returns a list ordered oldest-to-newest
(i.e. in acquisition order), or an empty list if the log is missing
or unreadable.
"""
log_file = os.path.join(
os.path.expanduser(self._TIMING_LOG_DIR), self._PROJECTION_TIMING_LOG
)
records: list[dict] = []
try:
with open(log_file, "r") as in_file:
lines = [ln for ln in in_file if ln.strip()]
for line in lines[-number_of_scans:]:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
continue
except FileNotFoundError:
print(f"scilog_last_scans: timing log not found at {log_file}")
except Exception as exc: # pylint: disable=broad-except
print(f"scilog_last_scans: could not read timing log: {exc}")
return records
def scilog_last_ptycho_scans(self, number_of_scans: int = 1) -> None:
"""Write a scilog entry with a user comment and recent scan info.
The user is prompted for a free-text comment, which appears at the
top of the entry. For each of the last ``number_of_scans`` projections
(1 to 4) the entry then lists the scan number(s), field of view,
exposure (counting) time and scan duration, read from the projection
timing log. Pass ``number_of_scans=0`` to send a comment only, with no
scan information.
Only ptycho projections write a timing record, so non-ptycho scans do
not appear; if fewer records exist than requested, whatever is
available is used and a note is added.
"""
if not isinstance(number_of_scans, int) or isinstance(number_of_scans, bool):
print("scilog_last_scans: number_of_scans must be an integer between 0 and 4.")
return
if not 0 <= number_of_scans <= 4:
print("scilog_last_scans: number_of_scans must be between 0 and 4.")
return
records = self._read_last_timing_records(number_of_scans) if number_of_scans else []
if number_of_scans and not records:
print(
"scilog_last_scans: no scans found in the timing log; "
"sending a comment-only entry."
)
comment = input("Enter a comment for the scilog entry: ").strip()
if not comment and not records:
print("scilog_last_scans: empty comment and no scans — nothing to write.")
return
lines = []
if comment:
lines.append(f"{comment}")
if records:
if comment:
lines.append("")
if len(records) < number_of_scans:
lines.append(
f"flOMNI summary of the last {len(records)} scan(s) "
f"(only {len(records)} of {number_of_scans} requested were found):"
)
else:
lines.append(f"flOMNI parameters:")
lines.append("")
for rec in records:
start = rec.get("start_scan_number")
end = rec.get("end_scan_number")
if start is not None and end is not None and end - start > 1:
scan_str = f"{start}-{end - 1}"
elif start is not None:
scan_str = f"{start}"
else:
scan_str = "?"
fovx = rec.get("fovx")
fovy = rec.get("fovy")
fov_str = f"{fovx} x {fovy} um" if fovx is not None and fovy is not None else "?"
exposure = rec.get("tomo_countingtime")
exp_str = f"{exposure} s" if exposure is not None else "?"
duration = rec.get("duration_s")
dur_str = self._format_duration(duration) if duration is not None else "?"
lines.append(
f"scan {scan_str}: FOV {fov_str}, exposure {exp_str}, duration {dur_str}"
)
content = "\n".join(lines)
print(content)
bec = builtins.__dict__.get("bec")
bec.messaging.scilog.new().add_text(content.replace("\n", "<br>")).add_tags(
"tomoscan"
).send()
def _write_tomo_scan_number(self, scan_number: int, angle: float, subtomo_number: int) -> None:
tomo_scan_numbers_file = os.path.expanduser("~/data/raw/logs/tomography_scannumbers.txt")
with open(tomo_scan_numbers_file, "a+") as out_file:
@@ -2781,6 +3151,8 @@ class Flomni(
self._current_scan_list = []
start_scan_number = bec.queue.next_scan_number
projection_start = time.perf_counter()
for stitch_x in range(-self.stitch_x, self.stitch_x + 1):
for stitch_y in range(-self.stitch_y, self.stitch_y + 1):
# pylint: disable=undefined-variable
@@ -2816,6 +3188,21 @@ class Flomni(
scans.flomni_fermat_scan(**scan_kwargs)
projection_duration = time.perf_counter() - projection_start
end_scan_number = bec.queue.next_scan_number
# Internal callers (_at_each_angle, tomo_alignment_scan) run inside a
# flow that already logs projection timing in _tomo_scan_at_angle, so
# only log here for direct CLI calls (e.g. commissioning) to avoid a
# duplicate record per projection.
if not _internal:
self._log_projection_timing(
angle=angle,
subtomo_number=0,
duration_s=projection_duration,
start_scan_number=start_scan_number,
end_scan_number=end_scan_number,
)
self.tomo_reconstruct(probe_propagation=probe_propagation)
def tomo_acquire_at_angle(self, angle: float, frames_per_trigger: int | None = None):
csaxs_bec/bec_ipython_client/plugins/flomni/flomni_optics_mixin.py
+36 -6
View File
@@ -72,19 +72,50 @@ class FlomniOpticsMixin:
else:
print("FZP is already at the in position.")
def ffzp_in(self):
return need_move_optics
def ffzp_in(self, force_feedback_reset=False):
"""
move in the flomni zone plate.
This will disable rt feedback, move the FZP and re-enabled the feedback.
This will disable rt feedback, move the FZP and re-enable the feedback.
The FZP move requires rt feedback OFF, and moving the FZP invalidates
the interferometer zero, so feedback is re-enabled *with reset*
afterwards. That reset is expensive: it re-zeros the interferometers
and moves you away from wherever the sample currently sits, which is
undesirable when the FZP is already in and feedback is already running
(e.g. repeated alignment scans, or interleaving an alignment run into a
tomogram).
Therefore the disable/move/reset cycle is skipped entirely when the FZP
does not actually need to move. Pass ``force_feedback_reset=True`` to
force the full disable + reset cycle even if the FZP is already in.
"""
if "rtx" in dev and dev.rtx.enabled:
rtx_present = "rtx" in dev and dev.rtx.enabled
# Only disable feedback if we're going to move the FZP (or a reset was
# explicitly requested). If the FZP is already in and feedback is
# already running, leave it untouched -- disabling and
# re-enabling-with-reset would needlessly re-zero the interferometers.
needs_move = not self._ffzp_is_in()
do_cycle = needs_move or force_feedback_reset
if rtx_present and do_cycle:
dev.rtx.controller.feedback_disable()
self._ffzp_in()
if "rtx" in dev and dev.rtx.enabled:
if rtx_present and do_cycle:
dev.rtx.controller.feedback_enable_with_reset()
def _ffzp_is_in(self, tol=0.003):
"""True if both FZP axes (foptx, fopty) are within ``tol`` of their IN position."""
foptx_in = self._get_user_param_safe("foptx", "in")
fopty_in = self._get_user_param_safe("fopty", "in")
return np.isclose(dev.foptx.readback.get(), foptx_in, atol=tol) and np.isclose(
dev.fopty.readback.get(), fopty_in, atol=tol
)
def foptics_in(self):
"""
Move in the flomni optics, including the FZP and the OSA.
@@ -290,5 +321,4 @@ class FlomniOpticsMixin:
print(
f" Note: OSA is {(-diff)*1000:.1f} um away from its IN position (likely parked OUT)."
)
print(f" Remaining space if OSA is moved to its IN position: \033[1m{remaining_at_in:.1f}\033[0m")
print(f" Remaining space if OSA is moved to its IN position: \033[1m{remaining_at_in:.1f}\033[0m")
csaxs_bec/bec_ipython_client/plugins/flomni/gui_tools.py
+29 -4
View File
@@ -22,6 +22,14 @@ class flomniGuiToolsError(Exception):
class flomniGuiTools:
# Screen assumed 2560x1440. Window is right-aligned with a small margin
# from the top; width/height tuned from the first test render (which
# came out taller than intended at 1000px).
_SCREEN_WIDTH = 2560
_WINDOW_WIDTH = 1500
_WINDOW_HEIGHT = 850
_WINDOW_TOP_MARGIN = 50
def __init__(self):
self.text_box = None
self.progressbar = None
@@ -31,6 +39,7 @@ class flomniGuiTools:
self.idle_text_box = None
self.camera_gripper_image = None
self.camera_overview_image = None
self.console = None
def set_client(self, client):
self.client = client
@@ -41,7 +50,10 @@ class flomniGuiTools:
self.flomni_window = self.gui.windows["flomni"]
self.gui.flomni.raise_window()
else:
self.flomni_window = self.gui.new("flomni")
# geometry: (pos_x, pos_y, w, h)
pos_x = self._SCREEN_WIDTH - self._WINDOW_WIDTH
geometry = (pos_x, self._WINDOW_TOP_MARGIN, self._WINDOW_WIDTH, self._WINDOW_HEIGHT)
self.flomni_window = self.gui.new("flomni", geometry=geometry)
time.sleep(1)
def flomnigui_stop_gui(self):
@@ -90,8 +102,10 @@ class flomniGuiTools:
def flomnigui_show_cameras(self):
self.flomnigui_show_gui()
if self._flomnigui_is_missing("camera_gripper_image") or self._flomnigui_is_missing(
"camera_overview_image"
if (
self._flomnigui_is_missing("camera_gripper_image")
or self._flomnigui_is_missing("camera_overview_image")
or self._flomnigui_is_missing("console")
):
self.flomnigui_remove_all_docks()
self.camera_gripper_image = self.gui.flomni.new("Image")
@@ -117,6 +131,16 @@ class flomniGuiTools:
else:
print("Cannot open camera_overview. Device does not exist.")
# Confirm/abort console, docked below the cameras.
self.console = self.gui.flomni.new(
"ConsoleButtonsWidget", object_name="console", where="bottom"
)
# set_layout_ratios uses relative weights, not pixels -- there is
# no width/height kwarg on dock_area.new(). [5, 1] gives the
# cameras most of the vertical space and keeps the console a
# slim strip at the bottom; adjust to taste once you see it.
self.gui.flomni.set_layout_ratios(vertical=[5, 1])
def flomnigui_remove_all_docks(self):
# dev.cam_flomni_overview.stop_live_mode()
# dev.cam_flomni_gripper.stop_live_mode()
@@ -130,6 +154,7 @@ class flomniGuiTools:
self.idle_text_box = None
self.camera_gripper_image = None
self.camera_overview_image = None
self.console = None
def flomnigui_idle(self):
self.flomnigui_show_gui()
@@ -316,4 +341,4 @@ if __name__ == "__main__":
flomni_gui = flomniGuiTools()
flomni_gui.set_client(client)
flomni_gui.flomnigui_show_gui()
flomni_gui.flomnigui_show_progress()
flomni_gui.flomnigui_show_progress()
csaxs_bec/bec_ipython_client/plugins/flomni/x_ray_eye_align.py
+31 -9
View File
@@ -28,9 +28,13 @@ if TYPE_CHECKING:
class XrayEyeAlign:
# pixel calibration, multiply to get mm
# Pixel calibration in mm/pixel (multiply pixel values to get mm).
# The live value is read from the cam_xeye device's ``pixel_calibration``
# user parameter (see the ``pixel_calibration`` property); this constant is
# only the fallback used when that device/parameter is unavailable.
test_wo_movements = False
PIXEL_CALIBRATION = 0.1 / 113 # .2 with binning
PIXEL_CALIBRATION_DEFAULT = 0.05 / 113 # mm per raw ROI pixel (.1/113 with binning)
PIXEL_CALIBRATION_USER_PARAM = "pixel_calibration"
# Sign for the automatic vertical-centering move in the height-centering
# branch of _align_impl (search for `_height_centered`).
@@ -62,6 +66,22 @@ class XrayEyeAlign:
def gui(self):
return self.flomni.xeyegui
@property
def pixel_calibration(self) -> float:
"""Pixel calibration in mm/pixel.
Reads the ``pixel_calibration`` user parameter from the cam_xeye device;
falls back to ``PIXEL_CALIBRATION_DEFAULT`` if the device or parameter is
unavailable.
"""
try:
mm_per_pixel = dev.cam_xeye.user_parameter.get(self.PIXEL_CALIBRATION_USER_PARAM)
except Exception:
mm_per_pixel = None
if mm_per_pixel is None:
return self.PIXEL_CALIBRATION_DEFAULT
return float(mm_per_pixel)
def _reset_init_values(self):
self.shift_xy = [0, 0]
self._xray_fov_xy = [0, 0]
@@ -183,6 +203,7 @@ class XrayEyeAlign:
# there.
self.flomni.reset_correction()
self.flomni.reset_tomo_alignment_fit()
self.flomni.manual_shift_y = 0
self.flomni.lights_off()
@@ -222,7 +243,7 @@ class XrayEyeAlign:
if dev.omny_xray_gui.submit.get() == 1:
self.alignment_values[k] = (
getattr(dev.omny_xray_gui, f"xval_x_{k}").get() / 2 * self.PIXEL_CALIBRATION
getattr(dev.omny_xray_gui, f"xval_x_{k}").get() * self.pixel_calibration
) # in mm
print(f"Clicked position {k}: x {self.alignment_values[k]}")
rtx_position = dev.rtx.readback.get() / 1000
@@ -297,8 +318,7 @@ class XrayEyeAlign:
delta_y_mm = (
self.HEIGHT_CENTERING_SIGN
* (fzp_center_y - mark_y)
/ 2
* self.PIXEL_CALIBRATION
* self.pixel_calibration
)
print(
f"Height centering: fzp_center_y={fzp_center_y}, mark_y={mark_y}, "
@@ -367,8 +387,8 @@ class XrayEyeAlign:
time.sleep(0.1)
self.write_output()
fovx = self._xray_fov_xy[0] * self.PIXEL_CALIBRATION * 1000 / 2
fovy = self._xray_fov_xy[1] * self.PIXEL_CALIBRATION * 1000 / 2
fovx = self._xray_fov_xy[0] * self.pixel_calibration * 1000
fovy = self._xray_fov_xy[1] * self.pixel_calibration * 1000
if keep_shutter_open:
if self.flomni.OMNYTools.yesno("Close the shutter now?", "y"):
@@ -376,10 +396,12 @@ class XrayEyeAlign:
self.gui.on_live_view_enabled(False)
print("setting 'XOMNYI-XEYE-ACQ:0'")
print(
self.flomni.OMNYTools.printgreenbold(
f"The largest field of view from the xrayeyealign was \nfovx = {fovx:.0f} microns, fovy"
f" = {fovy:.0f} microns"
f" = {fovy:.0f} microns. Adjusting the flomni.fovx to {fovx+6:.0f} microns."
)
self.flomni.fovx = fovx+6
print("Check the fit in the GUI...")
time.sleep(5)
csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py
+67
View File
@@ -93,6 +93,73 @@ class OMNYTools:
else:
print("Please expicitely confirm y or n.")
def gui_yesno(
self, message: str, gui, default="none", autoconfirm=0, poll_interval: float = 0.1
) -> bool:
"""
GUI-based alternative to yesno(), using a ConsoleButtonsWidget
(csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons.py)
instead of a blocking console input() prompt.
Not yet wired up as a replacement for yesno() anywhere -- this is
for standalone testing. Once confirmed working, call sites can be
switched over deliberately, one at a time.
Opening the GUI (must already exist and be passed in as `gui`;
this method does not create it):
gui.new("test", timeout=20)
console = gui.test.new("ConsoleButtonsWidget", object_name="console", timeout=20)
Then call, e.g. from an IPython session:
omny_tools.gui_yesno("Continue with sample transfer?", gui=console)
Args:
message (str): Question to display on the widget.
gui: The already-open ConsoleButtonsWidget RPC object (e.g. the
`console` object created above, or later a fixed instance
such as `gui.flomni.console` once wired into flomni's own
gui tools).
default (str): "y" or "n" -- only affects the autoconfirm path,
same as yesno(). There is no "just press enter" equivalent
for a button click, so this has no effect otherwise.
autoconfirm (int): if set together with default="y"/"n", skips
the GUI entirely and returns immediately, same as yesno().
poll_interval (float): seconds between response checks, same
style as the 0.1 s submit-poll in XrayEyeAlign.align().
Returns:
bool: True for "yes", False for "no".
Note on abort: the widget's ABORT button does not write a response
to poll for -- it sends a real SIGINT directly to this process (see
ConsoleButtonsWidget._on_abort), so pressing it raises
KeyboardInterrupt here exactly as a console Ctrl+C would, and
propagates normally out of this method without any special-casing.
"""
if autoconfirm and default == "y":
self.printgreen(message + " Automatically confirming default: yes")
return True
elif autoconfirm and default == "n":
self.printgreen(message + " Automatically confirming default: no")
return False
suffix = {"y": " [Y]/n?", "n": " y/[N]?"}.get(default, " y/n?")
gui.clear_response()
gui.message = message + suffix
while True:
response = gui.response()
if response == "yes":
gui.clear_response()
gui.message = ""
return True
if response == "no":
gui.clear_response()
gui.message = ""
return False
time.sleep(poll_interval)
def tweak_cursor(
self, dev1, step1: float, dev2="none", step2: float = "0", special_command="none"
):
csaxs_bec/bec_widgets/widgets/client.py
+50 -1
View File
@@ -12,6 +12,7 @@ logger = bec_logger.logger
_Widgets = {
"ConsoleButtonsWidget": "ConsoleButtonsWidget",
"SampleStorageWidget": "SampleStorageWidget",
"SAXSWidget": "SAXSWidget",
"SlitControlWidget": "SlitControlWidget",
@@ -20,6 +21,47 @@ _Widgets = {
}
class ConsoleButtonsWidget(RPCBase):
"""Small Yes / No / Abort control widget, intended as a GUI replacement for"""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons"
@property
@rpc_call
def message(self):
"""
None
"""
@message.setter
@rpc_call
def message(self):
"""
None
"""
@rpc_call
def response(self) -> "str":
"""
Current button response: "yes", "no", or "" if no button has been
pressed (or since the last clear_response()).
Note: intentionally a plain method, not @SafeProperty. SafeProperty
only becomes a real Qt property descriptor once a setter is chained
in the class body (see `message` above); a getter-only SafeProperty
is left as an unconverted internal wrapper object, which the RPC
generator then exposes as a callable stub anyway. Since `response`
has no RPC-facing setter, a plain method avoids that trap entirely -
same pattern XRayEye uses for its own read-only `active_roi`.
"""
@rpc_call
def clear_response(self):
"""
None
"""
class SampleStorageWidget(RPCBase):
"""View and correct the FlOMNI sample-storage records."""
@@ -28,7 +70,14 @@ class SampleStorageWidget(RPCBase):
@rpc_call
def refresh(self) -> "None":
"""
Re-read every slot from the device and update the cells.
Re-read all slots (one bulk device round-trip) and update only the
cells whose state actually changed.
Repainting every cell on every 2 s poll — even when nothing changed,
which is almost always the case for a sample magazine — was needless
work on the GUI thread. We diff against the last-seen state and only
call set_state() on cells that differ, so a steady-state poll does no
UI work at all.
"""
csaxs_bec/bec_widgets/widgets/console_buttons/__init__.py
View File
csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons.py
+171
View File
@@ -0,0 +1,171 @@
from __future__ import annotations
import os
import signal
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import VariableMessage
from bec_widgets import BECWidget, SafeProperty, SafeSlot
from bec_widgets.utils.rpc_decorator import rpc_timeout
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget
logger = bec_logger.logger
class ConsoleButtonsWidget(BECWidget, QWidget):
"""
Small Yes / No / Abort control widget, intended as a GUI replacement for
console prompts (e.g. ``OMNYTools.yesno()``) and as a general-purpose
emergency-stop control.
- Yes / No: set ``response`` to "yes" / "no". A blocking CLI script can
poll ``gui.<name>.response`` and reset it via ``clear_response()``.
- Abort: sends a real SIGINT to the BEC IPython client process (the
parent of the GUI server process), equivalent to pressing Ctrl+C in
the console. This works even if the client is blocked inside a motor
move or other long call, since it is a real OS signal rather than a
polled flag. The SIGINT is only sent if the parent process actually
looks like a BEC client -- it does not when the widget is opened
standalone from the launcher menu. In addition, 500 ms later a device
stop request is published to the device server (same mechanism as the
PositionerBox stop button): by default for ALL devices, so the widget
acts as a generic emergency stop from any context, even if the client
process is hung or dead.
"""
USER_ACCESS = ["message", "message.setter", "response", "clear_response"]
PLUGIN = True
def __init__(self, parent=None, **kwargs):
# Devices for which a backup stop request is published when ABORT is
# pressed. An empty list (default) means "stop ALL devices" -- the
# same device-server path BEC uses on scan abort -- which makes the
# widget a generic emergency stop, e.g. when opened standalone from
# the launcher menu. Stop-all also covers the flomni sample transfer:
# XQ#STOP via ftransy is controller-wide and halts the #GRGET/#GRPUT
# thread. Pass an explicit list to restrict the stop.
self._backup_stop_devices = list(kwargs.pop("backup_stop_devices", []))
super().__init__(parent=parent, **kwargs)
self._response = ""
# Captured once at construction time: the GUI server process is a
# direct child of the BEC IPython client (subprocess.Popen in
# bec_widgets.cli.client_utils), so getppid() here is the client's
# PID. Not re-read later, so a subsequent reparenting (e.g. if the
# client died) can't silently redirect the signal.
self._client_pid = os.getppid()
self._init_ui()
def _init_ui(self):
layout = QVBoxLayout(self)
button_row = QHBoxLayout()
self.yes_button = QPushButton("Yes", parent=self)
self.no_button = QPushButton("No", parent=self)
self.abort_button = QPushButton("ABORT", parent=self)
self.abort_button.setStyleSheet(
"background-color: #c0392b; color: white; font-weight: bold;"
)
button_row.addWidget(self.yes_button)
button_row.addWidget(self.no_button)
button_row.addWidget(self.abort_button)
layout.addLayout(button_row)
self.message_label = QLabel("", parent=self)
self.message_label.setWordWrap(True)
layout.addWidget(self.message_label)
self.yes_button.clicked.connect(self._on_yes)
self.no_button.clicked.connect(self._on_no)
self.abort_button.clicked.connect(self._on_abort)
# Start with Yes/No greyed out: with no message there is nothing to
# respond to, so there should be nothing clickable. ABORT is left
# always enabled -- it's an emergency stop and must work at any time,
# message or not.
self._set_yesno_enabled(False)
def _set_yesno_enabled(self, enabled: bool):
self.yes_button.setEnabled(enabled)
self.no_button.setEnabled(enabled)
@SafeSlot()
def _on_yes(self):
self._response = "yes"
@SafeSlot()
def _on_no(self):
self._response = "no"
def _client_is_bec_process(self) -> bool:
"""
Heuristic check whether the parent process (captured at construction
time) is a BEC IPython client. Only then does SIGINT make sense: when
the widget is opened from the standalone launcher instead of a
client-spawned GUI server, the parent is e.g. a shell or the launcher
process, which must not receive the signal.
"""
try:
with open(f"/proc/{self._client_pid}/cmdline", "rb") as f:
cmdline = f.read().replace(b"\x00", b" ").decode(errors="ignore")
except OSError:
return False
return ("bec" in cmdline) or ("ipython" in cmdline)
@SafeSlot()
def _on_abort(self):
if self._client_is_bec_process():
logger.warning(f"ConsoleButtonsWidget: sending SIGINT to client pid {self._client_pid}")
os.kill(self._client_pid, signal.SIGINT)
else:
logger.warning(
"ConsoleButtonsWidget: parent process does not look like a BEC client;"
" skipping SIGINT and only sending the device stop request."
)
# Backup: direct device stop via the device server, independent of
# the client process. Delayed so the SIGINT-triggered abort handler
# in the client (which still sees mntprgs=1 and aborts in a
# controlled way) wins the race: if the stop landed first, #STOP
# would clear mntprgs and the client transfer loop would exit
# "cleanly" into ensure_gripper_up, which must not happen
# mid-transfer.
QTimer.singleShot(500, self._send_backup_stop)
@SafeSlot()
def _send_backup_stop(self):
"""Publish a stop request for the configured devices to the device server."""
devices = self._backup_stop_devices
logger.warning(f"ConsoleButtonsWidget: sending backup stop request for {devices}")
self.client.connector.send(MessageEndpoints.stop_devices(), VariableMessage(value=devices))
@SafeProperty(str)
def message(self):
return self.message_label.text()
@message.setter
@rpc_timeout(20)
def message(self, text: str):
self.message_label.setText(text)
# A non-empty message means there's something to respond to, so
# enable Yes/No; an empty message greys them out again.
self._set_yesno_enabled(bool(text))
def response(self) -> str:
"""
Current button response: "yes", "no", or "" if no button has been
pressed (or since the last clear_response()).
Note: intentionally a plain method, not @SafeProperty. SafeProperty
only becomes a real Qt property descriptor once a setter is chained
in the class body (see `message` above); a getter-only SafeProperty
is left as an unconverted internal wrapper object, which the RPC
generator then exposes as a callable stub anyway. Since `response`
has no RPC-facing setter, a plain method avoids that trap entirely -
same pattern XRayEye uses for its own read-only `active_roi`.
"""
return self._response
@SafeSlot()
def clear_response(self):
self._response = ""
csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons_widget.pyproject
+1
View File
@@ -0,0 +1 @@
{'files': ['console_buttons.py']}
csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons_widget_plugin.py
+57
View File
@@ -0,0 +1,57 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons import ConsoleButtonsWidget
DOM_XML = """
<ui language='c++'>
<widget class='ConsoleButtonsWidget' name='console_buttons_widget'>
</widget>
</ui>
"""
class ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = ConsoleButtonsWidget(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(ConsoleButtonsWidget.ICON_NAME)
def includeFile(self):
return "console_buttons_widget"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "ConsoleButtonsWidget"
def toolTip(self):
return ""
def whatsThis(self):
return self.toolTip()
csaxs_bec/bec_widgets/widgets/console_buttons/register_console_buttons_widget.py
+15
View File
@@ -0,0 +1,15 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons_widget_plugin import ConsoleButtonsWidgetPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(ConsoleButtonsWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
main()
csaxs_bec/bec_widgets/widgets/designer_plugins.py
+5
View File
@@ -5,6 +5,10 @@ from __future__ import annotations
# pylint: skip-file
designer_plugins = {
"ConsoleButtonsWidget": (
"csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons",
"ConsoleButtonsWidget",
),
"SampleStorageWidget": (
"csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage",
"SampleStorageWidget",
@@ -22,6 +26,7 @@ designer_plugins = {
}
widget_icons = {
"ConsoleButtonsWidget": "widgets",
"SampleStorageWidget": "widgets",
"SAXSWidget": "table_chart",
"SlitControlWidget": "widgets",
csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py
+3
View File
@@ -339,6 +339,9 @@ class SampleStorageWidget(BECWidget, QWidget):
grid.setContentsMargins(0, 0, 0, 0)
for idx, slot in enumerate(STORAGE_SLOTS):
r, c = divmod(idx, STORAGE_COLS)
# display each row right-to-left (5..1, 10..6, ...) rather than
# left-to-right, to match the physical magazine orientation
c = STORAGE_COLS - 1 - c
cell = _SlotCell(slot, self, COLOR_SLOT_BORDER)
self._cells[slot] = cell
grid.addWidget(cell, r, c)
csaxs_bec/bec_widgets/widgets/xray_eye/x_ray_eye.py
+74 -1
View File
@@ -260,6 +260,13 @@ class XRayEye(BECWidget, QWidget):
ROI_LINE_COLOR = "blue"
ROI_LINE_WIDTH = 2
# Image scale read from the camera's ``pixel_calibration`` user parameter
# (mm/pixel, matching XrayEyeAlign.PIXEL_CALIBRATION in x_ray_eye_align.py).
# If the camera device or the parameter is missing, fall back to 1 (i.e.
# sizes are reported in raw pixels).
PIXEL_CALIBRATION_USER_PARAM = "pixel_calibration"
PIXEL_CALIBRATION_DEFAULT = 1.0
def __init__(self, parent=None, **kwargs):
super().__init__(parent=parent, **kwargs)
self._connected_motor = None
@@ -398,10 +405,21 @@ class XRayEye(BECWidget, QWidget):
form.addWidget(QLabel("Sample", parent=self), 0, 0)
form.addWidget(self.sample_name_line_edit, 0, 1)
self.message_line_edit = QTextEdit(parent=self)
self.message_line_edit.setFixedHeight(60)
self.message_line_edit.setFixedHeight(90)
self.message_line_edit.setReadOnly(True)
form.addWidget(QLabel("Message", parent=self), 1, 0)
form.addWidget(self.message_line_edit, 1, 1)
# Live ROI size readout (microns), updated as the active ROI is resized.
self.roi_size_line_edit = QLineEdit(parent=self)
self.roi_size_line_edit.setReadOnly(True)
form.addWidget(QLabel("ROI size", parent=self), 2, 0)
form.addWidget(self.roi_size_line_edit, 2, 1)
# Pixel calibration readout (microns/pixel), from the cam_xeye device.
self.calibration_line_edit = QLineEdit(parent=self)
self.calibration_line_edit.setReadOnly(True)
form.addWidget(QLabel("Calibration", parent=self), 3, 0)
form.addWidget(self.calibration_line_edit, 3, 1)
self._update_calibration_readout()
self.control_panel_layout.addLayout(form)
# Fix panel width and allow vertical expansion
@@ -475,6 +493,61 @@ class XRayEye(BECWidget, QWidget):
def _style_new_roi(self, roi):
"""Force a thinner outline on newly drawn ROIs (color is set via compact_color)."""
roi.line_width = self.ROI_LINE_WIDTH
# Live-update the microns readout while this ROI is drawn/resized.
roi.sigRegionChanged.connect(lambda r=roi: self._update_roi_size_readout(r))
self._update_roi_size_readout(roi)
def _microns_per_pixel(self):
"""Resolve microns/pixel and unit label from the camera user parameter.
Reads ``cam_xeye``'s ``pixel_calibration`` user parameter (mm/pixel) and
converts to microns. If the camera device or the parameter is not
available, falls back to ``PIXEL_CALIBRATION_DEFAULT`` and reports the
unit as raw pixels.
Returns:
tuple[float, str]: (scale factor applied to pixel sizes, unit label).
"""
try:
cam = getattr(self.dev, CAMERA[0])
mm_per_pixel = cam.user_parameter.get(self.PIXEL_CALIBRATION_USER_PARAM)
except Exception:
mm_per_pixel = None
if mm_per_pixel is None:
return self.PIXEL_CALIBRATION_DEFAULT, "px"
return float(mm_per_pixel) * 1000, "um"
def _update_calibration_readout(self):
"""Update the pixel-calibration readout from the camera user parameter."""
scale, unit = self._microns_per_pixel()
if unit == "um":
self.calibration_line_edit.setText(f"{scale:.4f} um/pixel")
else:
self.calibration_line_edit.setText("uncalibrated (pixels)")
def _update_roi_size_readout(self, roi=None):
"""Update the ROI size readout (microns, or pixels if uncalibrated)."""
self._update_calibration_readout()
if roi is None:
roi = self.roi_manager.single_active_roi
if roi is None:
self.roi_size_line_edit.setText("")
return
if isinstance(roi, RectangularROI):
coords = roi.get_coordinates(typed=True)
size_x = coords["width"]
size_y = coords["height"]
elif isinstance(roi, CircularROI):
# Read the raw x/y size so a non-circular (ellipse) state is reflected
# correctly; for an aspect-locked circle both values are equal.
size_x, size_y = roi.state["size"]
else:
self.roi_size_line_edit.setText("")
return
scale, unit = self._microns_per_pixel()
self.roi_size_line_edit.setText(
f"x = {abs(size_x) * scale:.1f} {unit}, y = {abs(size_y) * scale:.1f} {unit}"
)
def _create_separator(self):
sep = QFrame(parent=self)
csaxs_bec/device_configs/bl_endstation.yaml
+16
View File
@@ -52,6 +52,22 @@ fsh:
enabled: true
readoutPriority: monitored
##########################################################################
########################### FAST SHUTTER #################################
##########################################################################
shutter_es:
description: X12SA ES shutter status
deviceClass: ophyd_devices.EpicsSignalRO
deviceConfig:
read_pv: X12SA-OP-PSH1-EMLS-7010:OPEN
auto_monitor: true
onFailure: buffer
enabled: true
readoutPriority: baseline
readOnly: true
softwareTrigger: false
##########################################################################
######################## SMARACT STAGES ##################################
##########################################################################
csaxs_bec/device_configs/ptycho_flomni.yaml
+22 -20
View File
@@ -77,10 +77,11 @@ foptx:
connectionTimeout: 20
userParameter:
#170 micron, 60 nm
#in: -13.831
in: -13.831
out: -13.831
#250 micron, 30 nm, Abe structures
in: -13.8809375
out: -14.1809
# in: -13.8809375
# out: -14.1809
#250 micron, 30 nm, Tomas structures
# in: -14.5490625
# out: -14.1809
@@ -103,11 +104,11 @@ fopty:
connectionTimeout: 20
userParameter:
#170 micron, 60 nm
#in: 0.42
#out: 0.57
in: 0.42
out: 0.57
#250 micron, 30 nm, Abe structures
in: 2.8299
out: 2.8299
# in: 2.8299
# out: 2.8299
#250 micron, 30 nm, Tomas structures
# in: 2.8419921875
# out: 2.8419921875
@@ -321,11 +322,11 @@ fosax:
#in: 8.7568
#out: 5.1
#170 micron, 60 nm, 7.9 kev
# in: 8.731922
# out: 5.1
#250 micron, 30 nm, Abe structures
in: 8.7392
in: 8.727079
out: 5.1
#250 micron, 30 nm, Abe structures
# in: 8.7392
# out: 5.1
#250 micron, 30 nm, Tomas structures
# in: 9.420798
# out: 5.1
@@ -349,10 +350,10 @@ fosay:
userParameter:
#170 micron, 60 nm, 7.6 kev
#in: -0.0235
#170 micron, 60 nm, 7.6 kev
#in: -0.0422
#170 micron, 60 nm, 7.9 kev
in: -0.04603
#250 micron, 30 nm, Abe structures
in: -2.3684
# in: -2.3684
#250 micron, 30 nm, Tomas structures
# in: -2.383993
fosaz:
@@ -375,12 +376,12 @@ fosaz:
#170 micron, 60 nm, 7.6 kev
#in: 8.5
#out: 6
#170 micron, 60 nm, 7.9 kev, foptz 15.9
# in: 11.9
# out: 6
#170 micron, 60 nm, 7.9 kev, foptz 16.9, probe size 7.5 mu
in: 13.1
out: 6
# micron, 30 nm, 7.9 kev, very close to the sample. make sure foptz is 32.02 or smaller //abe's fzp's
in: 0.5
out: -5
# in: 0.5
# out: -5
############################################################
#################### flOMNI RT motors ######################
@@ -493,7 +494,8 @@ cam_xeye:
onFailure: buffer
readOnly: false
readoutPriority: async
userParameter:
pixel_calibration: 0.00044247787610619477 # mm/pixel (= 0.05 / 113)
# cam_ids_rgb:
# description: Camera flOMNI Xray eye ID203
# deviceClass: csaxs_bec.devices.ids_cameras.ids_camera.IDSCamera
docs/user/ptychography/flomni.md
+5 -1
View File
@@ -24,6 +24,10 @@ You will be asked to enter a sample name.
To load a new sample in the sample stage, in principle only one command is needed
`flomni.ftransfer_sample_change(<position new sample>)`
You will be asked where the previous sample should go with a suggestion for an empty position in the tray.
To remove the current sample from the sample stage __without__ mounting a new one, use
`flomni.ftransfer_sample_change(-1)`
You will be asked where to stow the removed sample, same as above.
Other commands:
`ftransfer_tray_in /_out (not yet implemented).`
@@ -54,7 +58,7 @@ If you see your sample already at the approximately correct height:
Otherwise adjust the height manually:
1. `umvr_fsamy_tracked(0.01)`, attention: unit <mm>, move the sample stage relative up (positive) or down (negative) until the sample is approximately vertically centered in xray eye screen
1. `flomni.umvr_fsamy_tracked(0.01)`, attention: unit <mm>, move the sample stage relative up (positive) or down (negative) until the sample is approximately vertically centered in xray eye screen
1. `flomni.xrayeye_update_frame()` will update the current image on the xray eye screen
#### Fine alignment