diff --git a/pxii_bec/macros/daily.py b/pxii_bec/macros/daily.py index d4d81c2..a6a19d3 100644 --- a/pxii_bec/macros/daily.py +++ b/pxii_bec/macros/daily.py @@ -1,11 +1,21 @@ -"""Files needed for the daily setup routine for local contacts""" +"""Daily setup routine at for local contacts""" + +# ============================================================================= +# Imports / context +# ============================================================================= import time +import subprocess +import shlex +import csv +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 dataclasses import dataclass +import matplotlib.pyplot as plt +from matplotlib.patches import Ellipse @dataclass @@ -15,38 +25,56 @@ class DailySetupContext: plot3: object plot4: object log: object + report: object progress: object messages: list progress_state: dict +# ============================================================================= +# Main daily setup +# ============================================================================= + + def daily_setup( energy=12400, *, + pre_checks=True, set_energy=True, - move_beam=False, + mono_scan=True, + move_beam=True, align_beamstop=False, - measure_beampos=False, + # measure_beampos=False, align_collimator=False, measure_flux=True, - save_positions=False, + save_scilog=True, ): """Run selected daily beamline setup tasks.""" selected = { + "Pre checks": pre_checks, "Set energy": set_energy, "Move beam": move_beam, "Align beamstop": align_beamstop, - "Measure beam position": measure_beampos, + # "Measure beam position": measure_beampos, "Align collimator": align_collimator, "Measure flux": measure_flux, - "Save new positions to BEC": save_positions, } results = {} + setup_start = time.perf_counter() ctx = get_daily_setup_context(selected) + # Reset the report for this run + report_file = bl.env.daily_setup_dir / "daily_report.pdf" + + create_empty_daily_report(report_file) + + ctx.report.load_pdf(str(report_file)) + ctx.report.fit_to_page() + + for task, enabled in selected.items(): ctx.progress_state[task] = "todo" if enabled else "skipped" @@ -54,47 +82,141 @@ def daily_setup( update_log(ctx.log, ctx.messages, "Starting daily beamline setup") try: + # --------------------------------------------------------- + # Pre-flight checks + # --------------------------------------------------------- + if pre_checks: + with time_step(results, "Pre checks"): + prechecks_ok, results["Pre checks"] = run_pre_checks(ctx) + + # Machine checks are critical: stop daily setup + if not prechecks_ok: + update_log( + ctx.log, + ctx.messages, + f"Daily setup stopped because machine pre-checks failed." + ) + total_duration = time.perf_counter() - setup_start + + if save_scilog: + log_daily_setup_to_scilog( + bec, + results, + ctx.progress_state, + total_duration=total_duration, + ) + return results + + # --------------------------------------------------------- + # Set energy + # --------------------------------------------------------- if set_energy: - results["Beamline energy"] = run_daily_energy(energy, ctx) + with time_step(results, "Set energy"): + results["Set energy"] = run_daily_energy( + energy, + mono_scan, + ctx, + ) + # --------------------------------------------------------- + # Move beam + # --------------------------------------------------------- if move_beam: - results["Move beam"] = run_daily_movebeam(ctx) + with time_step(results, "Move beam"): + results["Move beam"] = run_daily_movebeam(ctx) + # --------------------------------------------------------- + # Align beamstop + # --------------------------------------------------------- if align_beamstop: - results["Align beamstop"] = run_daily_beamstop(ctx) + with time_step(results, "Align beamstop"): + results["Align beamstop"] = run_daily_beamstop(ctx) - if measure_beampos: - results["Measure beam position"] = run_daily_beampos(ctx) + # # --------------------------------------------------------- + # # Measure beam position + # # --------------------------------------------------------- + # if measure_beampos: + # with time_step(results, "Measure beam position"): + # results["Measure beam position"] = run_daily_beamposition(ctx) + # --------------------------------------------------------- + # Align collimator + # --------------------------------------------------------- if align_collimator: - results["Align collimator"] = run_daily_collimator(ctx) + with time_step(results, "Align collimator"): + results["Align collimator"] = run_daily_collimator(ctx) + # --------------------------------------------------------- + # Measure flux + # --------------------------------------------------------- if measure_flux: - results["Flux"] = run_daily_flux(ctx) + with time_step(results, "Measure flux"): + results["Measure flux"] = run_daily_flux(ctx) - if save_positions: - run_save_positions(ctx) + # --------------------------------------------------------- + # Measure time taken + # --------------------------------------------------------- - log_daily_setup_to_scilog(bec,results) - return(results) + total_duration = time.perf_counter() - setup_start + update_log( + ctx.log, + ctx.messages, + f"Total duration: {format_duration(total_duration)}", + ) + + + # --------------------------------------------------------- + # Save daily diagnostic history + # --------------------------------------------------------- + log_daily_setup_to_csv(results) + + # --------------------------------------------------------- + # Create and display the PDF report + # --------------------------------------------------------- + report_file = ( + bl.env.daily_setup_dir + / "daily_report.pdf" + ) + + create_daily_report( + results, + report_file, + ) + + ctx.report.load_pdf(str(report_file)) + ctx.report.fit_to_page() + + # --------------------------------------------------------- + # SciLog + # --------------------------------------------------------- + if save_scilog: + log_daily_setup_to_scilog( + bec, + results, + ctx.progress_state, + total_duration = total_duration, + ) + + return results finally: + print("done") bl.planner.move_to(BeamlineState.ROBOT_SAMPLE_EXCHANGE) -def sanity_checks(): - pass def get_daily_setup_context(tasks=None): """Get widgets and state used by daily setup tasks.""" - plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() + plot1, plot2, plot3, plot4, log, progress, report = select_daily_setup_gui() + + daily_report = f" bl.env." if tasks is None: tasks = [ "Set energy", "Move beam", "Align beamstop", - "Measure beam position", + # "Measure beam position", "Align collimator", "Measure flux", "Save new positions to BEC", @@ -107,37 +229,477 @@ def get_daily_setup_context(tasks=None): plot4=plot4, log=log, progress=progress, + report=report, messages=[], progress_state={task: "todo" for task in tasks}, ) -def run_daily_energy(energy, ctx): - """Set and optimise beamline energy.""" +# ============================================================================= +# Pre-checks +# ============================================================================= + +def run_pre_checks(ctx): + """Run machine and experimental hutch pre-flight checks. + + Machine checks are critical and stop the setup if they fail. + Experimental hutch checks are advisory and do not stop the setup. + """ + + task = "Pre checks" + + clear_daily_plots() + set_progress( + ctx.progress, + ctx.progress_state, + task, + "running", + ) + + # --------------------------------------------------------- + # Machine checks - critical / fail fast + # --------------------------------------------------------- + machine_result = machine_prechecks() + + result = { + "ok": machine_result["ok"], + "warning": False, + "machine": machine_result, + "experiment_hutch": None, + } + + if not machine_result["ok"]: + update_log( + ctx.log, + ctx.messages, + format_prechecks_for_gui(result), + ) + + set_progress( + ctx.progress, + ctx.progress_state, + task, + "failed", + ) + + return False, result + + # --------------------------------------------------------- + # Experimental hutch checks - advisory + # --------------------------------------------------------- + eh_result = EH_prechecks() + + result["experiment_hutch"] = eh_result + + # 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), + ) + + status = "warning" if result["warning"] else "done" + + set_progress( + ctx.progress, + ctx.progress_state, + task, + status, + ) + + # EH warnings do NOT stop daily setup + return True, result + + +def format_prechecks_for_gui(result): + """Format pre-flight check results for the daily setup GUI.""" + + lines = [] + + mach = result.get("machine") + + if mach is not None: + lines.append("") + + for check in mach["checks"]: + symbol = "✓" if check["ok"] else "✗" + lines.append(f"{symbol} {check['message']}") + + eh = result.get("experiment_hutch") + + if eh is not None: + lines.append("") + + for check in eh["checks"]: + symbol = "✓" if check["ok"] else "✗" + lines.append(f"{symbol} {check['message']}") + + return "\n".join(lines) + + +# -------------------------------------------------------------------------- +# Check machine current and shutter status +# -------------------------------------------------------------------------- +def machine_prechecks(): + """Run machine checks in dependency order. + Stop immediately if any machine check fails""" + + checks = [check_machine_current, + check_attended_mode, + check_fe_shutter, + check_hutch_shutter, + ] + + results = [] + + for check in checks: + result = check() + results.append(result) + + + if result["ok"]: + print(f"✓ {result['message']}") + else: + print(f"✗ {result['message']}") + + return{ + "ok": False, + "message": "Machine checks failed", + "checks": results, + } + + return { + "ok": True, + "message": "Machine checks passed", + "checks": results, + } + + +def check_machine_current(): + """Checks current is >30 mA""" + current = get_value(dev.sls_current) + + if current <= 30: + return { + "ok": False, + "message": f"Machine current too low {current:.1f} mA)", + "current": current, + } + + return { + "ok": True, + "message": f"Machine current OK ({current:.1f} mA)", + "current": current, + } + + + +def check_attended_mode(): + """Check beamline is in attended more""" + state = caget("AGEOP-BL:STATUS-X10SA") + + if state == "offline": + return { + "ok": False, + "message": "Beamline is offline, please set to 'attended'", + } + + + return { + "ok": True, + "message": f"Beamline is not offline, (operation mode is {state})", + } + + + +def check_fe_shutter(): + """Check FE shutter is open""" + state = caget("X10SA-FE-PSH1-EMLS-0010:OPEN") + + if state != "Open": + return { + "ok": False, + "message": "FE shutter is not open'", + } + + return { + "ok": True, + "message": "FE shutter is open", + } + + + +def check_hutch_shutter(): + """Check hutch shutter is open""" + state = caget("X10SA-OP-PSH1-EMLS:STATE") + + if state != "OPEN": + return { + "ok": False, + "message": "Hucth shutter is not open'", + } + + return { + "ok": True, + "message": "Hutch shutter is open", + } + + + + +# -------------------------------------------------------------------------- +# Check Experimental Hutch (EH) equipment +# -------------------------------------------------------------------------- +def EH_prechecks(): + """Check experimental hutch devices.""" + + checks = [ + check_xray_eye_out, + check_cryo_temp, + check_detector_height, + check_beamstop_position, + ] + + results = [] + + for check in checks: + result = check() + results.append(result) + + if result["ok"]: + print(f"✓ {result['message']}") + else: + print(f"✗ {result['message']}") + + all_ok = all(result["ok"] for result in results) + + return { + "ok": all_ok, + "message": ( + "Experiment hutch checks passed" + if all_ok + else "Experiment hutch checks require attention" + ), + "checks": results, + } + + +def check_xray_eye_out(): + state = caget("X10SA-ES-SSXI:POS-SET") + try: + if state != "OUT": + umv(dev.ss_xi_x, -7) + except Exception as exc: + return { + "ok": False, + "message": f"Moving X-ray eye failed: {exc}" + } + return { + "ok": True, + "message": "SS Xray eye is OUT" + } + + +def check_bcubpm(): + pass + +def check_beamstop_position(): + """Check that the beamstop blocks the direct beam. + + If the check fails, align the beamstop and verify again. + """ + + try: + 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) + dev.bcu_shutter.put(1) + + time.sleep(0.5) + + i2 = abs(dev.i2.read()["i2"]["value"]) + cutoff = Beamstop.i2_limit + + # Beamstop already OK + if i2 < cutoff: + return { + "ok": True, + "message": f"Beamstop blocking beam: I2 = {i2:.2e}", + "value": i2, + "realigned": False, + } + + print( + f"Beamstop check failed (I2={i2:.2e}). " + "Running beamstop alignment..." + ) + + # Close shutter before handing over to alignment routine + dev.bcu_shutter.put(0) + + alignment = daily_beamstop() + + if not alignment["ok"]: + return { + "ok": False, + "message": "Beamstop alignment failed", + "value": i2, + "realigned": True, + } + + # Verify after alignment + dev.bcu_shutter.put(1) + time.sleep(0.2) + + i2_after = abs(dev.i2.read()["i2"]["value"]) + + if i2_after < cutoff: + return { + "ok": True, + "message": ( + "Beamstop was misaligned and was successfully " + f"realigned (I2={i2_after:.2e})" + ), + "value": i2_after, + "realigned": True, + } + + return { + "ok": False, + "message": ( + "Beamstop still not blocking beam after alignment " + f"(I2={i2_after:.2e})" + ), + "value": i2_after, + "realigned": True, + } + + except Exception as exc: + return { + "ok": False, + "message": f"Beamstop check failed: {exc}", + } + + finally: + dev.bcu_shutter.put(0) + +def check_cryo_temp(): + temp = caget("X10SA-ES-CS:TEMP_RBV") + + if temp is None: + return { + "ok":False, + "message": "Could not read cryostream temperature" + } + + if temp >= 110: + return { + "ok": False, + "message": f"Cryostream too warm ({temp} K)" + } + + return { + "ok":True, + "message": f"Cryostream temperature: {temp} K", + } + +def check_detector_height(): + height = get_value(dev.det_y) + tol = 0.1 + if abs(height-62) > tol: + return { + "ok":False, + "message": f"Detector not at operational height. Currently at {height} mm", + } + return { + "ok":True, + "message": f"Detector height: {height:.1f} mm" + } + +# ============================================================================= +# Energy +# ============================================================================= + + +def run_daily_energy(energy, mono_scan, ctx): + """Set beamline energy and log results.""" task = "Set energy" clear_daily_plots() set_progress(ctx.progress, ctx.progress_state, task, "running") - en = daily_energy(energy, plot_widget=ctx.plot1) + try: - result = { - "energy": en - } + result = daily_energy(energy, mono_scan, plot_widget=ctx.plot1) - update_log( - ctx.log, - ctx.messages, - format_energy_for_gui(result), - ) + except Exception as exc: + result = { + "ok": False, + "message": f"Energy setup failed: {exc}", + } + + + update_log(ctx.log, ctx.messages, format_energy_for_gui(result)) + + status = "done" if result["ok"] else "failed" + + set_progress(ctx.progress, ctx.progress_state, task, status) - set_progress(ctx.progress, ctx.progress_state, task, "done") return result +def daily_energy(energy, scan_mono, plot_widget=None): + """Set the beamline to the required energy and correct gap""" + + current_energy = get_current_energy() + + if abs(energy - current_energy) > 1: + bl_energy(energy, window_name="DailySetup", plot_widget=plot_widget) + + else: + 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, + } + + umv(dev.id_gap, gap_req) + + if scan_mono: + mono_pitch_scan(window_name="DailySetup", plot_widget=plot_widget) + + new_energy = get_current_energy() + + return { + "ok": True, + "message": f"Beamline energy set to {new_energy:.0f} eV", + "energy": new_energy, + } + + def format_energy_for_gui(result): - return f"Beamline energy {result['energy']:.0f} eV\n" + "Format energy change output for Daily GUI" + + icon = "✓" if result["ok"] else "✗" + + return f"\n{icon} {result['message']}" + + + +# ============================================================================= +# Move beam +# ============================================================================= def run_daily_movebeam(ctx): @@ -145,27 +707,284 @@ def run_daily_movebeam(ctx): task = "Move beam" - clear_daily_plots() set_progress(ctx.progress, ctx.progress_state, task, "running") - result = daily_movebeam() + try: - update_log( - ctx.log, - ctx.messages, - format_movebeam_for_gui(result), - ) + result = daily_movebeam() - set_progress(ctx.progress, ctx.progress_state, task, "done") + except Exception as exc: + result = { + "ok": False, + "message": f"Align beam step failed: {exc}", + } + + if result.get("ok"): + status = "done" + + elif result.get("warning"): + status = "warning" + + else: + status = "failed" + + update_log(ctx.log, ctx.messages, format_movebeam_for_gui(result)) + + set_progress(ctx.progress, ctx.progress_state, task, status) return result -def format_movebeam_for_gui(result): - if result["success"]: - return (f"Beam position relative to crosshair:\n" - f"Beam_X: {result["BeamX"]:.2g} microns, " - f"Beam_Y: {result["BeamY"]:.2g} microns.\n") + +def daily_movebeam(meas="scint", iterations=3): + """ + Move the beam onto the sample-camera crosshair using the VFM and HFM. + + The beam position is measured on the scintillator at 12.5X zoom. + VFM downstream Y ring motor controls beam Y. + HFM downstream X controls beam X. + + 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 = CameraZoom.X12_5.position + + 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 = 0.02 # mm + + # Starting mirror positions + vfm_start = get_value(dev.vfm_ydr) + hfm_start = get_value(dev.hfm_xd) + + print( + "Moving beam to scintillator crosshair\n" + f"Target: X={x_target:.1f}, Y={y_target:.1f} px" + ) + + # ------------------------------------------------------------------ + # Initial measurement + # ------------------------------------------------------------------ + + scint_result = read_scintXY(zoom) + + 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(dev.vfm_ydr) + 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( + dev.vfm_ydr, + current_vfm + vfm_change, + ) + + # Remeasure because VFM movement may also affect X + scint_result = read_scintXY(zoom) + + 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(dev.hfm_xd) + 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( + dev.hfm_xd, + current_hfm + hfm_change, + ) + + # Measure again ready for next iteration + scint_result = read_scintXY(zoom) + + if not scint_result["ok"]: + return { + "ok": False, + "warning": True, + "message": scint_result["message"], + } + + # ------------------------------------------------------------------ + # Final verification + # ------------------------------------------------------------------ + + final_scint = read_scintXY(zoom, 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(dev.vfm_ydr) + final_hfm = get_value(dev.hfm_xd) + + 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 DR: {vfm_start:.6f} -> {final_vfm:.6f}\n" + f"HFM XD: {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: - return f"Moving beam to crosshaor failed" + 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 DR": vfm_start, + "Initial HFM XD": hfm_start, + "Final VFM DR": final_vfm, + "Final HFM XD": final_hfm, + "image": final_scint["image"], + } + + +def format_movebeam_for_gui(result): + "Format move beam output for Daily GUI" + + icon = "✓" if result["ok"] else "✗" + return f"\n{icon} {result['message']}" + + +# ============================================================================= +# Beamstop alignment +# ============================================================================= def run_daily_beamstop(ctx): @@ -176,252 +995,127 @@ def run_daily_beamstop(ctx): # clear_daily_plots() set_progress(ctx.progress, ctx.progress_state, task, "running") - result = daily_beamstop() + try: + result = daily_beamstop() - update_log( - ctx.log, - ctx.messages, - format_beamstop_for_gui(result), - ) + except Exception as exc: + result = { + "ok": False, + "message": f"Beamstop alignment failed: {exc}", + } + + + status = "done" if result["ok"] else "failed" + + update_log(ctx.log, ctx.messages, format_beamstop_for_gui(result)) # save_current_position(dev.bs_x, "in") # save_current_position(dev.bs_y, "in") - set_progress(ctx.progress, ctx.progress_state, task, "done") + set_progress(ctx.progress, ctx.progress_state, task, status) return result -def format_beamstop_for_gui(result): - return ( - f"BS_X: {result['bs_x_pos']:.3g} mm, " - f"BS_Y: {result['bs_y_pos']:.3g} mm\n" - ) -def run_daily_beampos(ctx): +def daily_beamstop(): + """Centre the beamstop against the detector cover diode, i2""" + + plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() + + result = {} + + dev.transm.put(EnergyDefaults.default_trans) + + try: + 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) + + time.sleep(0.2) + + # Centre Y + y_result = find_feature( + dev.bs_y, + dev.i2, + -1, + 1, + 30, + settle=0.3, + feature="dip", + relative=True, + confirm=False, + window_name="DailySetup", + plot_widget=plot2, + ) + # Wait for bs_y to setlle + time.sleep(2) + # Centre X + x_result = find_feature( + dev.bs_x, + dev.i2, + -0.2, + 1.2, + 30, + feature="edge_offset", + offset=Beamstop.x_offset, + relative=True, + confirm=False, + window_name="DailySetup", + plot_widget=plot1, + ) + + result["bs_x_pos"] = x_result["centre"] + result["bs_y_pos"] = y_result["centre"] + result["intensity after alignment"] = read_i2() + + # message = ( + # "Intensity after Y scan: " + # f"{result['intensity after alignment']:.4g}\n" + # f"Centred position is {result["bs_x_pos"]:.5g}, {result["bs_y_pos"]:.5g}" + # ) + + message = f"BS_X: {result['bs_x_pos']:.3g} mm, " f"BS_Y: {result['bs_y_pos']:.3g} mm\n" + + print(message) + + result["ok"] = True + result["message"] = message + + return result + + finally: + dev.bcu_shutter.put(0) + + +def format_beamstop_for_gui(result): + "Format results for beamstop alignment" + icon = "✓" if result["ok"] else "✗" + return f"\n{icon} Beamstop aligned.\n{result['message']}" + + + +# ============================================================================= +# Beam-position measurement +# ============================================================================= + + +def run_daily_beamposition(ctx): """Measure the beam position at the scintillator at Aare zoom levels""" task = "Measure beam position" set_progress(ctx.progress, ctx.progress_state, task, "running") - beampos_result = daily_beampos() + beampos_result = daily_beamposition() - update_log( - ctx.log, - ctx.messages, - format_beam_positions(beampos_result), - ) + update_log(ctx.log, ctx.messages, format_beam_position(beampos_result)) set_progress(ctx.progress, ctx.progress_state, task, "done") return beampos_result -def format_beam_positions(pos_result): - """Format camera beam positions for display.""" - lines = [f"{'Zoom':<8} {'X / px':>10} {'Y / px':>10}", "-" * 30] - - for zoom, pos in pos_result.items(): - lines.append(f"{zoom:<8} {pos['x']:>10.1f} {pos['y']:>10.1f}") - - return "\n".join(lines) - - -def run_daily_collimator(ctx): - """Align the collimator and save positions""" - - task = "Align collimator" - set_progress(ctx.progress, ctx.progress_state, task, "running") - - result = daily_coll() - - update_log( - ctx.log, - ctx.messages, - format_collimator_for_gui(result), - ) - - set_progress(ctx.progress, ctx.progress_state, task, "done") - return result - -def format_collimator_for_gui(result): - """Format collimator result for the Daily Setup GUI.""" - - return ( - f"\nI1 intensity with collimator out: " - f"{result['intensity_coll_out']:.2e}\n" - f"Starting intensity with collimator in: " - f"{result['start_intensity_coll_in']:.2e}\n" - f"Intensity with collimator in after scans: " - f"{result['final_intensity_coll_in']:.2e}\n" - f"Scan Width X: {result['coll_x_width']:.3f} mm, " - f"Scan Width Y: {result['coll_y_width']:.3f} mm\n" - f"Coll_X: {result['coll_x_pos']:.3f} mm, " - f"Coll_Y: {result['coll_y_pos']:.3f} mm\n" - ) - - - -def run_daily_flux(ctx): - """Measure the beam intensity using BCU BPM.""" - - task = "Measure flux" - set_progress(ctx.progress, ctx.progress_state, task, "running") - - flux = bpm2flux() - - result = { - "current_flux": flux[0], - "full_flux": flux[1], - } - - update_log( - ctx.log, - ctx.messages, - format_flux_for_gui(result), - ) - - set_progress(ctx.progress, ctx.progress_state, task, "done") - return result - -def format_flux_for_gui(result): - return ( - f"Intensity at current transmission: " - f"{result['current_flux']:.2e} ph/s\n" - f"Intensity at 100% transmission: " - f"{result['full_flux']:.2e} ph/s" - ) - - -def select_daily_setup_gui(): - """Return widgets belonging to the Daily Setup GUI.""" - - window_name = "DailySetup" - open_windows = bec.gui.windows - - if open_windows.get(window_name) is None: - gui = bec.gui.new(window_name) - - # Row 1 - log = gui.new(widget="TextBox", object_name="Info") - - progress = gui.new( - widget="TextBox", object_name="Progress", relative_to="Info", where="right" - ) - - # Row 2 - start a new full-width row - plot1 = gui.new(widget="Waveform", object_name="Plot_1", where="bottom") - - plot2 = gui.new( - widget="Waveform", object_name="Plot_2", relative_to="Plot_1", where="right" - ) - - # Row 3 - start another full-width row - plot3 = gui.new(widget="Waveform", object_name="Plot_3", where="bottom") - - plot4 = gui.new( - widget="Waveform", object_name="Plot_4", relative_to="Plot_3", where="right" - ) - gui.set_layout_ratios(horizontal=[1, 1], vertical=[1, 1, 1]) - else: - gui = open_windows[window_name] - plot1 = gui.Plot_1 - plot2 = gui.Plot_2 - plot3 = gui.Plot_3 - plot4 = gui.Plot_4 - log = gui.Info - progress = gui.Progress - - return plot1, plot2, plot3, plot4, log, progress - - -def clear_daily_plots(): - plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() - plot1.clear_all() - plot2.clear_all() - plot3.clear_all() - plot4.clear_all() - - -def update_log(log_box, messages, message): - """Add new message to the daily setup log""" - messages.append(message) - log_box.set_plain_text("\n".join(messages)) - - -def update_progress(progress_box, progress_state): - """Sets up the status icons for the progress tracker""" - status = { - "todo": "\u25cb", - "running": "\u25b6", - "done": "\u2713", - "warning": "\u26a0", - "skipped": "–", - "failed": "\u2717", - } - - text = "\n".join(f"{status[state]} {label}" for label, state in progress_state.items()) - progress_box.set_plain_text(text) - - -def set_progress(progress_box, progress_state, step, status): - """Change the status of one step and update the display.""" - progress_state[step] = status - update_progress(progress_box, progress_state) - - -def plot_daily_setup_scan(motor_name, signal_name, plot="x", title=None): - """Plot live scan data in one of the Daily Setup waveform widgets.""" - - plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() - - if plot.lower() == "plot_1": - wf = plot1 - elif plot.lower() == "plot_2": - wf = plot2 - elif plot.lower() == "plot_3": - wf = plot3 - elif plot.lower() == "plot_4": - wf = plot4 - else: - raise ValueError("plot must be 'x' or 'y'") - - wf.clear_all() - - if title is None: - title = f"{motor_name} scan" - - wf.title = title - wf.x_label = motor_name - wf.y_label = signal_name - - wf.plot(device_x=motor_name, device_y=signal_name) - - -def daily_energy(energy, plot_widget=None): - """Set the beamline to the required energy""" - current_energy = get_current_energy() - if abs(energy - current_energy) > 1: - bl_energy(energy, window_name="DailySetup", plot_widget=plot_widget) - else: - gap_actual = dev.id_gap.position - gap_req = calc_gap(energy) - if abs(gap_actual - gap_req) > 0.05: - umv(dev.id_gap, gap_req) - mono_pitch_scan(window_name="DailySetup", plot_widget=plot_widget) - new_energy = get_current_energy() - return new_energy - - -def set_zoom(zoom): - """Set sample camera zoom level.""" - - if isinstance(zoom, str): - zoom = CameraZoom.from_label(zoom) - umv(dev.scam_zoom, zoom.position) - else: - umv(dev.scam_zoom, zoom) - - -def daily_beampos(): +def daily_beamposition(): """Measure beam positions at different zoom levels""" results = {} @@ -459,7 +1153,46 @@ def daily_beampos(): return results -def daily_coll(): +def format_beamposition(pos_result): + """Format camera beam positions for display.""" + + lines = [f"{'Zoom':<8} {'X / px':>10} {'Y / px':>10}", "-" * 30] + + for zoom, pos in pos_result.items(): + lines.append(f"{zoom:<8} {pos['x']:>10.1f} {pos['y']:>10.1f}") + + return "\n".join(lines) + + +# ============================================================================= +# Collimator alignment +# ============================================================================= + + +def run_daily_collimator(ctx): + """Align the collimator and save positions""" + + task = "Align collimator" + set_progress(ctx.progress, ctx.progress_state, task, "running") + + try: + result = daily_collimator() + + except Exception as exc: + result = { + "ok": False, + "message": f"Collimator alignment step failed: {exc}" + } + + update_log(ctx.log, ctx.messages, format_collimator_for_gui(result)) + + status = "done" if result["ok"] else "failed" + + set_progress(ctx.progress, ctx.progress_state, task, status) + return result + + +def daily_collimator(): """Centre the collimator in X and Y using the I1 diode.""" plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() @@ -524,7 +1257,14 @@ def daily_coll(): "Intensity with collimator in after scans: " f"{result['final_intensity_coll_in']:.3e}" ) - result["success"] = True + result["ok"] = True + + message = ( + f"Collimator X: {result['coll_x_pos']:.3f} mm, " + f"Collimator Y: {result['coll_y_pos']:.3f} mm" + ) + + result["message"] = message return result @@ -532,471 +1272,747 @@ def daily_coll(): dev.bcu_shutter.put(0) -def daily_beamstop(): - """Centre the beamstop against the detector cover, i2""" +def format_collimator_for_gui(result): + """Format collimator result for the Daily Setup GUI.""" + icon = "✓" if result["ok"] else "✗" + return f"\n{icon} Collimator aligned\n{result['message']}" - plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() - result = {} - dev.transm.put(EnergyDefaults.default_trans) +# ============================================================================= +# Flux measurement +# ============================================================================= + + +def run_daily_flux(ctx): + """Measure the beam intensity using BCU BPM.""" + + task = "Measure flux" + set_progress(ctx.progress, ctx.progress_state, task, "running") try: - if not bl.planner.is_state(BeamlineState.DET_DIODE): - bl.planner.move_to(BeamlineState.DET_DIODE) + flux, full_flux = bpm2flux() - bl.d["det_cov"].move("close") - dev.bcu_shutter.put(1) - - time.sleep(0.2) - - # Centre Y - y_result = find_feature( - dev.bs_y, - dev.i2, - -1, - 1, - 30, - settle=0.3, - feature="dip", - relative=True, - confirm=False, - window_name="DailySetup", - plot_widget=plot2, - ) - # Wait for bs_y to setlle - time.sleep(2) - # Centre X - x_result = find_feature( - dev.bs_x, - dev.i2, - -0.2, - 1.2, - 30, - feature="edge_offset", - offset=Beamstop.x_offset, - relative=True, - confirm=False, - window_name="DailySetup", - plot_widget=plot1, + message = ( + f"Intensity at current transmission: " + f"{flux:.2e} ph/s\n" + f"Intensity at 100% transmission: " + f"{full_flux:.2e} ph/s" ) - result["bs_x_pos"] = x_result["centre"] - result["bs_y_pos"] = y_result["centre"] - result["intensity after alignment"] = read_i2() + result = {"ok": True, + "message": message, + "current_flux": flux, + "full_flux": full_flux, + } - print( - "Intensity after Y scan: " - f"{result['intensity after alignment']:.4g}\n" - f"Centred position is {result["bs_x_pos"]:.5g}, {result["bs_y_pos"]:.5g}" - ) + except Exception as exc: + result = { + "ok": False, + "message": f"(Flux measurement failed: {exc})", + } - return result + status = "done" if result["ok"] else "failed" - finally: - dev.bcu_shutter.put(0) - - -def test_daily_movebeam(meas="scint", iterations=3): - - vfm_start = get_value(dev.vfm_ydr) - hfm_start = get_value(dev.hfm_xd) - - dev.transm.put(EnergyDefaults.default_trans) - - if meas == "scint": - - xbeam_target = BeamMove.scint_target_x - ybeam_target = BeamMove.scint_target_y - - print(f"Starting VFM DR: {vfm_start:.6f}") - print(f"Starting HFM XD: {hfm_start:.6f}") - - for i in range(iterations): - - print(f"\n--- Correction {i + 1} ---") - - # Fresh beam measurement at start of iteration - x, y = read_scintXY(CameraZoom.X12_5.position) - - diff_x = xbeam_target - x - diff_y = ybeam_target - y - - print( - f"Beam position: X={x:.2f}, Y={y:.2f}\n" - f"Residual: X={diff_x:+.2f} px, " - f"Y={diff_y:+.2f} px" - ) - - # ---------------------------------------------------------- - # VFM correction - # ---------------------------------------------------------- - - if abs(diff_y) > BeamMove.scint_tolerance: - - current_vfm = get_value(dev.vfm_ydr) - - vfm_change = diff_y / BeamMove.scint_y_slope - new_vfm = current_vfm + vfm_change - - print(f"VFM: {current_vfm:.6f} -> {new_vfm:.6f} " f"(change {vfm_change:+.6f})") - - beam_y_before = y - vfm_before = current_vfm - - umv(dev.vfm_ydr, new_vfm) - - # Remeasure because VFM may affect both X and Y - x, y = read_scintXY(CameraZoom.X12_5.position) - - vfm_after = get_value(dev.vfm_ydr) - - dy = y - beam_y_before - dvfm = vfm_after - vfm_before - - if abs(dvfm) > 1e-5: - observed_coeff = dy / dvfm - - print( - f"VFM observed movement:\n" - f" Motor: {dvfm:+.6f}\n" - f" Beam Y: {dy:+.2f} px\n" - f" Coefficient: {observed_coeff:.0f} px/mm" - ) - - else: - print(f"Y within tolerance ({diff_y:+.2f} px) " "- no VFM move") - - # ---------------------------------------------------------- - # HFM correction - # ---------------------------------------------------------- - - # x is either: - # - the initial measurement if VFM didn't move, or - # - the updated measurement after the VFM move - diff_x = xbeam_target - x - - if abs(diff_x) > BeamMove.scint_tolerance: - - current_hfm = get_value(dev.hfm_xd) - - hfm_change = diff_x / BeamMove.scint_x_slope - new_hfm = current_hfm + hfm_change - - print(f"HFM: {current_hfm:.6f} -> {new_hfm:.6f} " f"(change {hfm_change:+.6f})") - - umv(dev.hfm_xd, new_hfm) - - else: - print(f"X within tolerance ({diff_x:+.2f} px) " "- no HFM move") - - # Final check - x, y = read_scintXY(CameraZoom.X12_5.position) - - diff_x = xbeam_target - x - diff_y = ybeam_target - y - - print("\n--- Final result ---") - print( - f"Beam position: X={x:.2f}, Y={y:.2f}\n" - f"Residual: X={diff_x:+.2f} px, " - f"Y={diff_y:+.2f} px" - ) - - final_vfm = get_value(dev.vfm_ydr) - final_hfm = get_value(dev.hfm_xd) - - print( - f"VFM: {vfm_start:.6f} -> {final_vfm:.6f}\n" f"HFM: {hfm_start:.6f} -> {final_hfm:.6f}" - ) - - -def daily_movebeam(meas="scint", iterations=3): - """ - Move the beam onto the sample-camera crosshair using the VFM and HFM. - - The beam position is measured on the scintillator at 12.5X zoom. - VFM downstream Y ring motor controls beam Y. - HFM downstream X controls beam X. - - 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}") - - result = {"success": None, "BeamX": None, "BeamY": None} - - # Starting mirror positions - vfm_start = get_value(dev.vfm_ydr) - hfm_start = get_value(dev.hfm_xd) - - # Target position on sample camera - 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 = 0.02 # mm - - print( - f"Moving beam to scintillator crosshair\n" f"Target: X={x_target:.1f}, Y={y_target:.1f} px" - ) - - for i in range(iterations): - - # Measure current beam position - x, y = read_scintXY(CameraZoom.X12_5.position) - - 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(dev.vfm_ydr) - vfm_change = diff_y / BeamMove.scint_y_slope - - if abs(vfm_change) > max_move: - print( - f"VFM correction {vfm_change:+.5f} mm exceeds " - f"maximum allowed move of {max_move:.3f} mm." - ) - print("Beam alignment stopped.") - return False - - umv(dev.vfm_ydr, current_vfm + vfm_change) - - # Remeasure after VFM move since VFM may also affect X - x, y = read_scintXY(CameraZoom.X12_5.position) - - # ---------------------------------------------------------- - # HFM: horizontal beam position - # ---------------------------------------------------------- - - diff_x = x_target - x - - if abs(diff_x) > tolerance: - - current_hfm = get_value(dev.hfm_xd) - hfm_change = diff_x / BeamMove.scint_x_slope - - if abs(hfm_change) > max_move: - print( - f"HFM correction {hfm_change:+.5f} mm exceeds " - f"maximum allowed move of {max_move:.3f} mm." - ) - print("Beam alignment stopped.") - return False - - umv(dev.hfm_xd, current_hfm + hfm_change) - - # -------------------------------------------------------------- - # Final verification - # -------------------------------------------------------------- - - x, y = read_scintXY(CameraZoom.X12_5.position) - - diff_x = x_target - x - diff_y = y_target - y - - final_vfm = get_value(dev.vfm_ydr) - final_hfm = get_value(dev.hfm_xd) - - success = abs(diff_x) <= tolerance and abs(diff_y) <= tolerance - - print( - f"\nFinal beam position: X={x:.2f}, Y={y:.2f} px\n" - f"Final error: X={diff_x:+.2f}, Y={diff_y:+.2f} px\n" - f"VFM DR: {vfm_start:.6f} -> {final_vfm:.6f}\n" - f"HFM XD: {hfm_start:.6f} -> {final_hfm:.6f}" - ) - - if success: - print("Beam alignment successful.") - else: - print(f"Beam is not within ±{tolerance:.1f} px after " f"{iterations} corrections.") - - beamx = calc_scam_microns(diff_x, zoom=CameraZoom.X12_5.position) - beamy = calc_scam_microns(diff_y, zoom=CameraZoom.X12_5.position) - - result["success"] = success - result["BeamX"] = beamx - result["BeamY"] = beamy + update_log(ctx.log, ctx.messages, format_flux_for_gui(result)) + set_progress(ctx.progress, ctx.progress_state, task, status) return result +def format_flux_for_gui(result): + icon = "✓" if result["ok"] else "✗" + return f"\n{icon} Flux measured\n{result['message']}\n" -# def format_daily_setup_result(result): -# """Create readable text for a daily setup result.""" - -# if result is None: -# return "completed" - -# # Beam position dictionary -# if isinstance(result, dict): -# lines = [] - -# for key, value in result.items(): -# if ( -# isinstance(value, dict) -# and "x" in value -# and "y" in value -# ): -# lines.append( -# f"{key}: X={value['x']:.1f}, Y={value['y']:.1f}" -# ) -# else: -# lines.append(f"{key}: {value}") - -# return "\n".join(lines) - -# return str(result) - -# def build_daily_setup_log(results): -# """Build the text for a daily setup SciLog entry.""" - -# lines = ["PXII Daily Setup", ""] - -# for step, result in results.items(): -# lines.append(step) -# lines.append(format_daily_setup_result(result)) -# lines.append("") - -# return "\n".join(lines) +# ============================================================================= +# Daily-setup GUI helpers +# ============================================================================= -# def test_scilog(): -# test_results = { -# "Set energy": "12.40 keV", -# "Align collimator": { -# "x": 39.042, -# "y": 12.376, -# }, -# "Align beamstop": "completed", -# "Beam position": { -# "1x": {"x": 996.8, "y": 971.5}, -# "2x": {"x": 999.9, "y": 971.3}, -# "3x": {"x": 998.6, "y": 971.2}, -# "5.8x": {"x": 998.4, "y": 972.5}, -# "7.5x": {"x": 999.4, "y": 977.0}, -# "12.5x": {"x": 995.2, "y": 976.0}, -# }, -# "Measure flux": "1.24e12 ph/s", -# } -# return test_results +def select_daily_setup_gui(): + """Return widgets belonging to the Daily Setup GUI.""" -def add_energy_to_scilog(msg, result): - msg.add_text("Energy: ", bold=True) - msg.add_text(f"{result['energy']:.0f} eV\n") + window_name = "DailySetup" + open_windows = bec.gui.windows + + if open_windows.get(window_name) is None: + gui = bec.gui.new(window_name) + + # Ensure the placeholder report exists before loading the profile + filename = bl.env.daily_setup_dir / "daily_report.pdf" + if not filename.exists(): + create_empty_daily_report(filename) + + # Restore the standard Daily Setup layout + gui.load_profile("daily_setup") + gui.Report.load_pdf(str(filename)) + gui.Report.fit_to_page() + + else: + gui = open_windows[window_name] + + return ( + gui.Plot_1, + gui.Plot_2, + gui.Plot_3, + gui.Plot_4, + gui.Info, + gui.Progress, + gui.Report, + ) -def add_flux_to_scilog(msg, result): - msg.add_text("Current transmission: ", bold=True) - msg.add_text(f"{result['current_flux']:.2e} ph/s\n") - - msg.add_text("100% transmission: ", bold=True) - msg.add_text(f"{result['full_flux']:.2e} ph/s\n") - -def add_movebeam_to_scilog(msg, result): - msg.add_text("Beam position relative to crosshair: \n", bold=True) - msg.add_text(f"Beam_X: {result['BeamX']:.3f} microns\n") - msg.add_text(f"Beam_Y: {result['BeamY']:.3f} microns\n") +def clear_daily_plots(): + plot1, plot2, plot3, plot4, log, progress, report = select_daily_setup_gui() + plot1.clear_all() + plot2.clear_all() + plot3.clear_all() + plot4.clear_all() -def add_beamstop_to_scilog(msg, result): - msg.add_text("Beamstop X: ", bold=True) - msg.add_text(f"{result['bs_x_pos']:.3f} mm\n") - msg.add_text("Beamstop Y: ", bold=True) - msg.add_text(f"{result['bs_y_pos']:.3f} mm\n") +def update_log(log_box, messages, message): + """Add new message to the daily setup log""" + messages.append(message) + log_box.set_plain_text("\n".join(messages)) -def add_collimator_to_scilog(msg, result): - msg.add_text("Collimator X: ", bold=True) - msg.add_text(f"{result['coll_x_pos']:.3f} mm\n") - - msg.add_text("Collimator Y: ", bold=True) - msg.add_text(f"{result['coll_y_pos']:.3f} mm\n") - - msg.add_text("X width: ", bold=True) - msg.add_text(f"{result['coll_x_width']:.3f} mm\n") - - msg.add_text("Y width: ", bold=True) - msg.add_text(f"{result['coll_y_width']:.3f} mm\n") - -def add_beampos_to_scilog(msg, result): - for zoom, pos in result.items(): - msg.add_text(f"{zoom}: ", bold=True) - msg.add_text( - f"X={pos['x']:.1f} px, " - f"Y={pos['y']:.1f} px\n" - ) - -def log_daily_setup_to_scilog(bec, results): - """Send daily setup results to SciLog.""" - scilog_formatters = { - "Beamline energy": add_energy_to_scilog, - "Move beam": add_movebeam_to_scilog, - "Align beamstop": add_beamstop_to_scilog, - "Measure beam position": add_beampos_to_scilog, - "Align collimator": add_collimator_to_scilog, - "Flux": add_flux_to_scilog, +def update_progress(progress_box, progress_state): + """Sets up the status icons for the progress tracker""" + status = { + "todo": "\u25cb", + "running": "\u25b6", + "done": "\u2713", + "warning": "\u26a0", + "skipped": "–", + "failed": "\u2717", } + + text = "\n".join(f"{status[state]} {label}" for label, state in progress_state.items()) + progress_box.set_plain_text(text) + + +def set_progress(progress_box, progress_state, step, status): + """Change the status of one step and update the display.""" + progress_state[step] = status + update_progress(progress_box, progress_state) + + +def plot_daily_setup_scan(motor_name, signal_name, plot="x", title=None): + """Plot live scan data in one of the Daily Setup waveform widgets.""" + + plot1, plot2, plot3, plot4, log, progress = select_daily_setup_gui() + + if plot.lower() == "plot_1": + wf = plot1 + elif plot.lower() == "plot_2": + wf = plot2 + elif plot.lower() == "plot_3": + wf = plot3 + elif plot.lower() == "plot_4": + wf = plot4 + else: + raise ValueError("plot must be 'x' or 'y'") + + wf.clear_all() + + if title is None: + title = f"{motor_name} scan" + + wf.title = title + wf.x_label = motor_name + wf.y_label = signal_name + + wf.plot(device_x=motor_name, device_y=signal_name) + + +# ============================================================================= +# SciLog +# ============================================================================= + + +def log_daily_setup_to_scilog(bec, results, progress_state, total_duration=None): + """Send compact daily setup summary to SciLog.""" + msg = bec.messaging.scilog.new() - msg.add_text("PXII Daily Setup", bold=True) + date = datetime.now().strftime("%-d %B %Y") + + + 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) + msg.add_text("\n") - for step, result in results.items(): - msg.add_text(f"\n{step}\n", bold=True) + overall = get_daily_setup_status(progress_state) - formatter = scilog_formatters.get(step) + msg.add_text("Overall status: ", bold=True) + msg.add_text(f"{overall}\n") - if formatter: - formatter(msg, result) - else: - msg.add_text(str(result)) + if total_duration is not None: + msg.add_text("Total duration: ", bold=True) + msg.add_text(f"{format_duration(total_duration)}\n") + pdfname = bl.env.daily_setup_dir / "daily_report.pdf" - msg.add_tags(["daily-setup", "pxii"]) + print(f"Sending pdf from {pdfname} to scilog") + + msg.add_attachment(pdfname) + + msg.add_tags(["daily-setup", bl.env.beamline_name]) msg.send() -def preview_daily_setup_scilog(results): - """Preview the daily setup SciLog content without sending.""" + # Motor positions snapshot + log_daily_setup_positions(bec) - print("\nPXII Daily Setup\n") +def log_daily_setup_positions(bec): + """Record important motor positions after daily setup""" - for step, result in results.items(): - print(step) + bec.messaging.scilog.log_positions( + devices=[ + "vfm_y", + "vfm_pitch", + "vfm_ydr", + "hfm_x", + "hfm_pitch", + "hfm_xd", + "coll_x", + "bs_x", + "bs_y", + ], + title=f"{bl.env.beamline_name} Daily Setup - Motor Positions", + tags=["daily_setup", bl.env.beamline_name, "positions"] + ) - if isinstance(result, dict): - for key, value in result.items(): - if isinstance(value, dict) and "x" in value and "y" in value: - print( - f" {key}: " - f"X={value['x']:.1f}, " - f"Y={value['y']:.1f}" - ) - else: - print(f" {key}: {value}") +def add_prechecks_to_scilog(msg, results): + """Add pre-flight checks to the SciLog entry.""" + + if "Pre checks" not in results: + return + + result = results["Pre checks"] + + msg.add_text("Pre-flight\n", bold=True) + + machine = result.get("machine") + + if machine: + for check in machine["checks"]: + status = "OK" if check["ok"] else "FAILED" + msg.add_text( + f"{check['message']:<55} {status}\n" + ) + + eh = result.get("experiment_hutch") + + if eh: + for check in eh["checks"]: + status = "OK" if check["ok"] else "WARNING" + msg.add_text( + f"{check['message']:<55} {status}\n" + ) + + msg.add_text("\n") + +def add_setup_results_to_scilog(msg, results, progress_state): + """Add daily setup task results to SciLog.""" + + msg.add_text("Setup\n", bold=True) + + # --------------------------------------------------------- + # Energy + # --------------------------------------------------------- + if "Set energy" in results: + r = results["Set energy"] + + status = get_result_status("Set energy", r, progress_state) + + if r.get("energy") is not None: + detail = f"{r['energy'] / 1000:.3f} keV" else: - print(f" {result}") + detail = r.get("message", "") + + add_scilog_result( + msg, + "Set energy", + status, + detail, + ) + + # --------------------------------------------------------- + # Collimator + # --------------------------------------------------------- + if "Align collimator" in results: + r = results["Align collimator"] + + status = get_result_status( + "Align collimator", + r, + progress_state, + ) + + if r.get("ok"): + detail = ( + f"X={r['coll_x_pos']:.3f}, " + f"Y={r['coll_y_pos']:.3f} mm" + ) + else: + detail = r.get("message", "") + + add_scilog_result( + msg, + "Collimator alignment", + status, + detail, + ) + + # --------------------------------------------------------- + # Beamstop + # --------------------------------------------------------- + if "Align beamstop" in results: + r = results["Align beamstop"] + + status = get_result_status( + "Align beamstop", + r, + progress_state, + ) + + if r.get("ok"): + detail = ( + f"X={r['bs_x_pos']:.3f}, " + f"Y={r['bs_y_pos']:.3f} mm" + ) + else: + detail = r.get("message", "") + + add_scilog_result( + msg, + "Beamstop alignment", + status, + detail, + ) + + # --------------------------------------------------------- + # Move beam + # --------------------------------------------------------- + if "Move beam" in results: + r = results["Move beam"] + + status = get_result_status( + "Move beam", + r, + progress_state, + ) + + if r.get("ok"): + detail = ( + f"X={r['BeamX']:.1f}, " + f"Y={r['BeamY']:.1f} µm" + ) + else: + detail = r.get("message", "manual check required") + + add_scilog_result( + msg, + "Move beam", + status, + detail, + ) + + # --------------------------------------------------------- + # Flux + # --------------------------------------------------------- + if "Measure flux" in results: + r = results["Measure flux"] + + status = get_result_status( + "Measure flux", + r, + progress_state, + ) + + if r.get("full_flux") is not None: + detail = f"{r['full_flux']:.2e} ph/s" + else: + detail = r.get("message", "") + + add_scilog_result( + msg, + "Flux", + status, + detail, + ) + +def add_scilog_result(msg, label, status, detail=""): + """Add one formatted setup-result line.""" + + line = f"{label:<24} {status:<8}" + + if detail: + line += f" {detail}" + + msg.add_text(line + "\n") + + +def get_result_status(task, result, progress_state): + """Return human-readable status for a SciLog result.""" + + state = progress_state.get(task) + + if state == "warning": + return "WARNING" + + if state == "failed": + return "FAILED" + + if result.get("ok"): + return "OK" + + return "FAILED" + + +def get_daily_setup_status(progress_state): + """Return overall daily-setup status.""" + + states = progress_state.values() + + if "failed" in states: + return "FAILED" + + if "warning" in states: + return "PASSED WITH WARNINGS" + + return "PASSED" + + +# ============================================================================= +# Save values to csv file +# ============================================================================= + +def log_daily_setup_to_csv(results): + """Append key daily setup diagnostics to the beamline history CSV.""" + + filename = bl.env.daily_setup_dir / "daily_beamline_health.csv" + + move = results.get("Move beam", {}) + energy = results.get("Set energy", {}) + flux = results.get("Measure flux", {}) + row = { + "timestamp": datetime.now().isoformat(timespec="seconds"), + + # Energy / flux + "energy_eV": f"{energy.get("energy"):.0f}", + "flux_ph_s": f"{flux.get("full_flux"):.2e}", + + # Beam position before move_beam + "beam_start_x_px": f"{move.get('StartX'):.0f}", + "beam_start_y_px": f"{move.get("StartY"):.0f}", + + # Mirror positions before move_beam + "vfm_ydr_start_mm": f"{move.get("Initial VFM DR"):.5f}", + "hfm_xd_start_mm": f"{move.get("Initial HFM XD"):.5f}", + + # Mirror positions after move_beam + "vfm_ydr_final_mm": f"{move.get("Final VFM DR"):.5f}", + "hfm_xd_final_mm": f"{move.get("Final HFM XD"):.5f}", + + # Final beam displacement from crosshair + "beam_final_x_um": f"{move.get("BeamX"):.1f}", + "beam_final_y_um": f"{move.get("BeamY"):.1f}", + + # Current mirror state after daily setup + "vfm_y_mm": f"{get_value(dev.vfm_y):.5f}", + "vfm_ydr_mm": f"{get_value(dev.vfm_ydr):.5f}", + "vfm_pitch_mrad": f"{get_value(dev.vfm_pitch):.5f}", + + "hfm_x_mm": f"{get_value(dev.hfm_x):.5f}", + "hfm_xd_mm": f"{get_value(dev.hfm_xd):.5f}", + "hfm_pitch_mrad": f"{get_value(dev.hfm_pitch):.5f}", + } + + + file_exists = filename.exists() + + with open(filename, "a", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=row.keys()) + + if not file_exists: + writer.writeheader() + + writer.writerow(row) + + print(f"✓ Daily beamline history saved to {filename}") + +# ============================================================================= +# Create PDF report +# ============================================================================= + + +def create_daily_report(results, filename): + """Create the final Daily Setup PDF report.""" + + fig = plt.figure(figsize=(11, 7)) + + # Camera image on left + ax_image = fig.add_axes([0.06, 0.12, 0.55, 0.78]) + + # Results text on right + ax_text = fig.add_axes([0.66, 0.12, 0.30, 0.78]) + ax_text.axis("off") + + fig.suptitle( + f"{bl.env.beamline_name} Daily Setup", + fontsize=18, + weight="bold", + ) + + fig.text( + 0.06, + 0.94, + datetime.now().strftime("%d %B %Y %H:%M"), + fontsize=10, + ) + + # -------------------------------------------------------------- + # Beam image + # -------------------------------------------------------------- + + move_result = results.get("Move beam") + + if move_result and move_result.get("image") is not None: + + image = move_result["image"] + + ax_image.imshow(image) + + # Target crosshair + x_target = BeamMove.scint_target_x + y_target = BeamMove.scint_target_y + + ax_image.axvline( + x_target, + linestyle="--", + linewidth=1, + ) + ax_image.axhline( + y_target, + linestyle="--", + linewidth=1, + ) + + # Measured beam position + beam_x_px = move_result["FinalX"] + + beam_y_px = move_result["FinalY"] + + ax_image.plot( + beam_x_px, + beam_y_px, + marker="+", + markersize=12, + markeredgewidth=2, + ) + + ax_image.set_title("Final beam position") + ax_image.axis("off") + + else: + ax_image.text( + 0.5, + 0.5, + "No beam image available", + ha="center", + va="center", + transform=ax_image.transAxes, + ) + ax_image.axis("off") + + # -------------------------------------------------------------- + # Results summary + # -------------------------------------------------------------- + + y = 0.95 + + for task, result in results.items(): + + if result is None: + continue + + if result.get("warning"): + status = "WARNING" + elif result.get("ok"): + status = "OK" + else: + status = "FAILED" + + ax_text.text( + 0.0, + y, + f"{task:<22} {status}", + fontsize=12, + family="monospace", + va="top", + ) + + y -= 0.06 + + # Extra beam information + if move_result: + + y -= 0.04 + + ax_text.text( + 0.0, + y, + "Beam position", + fontsize=12, + weight="bold", + ) + y -= 0.05 + + if "BeamX" in move_result and "BeamY" in move_result: + ax_text.text( + 0.0, + y, + f"X offset: {move_result['BeamX']:+.1f} µm", + fontsize=11, + ) + y -= 0.04 + + ax_text.text( + 0.0, + y, + f"Y offset: {move_result['BeamY']:+.1f} µm", + fontsize=11, + ) + y -= 0.04 + + else: + ax_text.text( + 0.0, + y, + "Beam position not available", + fontsize=11, + ) + y -= 0.04 + + if "X FWHM" in move_result and "Y FWHM" in move_result: + ax_text.text( + 0.0, + y, + f"Beam size: " + f"{move_result['X FWHM']:.1f} × " + f"{move_result['Y FWHM']:.1f} µm", + fontsize=11, + ) + + fig.savefig( + filename, + format="pdf", + bbox_inches="tight", + ) + + plt.close(fig) + + return filename + +# ============================================================================= +# General helpers +# ============================================================================= + + +def move_with_offset(motor, target, offset=0.002): + """Move motor to target using an overshoot to overcome deadband/backlash.""" + + current = get_value(motor) + dmove = target - current + + if abs(dmove) < 1e-9: + return current + + direction = np.sign(dmove) + overshoot = target + direction * abs(offset) + + umv(motor, overshoot) + umv(motor, target) + + final = get_value(motor) + + print(f"{motor.name}: requested {current:.6f} -> {target:.6f} " f"({dmove * 1000:+.3f} µm)") + print( + f"{motor.name}: actual {current:.6f} -> {final:.6f} " + f"({(final-current) * 1000:+.3f} µm), " + f"error {(final-target) * 1000:+.3f} µm" + ) + + return final + + +def set_zoom(zoom): + """Set sample camera zoom level.""" + + if isinstance(zoom, str): + zoom = CameraZoom.from_label(zoom) + umv(dev.scam_zoom, zoom.position) + else: + umv(dev.scam_zoom, zoom) + + +@contextmanager +def time_step(results, name): + """Time a daily setup step.""" + start = time.perf_counter() + + try: + yield + finally: + elapsed = time.perf_counter() - start + + if name not in results: + results[name] = {} + + results[name]["duration"] = elapsed + + +def format_duration(seconds): + if seconds < 60: + return f"{seconds:.1f} s" + + minutes, seconds = divmod(seconds, 60) + return f"{int(minutes)} min {seconds:.0f} s" + + + +def caget(pv): + """Read an EPICS PV using caget and return its value.""" + + try: + result = subprocess.run( + ["caget", "-t", pv], + capture_output=True, + text=True, + check=True, + timeout=2, + ) + + fields = shlex.split(result.stdout.strip()) + + # Find the PV name and take the value immediately after it + try: + pv_index = fields.index(pv) + value = fields[pv_index + 1] + except (ValueError, IndexError): + return None + + # Convert numeric values to float + try: + return float(value) + except ValueError: + return value + + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + print(f"Could not read {pv}: {exc}") + return None - print() \ No newline at end of file diff --git a/pxii_bec/macros/mxcomm_utils.py b/pxii_bec/macros/mxcomm_utils.py new file mode 100644 index 0000000..1e52ca6 --- /dev/null +++ b/pxii_bec/macros/mxcomm_utils.py @@ -0,0 +1,293 @@ +"""General reusable utilities for MX beamline commissioning and operation.""" + +import epics +import time +import numpy as np +import matplotlib.pyplot as plt +from pxii_bec.scripts import beamline_context as bl + +# ============================================================================= +# Camera utilities +# ============================================================================= + +def read_scintXY(zoom=800, image=False): + """Get the X and Y beam positions from the sample camera at + a sepcified zoom. Default zoom is 800""" + + # Target position on sample camera + x_target = BeamMove.scint_target_x + y_target = BeamMove.scint_target_y + + results = {} + + # Ensure shutter and detector cover are closed before changing states + dev.bcu_shutter.put(0) + if bl.d["det_cov"].pos != "close": + bl.d["det_cov"].move("close") + try: + # set transmission to 10% + current_transm = get_value(dev.transm) + if abs(current_transm - 0.1) > 0.01: + dev.transm.put(0.1) + # set camera zoom + if get_value(dev.scam_zoom) != zoom: + umv(dev.scam_zoom, zoom) + # move to 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) + + # set autoexposure and check there is a reading in the image + time.sleep(0.5) + auto_exposure(cam="samcam", max_iter=25) + time.sleep(0.2) + + max_signal = get_value(dev.samcam_max) + results["max_signal"] = max_signal + + if max_signal < BeamMove.min_scint_signal: + return { + "ok": False, + "message": ( + f"No beam detected on the scintillator " + f"(max signal = {max_signal:.0f})" + ), + "max_signal": max_signal, + } + + + x_px = dev.samcam_x.read()["samcam_x"]["value"] + y_px = dev.samcam_y.read()["samcam_y"]["value"] + results["X_pixels"] = x_px + results["Y_pixels"] = y_px + + results["ok"] = True + results["message"] = ( + f"Beam detected at X={x_px:.1f}, Y ={y_px:.1f} px" + ) + + # Calculate difference from target position + diff_x = x_target - x_px + diff_y = y_target - y_px + results["X_diff_px"] = diff_x + results["Y diff_px"] = diff_y + + # Convert to pixels + diff_x_mic = calc_scam_microns(diff_x, zoom) + diff_y_mic = calc_scam_microns(diff_y, zoom) + results["X_diff_microns"] = diff_x_mic + results["Y_diff_microns"] = diff_y_mic + + # Measure beamsize + x_fwhm, y_fwhm = measure_beamsize() + results["X_FWHM"] = x_fwhm + results["Y_FWHM"] = y_fwhm + + print(f"Beam position: X = {results['X_pixels']:.1f} px, " + f"Y = {results['Y_pixels']:.1f} px\n" + f"Distance from crosshair: X = {results['X_diff_microns']:.1f} microns, " + f"Y = {results['Y_diff_microns']:.1f} microns\n", + f"X size (fwhm) = {results["X_FWHM"]:.1f} microns, " + f"Y size (fwhm) = {results["Y_FWHM"]:.1f} microns") + + if image: + imagedata = get_cam_image(Cameras.sample) + results["image"] = imagedata + + + return results + + finally: + # print("done") + dev.bcu_shutter.put(0) + + +def get_cam_image(prefix): + """Return the current RGB image from an EPICS areaDetector camera.""" + + width = int(epics.caget(f"{prefix}:image1:ArraySize1_RBV")) + height = int(epics.caget(f"{prefix}:image1:ArraySize2_RBV")) + + data = epics.caget(f"{prefix}:image1:ArrayData") + + if data is None: + raise RuntimeError(f"Could not read camera image from {prefix}") + + expected_size = width * height * 3 + + if data.size < expected_size: + raise RuntimeError( + f"Camera returned {data.size} bytes; " + f"expected {expected_size}" + ) + + return np.asarray( + data[:expected_size], + dtype=np.uint8, + ).reshape(height, width, 3) + +def save_cam_image( + image, + filename, + *, + crosshair=None, + beam_position=None, + crop=None, + title=None, +): + """Save a camera image, optionally with crosshair and beam position.""" + + fig, ax = plt.subplots(figsize=(4, 4)) + + ax.imshow(image) + + # Target crosshair + if crosshair is not None: + x, y = crosshair + ax.axvline(x, linewidth=1) + ax.axhline(y, linewidth=1) + + # Measured beam position + if beam_position is not None: + x, y = beam_position + ax.plot( + x, + y, + marker="+", + markersize=12, + markeredgewidth=2, + ) + + # Crop around crosshair + if crop is not None: + if crosshair is None: + raise ValueError( + "crosshair must be supplied when using crop" + ) + + x, y = crosshair + + ax.set_xlim(x - crop, x + crop) + + # Image coordinates have Y increasing downwards + ax.set_ylim(y + crop, y - crop) + + if title: + ax.set_title(title) + + ax.axis("off") + + fig.tight_layout() + fig.savefig(filename, dpi=150, bbox_inches="tight") + plt.close(fig) + + return filename + + + +def create_cam_report( + image, + pdf_filename, + *, + crosshair=None, + beam_position=None, + crop=None, + title="Camera snapshot", +): + """Create a one-page PDF containing an annotated camera image.""" + + fig, ax = plt.subplots(figsize=(6, 6)) + + ax.imshow(image) + + if crosshair is not None: + x, y = crosshair + ax.axvline(x, linewidth=1) + ax.axhline(y, linewidth=1) + + if beam_position is not None: + x, y = beam_position + ax.plot( + x, + y, + marker="+", + markersize=12, + markeredgewidth=2, + ) + + if crop is not None: + if crosshair is None: + raise ValueError( + "crosshair must be supplied when using crop" + ) + + x, y = crosshair + ax.set_xlim(x - crop, x + crop) + ax.set_ylim(y + crop, y - crop) + + ax.set_title(title) + ax.axis("off") + + fig.tight_layout() + + fig.savefig( + pdf_filename, + format="pdf", + bbox_inches="tight", + ) + + plt.close(fig) + + return pdf_filename + +# ============================================================================= +# Daily-setup PDF reports +# ============================================================================= + +def create_empty_daily_report(filename): + """Create the placeholder Daily Setup report.""" + + + fig, ax = plt.subplots(figsize=(8, 6)) + + ax.axis("off") + + ax.text( + 0.5, + 0.60, + "Daily Setup", + ha="center", + va="center", + fontsize=20, + weight="bold", + ) + + ax.text( + 0.5, + 0.48, + "Setup in progress.", + ha="center", + va="center", + fontsize=12, + ) + + ax.text( + 0.5, + 0.42, + "The report will appear here when complete.", + ha="center", + va="center", + fontsize=10, + ) + + fig.savefig( + filename, + format="pdf", + bbox_inches="tight", + ) + plt.close(fig) + + return filename +