bec/pxii_bec
0
1 Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

updates, bug fixes etc

Browse Source
This commit is contained in:
x10sa committed 2026-10-08 12:09:16 +02:00
1 parent 1cd9a87aa6
commit d4fafdc178
23 files changed
+6208 -1487
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pxii_bec/device_configs/pxii-devices.yaml
+51 -5
View File
@@ -55,8 +55,8 @@ aerotech:
deviceConfig:
prefix: http://mx-x10sa-queue-01:5234
deviceTags:
- motors
- aerotech
- motors
enabled: true
onFailure: buffer
readoutPriority: baseline
@@ -65,15 +65,15 @@ aerotech:
mount: 0
work: 0
x:
in: -8.1e-06
in: 0.05997465
out: -10
safe: -50
y:
mount: -0.0
work: -0.02600108
work: -0.02999903
z:
mount: 0
work: -0.00900383
work: -0.00900401
bcu_bpm1:
description: BCU BPM Channel 1
deviceClass: ophyd.EpicsSignalRO
@@ -330,7 +330,7 @@ bs_z:
max_blout: 70
min: 13
safe: 45.1
samp: 16.408
samp: 16.80557617
type: continuous
work_min: 20
bsf_f1_y:
@@ -1322,6 +1322,40 @@ samcam_x:
onFailure: buffer
readOnly: true
readoutPriority: monitored
samcam_xcurs:
description: Sample Camera X Cursor
deviceClass: ophyd.EpicsSignal
deviceConfig:
auto_monitor: true
read_pv: X10SA-ES-MS:Stats5:CursorX
deviceTags:
- scam
enabled: true
onFailure: buffer
readoutPriority: baseline
samcam_thresh:
description: Sample Camera Threshold
deviceClass: ophyd.EpicsSignal
deviceConfig:
auto_monitor: true
read_pv: X10SA-ES-MS:Stats5:CentroidThreshold
deviceTags:
- scam
enabled: true
onFailure: buffer
readoutPriority: baseline
samcam_xcursprof:
description: Sample Camera X Cursor Profile
deviceClass: ophyd.EpicsSignalRO
deviceConfig:
auto_monitor: true
read_pv: X10SA-ES-MS:Stats5:ProfileCursorX_RBV
deviceTags:
- scam
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: monitored
samcam_xsig:
description: Sample Camera X Sigma
deviceClass: ophyd.EpicsSignalRO
@@ -1346,6 +1380,18 @@ samcam_y:
onFailure: buffer
readOnly: true
readoutPriority: monitored
samcam_ycursprof:
description: Sample Camera Y Cursor Profile
deviceClass: ophyd.EpicsSignalRO
deviceConfig:
auto_monitor: true
read_pv: X10SA-ES-MS:Stats5:ProfileCursorY_RBV
deviceTags:
- scam
enabled: true
onFailure: buffer
readOnly: true
readoutPriority: monitored
samcam_ysig:
description: Sample Camera Y Sigma
deviceClass: ophyd.EpicsSignalRO
pxii_bec/macros/APscripts.py
+292 -256
View File
File diff suppressed because it is too large. Load diff
pxii_bec/macros/Undu_Helpers.py
+157 -41
View File
@@ -45,8 +45,42 @@ def exponential_func2(x, a, b, c, d):
##################################
# ##### Sept 2026 Data ######################################
def return_harmon():
# from 3rd to 19th Energy range from 6 keV to 30 keV
# harmonics, Sept 2026
import numpy as np
nharm = 19
h = [np.array([]) for _ in range(nharm + 1)]
h[0] = np.array([0])
h[1] = np.array([0])
h[2] = np.array([0])
h[3] = np.array([0.095, -0.276, 4.154]) # 3rd harmon
h[4] = np.array([0.00692593, -0.131504, 1.45763, -0.864603])
h[5] = np.array([0.00269444 , - 0.0616905 , 0.967925, - 0.295405])
h[6] = np.array([ 0.00248148 , - 0.0767908 , 1.2178 , - 1.91516 ])
h[7] = np.array([0.00110017 , - 0.0369968 , 0.77334 ,- 0.688198 ])
h[8] = np.array([0.000643939 , - 0.0226645 , 0.570453 , - 0.0479048 ])
h[9] = np.array([0.00021633 , -0.00619517 , 0.315392 , 0.971667 ])
h[10] = np.array([ 0.000708236 , - 0.0357576 , 0.851686 - 2.48765 ])
h[11] = np.array([ 0.000589549 , -0.0328631 , 0.837705 , - 2.90326 ])
h[12] = np.array([ 0.00218182 , 0.1212 , 1.74739 ])
h[13] = np.array([ 0.197702, 0.757869 ])
h[14] = np.array([ 0.183733, 0.762778 ])
h[15] = np.array([ 0.170381, 0.783155 ])
h[16] = np.array([ 0.157321, 0.850464 ])
h[17] = np.array([0.1466, 0.8906 ])
h[18] = np.array([0.1362, 0.9538 ])
h[19] = np.array([0.128, 0.985 ])
fitpar_u19 = np.array([ 2.22322523, 0.50396248, -0.00783873]) # measured from U19
return h
# ##### July 2025 Data ######################################
def return_harmon_old():
# from 3rd to 21th Energy range from 6 keV to 30 keV
# harmonics, July 2025
import numpy as np
@@ -97,7 +131,7 @@ def plot_harmon(e_start, e_end, h_no, pr_out=False):
plt.show()
def setu19(en, *harm_no, detune=0):
def setu19(en, *harm_no, detune=0, simul = 0, iexp = 0):
"""
set the U19 to the gaps defined in Jul2025, or the "theoretical" ones for higher
harmonics
@@ -106,10 +140,12 @@ def setu19(en, *harm_no, detune=0):
en in keV, possibly select a special harmonics, or detune [0/1] to a value
with a nicer beam shape but less flux
"""
g0 = dev.id_gap.readback.get()
fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255]) # measured from U19
if not simul:
g0 = dev.id_gap.readback.get()
#fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255]) # measured from U19
fitpar_u19 = np.array([ 2.22322523, 0.50396248, -0.00783873]) # measured from U19, Sept2026
h_all = return_harmon()
# dep on energy -> select harmonics
if harm_no:
@@ -120,18 +156,26 @@ def setu19(en, *harm_no, detune=0):
print("Using the default optimised harmonics")
if en <= 7:
harm = 3 # h = h[3]
elif 7 < en <= 10:
elif 7 < en <= 10.5:
harm = 5
elif 10 < en <= 13:
elif 10.5 < en < 13:
harm = 7
elif 13 < en <= 16:
elif 13 <= en <= 16.5:
harm = 9
elif 16 < en <= 19:
elif 16.5 < en <= 19.5:
harm = 11
elif 19 < en <= 22:
elif 19.5 < en <= 22.5:
harm = 13
elif 22.5 < en <= 25.5: ## 22
harm = 15
elif 25.5 < en <= 28.5: ## 22
harm = 17
elif 28.5 < en <= 30: ## 22
harm = 19
if en <= 22:
if (en <= 30 and harm <=19):
h = h_all[harm]
print(f"harmonics is {harm}.")
@@ -146,32 +190,33 @@ def setu19(en, *harm_no, detune=0):
# 29: 15, 17, 19
# 30: 19
##
if 23 < en <= 25:
n = 15
if (iexp and harm_no):
harm_selected = int(harm_no[0])
print("Selected harmonics is:", harm_selected)
n= harm_selected
enorm = en / n
g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
elif 25 < en <= 29:
n = 17
enorm = en / n
g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
elif 29 < en:
n = 19
elif (iexp):
n = harm # default for that energy range, as set before
enorm = en / n
print('exp fct fit used')
g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
print("Undulator has been ", g0, " mm")
if 4.5 <= g <= 20:
print("Moving Undulator gap to ", g, " mm")
else:
print("not a valid gap, do nothing")
if not simul:
print("Undulator has been ", g0, " mm")
if 4.5 <= g <= 20:
print("Moving Undulator gap to ", g, " mm")
else:
print("not a valid gap, do nothing")
if detune:
g = g * 0.996
print("moving to detuned gap value, slightly below max, about 0.15 % ")
if detune:
g = g * 0.996
print("moving to detuned gap value, slightly below max, about 0.15 % ")
# print("move disabled!!")
res = scans.umv(dev.id_gap, g, relative=False)
scans.umv(dev.id_gap, g, relative=False)
return
return round(g,4)
##################################
@@ -192,7 +237,8 @@ def harmon_walk(estart=7.5, end_en=13):
##################################
def gap_harm(e=12.4):
fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255])
# old values fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255])
fitpar_u19 = np.array([2.22322523, 0.50396248, -0.00783873])
fitpar_u17 = np.array([2.17078572, 0.46477267, -0.005766])
e = float(input("Please enter an energy between 6 and 30: "))
@@ -305,9 +351,26 @@ def long_gscan(estart=6, end_en=30.5, g_low=4.5, g_high=9.0, nsteps=1000, sett =
return
def plottest(wname):
#wname="GapScan"
existing_docks = bec.gui.windows
if existing_docks.get(wname) is None:
dock_area = bec.gui.new(name=wname, geometry = [4000, 900, 700, 700])
wr = dock_area.new(bec.gui.available_widgets.Waveform, object_name="Plot")
mot = dev.id_gap
det = dev.lu_bpmsum
else:
dock_area = open_docks[wname]
wr= dock_area.Plot
wr.plot(device_x=mot, device_y=det) ## names first !
# wr.plot(x=mot.name,y=det.name) ### this comes later
wr.x_label = mot.name
wr.y_label = det.name
##################################
def gscan(centre=0, gomax=0, detune=0,sett=1):
def gscan(deltag = 0.05,centre=0, gomax=0, detune=0,sett=1, silent=0, iplot=1, isave=0):
"""
Scan the ID GAP and go
to the max of lu_bpm intensity
@@ -319,34 +382,58 @@ def gscan(centre=0, gomax=0, detune=0,sett=1):
"""
import time
dock_area = bec.gui.new(name="GapScan", geometry = [4000, 900, 700, 700])
wr = dock_area.new(bec.gui.available_widgets.Waveform)
wname="GapScan"
mot = dev.id_gap
det = dev.lu_bpmsum
det = dev.lu_bpmsum
existing_docks = bec.gui.windows
if existing_docks.get(wname) is None:
dock_area = bec.gui.new(name=wname, geometry = [4000, 900, 700, 700])
wr = dock_area.new(bec.gui.available_widgets.Waveform, object_name="Plot")
else:
dock_area = existing_docks[wname]
wr= dock_area.Plot
wr.clear_all()
wr.plot(device_x=mot, device_y=det) ## names first !
# wr.plot(x=mot.name,y=det.name) ### this comes later
wr.x_label = mot.name
wr.y_label = det.name
g0 = dev.id_gap.readback.get()
deltag = 0.05
#deltag = 0.05
steps = 30
resol = 2*deltag/steps
if resol < 0.002:
resol = 0.003
steps = round(2*deltag/resol)
start=-deltag
stop = deltag
print(f"Gap resol is {resol} mm ")
ds = scans.line_scan(dev.id_gap, -deltag, deltag, steps=steps, settling_time=sett,exp_time=0.5, relative=True)
if (g0-deltag < 4.5):
start=g0-4.5
ds = scans.line_scan(dev.id_gap, start, stop, steps=steps, settling_time=sett,exp_time=0.1, relative=True)
gap_data = ds.scan.live_data.id_gap.id_gap.val
bpm_data = ds.scan.live_data.lu_bpmsum.lu_bpmsum.val
gap_data=np.array(gap_data)
bpm_data = np.array(bpm_data)
if isave:
en = round(getenergy())
dirname = "/sls/x10sa/config/commissioning/Data/GapsFE_0p3/"
combined = np.column_stack((gap_data, bpm_data))
filename = dirname + "gap" + str(en)+"_"+str(g0*100)[0:3]+ ".txt"
with open(filename, "w") as f:
np.savetxt(f, combined, delimiter=",", fmt="%5f")
maxy = max(bpm_data)
indmax = np.argmax(bpm_data)
gm = gap_data[indmax]
gmax = gap_data[indmax]
gcen, xm = fit_plot(gap_data, bpm_data, model="voigt")
gcen, xm = fit_plot(gap_data, bpm_data, model="voigt", silent=silent, iplot=iplot)
if gomax:
gm = xm
@@ -354,6 +441,8 @@ def gscan(centre=0, gomax=0, detune=0,sett=1):
gm = gcen
print("gap off by ", g0 - gm, " mm")
print(f"gcen = {gcen}, gmax = {xm} or {gmax}")
if detune:
gm = gm * 0.996
print("moving to detuned gap value, slightly (0.15 %) below max")
@@ -366,4 +455,31 @@ def gscan(centre=0, gomax=0, detune=0,sett=1):
else:
print("Fit too far off, try using option gomax=1")
return
return gm, xm
##################################
def bfield_x10sa(gap):
if gap < 4.5 or gap > 20:
print("no valid gap!")
return
### values for the U19
l_u = 19
a = 3.22482942
b = -0.30100054
c = 0.00808649
b0 = exponential_func0(gap, a,b,c )
b0= round(b0,4)
# K = e/mc * l_u/2pi * B
k = round(0.0934 *l_u * b0,2)
print(f'### X10SA and U19 ### ')
print(f'for a gap of {gap} mm : ')
print(f'B0 = {b0}')
print(f'K = {k}')
return b0, k
##################################
pxii_bec/macros/beamline_planner.py
+8 -24
View File
@@ -2,6 +2,7 @@
from collections import defaultdict, deque
from enum import Enum
# from pxii_bec.scripts import beamline_context as bl
# from enums import BeamlineState, TemperatureMode
@@ -74,7 +75,6 @@ class StateChangePlanner:
state_name = BeamlineState(state_name)
target = self._merged_state(state_name, modifier)
plan = self._plan(target)
print("PLAN:")
@@ -102,7 +102,7 @@ class StateChangePlanner:
def available_states(self):
"""Return a list of available states"""
return list(self.states.keys())
def state_devices(self):
"""Return the set of devices that participate in
beamline state definitions"""
@@ -111,14 +111,13 @@ class StateChangePlanner:
for state in self.states.values():
devices.update(state.keys())
return sorted(devices)
def get_positions(self):
"""Return current positions of devices that are used
in state transitions"""
return {name: self.devices[name].pos
for name in self.state_devices()}
return {name: self.devices[name].pos for name in self.state_devices()}
def print_positions(self):
"""Return current state of all devices"""
@@ -131,7 +130,7 @@ class StateChangePlanner:
"""Return a dict of {device: (before, after)} pairs for devices that changed state"""
after = self.get_positions()
return {k: (before[k], after[k]) for k in before if before[k] != after[k]}
def __get_closest_states(self):
"""
Return states ordered by number of mismatches.
@@ -148,21 +147,9 @@ class StateChangePlanner:
for dev, expected in config.items():
if not self.devices[dev].is_at(expected):
mismatches.append(
(
dev,
self.devices[dev].pos,
expected,
)
)
mismatches.append((dev, self.devices[dev].pos, expected))
results.append(
(
state.name,
len(mismatches),
mismatches,
)
)
results.append((state.name, len(mismatches), mismatches))
return sorted(results, key=lambda x: x[1])
@@ -173,10 +160,7 @@ class StateChangePlanner:
print(f"\n{state}: {count} mismatch(es)")
for dev, actual, expected in mismatches:
print(
f" {dev:15s} "
f"{actual:10s} -> {expected}"
)
print(f" {dev:15s} " f"{actual:10s} -> {expected}")
def current_state(self):
"""Return all current matching BeamlineState and TemperatureMode combinations,
pxii_bec/macros/build_devices.py
+2 -15
View File
@@ -63,14 +63,7 @@ def build_devices(yaml_file):
return state_devices
def update_position(
device,
position,
value,
axis=None,
force=False,
max_delta=0.5,
):
def update_position(device, position, value, axis=None, force=False, max_delta=0.5):
"""
Update a single saved position in memory.
@@ -189,13 +182,7 @@ def update_positions(updates):
return save_and_reload()
def save_current_position(
device,
position,
axis=None,
force=False,
max_delta=0.5,
):
def save_current_position(device, position, axis=None, force=False, max_delta=0.5):
"""
Save the current motor position as a named position.
"""
pxii_bec/macros/calculator.py
+9 -21
View File
@@ -2,6 +2,7 @@
from dataclasses import dataclass
import numpy as np
# from pxii_parameters import (EnergyDefaults, CamConversion)
@@ -60,9 +61,7 @@ def convert_to_degrees(angle_mrad: float) -> float:
return np.rad2deg(angle_mrad / 1000)
def create_conversion_result(
energy_ev: float, wavelength: float, bragg_angle_mrad: float
) -> dict:
def create_conversion_result(energy_ev: float, wavelength: float, bragg_angle_mrad: float) -> dict:
"""
Creates a dictionary containing converted values of energy and angles.
@@ -226,9 +225,7 @@ def validate_energy(energy_ev):
return energy_ev
def convert_from_bragg(
bragg_angle_mrad: float, temp=120, print_result: bool = False
) -> dict:
def convert_from_bragg(bragg_angle_mrad: float, temp=120, print_result: bool = False) -> dict:
"""
Convert the Bragg angle to wavelength and energy, returning the result as a dictionary.
@@ -274,9 +271,7 @@ def convert_from_energy(energy_ev: float, temp=120, print_result: bool = False)
"""
energy_ev = validate_energy(energy_ev)
wavelength = calculate_wavelength_from_energy(energy_ev)
bragg_angle_mrad = float(
calculate_bragg_angle_from_wavelength(wavelength, temp=temp)
)
bragg_angle_mrad = float(calculate_bragg_angle_from_wavelength(wavelength, temp=temp))
result = create_conversion_result(energy_ev, wavelength, bragg_angle_mrad)
if print_result:
print_conversion_result(result)
@@ -284,9 +279,7 @@ def convert_from_energy(energy_ev: float, temp=120, print_result: bool = False)
def convert_from_wavelength(
wavelength: float,
temp: float = 120,
print_result: bool = False,
wavelength: float, temp: float = 120, print_result: bool = False
) -> dict:
"""
Convert a given wavelength value into corresponding energy, Bragg angle, and
@@ -312,19 +305,14 @@ def convert_from_wavelength(
outside of the permitted range, returns None.
"""
energy_ev = calculate_energy_from_wavelength(wavelength)
bragg_angle_mrad = float(
calculate_bragg_angle_from_wavelength(wavelength, temp=temp)
)
bragg_angle_mrad = float(calculate_bragg_angle_from_wavelength(wavelength, temp=temp))
result = create_conversion_result(energy_ev, wavelength, bragg_angle_mrad)
if print_result:
print_conversion_result(result)
return result
def calc_perp_position(
energy_ev: float,
print_result: bool = False,
) -> float:
def calc_perp_position(energy_ev: float, print_result: bool = False) -> float:
"""
Calculate the perpendicular motor position based on provided energy in electron-volts (eV).
@@ -355,7 +343,7 @@ def calc_scam_microns(pixels, zoom=1000):
"""Convert pixels to microns for the sample camera"""
return float(pixels / (CamConversion.a * np.exp(CamConversion.b * zoom)))
def calc_sscam_microns(pixels):
"""Convert pixels to microns for the ss camera"""
return pixels*20
return pixels * 20
pxii_bec/macros/calibrate_bpm.py
+167 -47
View File
@@ -8,6 +8,8 @@ import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pxii_bec.scripts import beamline_context as bl
from pxii_bec.macros.beamline_planner import BeamlineState
# def compute_norm(n_readings=5):
# """Average normalized BPM X/Y from repeated sets of four channels."""
@@ -35,7 +37,161 @@ import matplotlib.pyplot as plt
# time.sleep(0.1)
# return float(np.mean(xn_values)), float(np.mean(yn_values))
def compute_norm_ss(n_readings=5):
"""Average normalized BPM X/Y from repeated sets of four channels."""
xn_values = []
yn_values = []
for _ in range(n_readings):
bpm1 = dev.ss_bpm1.read()["ss_bpm1"]["value"]
bpm2 = dev.ss_bpm2.read()["ss_bpm2"]["value"]
bpm3 = dev.ss_bpm3.read()["ss_bpm3"]["value"]
bpm4 = dev.ss_bpm4.read()["ss_bpm4"]["value"]
total = bpm1 + bpm2 + bpm3 + bpm4
if total == 0:
continue
xn = ((bpm1 + bpm2) - (bpm3 + bpm4)) / total
yn = ((bpm1 + bpm3) - (bpm2 + bpm4)) / total
xn_values.append(xn)
yn_values.append(yn)
time.sleep(0.1)
return float(np.mean(xn_values)), float(np.mean(yn_values))
def compute_norm_lu(n_readings=5):
"""Average normalized BPM X/Y from repeated sets of four channels."""
xn_values = []
yn_values = []
for _ in range(n_readings):
bpm1 = dev.lu_bpm1.read()["lu_bpm1"]["value"]
bpm2 = dev.lu_bpm2.read()["lu_bpm2"]["value"]
bpm3 = dev.lu_bpm3.read()["lu_bpm3"]["value"]
bpm4 = dev.lu_bpm4.read()["lu_bpm4"]["value"]
total = bpm1 + bpm2 + bpm3 + bpm4
if total == 0:
continue
xn = ((bpm1 + bpm2) - (bpm3 + bpm4)) / total
yn = ((bpm1 + bpm3) - (bpm2 + bpm4)) / total
xn_values.append(xn)
yn_values.append(yn)
time.sleep(0.1)
return float(np.mean(xn_values)), float(np.mean(yn_values))
def run_calibration_ss():
"""Scan bpm in x and y and record normalised x and y positions"""
centred_x = 2.8625
centred_y = -1.3874
now = datetime.now()
fnow = filenow()
print(f"moving bpm to {centred_x}, {centred_y}")
umv(dev.ss_bpm_y, centred_y)
umv(dev.ss_bpm_x, centred_x)
# # # Calibrate in X
# umv(dev.bsf_sl_xsize, 0.3)
# umv(dev.bsf_sl_xsize, 0.4)
# filename = bl.env.lut_dir / "bpm_calib" / f"ss_bpm_x_{fnow}.csv"
# xn_data = []
# x_positions = np.linspace(centred_x - 0.3, centred_x + 0.3, 51)
# with open(filename, "w", encoding="utf-8") as f:
# f.write(f"Xpos,Xnorm,BPM1,BPM2,BPM3,BPM4\n")
# for x in x_positions:
# umv(dev.ss_bpm_x, x)
# print(f"Moving BPM X to {x} mm")
# time.sleep(0.2)
# xn, yn = compute_norm_ss()
# xn_data.append(xn)
# with open(filename, "a", encoding="utf-8") as f:
# f.write(f"{x},{xn}\n")
# # Calibrate in Y
umv(dev.ss_bpm_x, centred_x)
umv(dev.bsf_sl_xsize, 0.3)
umv(dev.bsf_sl_ysize, 0.4)
filename = bl.env.lut_dir / "bpm_calib" / f"ss_bpm_y_{fnow}.csv"
y_positions = np.linspace(centred_y - 0.04, centred_y + 0.04, 41)
yn_data = []
with open(filename, "w", encoding="utf-8") as f:
f.write(f"Ypos,Ynorm,BPM1,BPM2,BPM3,BPM4\n")
for y in y_positions:
umv(dev.ss_bpm_y, y)
print(f"Moving BPM Y to {y} mm")
time.sleep(0.2)
xn, yn = compute_norm_ss()
yn_data.append(yn)
with open(filename, "a", encoding="utf-8") as f:
f.write(f"{y},{yn}\n")
umv(dev.ss_bpm_x, centred_x)
umv(dev.ss_bpm_y, centred_y)
umv(dev.bsf_sl_ysize, 3.0)
umv(dev.bsf_sl_xsize, 3.0)
def run_calibration_lu():
"""Scan bpm in x and y and record normalised x and y positions"""
centred_x = 2.1816
centred_y = 3.0175
now = datetime.now()
fnow = filenow()
print(f"moving bpm to {centred_x}, {centred_y}")
umv(dev.lu_bpm_y, centred_y)
umv(dev.lu_bpm_x, centred_x)
# # Calibrate in X
# umv(dev.bsf_sl_xsize, 0.3)
# umv(dev.bsf_sl_xsize, 0.4)
# filename = bl.env.lut_dir / "bpm_calib" / f"lu_bpm_x_{fnow}.csv"
# xn_data = []
# x_positions = np.linspace(centred_x - 0.3, centred_x + 0.3, 51)
# with open(filename, "w", encoding="utf-8") as f:
# f.write(f"Xpos,Xnorm\n")
# for x in x_positions:
# umv(dev.lu_bpm_x, x)
# print(f"Moving BPM X to {x} mm")
# time.sleep(0.2)
# xn, yn = compute_norm_lu()
# xn_data.append(xn)
# with open(filename, "a", encoding="utf-8") as f:
# f.write(f"{x},{xn}\n")
# # Calibrate in Y
umv(dev.lu_bpm_x, centred_x)
umv(dev.bsf_sl_xsize, 0.3)
umv(dev.bsf_sl_ysize, 0.4)
filename = bl.env.lut_dir / "bpm_calib" / f"lu_bpm_y_{fnow}.csv"
y_positions = np.linspace(centred_y - 0.04, centred_y + 0.04, 41)
yn_data = []
with open(filename, "w", encoding="utf-8") as f:
f.write(f"Ypos,Ynorm,BPM1,BPM2,BPM3,BPM4\n")
for y in y_positions:
umv(dev.lu_bpm_y, y)
print(f"Moving BPM Y to {y} mm")
time.sleep(0.2)
xn, yn = compute_norm_lu()
yn_data.append(yn)
with open(filename, "a", encoding="utf-8") as f:
f.write(f"{y},{yn}\n")
umv(dev.lu_bpm_x, centred_x)
umv(dev.lu_bpm_y, centred_y)
umv(dev.bsf_sl_ysize, 3.0)
umv(dev.bsf_sl_xsize, 3.0)
def run_calibration(env):
"""Scan bpm in x and y and record normalised x and y positions"""
@@ -86,27 +242,20 @@ def run_calibration(env):
umv(dev.bcu_bpm_y, centred_y)
umv(dev.ss_sl_ysize, 3.0)
def read_bpm_calibration_data(
env,
x_y,
rundate,
bpm_name="bcu_bpm",
):
def read_bpm_calibration_data(env, x_y, rundate, bpm_name="bcu_bpm"):
"""Read BPM calibration data from CSV."""
filename = env.lut_dir / f"bpm_calib/{bpm_name}_{x_y}_{rundate}.csv"
data = np.loadtxt(
filename,
delimiter=",",
skiprows=1,
)
data = np.loadtxt(filename, delimiter=",", skiprows=1)
position = data[:, 0]
bpm_norm = data[:, 1]
return position, bpm_norm
# def fit_bpm_linear(env, filename, x_y, plot=True):
# """
# Fit the linear region of a BPM calibration scan.
@@ -200,31 +349,17 @@ def read_bpm_calibration_data(
# plt.tight_layout()
# plt.show()
# return result
# return result
def polynomial_calibration(
env,
x_y,
rundate,
bpm_name="bcu_bpm",
degree=5,
signal_limit=0.9,
plot=True,
env, x_y, rundate, bpm_name="bcu_bpm", degree=5, signal_limit=0.9, plot=True
):
# BPM centre during calibration run
bpm_centre = {"x": 0.441,"y": -1.5405,}
bpm_centre = {"x": 0.441, "y": -1.5405}
xdata, ydata = read_bpm_calibration_data(
env,
x_y,
rundate=rundate,
bpm_name=bpm_name,
)
xdata, ydata = read_bpm_calibration_data(env, x_y, rundate=rundate, bpm_name=bpm_name)
# Keep useful BPM range
mask = np.abs(ydata) < signal_limit
@@ -262,26 +397,13 @@ def polynomial_calibration(
print(f" crosshair_bpm_position = None")
if plot:
signal_fit = np.linspace(
signal.min(),
signal.max(),
500,
)
signal_fit = np.linspace(signal.min(), signal.max(), 500)
position_fit = np.polyval(coeffs, signal_fit)
plt.figure(figsize=(7, 5))
plt.scatter(
ydata,
xdata,
alpha=0.4,
label="Calibration data",
)
plt.plot(
signal_fit,
position_fit,
label=f"{degree}th-order polynomial",
)
plt.scatter(ydata, xdata, alpha=0.4, label="Calibration data")
plt.plot(signal_fit, position_fit, label=f"{degree}th-order polynomial")
plt.xlabel(f"BPM {x_y} normalized signal")
plt.ylabel(f"BPM {x_y} position (mm)")
@@ -292,5 +414,3 @@ def polynomial_calibration(
plt.show()
return coeffs
pxii_bec/macros/daily.py
+2353 -408
View File
File diff suppressed because it is too large. Load diff
pxii_bec/macros/devices_manager.py
+3 -3
View File
@@ -1,9 +1,9 @@
"""
This module manages the initialization of devices."""
#from guards import attach_guards
#from policies import attach_policies
#from build_devices import build_devices
# from guards import attach_guards
# from policies import attach_policies
# from build_devices import build_devices
class DeviceManager:
pxii_bec/macros/enScan.py
+31 -30
View File
@@ -11,7 +11,7 @@ def en_plothist(hindex: int, signal_name: str, isave=1):
s=scans.line_scan(dev.dcm_bragg, 178.5, 179.5, steps=240, relative=False
data_x, data_y = en_plothist(-1,"bcu_bpmsum")
scanned with
scanned with
scans.line_scan(dev.aerotech.y,-0.05,0.05, steps=101, settling_time=0.5, relative=True)
"""
@@ -21,54 +21,56 @@ def en_plothist(hindex: int, signal_name: str, isave=1):
if hindex < 0:
h = bec.history[hindex]
print('by scan number')
print("by scan number")
else:
h = bec.history.get_by_scan_number(hindex)
print('by scan number')
print("by scan number")
md = h.metadata["bec"]
sno = md['dataset_number']
sno = md["dataset_number"]
print(f"number of scan = {sno}")
scanvar = list(md["args"].keys())[0] # string, returns the variable of the last performed scan
#scanvar = list(md["args"].keys())[0] # string, returns the variable of the last performed scan
# data = h.devices[device_name][signal].read()["value"]
# data_x = h.devices.dcm_pitch.dcm_pitch.read()["value"]
# data_y = h.devices.lu_bpmsum.lu_bpmsum.read()["value"]
# data_x = h.devices[device_name][device_name].read()["value"]
# that works ...
#print(list(md['args'].keys()))
val = list(md['args'].keys())
#val[0] = 'aerotech.x'
devicenam = val[0]
print('device = ', devicenam)
# that works ...
# print(list(md['args'].keys()))
## val = list(md["args"].keys()) no longer !!!!! to be checked !!!
# val[0] = 'aerotech.x'
##devicenam = val[0]
##print("device = ", devicenam)
devicenam = md["scan_report_devices"][0].decode()
data_xd = md["positions"] # last scan knows which device ... however, double array
data_x = data_xd.flatten()
print("dx = ", data_x)
#data_y = h.devices[signal_name][signal_name].read()["value"]
# data_y = h.devices[signal_name][signal_name].read()["value"]
data_y1 = h.devices["bcu_bpmsum"]["bcu_bpmsum"].read()["value"]
data_y2 = h.devices["ss_bpmsum"]["ss_bpmsum"].read()["value"]
#print("dy = ", data_y)
data_y = data_y1/data_y2
rev = 0
#if val[0][9] == 'x':
# print("dy = ", data_y)
data_y = data_y1 / data_y2
rev = 0
# if val[0][9] == 'x':
# rev = 1
smoothplotgrad(data_x, data_y, reverse = rev)
# g, xm = justfit(data_x, data_y, model=model, ibg=ibg)
smoothplotgrad(data_x, data_y, reverse=rev)
# g, xm = justfit(data_x, data_y, model=model, ibg=ibg)
# plt.ion()
# plt.figure()
# plt.plot(data_x, data_y, "*")
# plt.plot(data_x, g.best_fit)
# plt.xlabel(scanvar)
# plt.ylabel(signal_name)
# plt.show()
# plt.ion()
# plt.figure()
# plt.plot(data_x, data_y, "*")
# plt.plot(data_x, g.best_fit)
# plt.xlabel(scanvar)
# plt.ylabel(signal_name)
# plt.show()
# gcen = g.pars["center"].value
# gcen = g.pars["center"].value
if isave:
if isave:
ans = "n"
ans = input("Store data in csv file? y/n ")
if ans == "y":
@@ -79,5 +81,4 @@ def en_plothist(hindex: int, signal_name: str, isave=1):
with open(filename, "w") as f:
np.savetxt(f, combined, delimiter=",", fmt="%.9e")
return data_x, data_y
pxii_bec/macros/flux_macros.py
+22 -8
View File
@@ -256,7 +256,7 @@ def bpm2flux(energy=None, bpmsum=None, transm=None):
"""
calibration_min_energy = 6500
calibration_max_energy = 21000
# calibration_max_energy = 21000
if energy is None:
energy = get_current_energy()
@@ -266,14 +266,14 @@ def bpm2flux(energy=None, bpmsum=None, transm=None):
# bpmsum = read_av(dev.bcu_bpmsum,10) # do not add to traffic jam
if transm is None:
transm = dev.transm.read()["transm"]["value"]
transm = dev.transm.read(cached=True)["transm"]["value"]
if not calibration_min_energy <= energy <= calibration_max_energy:
raise ValueError(
f"Energy {energy:.0f} eV is outside the calibrated range "
f"{calibration_min_energy:.0f}–"
f"{calibration_max_energy:.0f} eV"
)
# if not calibration_min_energy <= energy <= calibration_max_energy:
# raise ValueError(
# f"Energy {energy:.0f} eV is outside the calibrated range "
# f"{calibration_min_energy:.0f}–"
# f"{calibration_max_energy:.0f} eV"
# )
# coeffs = [
# 1.06436277e5,
@@ -296,3 +296,17 @@ def bpm2flux(energy=None, bpmsum=None, transm=None):
)
return float(flux), float(full_flux)
def transmission_for_flux(target_flux):
"""
Calculate the transmission required to give target_flux
at the current energy.
Transmission is returned as a fraction from 0 to 1.
"""
_, full_flux = bpm2flux()
transmission = target_flux / full_flux
return float(np.clip(transmission, 0.0, 1.0))
pxii_bec/macros/katscripts.py
+1957 -439
View File
File diff suppressed because it is too large. Load diff
pxii_bec/macros/luts/roll_6000.csv
+9 -11
View File
@@ -1,12 +1,10 @@
Roll, Y pos
2,125.4
2.5,194.1
3,260.1
3.5,325.8
4,387.4
4.5,446.9
5,504.4
5.5,558.2
6,609.9
6.5,658.1
7,705.2
4,802
4.375,890.3
4.75,994.2
5.125,1096
5.5,1190
5.875,1276
6.25,1359
6.625,1443
7,1524
1 Roll Y pos
2 2 4 125.4 802
3 2.5 4.375 194.1 890.3
4 3 4.75 260.1 994.2
5 3.5 5.125 325.8 1096
6 4 5.5 387.4 1190
7 4.5 5.875 446.9 1276
8 5 6.25 504.4 1359
9 5.5 6.625 558.2 1443
10 6 7 609.9 1524
6.5 658.1
7 705.2
pxii_bec/macros/mxcomm_analysis.py
+51 -7
View File
@@ -101,11 +101,38 @@ def fit(data, fit_params):
params["base_intercept"].set(value=y_min)
params["base_slope"].set(value=0)
# # Add peak-specific parameters
# params.update(peak_model.guess(processed_data["y_to_fit"], x=processed_data["x_data"]))
# # Perform the fitting
# lmfit_result = full_model.fit(processed_data["y_to_fit"], params, x=processed_data["x_data"])
# Add peak-specific parameters
params.update(peak_model.guess(processed_data["y_to_fit"], x=processed_data["x_data"]))
x_data = np.asarray(processed_data["x_data"])
params.update(
peak_model.guess(
processed_data["y_to_fit"],
x=x_data,
)
)
# Constrain peak width to something physically resolvable by the scan
step = np.median(np.diff(np.sort(x_data)))
scan_range = np.ptp(x_data)
if "peak_sigma" in params:
params["peak_sigma"].set(
min=0.5 * step,
max=0.5 * scan_range,
)
# Perform the fitting
lmfit_result = full_model.fit(processed_data["y_to_fit"], params, x=processed_data["x_data"])
lmfit_result = full_model.fit(
processed_data["y_to_fit"],
params,
x=x_data,
)
# Find the X that gives the max Y
max_index = np.argmax(processed_data["y_to_fit"])
@@ -125,6 +152,7 @@ def fit(data, fit_params):
centre_stderr = lmfit_result.params["peak_center"].stderr
height_stderr = lmfit_result.params["peak_height"].stderr
snr = height / noise if noise > 0 else np.inf
@@ -151,7 +179,7 @@ def fit(data, fit_params):
}
def check_peak_fit(x_data, result):
def check_peak_fit(x_data, result, min_fwhm_steps=3):
"""Check whether a fitted peak is reliable
Parameters: X data, results from the fit step
Returns: success or problems"""
@@ -166,13 +194,18 @@ def check_peak_fit(x_data, result):
centre = result["centre"]
fwhm = result["fwhm"]
fwhm_steps = fwhm / step
problems = []
if not x_min + 0.1 * scan_range < centre < x_max - 0.1 * scan_range:
problems.append("peak centre is too close to scan edge")
if fwhm < 3 * step:
problems.append("peak is narrower than 3 scan steps")
if fwhm < min_fwhm_steps * step:
problems.append(
f"Peak FWHM spans only {fwhm_steps:.2f} scan steps " f"(minimum {min_fwhm_steps})"
)
if fwhm > 0.7 * scan_range:
problems.append("peak is too broad for scan range")
@@ -184,8 +217,19 @@ def check_peak_fit(x_data, result):
if centre_stderr is None:
problems.append("centre uncertainty could not be estimated")
elif centre_stderr > 0.2 * fwhm:
problems.append(f"centre uncertainty is large " f"({centre_stderr:.3g})")
elif centre_stderr > 0.5 * step:
problems.append(
f"centre uncertainty is large "
f"({centre_stderr:.3g}, "
f"{centre_stderr / step:.2f} scan steps)"
)
print("FIT DEBUG")
print(f"centre : {centre}")
print(f"fwhm : {fwhm}")
print(f"step : {step}")
print(f"fwhm/step : {fwhm / step}")
print(f"centre error : {result.get('centre_stderr')}")
return {"success": not problems, "problems": problems}
pxii_bec/macros/mxcomm_data.py
+3 -3
View File
@@ -81,17 +81,17 @@ def get_value(dev_name, details=False):
root = "scam"
live_root = "scam"
name = "scam_contrast"
value = dev_name.read()[name]["value"]
value = dev_name.read(cached=True)[name]["value"]
else:
try:
name = live_root = root = dev_name.name
value = dev_name.read()[name]["value"]
value = dev_name.read(cached=True)[name]["value"]
except:
print("non-standard device")
root = dev_name.dotted_name
name = dev_name.full_name
live_root = dev_name.parent.name
value = dev_name.read()[name]["value"]
value = dev_name.read(cached=True)[name]["value"]
if details:
return {"value": value, "root": root, "name": name, "live_root": live_root}
pxii_bec/macros/mxcomm_plot.py
+4 -12
View File
@@ -76,7 +76,7 @@ def plot_fitted_data(data, fit_result, window_name="Fitting", plot_widget=None,
f"FWHM = {fit_result['fwhm']:.5f}\n"
f"Height = {fit_result['height']:.4g}\n"
f"Model = {fit_result['model']}\n"
f"Chi sq = {fit_result['chi_sq']:.3g}\n"
f"Centre_StdErr = {fit_result['centre_stderr']:.3g}\n"
f"SNR = {fit_result['snr']:.3g}"
)
@@ -94,12 +94,8 @@ def plot_fitted_data(data, fit_result, window_name="Fitting", plot_widget=None,
# Display the fitted curve as a line without symbols.
wf.get_curve("Fit").set(symbol_size=0)
def plot_scan_data(
data,
window_name="Fitting",
plot_widget=None,
title=None,
):
def plot_scan_data(data, window_name="Fitting", plot_widget=None, title=None):
"""Plot completed scan data without a fitted curve."""
if isinstance(plot_widget, str):
@@ -120,8 +116,4 @@ def plot_scan_data(
wf.x_label = data["motor_name"]
wf.y_label = data["signal_name"]
wf.plot(
x=data["x_data"],
y=data["y_data"],
label="Data",
)
wf.plot(x=data["x_data"], y=data["y_data"], label="Data")
pxii_bec/macros/mxcomm_scan.py
+197 -68
View File
@@ -6,7 +6,6 @@
from dataclasses import dataclass
import numpy as np
# from mxcomm_analysis import (
# create_fit_parameters,
# get_data_from_history,
@@ -26,6 +25,7 @@ import numpy as np
# dev
# )
@dataclass(frozen=True)
class ScanDefaults:
"""Default values for fitting routines"""
@@ -100,9 +100,11 @@ def move_to_scan_position(motor_device, motor_name: str, position: float, data:
scans.umv(motor_device, motor_position, relative=False)
print(f"\n Moving {motor_name} to position {motor_position: .3f}")
def restore_position(motor_device, position):
"""Return a motor to its position before the scan."""
scans.umv(motor_device, position, relative=False)
scans.umv(motor_device, position, relative=False)
def go_to_peak(
motor_device,
@@ -117,87 +119,152 @@ def go_to_peak(
gomax: bool = False,
negative: bool = False,
gap: bool = False,
min_fwhm_steps: float = 3,
window_name: str = "Fitting",
plot_widget=None,
):
"""
Go to the peak of a signal by scanning a motor within a specified range and
identifying the optimal position based on signal peak data.
Scan a motor and move either to the fitted peak centre or the
highest measured data point.
Parameters:
motor_device: The motor device to be scanned.
signal_device: The signal device to monitor during the scan.
start (float): The starting position of the scan. Ignored if `relative` is True.
stop (float): The ending position of the scan. Ignored if `relative` is True.
steps (int): The number of steps to divide the scan range into.
relative (bool, optional): If True, interpret `start` and `stop` as relative to
the current motor position. Defaults to RELATIVE_MODE constant.
plot (bool, optional): If True, plot the scan data and the fitted results.
Defaults to True.
settle (float, optional): The time in seconds to wait after each step for the
signal to stabilize. Defaults to DEFAULT_SETTLE_TIME constant.
confirm (bool, optional): If True, ask for user confirmation before starting
the scan. Defaults to True.
If gomax=False:
Fit the scan, validate the fit using check_peak_fit(),
and move to the fitted centre.
Raises:
Exception: Raises exceptions potentially raised by dependent functions or
operations such as plotting, fitting, or motor movement.
If gomax=True:
Skip peak-fit validation and move directly to the highest
measured signal value. Information about whether the maximum
occurred at a scan boundary is returned.
Returns:
fit parameters
Parameters
----------
motor_device
Motor device to scan.
signal_device
Signal device to monitor.
start : float
Scan start position.
stop : float
Scan end position.
steps : int
Number of scan steps.
relative : bool
If True, start and stop are relative to current motor position.
plot : bool
Plot scan data/results.
settle : float
Settling time at each scan point.
confirm : bool
Ask for confirmation before starting scan.
gomax : bool
If True, move to highest measured point rather than fitted centre.
negative : bool
Fit a negative-going peak.
gap : bool
Apply minimum-gap protection when using fitted peak position.
min_fwhm_steps : float
Minimum allowed fitted FWHM in scan steps. Only used when
gomax=False.
window_name : str
BEC GUI window name.
plot_widget
Optional explicit plot widget.
Returns
-------
dict
Result information.
For a successful fitted peak:
success
problems
centre
fwhm
height
chi_sq
centre_stderr
snr
For gomax=True:
success
problems
x_max
max_index
at_lower_boundary
at_upper_boundary
For a rejected fit:
success=False
problems=list of fit rejection reasons
"""
# ------------------------------------------------------------
# Get motor and signal names
# ------------------------------------------------------------
mres = get_value(motor_device, True)
motor_name = mres["root"]
sres = get_value(signal_device, True)
signal_name = sres["root"]
print(f"Signal name is {signal_name}")
# if plot:
# if plot_widget is not None:
# plot_live_data(plot_widget, motor_name, signal_name, title=f"{motor_name} scan")
# else:
# plot_live_data_bec(motor_name, signal_name, window_name=window_name)
# ------------------------------------------------------------
# Set up live plotting
# ------------------------------------------------------------
if plot:
plot_live_data(
motor_name,
motor_name,
signal_name,
window_name=window_name,
plot_widget=plot_widget,
title=f"{motor_name} scan")
plot_widget=plot_widget,
title=f"{motor_name} scan",
)
# ------------------------------------------------------------
# Validate scan range and get starting position
# ------------------------------------------------------------
# Validate and calculate step size
step_size = calculate_step_size(start, stop, steps)
# Confirm the scan range
starting_position = mres["value"]
print(f"Starting motor position = {starting_position}")
if confirm:
if relative:
scan_start = starting_position + start
scan_end = starting_position + stop
print(
f"\nScanning from {scan_start: .6g} to {scan_end: .6g} in "
f"{steps} steps of size {step_size}"
f"\nScanning from {scan_start:.6g} to {scan_end:.6g} "
f"in {steps} steps of size {step_size}"
)
print(f"Relative mode = {relative}")
else:
print(
f"\nScanning from {start: .5g} to {stop: .5g} in {steps} steps of size {step_size}"
f"\nScanning from {start:.5g} to {stop:.5g} "
f"in {steps} steps of size {step_size}"
)
print(f"Relative mode = {relative}")
input("Press Enter to continue...")
# Perform the scan
# ------------------------------------------------------------
# Perform scan
# ------------------------------------------------------------
scan_result = scans.line_scan(
motor_device, start, stop, steps=steps, relative=relative, settling_time=settle
)
motor_data = scan_result.scan.live_data[mres["live_root"]][mres["name"]].val
# # motor_data = scan_result.scan.live_data[motor_name][motor_name].val
# sres = get_value(signal_name, True)
signal_data = scan_result.scan.live_data[sres["root"]][sres["name"]].val
scan_number = "Current"
data = {
@@ -209,16 +276,75 @@ def go_to_peak(
"scan_number": scan_number,
}
# Define and fit model to scan data
# ============================================================
# GOMAX MODE
# ============================================================
#
# Do not fit or apply peak-shape validation.
# Simply use the highest measured point.
#
if gomax:
if negative:
max_index = int(np.argmin(data["y_data"]))
else:
max_index = int(np.argmax(data["y_data"]))
x_max = float(data["x_data"][max_index])
at_lower_boundary = max_index == 0
at_upper_boundary = max_index == len(data["x_data"]) - 1
print(f"Maximum measured signal at {x_max:.6g}")
if at_lower_boundary:
print("Maximum is at the lower scan boundary")
if at_upper_boundary:
print("Maximum is at the upper scan boundary")
# Plot raw scan rather than trying to assess a fitted peak
if plot:
plot_scan_data(
data,
window_name=window_name,
plot_widget=plot_widget,
title=f"{motor_name} scan - maximum at {x_max:.6g}",
)
# Move to measured maximum
move_to_scan_position(data["motor_device"], data["motor_name"], x_max, data)
return {
"success": True,
"problems": [],
"x_max": x_max,
"max_index": max_index,
"at_lower_boundary": at_lower_boundary,
"at_upper_boundary": at_upper_boundary,
}
# ============================================================
# FITTED PEAK MODE
# ============================================================
fit_params = create_fit_parameters(
deriv=False, negative=False, model=FitDefaults.MODEL, baseline=FitDefaults.BASELINE
deriv=False, negative=negative, model=FitDefaults.MODEL, baseline=FitDefaults.BASELINE
)
fit_result = fit(data, fit_params)
check = check_peak_fit(data["x_data"], fit_result)
# ------------------------------------------------------------
# Validate fitted peak
# ------------------------------------------------------------
check = check_peak_fit(data["x_data"], fit_result, min_fwhm_steps=min_fwhm_steps)
if not check["success"]:
print("Peak fit rejected:")
for problem in check["problems"]:
print(f" - {problem}")
@@ -229,42 +355,45 @@ def go_to_peak(
plot_widget=plot_widget,
title=f"{motor_name} scan - FIT REJECTED",
)
print(f"Returning motor to start position, {starting_position:.5g}")
restore_position(
motor_device,
starting_position,
)
return None
print(f"Returning motor to start position, " f"{starting_position:.5g}")
restore_position(motor_device, starting_position)
return {"success": False, "problems": check["problems"]}
# ------------------------------------------------------------
# Plot accepted fit
# ------------------------------------------------------------
# Plot the fitted data if plot = True
if plot:
plot_fitted_data(data, fit_result, window_name=window_name, plot_widget=plot_widget)
# If gomax is set then move to the maximum value, rather than the fit centre
if gomax:
value = fit_result["x_max"]
print(f"Max position is at {value}")
move_to_scan_position(data["motor_device"], data["motor_name"], fit_result["x_max"], data)
# ------------------------------------------------------------
# Move to fitted peak centre
# ------------------------------------------------------------
centre = fit_result["centre"]
if gap and centre < 4.5:
print(f"Fitted gap position {centre:.5g} mm is below " "minimum gap of 4.5 mm")
umv(dev.id_gap, 4.5)
else:
if gap:
if fit_result["centre"] < 4.5:
umv(dev.id_gap, 4.5)
else:
move_to_scan_position(
data["motor_device"], data["motor_name"], fit_result["centre"], data
)
else:
# Safely move the motor to the peak position
move_to_scan_position(data["motor_device"], data["motor_name"], fit_result["centre"], data)
move_to_scan_position(data["motor_device"], data["motor_name"], centre, data)
# ------------------------------------------------------------
# Return successful fit information
# ------------------------------------------------------------
return {
"centre": fit_result["centre"],
"success": True,
"problems": [],
"centre": centre,
"fwhm": float(fit_result["fwhm"]),
"height": float(fit_result["height"]),
"chi_sq": float(fit_result["chi_sq"]),
"centre_stderr": float(fit_result["centre_stderr"]),
"snr": float(fit_result["snr"]),
}
@@ -376,9 +505,9 @@ def find_feature(
plot_live_data(
motor_name,
signal_name,
window_name = window_name,
plot_widget = plot_widget,
title = f"{motor_name} scans"
window_name=window_name,
plot_widget=plot_widget,
title=f"{motor_name} scans",
)
# Perform scan
scan_result = scans.line_scan(
pxii_bec/macros/mxcomm_utils.py
+348
View File
@@ -0,0 +1,348 @@
"""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 predict_scint_transmission(target_flux=None):
"""Predict transmission required for a scintillator measurement."""
if target_flux is None:
target_flux = BeamMove.scint_target_flux
_, full_flux = bpm2flux()
if not np.isfinite(full_flux) or full_flux <= 0:
raise ValueError(
f"Cannot predict scintillator transmission: "
f"invalid full flux ({full_flux:.3g} ph/s)"
)
transmission = target_flux / full_flux
# Prevent an unreasonable attenuator setting
transmission = np.clip(transmission, 0.001, 1.0)
return float(transmission)
def prepare_scint_measurement(autoexp=True, zoom=1000):
"""Prepare for taking a reading of the beam from the scintillator"""
# set transmission
transmission = predict_scint_transmission()
print(
f"Predicted scintillator transmission: "
f"{transmission:.2%}"
)
dev.transm.put(transmission)
time.sleep(3)
# 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)
time.sleep(0.5)
# set autoexposure and check there is a reading in the image
if autoexp:
auto_exposure(cam="samcam", max_iter=25)
return zoom
def read_scintXY(image=False, measure_size=True):
"""Get the X and Y beam positions from the sample camera at
a sepcified zoom. Default zoom is 1000"""
# Target position on sample camera
x_target = BeamMove.scint_target_x
y_target = BeamMove.scint_target_y
zoom = get_value(dev.scam_zoom)
results = {}
max_signal = get_value(dev.samcam_max)
results["max_signal"] = max_signal
thresh = get_value(dev.samcam_thresh)
if max_signal < thresh:
return {
"ok": False,
"message": (
f"Insufficient signal for centroid calculation "
f"(max signal = {max_signal:.0f}, threshold = {thresh:.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
if measure_size:
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")
if image:
imagedata = get_cam_image(Cameras.sample)
results["image"] = imagedata
return results
def finish_scint_measurement():
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
def test_predicted_trans():
filename = bl.env.lut_dir / f"predicted_trans.csv"
with open(filename, "w", encoding="utf-8") as f:
f.write(f"Energy,PredictedT,ExposureTime,max,auto_expT,max2\n")
energies = [8000, 10000, 12400, 16000, 18000, 20000, 24000]
for energy in energies:
bl_energy(energy)
dev.transm.put(0.1)
time.sleep(3)
exp_time =0.005
dev.samcam_exp.put(exp_time)
transmission = predict_scint_transmission()
dev.transm.put(transmission)
time.sleep(5)
max1 = get_value(dev.samcam_max)
auto_exposure()
aut_exp_time = get_value(dev.samcam_exp)
max2 = get_value(dev.samcam_max)
with open(filename, "a", encoding="utf-8") as f:
f.write(f"{energy:.0f},{transmission:.3f},{exp_time:.4f},{max1:.0f},{aut_exp_time:.4f},{max2:.0f}\n")
pxii_bec/macros/position_device.py
+11 -6
View File
@@ -5,6 +5,7 @@ from typing import Callable, List, Dict, Optional, Union
from bec_lib.logger import bec_logger
from bec_lib.device import Device
def motor_resolver(bec_name):
"""Used to handle aerotech and smargon motors"""
@@ -24,6 +25,7 @@ def motor_resolver(bec_name):
raise ValueError(f"Cannot resolve motor for '{bec_name}'")
@dataclass
class PositionDevice:
"""Generic device that moves between named or numeric positions"""
@@ -53,14 +55,20 @@ class PositionDevice:
if isinstance(target, str):
name = target.lower()
try:
bec_logger.logger.info(f"attempting to find position {name} from bec device {self.bec_name} userParameter.")
bec_logger.logger.info(
f"attempting to find position {name} from bec device {self.bec_name} userParameter."
)
bec_device: Device = dev[self.bec_name]
position = bec_device.user_parameter[target]
bec_logger.logger.info(f"obtained position {position} - compare to cached value {self.positions.get(target)}")
bec_logger.logger.info(
f"obtained position {position} - compare to cached value {self.positions.get(target)}"
)
return position
except Exception as e:
bec_logger.logger.debug(f"Failed to fetch user parameter for device {self.bec_name} and target {target} from bec, falling back to cached position {e}")
bec_logger.logger.debug(
f"Failed to fetch user parameter for device {self.bec_name} and target {target} from bec, falling back to cached position {e}"
)
if name not in self.positions:
raise ValueError(f"Unknown position '{target}'")
@@ -100,7 +108,6 @@ class PositionDevice:
return scans.mv(self.mot_device, pos, relative=False)
# def set_position(self, target: Union[str, float]):
# """Only to be used for testing purposes, bypasses guards"""
# pos = self._resolve_target(target)
@@ -129,5 +136,3 @@ class PositionDevice:
pos = self._resolve_target(target)
return abs(self.actual - pos) <= self.tol
pxii_bec/macros/pxii_energy.py
+74 -33
View File
@@ -25,7 +25,7 @@ def get_current_energy():
"""
Returns the energy in eV from the current bragg angle.
"""
current_bragg_angle = dev.dcm_bragg.user_readback.get()
current_bragg_angle = dev.dcm_bragg.user_readback.get(cached=True)
current_energy = convert_from_bragg(current_bragg_angle, print_result=False)["energy_ev"]
return current_energy
@@ -88,8 +88,7 @@ def mono_pitch_scan(plot=True, window_name="Fitting", plot_widget=None):
# Set fine pitch to middle of its' travel range
scans.umv(EnergyDefaults.mono_fpitch, pos["dcm_fine_pitch"], relative=False)
# Move DCM pitch to starting value
scans.umv(EnergyDefaults.mono_pitch, pos["dcm_pitch"], relative=False)
scans.umv(EnergyDefaults.mono_pitch, pos["dcm_pitch"]-0.04, relative=False)
# Go to peak using default parameters from EnergyDefaults
if plot:
@@ -104,9 +103,32 @@ def mono_pitch_scan(plot=True, window_name="Fitting", plot_widget=None):
settle=0.01,
plot=True,
confirm=False,
min_fwhm_steps=1.1,
window_name=window_name,
plot_widget=plot_widget
plot_widget=plot_widget,
)
result["extended_scan"] = False
if not result["success"]:
extended_range = 2 * EnergyDefaults.pitch_scan["halfwidth"]
extended_steps = 2 * EnergyDefaults.pitch_scan["steps"]
print("Scanning monochromator pitch over a broader range.")
result = go_to_peak(
EnergyDefaults.mono_pitch,
EnergyDefaults.signals["sig1"],
-extended_range,
extended_range,
steps=extended_steps,
relative=True,
settle=0.01,
plot=True,
confirm=False,
min_fwhm_steps=1.1,
window_name=window_name,
plot_widget=plot_widget,
)
result["extended_scan"] = True
else:
print("Scanning monochromator pitch and moving to peak, without plotting.")
result = go_to_peak(
@@ -118,9 +140,28 @@ def mono_pitch_scan(plot=True, window_name="Fitting", plot_widget=None):
relative=True,
settle=0.01,
plot=False,
min_fwhm_steps=1.1,
confirm=False,
)
result["extended_scan"] = False
if not result["success"]:
extended_range = 2 * EnergyDefaults.pitch_scan["halfwidth"]
extended_steps = 2 * EnergyDefaults.pitch_scan["steps"]
print("Scanning monochromator pitch over a broader range.")
result2 = go_to_peak(
EnergyDefaults.mono_pitch,
EnergyDefaults.signals["sig1"],
-extended_range,
extended_range,
steps=extended_steps,
relative=True,
settle=0.01,
plot=True,
confirm=False,
min_fwhm_steps=1.1,
)
result["extended_scan"] = True
return result
@@ -178,7 +219,13 @@ def get_dcm_motors_positions(energy_ev):
perp = calc_perp_position(energy_ev, print_result=False)
bragg_angle = convert_from_energy(energy_ev, print_result=False)["bragg_angle_mrad"]
dcm_motor_values.update(
{"bragg_angle": bragg_angle, "perp": perp, "dcm_pitch": pitch, "dcm_roll": roll, "dcm_fine_pitch": fpitch}
{
"bragg_angle": bragg_angle,
"perp": perp,
"dcm_pitch": pitch,
"dcm_roll": roll,
"dcm_fine_pitch": fpitch,
}
)
return dcm_motor_values
@@ -202,7 +249,7 @@ def move_dcm_motors(energy_ev):
# umv(EnergyDefaults.mono_pitch, dcm_pos["dcm_pitch"])
# umv(EnergyDefaults.mono_roll, dcm_pos["dcm_roll"])
# print("\n***DCM Roll movement is currently disabled ***\n")
scans.umv(EnergyDefaults.mono_fpitch,dcm_pos["dcm_fine_pitch"], relative=False )
scans.umv(EnergyDefaults.mono_fpitch, dcm_pos["dcm_fine_pitch"], relative=False)
scans.umv(
EnergyDefaults.energy,
dcm_pos["bragg_angle"],
@@ -216,12 +263,9 @@ def move_dcm_motors(energy_ev):
)
def bl_energy(energy_ev,
move_gap=True,
mono_scan=True,
plot=True,
window_name="Fitting",
plot_widget=None):
def bl_energy(
energy_ev, move_gap=True, mono_scan=True, plot=True, window_name="Fitting", plot_widget=None
):
"""
Adjusts the beamline's energy to the specified energy in electron volts (eV).
The function validates the target energy, checks the current energy, and makes
@@ -248,26 +292,29 @@ def bl_energy(energy_ev,
print(f"Energy change of {energy_diff:.2f} eV is too small, not changing energy.")
return
transmission = get_value(dev.transm)
# Step 1: Move the gap if needed.
move_gap_if_needed(energy_ev, move_gap)
# Step 2: Move and set the DCM motors.
move_dcm_motors(energy_ev)
# Step 3: Update the mirror stripe.
# Step 3: Reset the transmission after the energy change
print(f"Setting transmission to {transmission}")
dev.transm.put(transmission)
# Step 4: Update the mirror stripe.
set_mirror_stripe(energy_ev)
# Step 4: Perform DCM pitch scan and move to peak.
# Step 5: Perform DCM pitch scan and move to peak.
if mono_scan:
if plot:
result = mono_pitch_scan(plot=True,
window_name=window_name,
plot_widget=plot_widget)
result = mono_pitch_scan(plot=True, window_name=window_name, plot_widget=plot_widget)
return result
else:
mono_pitch_scan(plot=False)
# def new_calc_gap(energy_ev):
# """
@@ -301,7 +348,8 @@ def bl_energy(energy_ev,
# # raise ValueError(msg)
# return gap
def new_calc_gap(energy_ev, print_result=True):
def calc_gap(energy_ev, print_result=True):
"""Calculate the undulator gap for the given energy."""
harmonic_values = Gap.get_harmonic_values(energy_ev)
@@ -314,33 +362,26 @@ def new_calc_gap(energy_ev, print_result=True):
if print_result:
harmonic = Gap.get_harmonic_by_energy(energy_ev)
print(
f"Calculated gap is {gap:.3f} mm "
f"using {harmonic}"
)
print(f"Calculated gap is {gap:.3f} mm " f"using {harmonic}")
return gap
def set_gap(energy_ev, print_result=False):
"""Set the undulator gap."""
gap = new_calc_gap(energy_ev)
gap = calc_gap(energy_ev)
if gap is not None and Gap.minimum_gap <= gap <= 20:
umv(dev.id_gap, gap)
if print_result:
harmonic = Gap.get_harmonic_by_energy(energy_ev)
print(
f"Setting gap to {gap:.3f} mm "
f"using {harmonic}"
)
print(f"Setting gap to {gap:.3f} mm " f"using {harmonic}")
else:
raise ValueError(
f"Cannot calculate a valid gap for {energy_ev:.0f} eV. "
f"Calculated gap: {gap}"
)
raise ValueError(
f"Cannot calculate a valid gap for {energy_ev:.0f} eV. " f"Calculated gap: {gap}"
)
return gap
pxii_bec/macros/pxii_parameters.py
+103 -49
View File
@@ -4,6 +4,7 @@ from dataclasses import dataclass
import numpy as np
from enum import Enum
@dataclass(frozen=True)
class EnergyDefaults:
"""Parameters for PXII energy changes"""
@@ -12,14 +13,14 @@ class EnergyDefaults:
min_energy_ev = 4800
max_energy_ev = 30002
beam_offset = 6
default_trans = 0.1
signals = {"sig1": dev.lu_bpmsum, "sig2": dev.ss_bpmsum, "sig3": dev.bcu_bpmsum}
energy = dev.dcm_bragg
mono_pitch = dev.dcm_pitch
mono_fpitch = dev.dcm_fpitch
mono_perp = dev.dcm_perp
mono_roll = dev.dcm_froll
mono_roll_value = 4.75
LUT_table = "luts/energy_lut.csv"
mono_roll_value = 4.36
stripe_thresholds = {"silicon": 9000, "rhodium": 20000, "platinum": 40000}
pitch_scan = {"halfwidth": 0.075, "steps": 20}
@@ -28,25 +29,11 @@ class EnergyDefaults:
class Calibration:
"""Calibration parameters for PXII optics"""
# pitch = np.array([4.61823701e-14, -1.97330772e-09, 2.89694543e-05, -5.34468669e00])
# pitch_calib = [-5.89030403e-19,
# 4.85665619e-14,
# -1.49364459e-09,
# 2.08652765e-05,
# -5.59843072e+00]
pitch_calib = [-4.09746242e-18,
2.32320520e-13,
-4.95650128e-09,
4.85168715e-05,
-5.67418203e+00]
pitch_calib = [-7.60082220e-19, 6.09893342e-14, -1.81702221e-09, 2.45457520e-05, -5.61275367e00]
# roll = np.array([2.28291039e-03, -2.41928101e01])
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class Gap:
"""Fit parameters to calculate undulator gap from energy."""
@@ -56,15 +43,15 @@ class Gap:
correction_factor = 0.9994
harmonics = {
"H3": np.array([9.57400000e-04, 1.62100000e-01]),
"H5": np.array([5.03757576e-04, 8.53992727e-01]),
"H7": np.array([3.57624242e-04, 8.74460000e-01]),
"H9": np.array([2.76218182e-04, 9.00565455e-01]),
"H11": np.array([2.25662626e-04, 9.04630909e-01]),
"H13": np.array([1.89440404e-04, 9.35049697e-01]),
"H15": np.array([1.64179798e-04, 9.35255758e-01]),
"H17": np.array([1.44460317e-04, 9.47441667e-01]),
"H19": np.array([1.26071429e-04, 1.04085357e00]),
"H3": np.array([9.37600000e-04, 2.98800000e-01]),
"H5": np.array([5.10102857e-04, 7.99666667e-01]),
"H7": np.array([3.58874286e-04, 8.60410476e-01]),
"H9": np.array([2.74520000e-04, 9.25540000e-01]),
"H11": np.array([2.24457143e-04, 9.25714286e-01]),
"H13": np.array([1.88834286e-04, 9.47505714e-01]),
"H15": np.array([1.63434286e-04, 9.52602857e-01]),
"H17": np.array([1.43320000e-04, 9.77500000e-01]),
"H19": np.array([1.27940000e-04, 9.85770000e-01]),
}
@classmethod
@@ -75,7 +62,7 @@ class Gap:
"""
slope, intercept = cls.harmonics[harmonic]
# new_calc_gap() applies the correction factor afterwards,
# calc_gap() applies the correction factor afterwards,
# so account for it here as well.
raw_minimum_gap = cls.minimum_gap / cls.correction_factor
@@ -124,9 +111,7 @@ class Gap:
harmonic = cls.get_harmonic_by_energy(energy_ev)
if harmonic is None:
raise ValueError(
f"{energy_ev:.0f} eV is below the minimum calibrated energy."
)
raise ValueError(f"{energy_ev:.0f} eV is below the minimum calibrated energy.")
slope, intercept = cls.harmonics[harmonic]
@@ -141,8 +126,6 @@ class Gap:
)
return harmonic, gap
@dataclass(frozen=True)
@@ -169,34 +152,69 @@ class Harmonics:
@dataclass(frozen=True)
class CamConversion:
"""Convert pixels to microns for sam cam"""
"""Convert pixels to microns for sample camera"""
a = 0.5208
b = 0.002586
# @dataclass(frozen=True)
class BPM_X_Calibration:
coeffs = (0.11079256, -0.02493623, -0.00799871, -0.00482772, 0.16569134, 0.44192518)
class BCU_BPM_X_Calibration:
coeffs = (0.11079256, -0.02493623, -0.00799871, -0.00482772, 0.16569134, 0.44192518)
signal_limit = 0.90
# centred_position = 0.441
centred_position = -0.009 # updated on 31st Aug26
crosshair_signal = 0.000731
crosshair_bpm_position = 0.444
centred_position = -0.009 # where the bpm was during calibration
crosshair_signal = -0.00339158
crosshair_bpm_position = 0.0532
# @dataclass(frozen=True)
class BPM_Y_Calibration:
class BCU_BPM_Y_Calibration:
coeffs = (-0.14645507, 0.010016, 0.08258468, -0.00547233, -0.02715706, -1.54419084)
signal_limit = 0.95
centred_position = -1.540 # where the bpm was during calibration
crosshair_signal = -0.00392027
crosshair_bpm_position = -1.548
# @dataclass(frozen=True)
class SS_BPM_X_Calibration:
coeffs = (4.04882, -10.9093)
signal_limit = 0.90
centred_position = -1.540
crosshair_signal = -0.01383
crosshair_bpm_position = -1.540
centred_position = 2.8625 # updated on 26th Sep26
# @dataclass(frozen=True)
class SS_BPM_Y_Calibration:
coeffs = (-23.8286, -32.7051)
signal_limit = 0.90
centred_position = -1.3876
# @dataclass(frozen=True)
class LU_BPM_X_Calibration:
coeffs = (4.42860, -9.63986)
signal_limit = 0.90
# centred_position = 0.441
centred_position = 2.1816 # updated on 26th Sep26
# @dataclass(frozen=True)
class LU_BPM_Y_Calibration:
coeffs = (-27.6544, 83.2683)
signal_limit = 0.8
centred_position = 3.0175
# @dataclass(frozen=True)
class Beamstop:
x_offset = -0.65
x_offset = -0.55
i2_limit = 5e-8
@dataclass(frozen=True)
class Cameras:
sample = "X10SA-ES-MS"
ss_xray_eye = "X10SA-ES-SSCAM"
det_xray_eye = "X10SA-ES-XEYE"
class CameraZoom(Enum):
X1 = ("1x", 1)
@@ -207,7 +225,7 @@ class CameraZoom(Enum):
X12_5 = ("12.5x", 1000)
def __init__(self, label, position):
self.label= label
self.label = label
self.position = position
@classmethod
@@ -216,9 +234,7 @@ class CameraZoom(Enum):
for zoom in cls:
if abs(position - zoom.position) <= tolerance:
return zoom
raise ValueError(
f"No camera zoom levle found near motor position {position}"
)
raise ValueError(f"No camera zoom level found near motor position {position}")
@classmethod
def from_label(cls, label):
@@ -228,7 +244,45 @@ class CameraZoom(Enum):
return zoom
raise ValueError(f"Unknown camera zoom: {label}")
# @dataclass(frozen=True)
class PreChecks:
mode_pv = "AGEOP-BL:STATUS-X10SA"
fe_shutter_pv = "X10SA-FE-PSH1-EMLS-0010:OPEN"
hutch_shutter_pv = "X10SA-OP-PSH1-EMLS:STATE"
# @dataclass(frozen=True)
class EHChecks:
cryo_pv = "X10SA-ES-CS:TEMP_RBV"
ss_xi_pv = "X10SA-ES-SSXI:POS-SET"
ss_mot = dev.ss_xi_x
# @dataclass(frozen=True)
class BeamMove:
"""Values used for moving beam to the crosshair using
the VFM DR motor and the HFM XD motors. Slope values apply to 12.5X
zoom when the scintillator is used for measurments"""
# bcu_x_coeffs = (2519.3,-5184.1)
# bcu_y_coeffs = (1324.4,1108.7)
zoom = CameraZoom.X12_5.position
vfm_mot = dev.vfm_ydr
hfm_mot = dev.hfm_xd
max_move = 0.02
scint_target_x = 993
scint_target_y = 969
scint_x_slope = -19450
scint_y_slope = -37800
scint_tolerance = 7
scint_target_flux = 8e11
# @dataclass(frozen=True)
class BLSetup:
"""Values used for the weekly setup script"""
energy = 12400
fe_xsize = 0.6
fe_ysize = 0.6
detector_height = 0
pxii_bec/macros/weekly.py
+347
View File
@@ -0,0 +1,347 @@
# """Weekly beamline checks"""
"""Weekly beamline checks."""
import csv
from dataclasses import dataclass
from datetime import datetime
from pxii_bec.scripts import beamline_context as bl
from pxii_bec.macros.beamline_planner import BeamlineState
# =============================================================================
# Setup
# =============================================================================
def setup_weekly():
"""Put the beamline into standard conditions for weekly checks."""
print("\nSetting up standard weekly-check conditions...")
# Set standard energy and run mono pitch scan
energy_result = daily_energy(
BLSetup.energy,
scan_mono=True,
plot_widget="Plot_1"
)
return {
"ok": True,
"energy": BLSetup.energy,
"gap": get_value(dev.id_gap),
"energy_result": energy_result,
}
# =============================================================================
# Slit scans
# =============================================================================
def weekly_fe_scan(window_name="DailySetup"):
"""Scan the FE slits to find the centre values."""
print("\nScanning FE slits...")
try:
results = scan_fe(window_name)
return {
"ok": True,
"x": results["fe_xcen"],
"y": results["fe_ycen"],
}
except Exception as exc:
reason = str(exc)
print(f"FE slit scan failed: {reason}")
return {
"ok": False,
"error": reason,
}
def scan_fe(window_name="Fitting", plot_widget=None):
"""Scan the FE slits in X and Y"""
try:
print("Scanning FE slits in X")
umv(dev.fe_sl_xsize, 0.05, dev.fe_sl_ysize, 0.6)
go_to_peak(dev.fe_sl_xcen, dev.lu_bpmsum, -1, 1, 25, confirm=False, relative=True, window_name=window_name, plot_widget=1)
xcen = get_value(dev.fe_sl_xcen)
print("Scanning FE slits in Y")
umv(dev.fe_sl_xsize, 0.6, dev.fe_sl_ysize, 0.05)
go_to_peak(dev.fe_sl_ycen, dev.lu_bpmsum, -1, 1, 25, confirm=False, relative=True, window_name=window_name, plot_widget=2)
ycen = get_value(dev.fe_sl_ycen)
return {
"fe_xcen": xcen,
"fe_ycen": ycen,
}
finally:
umv(dev.fe_sl_xsize, WeeklySetup.fe_xsize)
umv(dev.fe_sl_ysize, WeeklySetup.fe_ysize)
def weekly_bsf_slits_scan():
"""Scan the BSF slits to find the centre values."""
print("\nScanning BSF slits...")
try:
results = centre_one("bsf", dev.lu_bpmsum)
return {
"ok": True,
"x": get_value(dev.bsf_sl_xcen),
"y": get_value(dev.bsf_sl_ycen),
# "result": results,
}
except Exception as exc:
reason = str(exc)
print(f"BSF slit scan failed: {reason}")
return {
"ok": False,
"error": reason,
}
def weekly_ss_slits_scan():
"""Scan the secondary slits to find the centre values."""
print("\nScanning secondary slits...")
try:
results = centre_one("ss", dev.bcu_bpmsum)
return {
"ok": True,
"x": get_value(dev.ss_sl_xcen),
"y": get_value(dev.ss_sl_ycen),
# "result": results,
}
except Exception as exc:
reason = str(exc)
print(f"SS slit scan failed: {reason}")
return {
"ok": False,
"error": reason,
}
# =============================================================================
# BPM readings
# =============================================================================
def measure_bpm_positions():
lubpm = compute_norm_lu()
ssbpm = compute_norm_ss()
bcubpm = get_bcu_beampos()
results = {}
results["lu"]["X norm"] = lubpm["lu_Xnorm"]
results["lu"]["Y norm"] = lubpm["lu_Ynorm"]
results["ss"]["X norm"] = ssbpm["ss_Xnorm"]
results["ss"]["Y norm"] = ssbpm["ss_Ynorm"]
results["bcu"]["BCU Xnorm"] = bcubpm["bcu_Xnorm"]
results["bcu"]["Ynorm"] = bcubpm["bcu_Xnorm"]
results["bcu"]["bcu_x_um"] = bcubpm["bcu_x_um"]
results["bcu"]["bcu_y_um"] = bcubpm["bcu_y_um"]
return results
# =============================================================================
# Save results
# =============================================================================
def save_weekly_results(results):
"""Append weekly beamline diagnostics to the history CSV."""
filename = bl.env.daily_setup_dir / "weekly_beamline_health.csv"
setup = results.get("setup", {})
energy_result = setup.get("energy_result", {})
fe = results.get("FE", {})
bsf = results.get("BSF", {})
ss = results.get("SS", {})
lubpm = results.get("lubpm", {})
ssbpm = results.get("ssbpm", {})
bcubpm = results.get("bcubpm", {})
row = {
"timestamp": datetime.now().isoformat(timespec="seconds"),
# Energy / machine settings
"energy_eV": format_value(
energy_result.get("energy"), ".0f"
),
"id_gap_mm": format_value(
setup.get("gap"), ".3f"
),
"pitch_offset_mrad": format_value(
energy_result.get("pitch_offset"), ".4f"
),
# Slit centres
"fe_x": format_value(
fe.get("x"), ".3f"
),
"fe_y": format_value(
fe.get("y"), ".3f"
),
"bsf_x": format_value(
bsf.get("x"),".3f"
),
"bsf_y": format_value(
bsf.get("y"), ".3f"
),
"ss_x": format_value(
ss.get("x"), ".3f"
),
"ss_y": format_value(
ss.get("y"),".3f"
),
"lu_Xnorm": format_value(
lubpm.get("lu_Xnorm"), ".4f"
),
"lu_Ynorm": format_value(
lubpm.get("lu_Ynorm"), ".4f"
),
"ss_Xnorm": format_value(
ssbpm.get("ss_Xnorm"), ".4f"
),
"ss_Ynorm": format_value(
ssbpm.get("ss_Ynorm"), ".4f"
),
"bcu_Xnorm": format_value(
bcubpm.get("bcu_Xnorm"), ".4f"
),
"bcu_Ynorm": format_value(
bcubpm.get("bcu_Ynorm"), ".4f"
),
"bcu_beamX": format_value(
bcubpm.get("bcu_x_um"), ".1f"
),
"bcu_beamY": format_value(
bcubpm.get("bcu_y_um"), ".1f"
),
}
fieldnames = list(row.keys())
file_exists = filename.exists()
with open(filename, "a", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
if not file_exists:
writer.writeheader()
writer.writerow(row)
print(f"\nWeekly results saved to:")
print(filename)
return filename
# =============================================================================
# SciLog results
# =============================================================================
def send_week_to_scilog():
filename = bl.env.daily_setup_dir / "weekly_beamline_health.csv"
msg = bec.messaging.scilog.new()
msg.add_text("Weekly PXII checks")
msg.add_attachment(filename)
msg.add_tags("weekly-check", "PXII")
msg.send()
# =============================================================================
# Main weekly check
# =============================================================================
def weekly_checks():
"""Run standard weekly beamline checks."""
print("\n========================================")
print("PXII WEEKLY CHECKS")
print("========================================")
results = {}
# ------------------------------------------------------------------
# Standard conditions
# ----------------------------------SS--------------------------------
try:
results["setup"] = setup_weekly()
except Exception as exc:
reason = str(exc)
print(f"Weekly setup failed: {reason}")
results["setup"] = {
"ok": False,
"error": reason,
}
# Don't continue with slit scans if we couldn't establish
# the standard starting conditions.
save_weekly_results(results)
return results
# ------------------------------------------------------------------
# BPM positions
# ------------------------------------------------------------------
results["lubpm"] = compute_norm_lu()
results["ssbpm"] = compute_norm_ss()
results["bcubpm"] = get_bcu_beampos()
# ------------------------------------------------------------------
# Slit scans
# ------------------------------------------------------------------
results["FE"] = weekly_fe_scan()
results["BSF"] = weekly_bsf_slits_scan()
results["SS"] = weekly_ss_slits_scan()
# ------------------------------------------------------------------
# Save results
# ------------------------------------------------------------------
results["filename"] = save_weekly_results(results)
# ------------------------------------------------------------------
# Edge scan
# ------------------------------------------------------------------
results["edge"] = scan_ge(11103, mode="coarse")
# ------------------------------------------------------------------
# Save results to SciLog
# ------------------------------------------------------------------
print("\nWeekly checks complete.")
return results
pxii_bec/scripts/init_beamline.py
+9 -1
View File
@@ -14,16 +14,19 @@ class Environment:
beamline_configs = {
"x06sa": {
"name": "PXI",
"package": "pxi_bec",
"devices": "pxi-devices.yaml",
"states": "pxi-beamline-states.yaml",
},
"x10sa": {
"name": "PXII",
"package": "pxii_bec",
"devices": "pxii-devices.yaml",
"states": "pxii-beamline-states.yaml",
},
"x06da": {
"name": "PXIII",
"package": "pxiii_bec",
"devices": "pxiii-devices.yaml",
"states": "pxiii-beamline-states.yaml",
@@ -40,8 +43,9 @@ class Environment:
)
cfg = self.beamline_configs[self.beamline]
self.beamline_name = cfg["name"]
# Import whichever deplyment BEV has already made available
# Import whichever deployment BEC has already made available
package = importlib.import_module(cfg["package"])
self.bec_dir = Path(package.__file__).resolve().parent
@@ -49,6 +53,10 @@ class Environment:
self.device_config_dir = self.bec_dir / "device_configs"
self.lut_dir = self.bec_dir / "macros" / "luts"
# Runtime files generated by daily setup
self.daily_setup_dir = self.bec_dir / "daily_setup_files"
self.daily_setup_dir.mkdir(exist_ok=True)
self.devices_file = self.device_config_dir / cfg["devices"]
self.states_file = self.device_config_dir / cfg["states"]