From e98cd432ad1f32fcf975119c02b06eee8cdf5aff Mon Sep 17 00:00:00 2001 From: x10sa Date: Mon, 21 Sep 2026 09:04:42 +0200 Subject: [PATCH] moved some common functions from here into mxcomm --- pxii_bec/macros/katscripts.py | 695 +++++++++++++++------------------- 1 file changed, 298 insertions(+), 397 deletions(-) diff --git a/pxii_bec/macros/katscripts.py b/pxii_bec/macros/katscripts.py index cbf07a4..ff09ee3 100755 --- a/pxii_bec/macros/katscripts.py +++ b/pxii_bec/macros/katscripts.py @@ -13,6 +13,7 @@ from pathlib import Path from pxii_bec.scripts import beamline_context as bl from pxii_bec.macros.beamline_planner import BeamlineState + def bragg_scan(): """Energy scans from 6 to 18 keV at various gap positions""" @@ -31,6 +32,7 @@ def bragg_scan(): ) gap += 0.5 + def scan_bpm(bpmname): """ Runs a grid scan of a BPM in x and y, and plots each channel @@ -217,6 +219,7 @@ def scan_fe(): # ) # close_fe_slits(False) + def scan_gap(gomax=False, close=True, fine=False, min_fwhm_steps=1.5): """ Scan the undulator gap and monitor the LU BPM sum. @@ -278,20 +281,13 @@ def scan_gap(gomax=False, close=True, fine=False, min_fwhm_steps=1.5): min_gap_reached = False if gomax: - min_gap_reached = ( - coarse_result["at_lower_boundary"] - and np.isclose(coarse_result["x_max"], MIN_GAP) + min_gap_reached = coarse_result["at_lower_boundary"] and np.isclose( + coarse_result["x_max"], MIN_GAP ) if min_gap_reached: - print( - "\nMaximum occurs at the minimum gap of " - f"{MIN_GAP:.2f} mm." - ) - print( - "The optimum for this harmonic may lie below " - "the accessible gap." - ) + print("\nMaximum occurs at the minimum gap of " f"{MIN_GAP:.2f} mm.") + print("The optimum for this harmonic may lie below " "the accessible gap.") # -------------------------------------------------------- # Fine scan @@ -313,10 +309,7 @@ def scan_gap(gomax=False, close=True, fine=False, min_fwhm_steps=1.5): ) elif fine and min_gap_reached: - print( - "Skipping fine gap scan because the coarse maximum " - "is at the minimum gap." - ) + print("Skipping fine gap scan because the coarse maximum " "is at the minimum gap.") return { "success": True, @@ -328,6 +321,7 @@ def scan_gap(gomax=False, close=True, fine=False, min_fwhm_steps=1.5): finally: close_fe_slits(False) + def record_pitch(): """Record the pitch and perp distance to file""" with open("luts/pitch_record.csv", "a", encoding="utf-8") as f: @@ -367,13 +361,7 @@ def beam_centre_from_ss(): return beampos -def generate_dcm_lut( - start_energy=6000, - end_energy=30000, - step=300, - scan_id_gap=True, - scint=True, -): +def generate_dcm_lut(start_energy=6000, end_energy=30000, step=300, scan_id_gap=True, scint=True): """Generate a DCM calibration lookup table. Parameters @@ -397,9 +385,7 @@ def generate_dcm_lut( """ filename = bl.env.lut_dir / f"dcm_calibration_{filenow(date_only=True)}.csv" - failure_filename = ( - bl.env.lut_dir / f"dcm_calibration_failures_{filenow(date_only=True)}.csv" - ) + failure_filename = bl.env.lut_dir / f"dcm_calibration_failures_{filenow(date_only=True)}.csv" failures = [] @@ -409,30 +395,21 @@ def generate_dcm_lut( if scan_id_gap: with open(filename, "w", encoding="utf-8") as f: if scint: - f.write( - f"energy,harmonic,gap,pitch," - f"samcam_x,samcam_y,zoom={zoom}\n" - ) + f.write(f"energy,harmonic,gap,pitch," f"samcam_x,samcam_y,zoom={zoom}\n") else: - f.write( - "energy,harmonic,gap,pitch\n" - ) + f.write("energy,harmonic,gap,pitch\n") else: with open(filename, "w", encoding="utf-8") as f: if scint: - f.write( - f"energy,pitch,samcam_x,samcam_y,zoom={zoom}\n" - ) + f.write(f"energy,pitch,samcam_x,samcam_y,zoom={zoom}\n") else: - f.write( - "energy,pitch\n" - ) + f.write("energy,pitch\n") # Create failure log with open(failure_filename, "w", encoding="utf-8") as f: f.write("energy,scan,reason\n") try: - for energy in range(start_energy, end_energy+1, step): + for energy in range(start_energy, end_energy + 1, step): print("\n" + "=" * 60) print(f"Moving to {energy:.0f} eV") @@ -450,18 +427,12 @@ def generate_dcm_lut( print(f"Energy move failed at {energy:.0f} eV: {reason}") - failure = { - "energy": energy, - "scan": "energy", - "reason": reason, - } + failure = {"energy": energy, "scan": "energy", "reason": reason} failures.append(failure) with open(failure_filename, "a", encoding="utf-8") as f: - f.write( - f"{energy:.0f},energy,{reason}\n" - ) + f.write(f"{energy:.0f},energy,{reason}\n") # Cannot calibrate this energy if the energy move failed continue @@ -473,42 +444,22 @@ def generate_dcm_lut( if scan_id_gap: try: - scan_gap( - close=True, - gomax=True, - fine=True, - ) + scan_gap(close=True, gomax=True, fine=True) except Exception as exc: reason = str(exc) - print( - f"Gap scan failed at {energy:.0f} eV: " - f"{reason}" - ) + print(f"Gap scan failed at {energy:.0f} eV: " f"{reason}") - failure = { - "energy": energy, - "scan": "gap", - "reason": reason, - } + failure = {"energy": energy, "scan": "gap", "reason": reason} failures.append(failure) - with open( - failure_filename, - "a", - encoding="utf-8", - ) as f: - f.write( - f"{energy:.0f},gap,{reason}\n" - ) + with open(failure_filename, "a", encoding="utf-8") as f: + f.write(f"{energy:.0f},gap,{reason}\n") # Fall back to calculated gap - print( - f"Setting calculated gap for " - f"{energy:.0f} eV" - ) + print(f"Setting calculated gap for " f"{energy:.0f} eV") set_gap(energy) # ------------------------------------------------------------ @@ -522,56 +473,29 @@ def generate_dcm_lut( reason = "; ".join(result["problems"]) - print( - f"Mono pitch scan rejected at " - f"{energy:.0f} eV:" - ) + print(f"Mono pitch scan rejected at " f"{energy:.0f} eV:") for problem in result["problems"]: print(f" - {problem}") - failure = { - "energy": energy, - "scan": "mono_pitch", - "reason": reason, - } + failure = {"energy": energy, "scan": "mono_pitch", "reason": reason} failures.append(failure) - with open( - failure_filename, - "a", - encoding="utf-8", - ) as f: - f.write( - f"{energy:.0f},mono_pitch,{reason}\n" - ) + with open(failure_filename, "a", encoding="utf-8") as f: + f.write(f"{energy:.0f},mono_pitch,{reason}\n") except Exception as exc: reason = str(exc) - print( - f"Mono pitch scan failed at " - f"{energy:.0f} eV: {reason}" - ) + print(f"Mono pitch scan failed at " f"{energy:.0f} eV: {reason}") - failure = { - "energy": energy, - "scan": "mono_pitch_exception", - "reason": reason, - } + failure = {"energy": energy, "scan": "mono_pitch_exception", "reason": reason} failures.append(failure) - with open( - failure_filename, - "a", - encoding="utf-8", - ) as f: - f.write( - f"{energy:.0f},mono_pitch_exception," - f"{reason}\n" - ) + with open(failure_filename, "a", encoding="utf-8") as f: + f.write(f"{energy:.0f},mono_pitch_exception," f"{reason}\n") # ------------------------------------------------------------ # Record calibration values @@ -601,27 +525,14 @@ def generate_dcm_lut( f"{y:.4g}\n" ) else: - f.write( - f"{energy:.0f}," - f"{h}," - f"{gap_value:.5g}," - f"{pitch:.5g}\n" - ) + f.write(f"{energy:.0f}," f"{h}," f"{gap_value:.5g}," f"{pitch:.5g}\n") else: if scint: - f.write( - f"{energy:.0f}," - f"{pitch:.5g}," - f"{x:.4g}," - f"{y:.4g}\n" - ) + f.write(f"{energy:.0f}," f"{pitch:.5g}," f"{x:.4g}," f"{y:.4g}\n") else: - f.write( - f"{energy:.0f}," - f"{pitch:.5g}\n" - ) + f.write(f"{energy:.0f}," f"{pitch:.5g}\n") finally: print("\nReturning beamline to safe conditions...") @@ -636,7 +547,6 @@ def generate_dcm_lut( except Exception as exc: print(f"WARNING: Could not open undulator gap: {exc}") - # ------------------------------------------------------------ # Summary # ------------------------------------------------------------ @@ -651,28 +561,19 @@ def generate_dcm_lut( for failure in failures: print( - f"{failure['energy']:5.0f} eV " - f"{failure['scan']:22s} " - f"{failure['reason']}" + f"{failure['energy']:5.0f} eV " f"{failure['scan']:22s} " f"{failure['reason']}" ) - print( - f"\nFailure log written to:\n" - f"{failure_filename}" - ) + print(f"\nFailure log written to:\n" f"{failure_filename}") else: print("\nNo scan failures.") - print( - f"\nCalibration data written to:\n" - f"{filename}" - ) + print(f"\nCalibration data written to:\n" f"{filename}") return failures - def write_lut(): """Write DCM parameters to file""" filename = "luts/manual_big.csv" @@ -718,7 +619,6 @@ def write_beamsize(): f.write(f"{lens_config},{sx:.4g},{sy:.4g},{i1_reading:.4g}\n") - def scan_Ge(): scanbefore = 12 # eV scanafter = 12 # eV @@ -734,15 +634,14 @@ def scan_Ge(): ydata = [] filename = f"{bl.env.lut_dir}/{filenow(date_only=True)}_ge_edge.csv" print(filename) - with open(filename, 'w') as f: + with open(filename, "w") as f: f.write(f"Energy,Bragg,bcu_bpmsum\n") - - plot_live_data(dev.dcm_bragg,dev.bcu_bpmsum) + plot_live_data(dev.dcm_bragg, dev.bcu_bpmsum) scanpoints = np.linspace(current_bragg - 0.5, current_bragg + 0.5, 100) - start_energy = convert_from_bragg(current_bragg+0.5)["energy_ev"] + start_energy = convert_from_bragg(current_bragg + 0.5)["energy_ev"] bl_energy(start_energy) - dev.transm.put(1.0) # Take all the filters out + dev.transm.put(1.0) # Take all the filters out umv(dev.ss_f4_x, 6) # Put Ge filter in place for bragg in scanpoints: @@ -751,7 +650,7 @@ def scan_Ge(): inten = average_inten() xdata.append(energy) ydata.append(inten) - with open(filename, 'a') as f: + with open(filename, "a") as f: f.write(f"{energy:.2f},{bragg:.2f},{inten:.2f}\n") data = { "x_data": np.array(xdata), @@ -761,25 +660,22 @@ def scan_Ge(): "motor_device": dev.dcm_bragg, "scan_number": "Current", } - dev.transm.put(0.1) # set transmission back to 0.1 + dev.transm.put(0.1) # set transmission back to 0.1 # Define and fit model to scan data - fit_params = create_fit_parameters(deriv = True, - negative = False, - model = FitDefaults.MODEL, - baseline = FitDefaults.BASELINE) + fit_params = create_fit_parameters( + deriv=True, negative=False, model=FitDefaults.MODEL, baseline=FitDefaults.BASELINE + ) fit_result = fit(data, fit_params) # Plot the fitted data if plot = True - + plot_fitted_data(data, fit_result) - - def cal_samcam_fixed_x(): # lat = dev.hfm_x.read()["hfm_x"]["value"] # pitch = dev.hfm_pitch.read()["hfm_pitch"]["value"] - fpitch = dev.dcm_fpitch.read()['dcm_fpitch']['value'] + fpitch = dev.dcm_fpitch.read()["dcm_fpitch"]["value"] scanpoints = np.linspace(3.75, 4.75, 7) zoom_high = 800 zoom_low = 200 @@ -1045,12 +941,13 @@ def correlate_hfm_xd(meas="bcu"): if meas == "bcu": plt.ylabel("beam X (from BCU BPM)") elif meas == "scint": - plt.ylabel("beam X (from Scintillator)") + plt.ylabel("beam X (from Scintillator)") plt.legend() plt.grid(True) plt.tight_layout() plt.show() + def correlate_vfm_ydr(meas="bcu"): y_correlation = [] starty = get_value(dev.vfm_ydr) @@ -1058,7 +955,7 @@ def correlate_vfm_ydr(meas="bcu"): scanpoints = np.linspace(starty - 0.005, starty + 0.005, 10) elif meas == "scint": scanpoints = np.linspace(starty - 0.02, starty + 0.02, 10) - + for pos in scanpoints: umv(dev.vfm_ydr, pos) if meas == "bcu": @@ -1097,7 +994,6 @@ def correlate_vfm_ydr(meas="bcu"): plt.show() - def correlate_hfm_pitch(): x_correlation = [] pitch = dev.hfm_pitch.read()["hfm_pitch"]["value"] @@ -1277,17 +1173,18 @@ def filters(pos="out"): if SE.scin.checkpos(): auto_exposure(cam="samcam", max_iter=20) -def filenow(date_only = False): + +def filenow(date_only=False): """Return current time formatted for filenames.""" if date_only: return datetime.now().strftime("%Y%m%d") else: - return datetime.now().strftime("%Y%m%d-%H%M") + return datetime.now().strftime("%Y%m%d-%H%M") def energy_vs_height(start_energy=11000, end_energy=25000, step=1000): """Correlate beam height with energy from start_energy to end_energy in steps of step""" - filedir, fnow = fnow(dir = "test") + filedir, fnow = fnow(dir="test") filename = f"{filedir}/{fnow}_energy-height-ss.csv" with open(filename, "w", encoding="utf-8") as f: # f.write("energy,bcu_norm_y,bcu_y,samcam_ypix,samcam_y\n") @@ -1317,7 +1214,6 @@ def energy_vs_height(start_energy=11000, end_energy=25000, step=1000): # umv(dev.id_gap, 20.0) - def scan_in_x(xsize, xcen, mon, small, large, width, steps): umv(xsize, small) go_to_peak(xcen, mon, -(width * 0.5), (width * 0.5), steps=steps, relative=False) @@ -1335,16 +1231,7 @@ def trans_fils(inc): x2 = dev.ss_f2_x.read()["ss_f2_x"]["value"] x3 = dev.ss_f3_x.read()["ss_f3_x"]["value"] x4 = dev.ss_f4_x.read()["ss_f4_x"]["value"] - umv( - dev.ss_f1_x, - x1 + inc, - dev.ss_f2_x, - x2 + inc, - dev.ss_f3_x, - x3 + inc, - dev.ss_f4_x, - x4 + inc, - ) + umv(dev.ss_f1_x, x1 + inc, dev.ss_f2_x, x2 + inc, dev.ss_f3_x, x3 + inc, dev.ss_f4_x, x4 + inc) def calc_from_centre(cam="samcam"): @@ -1387,14 +1274,13 @@ def get_cam(low_zoom, high_zoom): # else: # print("Adjust X tilt in anticlockwise direction") if -15 <= ydiff <= 15: - print("Y is OK") + print("Y is OK") elif ydiff > 15: - print("Adjust Y tilt in anticlockwise direction") + print("Adjust Y tilt in anticlockwise direction") else: print("Adjust Y tilt in clockwise direction") - # def read_currents(): # """Read the individual bpm channels""" # bpm1 = dev.bcu_bpm1.read()["bcu_bpm1"]["value"] @@ -1414,17 +1300,19 @@ def get_cam(low_zoom, high_zoom): # yn = ((readings["bpm1"] + readings["bpm3"]) - (readings["bpm2"] + readings["bpm4"])) / total # return xn, yn -def average_inten(): - vals_inten = [] - for _ in range(5): - vals_inten.append(dev.bcu_bpmsum.read()['bcu_bpmsum']['value']) - time.sleep(0.1) - inten = np.mean(vals_inten) - return inten -def mirror_pitch(mirror = 'vfm'): - - if mirror == 'vfm': +def average_inten(): + vals_inten = [] + for _ in range(5): + vals_inten.append(dev.bcu_bpmsum.read()["bcu_bpmsum"]["value"]) + time.sleep(0.1) + inten = np.mean(vals_inten) + return inten + + +def mirror_pitch(mirror="vfm"): + + if mirror == "vfm": pitch = dev.vfm_pitch.position mot = dev.vfm_pitch start = 2.94 @@ -1433,9 +1321,9 @@ def mirror_pitch(mirror = 'vfm'): mot = dev.hfm_pitch start = 3.3 print(f"Scanning {mirror}_pitch from 10-29 keV") - pitch_range = [start - 0.25, start - 0.2, start -0.15, start - 0.1, start -0.05] + pitch_range = [start - 0.25, start - 0.2, start - 0.15, start - 0.1, start - 0.05] - umv(mot,start) + umv(mot, start) results = [] now = datetime.now() @@ -1457,55 +1345,19 @@ def mirror_pitch(mirror = 'vfm'): vals_diode = [] for _ in range(5): - vals_diode.append(dev.xidiode.read()['xidiode']['value']) + vals_diode.append(dev.xidiode.read()["xidiode"]["value"]) time.sleep(0.1) diode = np.mean(vals_diode) - results.append([ - energy, - p, - diode, - ]) + results.append([energy, p, diode]) with open(csvfile, "a") as f: - f.write( - f"{energy:.2f},{p:.5f},{diode:.5g}\n" - ) - umv(mot,start) + f.write(f"{energy:.2f},{p:.5f},{diode:.5g}\n") + umv(mot, start) bl_energy(12400) results = np.array(results) -def read_scintXY(zoom=800): - """ Get the X and Y beam positions from the sample camera at - a sepcified zoom. Default zoom is 800""" - # 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 - auto_exposure(cam="samcam", max_iter=25) - time.sleep(0.2) - x = dev.samcam_x.read()['samcam_x']['value'] - y = dev.samcam_y.read()['samcam_y']['value'] - return x, y - finally: - dev.bcu_shutter.put(0) - def vfm_yaw_check(): @@ -1526,18 +1378,18 @@ def vfm_yaw_check(): # Point 1 print(f"Moving slits to position 1") umv(dev.ss_sl_ycen, slit_point1) - scans.line_scan(dev.vfm_x, mirror_start, mirror_end, steps = 20, relative=False) + scans.line_scan(dev.vfm_x, mirror_start, mirror_end, steps=20, relative=False) fit_result = fit_history(-1, "xidiode", deriv=True) - centre = fit_result['centre'] - print(f'Yaw = {yaw}, position = {slit_point1}, centre = {centre}') + centre = fit_result["centre"] + print(f"Yaw = {yaw}, position = {slit_point1}, centre = {centre}") point1.append(centre) # Point 2 print(f"Moving slits to position 2") umv(dev.ss_sl_ycen, slit_point2) - scans.line_scan(dev.vfm_x, mirror_start, mirror_end, steps = 20, relative=False) + scans.line_scan(dev.vfm_x, mirror_start, mirror_end, steps=20, relative=False) fit_result = fit_history(-1, "xidiode", deriv=True) - centre = fit_result['centre'] - print(f'Yaw = {yaw}, position = {slit_point2}, centre = {centre}') + centre = fit_result["centre"] + print(f"Yaw = {yaw}, position = {slit_point2}, centre = {centre}") point2.append(centre) print(f"Point 1 values are {point1}") print(f"Point 2 values are {point2}") @@ -1546,7 +1398,6 @@ def vfm_yaw_check(): y_point1 = np.array(point1) y_point2 = np.array(point2) - # # Fit linear models: y = m*x + b m_point1, b_point1 = np.polyfit(yaw_values, y_point1, 1) m_point2, b_point2 = np.polyfit(yaw_values, y_point2, 1) @@ -1560,24 +1411,25 @@ def vfm_yaw_check(): print(f" Y = {y_intersect:.2f}") # # --- Plot --- - plt.figure(figsize=(7,5)) - plt.scatter(yaw_values, y_point1, color='red', label=f'point1') - plt.scatter(yaw_values, y_point2, color='blue', label=f'point2') + plt.figure(figsize=(7, 5)) + plt.scatter(yaw_values, y_point1, color="red", label=f"point1") + plt.scatter(yaw_values, y_point2, color="blue", label=f"point2") x_fit = np.linspace(min(yaw_values), max(yaw_values), 100) - plt.plot(x_fit, m_point1*x_fit + b_point1, 'r--', label=f'point1') - plt.plot(x_fit, m_point2*x_fit + b_point2, 'b--', label=f'point2') + plt.plot(x_fit, m_point1 * x_fit + b_point1, "r--", label=f"point1") + plt.plot(x_fit, m_point2 * x_fit + b_point2, "b--", label=f"point2") - plt.scatter(x_intersect, y_intersect, color='green', s=80, zorder=5, label='Intersection') + plt.scatter(x_intersect, y_intersect, color="green", s=80, zorder=5, label="Intersection") - plt.xlabel('VFM yaw') - plt.ylabel('Edge') - plt.title('VFM Yaw Calibration') + plt.xlabel("VFM yaw") + plt.ylabel("Edge") + plt.title("VFM Yaw Calibration") plt.legend() plt.grid(True) plt.tight_layout() plt.show() + from dataclasses import dataclass import time @@ -1585,9 +1437,9 @@ import time @dataclass class SlitAxis: name: str - centre_motor: object # e.g. dev.slit1_x - size_motor: object # e.g. dev.slit1_xsize - scan_func: callable # your existing peak scan function + centre_motor: object # e.g. dev.slit1_x + size_motor: object # e.g. dev.slit1_xsize + scan_func: callable # your existing peak scan function scan_range: float = 2.0 open_size: float = 2.0 min_size: float = -0.4 @@ -1609,7 +1461,6 @@ def read_intensity(detector, n=5, delay=0.1): return sum(values) / len(values) - def reduce_to_transmission(axis, detector, baseline, target=0.15, step=0.05): """ Close slit size until intensity is around target * baseline. @@ -1621,8 +1472,7 @@ def reduce_to_transmission(axis, detector, baseline, target=0.15, step=0.05): transmission = intensity / baseline print( - f"{axis.name}: size={size:.4f}, " - f"I={intensity:.3g}, transmission={transmission:.3f}" + f"{axis.name}: size={size:.4f}, " f"I={intensity:.3g}, transmission={transmission:.3f}" ) if transmission <= target: @@ -1645,8 +1495,7 @@ def open_to_transmission(axis, detector, baseline, target=0.99, step=0.05): transmission = intensity / baseline print( - f"{axis.name}: size={size:.4f}, " - f"I={intensity:.3g}, transmission={transmission:.3f}" + f"{axis.name}: size={size:.4f}, " f"I={intensity:.3g}, transmission={transmission:.3f}" ) if transmission >= target: @@ -1660,6 +1509,7 @@ def open_to_transmission(axis, detector, baseline, target=0.99, step=0.05): raise RuntimeError(f"{axis.name}: reached maximum slit size without restoring intensity") + # def approx_centre_axis(axis, detector, slitsize=0.5): # umv(axis.size_motor, slitsize) # go_to_peak(axis.centre_motor, detector, -4, 4, 50, confirm=False) @@ -1678,25 +1528,19 @@ def centre_axis(axis, detector, close_fraction=0.10, open_fraction=0.95): # if baseline <= 0: # raise RuntimeError(f"{axis.name}: invalid baseline intensity") - reduce_to_transmission( - axis, - detector, - baseline, - target=close_fraction, - ) + reduce_to_transmission(axis, detector, baseline, target=close_fraction) - go_to_peak(axis.centre_motor, detector, axis.centre_motor.position - axis.scan_range, - axis.centre_motor.position + axis.scan_range, - 50, - confirm=False) + go_to_peak( + axis.centre_motor, + detector, + axis.centre_motor.position - axis.scan_range, + axis.centre_motor.position + axis.scan_range, + 50, + confirm=False, + ) print(f"{axis.name}: moving centre to peak position") - open_to_transmission( - axis, - detector, - baseline, - target=open_fraction, - ) + open_to_transmission(axis, detector, baseline, target=open_fraction) final_intensity = read_intensity(detector) print( @@ -1704,6 +1548,7 @@ def centre_axis(axis, detector, close_fraction=0.10, open_fraction=0.95): f"transmission = {final_intensity / baseline:.3f}" ) + def open_slits(size=3): slits = define_slits() for slit in slits: @@ -1728,12 +1573,14 @@ def centre_all_slits(slits, detector): print(f"WARNING: failed to align {slit.name}: {exc}") print("Continuing with next slit.") + def centre_one(name, detector): slits = define_slits() for slit in slits: if slit.name == name: centre_slit(slit, detector) + def define_slits(): slits = [ Slit( @@ -1781,9 +1628,10 @@ def define_slits(): scan_func=go_to_peak, ), ), - ] + ] return slits + def get_mirror_data(history_index: int): """Read data from the BEC history and return the X and Y data as arrays.""" motor1 = "vfm_bu" @@ -1793,14 +1641,13 @@ def get_mirror_data(history_index: int): md = scan.metadata["bec"] x1_data = scan.devices[motor1][motor1].read()["value"] x2_data = scan.devices[motor2][motor2].read()["value"] - y1_data = scan.devices['microscope_stats']['microscope_stats_data_sigma_y'].read()['value'] - y2_data = scan.devices['microscope_stats']['microscope_stats_data_eccentricity'].read()['value'] + y1_data = scan.devices["microscope_stats"]["microscope_stats_data_sigma_y"].read()["value"] + y2_data = scan.devices["microscope_stats"]["microscope_stats_data_eccentricity"].read()["value"] np.savetxt("luts/x1_data", x1_data) np.savetxt("luts/x2_data", x2_data) np.savetxt("luts/y1_data", y1_data) np.savetxt("luts/y2_data", y2_data) - # plt.imshow(y1_data, cmap="hot") # plt.clf # return { @@ -1820,13 +1667,12 @@ def record_beampos_until(stop_time_str, interval_minutes=10): # Prepare the file filename = bl.env.lut_dir / "beampos" / f"beampos_{filenow()}.csv" - scam_zoom = dev.scam_zoom.read()['scam_zoom']['value'] + scam_zoom = dev.scam_zoom.read()["scam_zoom"]["value"] umv(dev.scam_zoom, 800) beam_size = "25 x 25" energy = get_current_energy() - # Write the header to the file with open(filename, "w", encoding="utf-8") as f: f.write(f"Beam size: {beam_size}, Energy: {energy:.0f} eV, Zoom: {scam_zoom}\n") @@ -1860,8 +1706,6 @@ def record_beampos_until(stop_time_str, interval_minutes=10): print(f"Finished — reached {end_time:%H:%M}.") bl.planner.move_to(BeamlineState.MANUAL_SAMPLE_EXCHANGE) - - def record_beam(filename): @@ -1873,7 +1717,7 @@ def record_beam(filename): umv(dev.bcu_bpm_x, BPM_X_Calibration.centred_position) umv(dev.bcu_bpm_y, BPM_Y_Calibration.centred_position) - try: + try: # Open shutter and record positions dev.bcu_shutter.put(1) auto_exposure(cam="samcam") @@ -1886,7 +1730,9 @@ def record_beam(filename): xpos_cam = dev.samcam_x.read()["samcam_x"]["value"] ypos_cam = dev.samcam_y.read()["samcam_y"]["value"] xpos_bpm, ypos_bpm = get_bcu_beampos() - print(f"Time is {fnow}, ScinX position is {xpos_cam:.5g}, ScinY position is {ypos_cam:.5g}") + print( + f"Time is {fnow}, ScinX position is {xpos_cam:.5g}, ScinY position is {ypos_cam:.5g}" + ) print(f"BPM X is: {xpos_bpm}, BPM Y is {ypos_bpm}") f.write( # f"{fnow}, {xpos: .5g}, {ypos: .4g}, {xpos_cam: .5g}, {ypos_cam: .4g}\n" @@ -1920,7 +1766,6 @@ def analyse_beampos(file): data["Scin_x"] = (data["Scin_x"] - ref_scinx) * calc_scam_microns(1) data["Scin_y"] = (data["Scin_y"] - ref_sciny) * calc_scam_microns(1) - # Extract Y-position columns (all columns except the first column) y_positions = data.iloc[:, 1:] # Exclude the first column @@ -1940,9 +1785,7 @@ def analyse_beampos(file): # Plot the line and scatter with the chosen color ax.plot(meas_time, y_positions[column], color=color, linewidth=0.5) # Line plot - ax.scatter( - meas_time, y_positions[column], color=color, label=column - ) # Scatter plot + ax.scatter(meas_time, y_positions[column], color=color, label=column) # Scatter plot # Set x-axis tick positions and labels tick_positions = range(0, len(meas_time), 7) # Adjust the step size as needed @@ -1959,14 +1802,14 @@ def analyse_beampos(file): # Show the resulting figure plt.show() + import time import numpy as np import matplotlib.pyplot as plt from epics import caget - -def scan_scin_focus(planner, type = 'fine'): +def scan_scin_focus(planner, type="fine"): dev.bcu_shutter.put(0) planner.move_to(BeamlineState.BEAM_VISUALISATION) dev.bcu_shutter.put(1) @@ -1975,10 +1818,9 @@ def scan_scin_focus(planner, type = 'fine'): min = 0.1 max = 9.9 - - if type == 'fine': + if type == "fine": steps = 100 - elif type == 'coarse': + elif type == "coarse": steps = 30 else: steps = 60 @@ -2007,7 +1849,7 @@ def bpm_errors(ch1, ch2, ch3, ch4): def find_bpm_centre(positions, errors): """Fit the BPM difference signal and return its zero crossing.""" - + slope, intercept = np.polyfit(positions, errors, 1) if abs(slope) < 1e-6: @@ -2017,12 +1859,13 @@ def find_bpm_centre(positions, errors): return centre, slope, intercept + def centre_bcu_bpm_x(): positions = [] errors = [] startx = dev.bcu_bpm_x.position starty = dev.bcu_bpm_y.position - scanpoints_x = np.linspace(startx-0.15, startx+0.15, 20) + scanpoints_x = np.linspace(startx - 0.15, startx + 0.15, 20) for x in scanpoints_x: umv(dev.bcu_bpm_x, x) result = read_currents() @@ -2037,16 +1880,17 @@ def centre_bcu_bpm_x(): positions.append(x) errors.append(x_error) - centre, slope, intercept = find_bpm_centre(positions,errors) + centre, slope, intercept = find_bpm_centre(positions, errors) print(f"BPM centre in X is at {centre} mm") umv(dev.bcu_bpm_x, centre) return centre + def centre_bcu_bpm_y(): positions = [] errors = [] starty = dev.bcu_bpm_y.position - scanpoints_y = np.linspace(starty-0.15, starty+0.15, 20) + scanpoints_y = np.linspace(starty - 0.15, starty + 0.15, 20) for y in scanpoints_y: umv(dev.bcu_bpm_y, y) result = read_currents() @@ -2061,20 +1905,24 @@ def centre_bcu_bpm_y(): positions.append(y) errors.append(y_error) - centre, slope, intercept = find_bpm_centre(positions,errors) + centre, slope, intercept = find_bpm_centre(positions, errors) print(f"BPM centre in Y is at {centre} mm") umv(dev.bcu_bpm_y, centre) return centre + def read_i1(): - return dev.i1.read()['i1']['value'] + return dev.i1.read()["i1"]["value"] + def read_i2(): - return dev.i2.read()['i2']['value'] + return dev.i2.read()["i2"]["value"] + def quick_test(): bl.planner.closest_states() + import time import numpy as np @@ -2097,13 +1945,7 @@ def _smooth_profile(profile, window=7): return np.convolve(profile, kernel, mode="same") -def find_dark_feature( - profile, - search_min, - search_max, - smooth_window=7, - threshold_fraction=0.35, -): +def find_dark_feature(profile, search_min, search_max, smooth_window=7, threshold_fraction=0.35): """ Find the centre and edges of a dark feature in a 1-D intensity profile. @@ -2224,7 +2066,7 @@ def measure_omega_runout( for angle in angles: print(f"\nOmega = {angle:.1f}°") - umv(omega_motor,(float(angle))) + umv(omega_motor, (float(angle))) time.sleep(settle) # ------------------------------------------------------ @@ -2264,11 +2106,9 @@ def measure_omega_runout( result = { "omega": float(angle), - "x": needle_x, "x_width": x_result["width"], "x_depth": x_result["depth"], - "y": needle_y, "y_width": y_result["width"], "y_depth": y_result["depth"], @@ -2276,14 +2116,8 @@ def measure_omega_runout( results.append(result) - print( - f" X = {needle_x:8.2f} px" - f" width = {x_result['width']:6.1f}" - ) - print( - f" Y = {needle_y:8.2f} px" - f" width = {y_result['width']:6.1f}" - ) + print(f" X = {needle_x:8.2f} px" f" width = {x_result['width']:6.1f}") + print(f" Y = {needle_y:8.2f} px" f" width = {y_result['width']:6.1f}") # ---------------------------------------------------------- # Summary @@ -2298,14 +2132,9 @@ def measure_omega_runout( print(f"Y peak-to-peak : {np.ptp(ys):.2f} px") if len(results) > 1: - return_error = np.hypot( - xs[-1] - xs[0], - ys[-1] - ys[0], - ) + return_error = np.hypot(xs[-1] - xs[0], ys[-1] - ys[0]) - print( - f"0° -> 360° error: {return_error:.2f} px" - ) + print(f"0° -> 360° error: {return_error:.2f} px") return results @@ -2315,29 +2144,18 @@ def dcm_pitch_calibration(start_energy=6000, end_energy=31000, step=3000): with open(filename, "w", encoding="utf-8") as f: f.write("energy,pitch\n") for energy in range(start_energy, end_energy, step): - # energy_list = [29000,26000,23000,20000,17000,14000,11000,8000,6000] - # for energy in energy_list: + # energy_list = [29000,26000,23000,20000,17000,14000,11000,8000,6000] + # for energy in energy_list: bl_energy(energy, mono_scan=False) - result = go_to_peak(dev.dcm_pitch, - dev.lu_bpmsum, - -0.1, - 0.4, - 40, - min_fwhm_steps=1, - confirm=False - ) + result = go_to_peak( + dev.dcm_pitch, dev.lu_bpmsum, -0.1, 0.4, 40, min_fwhm_steps=1, confirm=False + ) pitch_new = get_value(dev.dcm_pitch) with open(filename, "a", encoding="utf-8") as f: f.write(f"{energy:.0f},{pitch_new:.5g}\n") - -def scan_roll( - energy, - start, - stop, - steps=9, - settle=0.2, - overwrite=False, -): + + +def scan_roll(energy, start, stop, steps=9, settle=0.2, overwrite=False): """Measure scintillator Y position as a function of DCM fine roll.""" scanpoints = np.linspace(start, stop, steps) @@ -2359,33 +2177,22 @@ def scan_roll( _, ypos = read_scintXY() - print( - f"{energy / 1000:.1f} keV | " - f"roll = {roll:.3f} | " - f"Y = {ypos:.2f}" - ) + print(f"{energy / 1000:.1f} keV | " f"roll = {roll:.3f} | " f"Y = {ypos:.2f}") f.write(f"{roll:.6f},{ypos:.4f}\n") return filename + def analyse_roll(low=6000, high=25000): """Find the DCM fine-roll value where low/high-energy beam positions intersect.""" low_file = bl.env.lut_dir / f"roll_{low:.0f}.csv" high_file = bl.env.lut_dir / f"roll_{high:.0f}.csv" - low_data = np.loadtxt( - low_file, - delimiter=",", - skiprows=1, - ) + low_data = np.loadtxt(low_file, delimiter=",", skiprows=1) - high_data = np.loadtxt( - high_file, - delimiter=",", - skiprows=1, - ) + high_data = np.loadtxt(high_file, delimiter=",", skiprows=1) low_roll = low_data[:, 0] low_y = low_data[:, 1] @@ -2408,44 +2215,20 @@ def analyse_roll(low=6000, high=25000): # Plot plt.figure(figsize=(7, 5)) - plt.scatter( - low_roll, - low_y, - label=f"{low / 1000:.1f} keV", - ) + plt.scatter(low_roll, low_y, label=f"{low / 1000:.1f} keV") - plt.scatter( - high_roll, - high_y, - label=f"{high / 1000:.1f} keV", - ) + plt.scatter(high_roll, high_y, label=f"{high / 1000:.1f} keV") x_min = min(low_roll.min(), high_roll.min()) x_max = max(low_roll.max(), high_roll.max()) x_fit = np.linspace(x_min, x_max, 200) - plt.plot( - x_fit, - m_low * x_fit + b_low, - "--", - label=f"{low / 1000:.1f} keV fit", - ) + plt.plot(x_fit, m_low * x_fit + b_low, "--", label=f"{low / 1000:.1f} keV fit") - plt.plot( - x_fit, - m_high * x_fit + b_high, - "--", - label=f"{high / 1000:.1f} keV fit", - ) + plt.plot(x_fit, m_high * x_fit + b_high, "--", label=f"{high / 1000:.1f} keV fit") - plt.scatter( - x_intersect, - y_intersect, - s=80, - zorder=5, - label="Intersection", - ) + plt.scatter(x_intersect, y_intersect, s=80, zorder=5, label="Intersection") plt.xlabel("DCM fine roll") plt.ylabel("samcam Y pos") @@ -2457,34 +2240,152 @@ def analyse_roll(low=6000, high=25000): return x_intersect -def find_best_roll( - low=6000, - high=25000, - low_range=(4.0, 7.0), - high_range=(1.0, 8.0), - steps=9, -): + +def find_best_roll(low=6000, high=25000, low_range=(4.0, 7.0), high_range=(1.0, 8.0), steps=9): """Collect missing roll scans and calculate optimum roll.""" low_file = bl.env.lut_dir / f"roll_{low:.0f}.csv" high_file = bl.env.lut_dir / f"roll_{high:.0f}.csv" if not low_file.exists(): - scan_roll( - low, - *low_range, - steps=steps, - ) + scan_roll(low, *low_range, steps=steps) else: print(f"Using existing {low_file.name}") if not high_file.exists(): - scan_roll( - high, - *high_range, - steps=steps, - ) + scan_roll(high, *high_range, steps=steps) else: print(f"Using existing {high_file.name}") - return analyse_roll(low, high) \ No newline at end of file + return analyse_roll(low, high) + + +def test_hfm(): + target = current_hfm + correction + + print(f"HFM requested: {current_hfm:.6f} -> {target:.6f}") + umv(dev.hfm_xd, target) + + actual = get_value(dev.hfm_xd) + print( + f"HFM actual: {actual:.6f} " + f"(requested move {(target-current_hfm)*1000:+.3f} µm, " + f"actual move {(actual-current_hfm)*1000:+.3f} µm)" + ) + + +def test_hfm2(dmove, offset): + current = get_value(dev.hfm_xd) + target = current + dmove + print( + f"Current HFM = {current:.4f}, target position = {target:.4f}, difference = {target-current:.4f}" + ) + # Step 1, add the offset to dmove and add to current position + move_to_1 = current + dmove + offset + umv(dev.hfm_xd, move_to_1) + print(f"Moving to {move_to_1:.4f}, difference of {current-move_to_1:.4f}") + # Step 2, move back by the offset amount + move_to_2 = move_to_1 - offset + umv(dev.hfm_xd, move_to_2) + print(f"Moving to {move_to_2:.4f}, difference of {current-move_to_2:.4f}") + # Print actual move + new = get_value(dev.hfm_xd) + print(f"Requested move was {dmove:.4f}, actual move was {new-current:.4f}") + + +def test_hfm3(dmove, offset=0.002): + current = get_value(dev.hfm_xd) + target = current + dmove + + # Overshoot in the same direction as the requested move + signed_offset = np.sign(dmove) * abs(offset) + overshoot = target + signed_offset + + print( + f"Current HFM = {current:.6f}\n" + f"Target HFM = {target:.6f}\n" + f"Requested move = {dmove*1000:+.3f} µm" + ) + + umv(dev.hfm_xd, overshoot) + + after_overshoot = get_value(dev.hfm_xd) + print(f"Overshoot requested = {overshoot:.6f}, " f"actual = {after_overshoot:.6f}") + + umv(dev.hfm_xd, target) + + final = get_value(dev.hfm_xd) + + print( + f"Final requested = {target:.6f}, actual = {final:.6f}\n" + f"Requested net move = {dmove*1000:+.3f} µm\n" + f"Actual net move = {(final-current)*1000:+.3f} µm\n" + f"Position error = {(final-target)*1000:+.3f} µm" + ) + + +def test_vfm3(dmove, offset=0.002): + current = get_value(dev.vfm_ydr) + target = current + dmove + + direction = np.sign(dmove) + overshoot = target + direction * abs(offset) + + print( + f"Current VFM = {current:.6f}\n" + f"Target VFM = {target:.6f}\n" + f"Requested move = {dmove*1000:+.3f} µm" + ) + + umv(dev.vfm_ydr, overshoot) + + after_overshoot = get_value(dev.vfm_ydr) + print(f"Overshoot requested = {overshoot:.6f}, " f"actual = {after_overshoot:.6f}") + + umv(dev.vfm_ydr, target) + + final = get_value(dev.vfm_ydr) + + print( + f"Final requested = {target:.6f}, actual = {final:.6f}\n" + f"Requested net move = {dmove*1000:+.3f} µm\n" + f"Actual net move = {(final-current)*1000:+.3f} µm\n" + f"Position error = {(final-target)*1000:+.3f} µm" + ) + +def beamstop_check_energies(): + + bs_in = 0.904 + bs_out = 0 + + start_energy = 30000 + end_energy = 5000 + + filename = bl.env.lut_dir / f"bs_checks_{filenow(date_only=True)}.csv" + with open(filename, "w", encoding="utf-8") as f: + f.write("Energy,I2_in,I2_out\n") + + try: + for energy in range (start_energy, end_energy, -2000): + print(f"Moving to energy {energy}") + bl_energy(energy) + dev.transm.put(0.1) + trans = get_value(dev.transm) + print(f"Transmission is {trans}") + dev.bcu_shutter.put(1) + time.sleep(1) + invalue = get_value(dev.i2) + print(f"Reading with beamstop in position is {invalue:.4g}") + # Check reading with beamstop out + umv(dev.bs_y, bs_out) + time.sleep(1) + outvalue = get_value(dev.i2) + dev.bcu_shutter.put(0) + print(f"Reading with beamstop out is {outvalue:.4g}") + umv(dev.bs_y, bs_in) + with open(filename, "a", encoding="utf-8") as f: + f.write(f"{energy:.0f},{invalue:.4g}, {outvalue:.4g}\n") + finally: + dev.bcu_shutter.put(0) + umv(dev.bs_y, bs_in) + bl_energy(12400)