daily setup routine for pxiii
CI for pxiii_bec / test (push) Successful in 1m6s

This commit is contained in:
x06da committed 2026-10-04 16:57:36 +02:00
1 parent d85abc533c
commit e1aa898fbe
1 file changed
+139 -351
+139 -351
View File
@@ -14,8 +14,8 @@ import numpy as np
from datetime import datetime
from contextlib import contextmanager
from dataclasses import dataclass
# from pxii_bec.scripts import beamline_context as bl
# from pxii_bec.macros.beamline_planner import BeamlineState
from pxiii_bec.scripts import beamline_context as bl
from pxiii_bec.macros.beamline_planner import BeamlineState
import matplotlib.pyplot as plt
from matplotlib.patches import Ellipse
@@ -38,10 +38,10 @@ class DailySetupContext:
def daily_setup(
energy=DailySetup.energy,
energy=None,
*,
pre_checks=True,
set_energy=False, #until PXIII energy is tested
set_energy=True,
mono_scan=True,
move_beam=True, # beam check only to start with
align_beamstop=False,
@@ -62,6 +62,9 @@ def daily_setup(
"Measure flux": measure_flux,
}
if energy is None:
energy = DailySetup.energy
results = {}
setup_start = time.perf_counter()
@@ -199,7 +202,7 @@ def daily_setup(
finally:
print("done")
planner.move_to(BeamlineState.ROBOT_SAMPLE_EXCHANGE)
# bl.planner.move_to(BeamlineState.ROBOT_SAMPLE_EXCHANGE)
def get_daily_setup_context(tasks=None):
@@ -287,31 +290,31 @@ def run_pre_checks(ctx):
# ---------------------------------------------------------
# Experimental hutch checks - advisory
# ---------------------------------------------------------
# eh_result = EH_prechecks()
eh_result = EH_prechecks()
# result["experiment_hutch"] = eh_result
result["experiment_hutch"] = eh_result
# # EH problems are warnings, not setup failures
# result["ok"] = True
# result["warning"] = not eh_result["ok"]
# EH problems are warnings, not setup failures
result["ok"] = True
result["warning"] = not eh_result["ok"]
# update_log(
# ctx.log,
# ctx.messages,
# format_prechecks_for_gui(result),
# )
update_log(
ctx.log,
ctx.messages,
format_prechecks_for_gui(result),
)
# status = "warning" if result["warning"] else "done"
status = "warning" if result["warning"] else "done"
# set_progress(
# ctx.progress,
# ctx.progress_state,
# task,
# status,
# )
set_progress(
ctx.progress,
ctx.progress_state,
task,
status,
)
# # EH warnings do NOT stop daily setup
# return True, result
# EH warnings do NOT stop daily setup
return True, result
def format_prechecks_for_gui(result):
@@ -460,10 +463,10 @@ def EH_prechecks():
"""Check experimental hutch devices."""
checks = [
check_xray_eye_out,
# check_xray_eye_out,
check_cryo_temp,
check_detector_height,
check_beamstop_position,
# check_detector_height,
# check_beamstop_position,
]
results = []
@@ -519,8 +522,8 @@ def check_beamstop_position():
"""
try:
if not planner.is_state(BeamlineState.DET_DIODE):
planner.move_to(BeamlineState.DET_DIODE)
if not bl.planner.is_state(BeamlineState.DET_DIODE):
bl.planner.move_to(BeamlineState.DET_DIODE)
bl.d["det_cov"].move("close")
dev.transm.put(EnergyDefaults.default_trans)
@@ -595,7 +598,7 @@ def check_beamstop_position():
dev.bcu_shutter.put(0)
def check_cryo_temp():
cryo_pv = EHChecks.cryo_pv
cryo_pv = PreChecks.cryo_pv
temp = caget(cryo_pv)
if temp is None:
@@ -683,25 +686,26 @@ def daily_energy(energy, scan_mono, plot_widget=None):
bl_energy(energy, window_name="DailySetup", plot_widget=plot_widget)
else:
gap_actual = dev.id_gap.position
gap_req = calc_gap(energy)
if bl.env.beamline_name != "PXIII":
gap_actual = dev.id_gap.position
gap_req = calc_gap(energy)
if abs(gap_actual - gap_req) > 0.05:
if dev.id_gap.moving():
return {
"ok": False,
"message": "Undulator gap is moving",
"energy": current_energy,
}
if abs(gap_actual - gap_req) > 0.05:
if dev.id_gap.moving():
return {
"ok": False,
"message": "Undulator gap is moving",
"energy": current_energy,
}
umv(dev.id_gap, gap_req)
umv(dev.id_gap, gap_req)
if scan_mono:
mono_pitch_scan(window_name="DailySetup", plot_widget=plot_widget)
new_energy = get_current_energy()
calib_pitch = get_dcm_motors_positions(energy)["dcm_pitch"]
pitch_offset = get_value(dev.dcm_pitch) - calib_pitch
calib_pitch = get_dccm_motors_positions(energy)["theta2_angle"]
pitch_offset = get_value(dev.dccm_theta2) - calib_pitch
return {
"ok": True,
@@ -743,12 +747,12 @@ def run_daily_movebeam(ctx):
"message": f"Align beam step failed: {exc}",
}
if result.get("ok"):
status = "done"
elif result.get("warning"):
if result.get("warning"):
status = "warning"
elif result.get("ok"):
status = "done"
else:
status = "failed"
@@ -758,296 +762,80 @@ def run_daily_movebeam(ctx):
return result
# def daily_movebeam(meas="scint", iterations=5):
# """
# Move the beam onto the sample-camera crosshair using the VFM and HFM.
# The sample crosshair parameters are defined in BeamMove class.
def daily_movebeam():
"""Move the beam onto the sample-camera crosshair using PXIII beam steering."""
# The beam position is measured on the scintillator at 12.5X zoom.
# VFM motor to move the beam vertically is defined in BeamMove class.
# HFM motor to move beam horizontally is defined in BeamMove class.
prepare_scint_measurement()
# Corrections are repeated until the beam is within tolerance or the
# maximum number of iterations is reached.
# """
# if meas != "scint":
# raise ValueError(f"Unknown beam position measurement: {meas}")
# # ------------------------------------------------------------------
# # Setup
# # ------------------------------------------------------------------
# zoom = BeamMove.zoom
# vfm_mot = BeamMove.vfm_mot
# hfm_mot = BeamMove.hfm_mot
# x_target = BeamMove.scint_target_x
# y_target = BeamMove.scint_target_y
# tolerance = BeamMove.scint_tolerance
# # Safety limit for a single mirror-jack correction
# max_move = BeamMove.max_move
# # Starting mirror positions
# vfm_start = get_value(vfm_mot)
# hfm_start = get_value(hfm_mot)
# print(
# "Moving beam to scintillator crosshair\n"
# f"Target: X={x_target:.1f}, Y={y_target:.1f} px"
# )
# # ------------------------------------------------------------------
# # Initial measurement
# # ------------------------------------------------------------------
# prepare_scint_measurement(zoom=zoom)
# try:
# scint_result = read_scintXY()
# if not scint_result["ok"]:
# print(f"✗ {scint_result['message']}")
# return {
# "ok": False,
# "warning": True,
# "message": scint_result["message"],
# }
# x_start = scint_result["X_pixels"]
# y_start = scint_result["Y_pixels"]
# # ------------------------------------------------------------------
# # Beam corrections
# # ------------------------------------------------------------------
# for i in range(iterations):
# x = scint_result["X_pixels"]
# y = scint_result["Y_pixels"]
# diff_x = x_target - x
# diff_y = y_target - y
# print(
# f"\nCorrection {i + 1}: "
# f"X={x:.2f}, Y={y:.2f} px "
# f"(error X={diff_x:+.2f}, Y={diff_y:+.2f})"
# )
# # Already aligned
# if abs(diff_x) <= tolerance and abs(diff_y) <= tolerance:
# print("Beam is within tolerance.")
# break
# # --------------------------------------------------------------
# # VFM: vertical beam position
# # --------------------------------------------------------------
# if abs(diff_y) > tolerance:
# current_vfm = get_value(vfm_mot)
# vfm_change = diff_y / BeamMove.scint_y_slope
# if abs(vfm_change) > max_move:
# message = (
# f"VFM correction {vfm_change:+.5f} mm exceeds "
# f"maximum allowed move of {max_move:.3f} mm. "
# "Manual check required."
# )
# print(message)
# return {
# "ok": False,
# "warning": True,
# "message": message,
# "BeamX": None,
# "BeamY": None,
# }
# move_with_offset(
# vfm_mot,
# current_vfm + vfm_change,
# )
# # Remeasure because VFM movement may also affect X
# scint_result = read_scintXY()
# if not scint_result["ok"]:
# return {
# "ok": False,
# "warning": True,
# "message": scint_result["message"],
# }
# x = scint_result["X_pixels"]
# y = scint_result["Y_pixels"]
# # --------------------------------------------------------------
# # HFM: horizontal beam position
# # --------------------------------------------------------------
# diff_x = x_target - x
# if abs(diff_x) > tolerance:
# current_hfm = get_value(hfm_mot)
# hfm_change = diff_x / BeamMove.scint_x_slope
# if abs(hfm_change) > max_move:
# message = (
# f"HFM correction {hfm_change:+.5f} mm exceeds "
# f"maximum allowed move of {max_move:.3f} mm. "
# "Manual check required."
# )
# print(message)
# return {
# "ok": False,
# "warning": True,
# "message": message,
# "BeamX": None,
# "BeamY": None,
# }
# move_with_offset(
# hfm_mot,
# current_hfm + hfm_change,
# )
# # Measure again ready for next iteration
# scint_result = read_scintXY()
# if not scint_result["ok"]:
# return {
# "ok": False,
# "warning": True,
# "message": scint_result["message"],
# }
# # ------------------------------------------------------------------
# # Final verification
# # ------------------------------------------------------------------
# final_scint = read_scintXY(image=True)
# if not final_scint["ok"]:
# return {
# "ok": False,
# "warning": True,
# "message": final_scint["message"],
# }
# final_x = final_scint["X_pixels"]
# final_y = final_scint["Y_pixels"]
# diff_x = x_target - final_x
# diff_y = y_target - final_y
# beamx = final_scint["X_diff_microns"]
# beamy = final_scint["Y_diff_microns"]
# final_vfm = get_value(vfm_mot)
# final_hfm = get_value(hfm_mot)
# success = (
# abs(diff_x) <= tolerance
# and abs(diff_y) <= tolerance
# )
# print(
# f"\nStarting beam position: X={x_start:.2f}, Y={y_start:.2f} px\n"
# f"Final beam position: X={final_x:.2f}, Y={final_y:.2f} px\n"
# f"Final error: X={diff_x:+.2f}, Y={diff_y:+.2f} px\n"
# f"VFM MOT: {vfm_start:.6f} -> {final_vfm:.6f}\n"
# f"HFM MOT: {hfm_start:.6f} -> {final_hfm:.6f}"
# )
# if success:
# message = (
# "Beam aligned\n"
# f"Displacement: Beam X: {beamx:.1f} microns, "
# f"Beam Y: {beamy:.1f} microns"
# )
# warning = False
# else:
# message = (
# f"Beam is not within ±{tolerance:.1f} px after "
# f"{iterations} corrections."
# )
# warning = True
# print(message)
# # ------------------------------------------------------------------
# # Results
# # ------------------------------------------------------------------
# return {
# "ok": success,
# "warning": warning,
# "message": message,
# "BeamX": beamx,
# "BeamY": beamy,
# "FinalX": final_x,
# "FinalY": final_y,
# "X FWHM": final_scint["X_FWHM"],
# "Y FWHM": final_scint["Y_FWHM"],
# "StartX": x_start,
# "StartY": y_start,
# "Initial VFM MOT": vfm_start,
# "Initial HFM MOT": hfm_start,
# "Final VFM MOT": final_vfm,
# "Final HFM MOT": final_hfm,
# "image": final_scint["image"],
# }
# finally:
# finish_scint_measurement()
def daily_movebeam(meas="scint"):
"""Measure beam position relative to the scintillator crosshair."""
if meas != "scint":
raise ValueError(f"Unknown beam position measurement: {meas}")
zoom = BeamMove.zoom
prepare_scint_measurement(zoom=zoom)
bp = dev.beam_profile_centroid
bs = dev.beam_steering
try:
result = read_scintXY(image=True)
target_x = bs.sample_loc_x_px.get()
target_y = bs.sample_loc_y_px.get()
tolerance = bs.px_tolerance.get()
if not result["ok"]:
return {
"ok": False,
"warning": True,
"message": result["message"],
}
# Starting position
x_start = bp.x_pos_px.get()
y_start = bp.y_pos_px.get()
print(f"Beam before steering: {x_start:.1f}, {y_start:.1f} px")
# Move beam
# bs.trigger()
# Final position
x_final = bp.x_pos_px.get()
y_final = bp.y_pos_px.get()
dx_px = x_final - target_x
dy_px = y_final - target_y
dx_um = calc_scam_microns(dx_px, zoom=1000)
dy_um = calc_scam_microns(dy_px, zoom=1000)
warning = (
abs(dx_px) > tolerance
or abs(dy_px) > tolerance
)
if warning:
message = (
f"Beam is outside crosshair tolerance "
f"(±{tolerance:.0f} px). "
f"Offset: X={dx_um:+.1f} µm, Y={dy_um:+.1f} µm "
f"({dx_px:+.1f}, {dy_px:+.1f} px)."
)
else:
message = (
f"Beam is within crosshair tolerance. "
f"Offset: X={dx_um:+.1f} µm, Y={dy_um:+.1f} µm."
)
# Capture final beam image
image = get_cam_image(Cameras.sample)
beamx = result["X_diff_microns"]
beamy = result["Y_diff_microns"]
return {
"ok": True,
"warning": False,
"message": (
"Beam position measured\n"
f"Displacement: Beam X: {beamx:.1f} microns, "
f"Beam Y: {beamy:.1f} microns"
),
"BeamX": beamx,
"BeamY": beamy,
"FinalX": result["X_pixels"],
"FinalY": result["Y_pixels"],
"X FWHM": result["X_FWHM"],
"Y FWHM": result["Y_FWHM"],
"image": result["image"],
"warning": warning,
"message": message,
"BeamX": dx_um,
"BeamY": dy_um,
"StartX_px": x_start,
"StartY_px": y_start,
"FinalX_px": x_final,
"FinalY_px": y_final,
"TargetX_px": target_x,
"TargetY_px": target_y,
"image": image,
}
finally:
finish_scint_measurement()
def format_movebeam_for_gui(result):
"Format move beam output for Daily GUI"
@@ -1099,8 +887,8 @@ def daily_beamstop():
dev.transm.put(EnergyDefaults.default_trans)
try:
if not planner.is_state(BeamlineState.DET_DIODE):
planner.move_to(BeamlineState.DET_DIODE)
if not bl.planner.is_state(BeamlineState.DET_DIODE):
bl.planner.move_to(BeamlineState.DET_DIODE)
bl.d["det_cov"].move("close")
dev.bcu_shutter.put(1)
@@ -1197,8 +985,8 @@ def daily_beamposition():
dev.transm.put(EnergyDefaults.default_trans)
# move to beam visualisation
if not planner.is_state(BeamlineState.BEAM_VISUALISATION):
planner.move_to(BeamlineState.BEAM_VISUALISATION)
if not bl.planner.is_state(BeamlineState.BEAM_VISUALISATION):
bl.planner.move_to(BeamlineState.BEAM_VISUALISATION)
# open shutter
dev.bcu_shutter.put(1)
@@ -1272,8 +1060,8 @@ def daily_collimator():
result = {}
try:
if not planner.is_state(BeamlineState.FLUX_MEASUREMENT):
planner.move_to(BeamlineState.FLUX_MEASUREMENT)
if not bl.planner.is_state(BeamlineState.FLUX_MEASUREMENT):
bl.planner.move_to(BeamlineState.FLUX_MEASUREMENT)
dev.transm.put(EnergyDefaults.default_trans)
dev.bcu_shutter.put(1)
@@ -1510,7 +1298,7 @@ def log_daily_setup_to_scilog(
date = datetime.now().strftime("%-d %B %Y")
msg.add_text(f"{DailySetup.beamline_name} Daily Setup — {date}\n\n", bold=True)
msg.add_text(f"{bl.env.beamline_name} Daily Setup — {date}\n\n", bold=True)
add_prechecks_to_scilog(msg, results)
add_setup_results_to_scilog(msg, results, progress_state)
@@ -1540,7 +1328,7 @@ def log_daily_setup_to_scilog(
for message in messages:
msg.add_text(f"{message}\n")
msg.add_tags(["daily-setup", DailySetup.beamline_name])
msg.add_tags(["daily-setup", bl.env.beamline_name])
msg.send()
# Motor positions snapshot
@@ -1554,10 +1342,8 @@ def log_daily_setup_positions(bec):
bec.messaging.scilog.log_positions(
devices=[
"dcm_bragg",
"dcm_froll",
"dcm_pitch",
"dcm_fpitch",
"dccm_theta1",
"dccm_theta2",
"vfm_y",
"vfm_pitch",
vfm_mot.name,
@@ -1568,8 +1354,8 @@ def log_daily_setup_positions(bec):
"bs_x",
"bs_y",
],
title=f"{DailySetup.beamline_name} Daily Setup - Motor Positions",
tags=["daily_setup", DailySetup.beamline_name, "positions"]
title=f"{bl.env.beamline_name} Daily Setup - Motor Positions",
tags=["daily_setup", bl.env.beamline_name, "positions"]
)
def add_prechecks_to_scilog(msg, results):
@@ -1787,10 +1573,10 @@ def get_daily_setup_status(progress_state):
def log_daily_setup_to_csv(results):
"""Append key daily setup diagnostics to the beamline history CSV."""
vfm_mot = BeamMove.vfm_mot
hfm_mot = BeamMove.hfm_mot
vfm_mot = DailySetup.vfm_mot
hfm_mot = DailySetup.hfm_mot
filename = DailySetup.setup_dir / "daily_beamline_health.csv"
filename = bl.env.daily_setup_dir / "daily_beamline_health.csv"
move_results = results.get("Move beam", {})
energy_result = results.get("Set energy", {})
@@ -1988,21 +1774,23 @@ def save_daily_beam_image(results):
return None
# timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = DailySetup.setup_dir / f"daily_beam_{energy_result['energy']:.0f}eV.png"
filename = bl.env.daily_setup_dir / f"daily_beam_{energy_result['energy']:.0f}eV.png"
fig, ax = plt.subplots(figsize=(8, 6))
ax.imshow(image)
x = BeamMove.scint_target_x
y = BeamMove.scint_target_y
length = 40
gap = 8
x = move_result.get("TargetX_px")
y = move_result.get("TargetY_px")
# Target crosshair
ax.plot([x - length, x - gap], [y, y], color="red", linewidth=1.0)
ax.plot([x + gap, x + length], [y, y], color="red", linewidth=1.0)
ax.plot([x, x], [y - length, y - gap], color="red", linewidth=1.0)
ax.plot([x, x], [y + gap, y + length], color="red", linewidth=1.0)
if x is not None and y is not None:
length = 40
gap = 8
# Target crosshair
ax.plot([x - length, x - gap], [y, y], color="red", linewidth=1.0)
ax.plot([x + gap, x + length], [y, y], color="red", linewidth=1.0)
ax.plot([x, x], [y - length, y - gap], color="red", linewidth=1.0)
ax.plot([x, x], [y + gap, y + length], color="red", linewidth=1.0)
ax.axis("off")
@@ -2119,7 +1907,7 @@ def create_daily_html_report(results, total_duration=None, messages=None):
result_rows.append(
(
"DCM pitch offset",
"Mono pitch offset",
html_value(
energy_result.get("pitch_offset"),
".4f",
@@ -2359,7 +2147,7 @@ def create_daily_html_report(results, total_duration=None, messages=None):
<body>
<h1>{DailySetup.beamline_name} Daily Setup</h1>
<h1>{bl.env.beamline_name} Daily Setup</h1>
<p class="date">{date}</p>