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

Integrate latest PXII production calibrations and fixes
CI for pxii_bec / test (push) Successful in 1m2s
Details
CI for pxii_bec / test (pull_request) Successful in 1m5s
Details

Browse Source
This commit is contained in:
x10sa committed 2026-10-28 20:18:25 +01:00
1 parent 225ea18da2
commit 13b2eb32d9
6 files changed
+236 -225
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pxii_bec/device_configs/pxii-devices.yaml
+5 -5
View File
@@ -65,15 +65,15 @@ aerotech:
mount: 0
work: 0
x:
in: 0.0599968
in: 0.06
out: -10
safe: -50
y:
mount: -0.0
work: -0.03699256
work: -0.035
z:
mount: 0
work: -0.00900414
work: -0.009
bcu_bpm1:
description: BCU BPM Channel 1
deviceClass: ophyd.EpicsSignalRO
@@ -302,7 +302,7 @@ bs_x:
onFailure: buffer
readoutPriority: baseline
userParameter:
in: 0.44000227
in: 0.34999599
bs_y:
description: Beamstop Y
deviceClass: ophyd_devices.EpicsMotor
@@ -314,7 +314,7 @@ bs_y:
onFailure: buffer
readoutPriority: baseline
userParameter:
in: -3.3000017
in: -3.5
bs_z:
description: Beamstop Z
deviceClass: ophyd_devices.EpicsMotor
pxii_bec/macros/daily.py
+30 -7
View File
@@ -1182,7 +1182,7 @@ def daily_beamstop():
result["bs_x_pos"] = x_result["centre"]
result["bs_y_pos"] = y_result["centre"]
result["intensity after alignment"] = read_i2()
result['intensity after alignment'] = read_i2()
# message = (
# "Intensity after Y scan: "
@@ -1324,8 +1324,8 @@ def daily_collimator():
get_value(dev.i1)
result["intensity_coll_out"] = get_value(dev.i1)
print(f"Starting intensity is {result["intensity_coll_out"]:.3e}")
result['intensity_coll_out'] = get_value(dev.i1)
print(f"Starting intensity is {result['intensity_coll_out']:.3e}")
bl.d["coll_y"].move("in")
@@ -1448,7 +1448,7 @@ def read_bcu_beampos_safe():
"""Read calibrated BCU beam position without failing beam alignment."""
try:
result = get_bcu_beampos()
return result["bcu_x_um"], result["bcu_y_um"], None
return result['bcu_x_um'], result['bcu_y_um'], None
except Exception as exc:
warning = f"BCU BPM position unavailable: {exc}"
@@ -1930,11 +1930,34 @@ def log_daily_setup_to_csv(results):
get_value(dev.hfm_pitch), ".5f"
),
}
fieldnames = list(row.keys())
file_exists = filename.exists()
# Check whether the existing CSV has the current schema
if file_exists:
with open(filename, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
existing_header = next(reader, None)
if existing_header != fieldnames:
# Archive the old CSV rather than writing incompatible rows to it
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_filename = filename.with_name(
f"{filename.stem}_archive_{timestamp}{filename.suffix}"
)
filename.rename(archive_filename)
print(
"⚠ Daily beamline CSV schema has changed.\n"
f" Previous history archived as: {archive_filename.name}\n"
f" Starting new history file: {filename.name}"
)
file_exists = False
with open(filename, "a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=row.keys())
writer = csv.DictWriter(f, fieldnames=fieldnames)
if not file_exists:
writer.writeheader()
@@ -2478,4 +2501,4 @@ def create_daily_html_report(results, total_duration=None, messages=None):
</html>
"""
return html
return html
pxii_bec/macros/katscripts.py
+161 -200
View File
@@ -190,7 +190,6 @@ def energy_checks(start_energy=8000, end_energy=20000):
f.close()
# def scan_gap(gomax = False, close = True, fine = False):
# """Scan the gap and monitor on lu_bpmsum or bcu_bpmsum"""
# close_fe_slits(close)
@@ -352,14 +351,13 @@ def beam_centre_from_ss():
def generate_dcm_lut(
start_energy=6000,
end_energy=30000,
step=1000,
scan_id_gap=False,
scint=True,
prefix = "dcm_calibration",
start_energy=6000,
end_energy=30000,
step=1000,
scan_id_gap=False,
scint=True,
prefix="dcm_calibration",
):
"""Generate a DCM calibration lookup table.
Parameters
@@ -393,13 +391,19 @@ 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,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n")
f.write(
f"energy,harmonic,gap,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n"
)
else:
f.write("energy,harmonic,gap,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n")
f.write(
"energy,harmonic,gap,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n"
)
else:
with open(filename, "w", encoding="utf-8") as f:
if scint:
f.write(f"energy,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n")
f.write(
f"energy,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y,samcam_x,samcam_y,zoom={zoom}\n"
)
else:
f.write("energy,pitch,lu_Xnorm,lu_Ynorm,ss_Xnorm,ss_Ynorm,bcu_x,bcu_y\n")
@@ -500,15 +504,14 @@ def generate_dcm_lut(
# ------------------------------------------------------------
res = get_bcu_beampos()
bcu_x = res['bcu_x_um']
bcu_y = res['bcu_y_um']
bcu_x = res["bcu_x_um"]
bcu_y = res["bcu_y_um"]
ss_res = compute_norm_ss()
ss_Xnorm = ss_res['ss_Xnorm']
ss_Ynorm = ss_res['ss_Ynorm']
ss_Xnorm = ss_res["ss_Xnorm"]
ss_Ynorm = ss_res["ss_Ynorm"]
lu_res = compute_norm_lu()
lu_Xnorm = lu_res['lu_Xnorm']
lu_Ynorm = lu_res['lu_Ynorm']
lu_Xnorm = lu_res["lu_Xnorm"]
lu_Ynorm = lu_res["lu_Ynorm"]
pitch = get_value(dev.dcm_pitch)
@@ -547,7 +550,7 @@ def generate_dcm_lut(
f.write(
f"{energy:.0f},{h},{gap_value:.5g},{pitch:.5g},"
f"{ss_Xnorm:.5g},{ss_Ynorm:.5g},{bcu_x:.2f},{bcu_y:.2f}\n"
)
)
else:
@@ -555,12 +558,13 @@ def generate_dcm_lut(
f.write(
f"{energy:.0f},{pitch:.5g},{lu_Xnorm:.5g},{lu_Ynorm:.5g},"
f"{ss_Xnorm:.5g},{ss_Ynorm:.5g},{bcu_x:.2f},{bcu_y:.2f},{x:.4g},{y:.4g}\n"
)
)
else:
f.write(
f"{energy:.0f}," f"{pitch:.5g},{lu_Xnorm:.5g},{lu_Ynorm:.5g},"
f"{energy:.0f},"
f"{pitch:.5g},{lu_Xnorm:.5g},{lu_Ynorm:.5g},"
f"{ss_Xnorm:.5g},{ss_Ynorm:.5g},{bcu_x:.2f},{bcu_y:.2f}\n"
)
)
finally:
print("\nReturning beamline to safe conditions...")
@@ -600,45 +604,24 @@ def generate_dcm_lut(
print(f"\nCalibration data written to:\n" f"{filename}")
return {
"filename": filename,
"failure_filename": failure_filename,
"failures": failures,
}
return {"filename": filename, "failure_filename": failure_filename, "failures": failures}
def calibrate_dcm_pitch(
start_energy=6000,
end_energy=30000,
step=2000,
fit_order=3,
):
def calibrate_dcm_pitch(start_energy=6000, end_energy=30000, step=2000, fit_order=3):
"""Quickly recalibrate the DCM pitch lookup table."""
print("\nRunning DCM pitch calibration")
print(
f"{start_energy:.0f}–{end_energy:.0f} eV "
f"in {step:.0f} eV steps"
)
print(f"{start_energy:.0f}–{end_energy:.0f} eV " f"in {step:.0f} eV steps")
failures = generate_dcm_lut(
start_energy=start_energy,
end_energy=end_energy,
step=step,
scan_id_gap=False,
scint=False,
start_energy=start_energy, end_energy=end_energy, step=step, scan_id_gap=False, scint=False
)
print("\nAnalysing DCM pitch calibration...")
fit = pitch_analysis(
filedate=filenow(date_only=True),
fit_order=fit_order,
)
fit = pitch_analysis(filedate=filenow(date_only=True), fit_order=fit_order)
return {
"fit": fit,
"failures": failures,
}
return {"fit": fit, "failures": failures}
def write_lut():
@@ -657,8 +640,6 @@ def write_lut():
f.write(f"{energy:.0f},{gap:.5g},{pitch:.5g},{roll:.4g},{perp:.5g},{x:.4g},{y:.4g}\n")
def write_beamsize():
"""Write beam size and flux to file"""
filename = "luts/beamsizes.csv"
@@ -677,8 +658,6 @@ def write_beamsize():
f.write(f"{lens_config},{sx:.4g},{sy:.4g},{i1_reading:.4g}\n")
def cal_samcam_fixed_x():
# lat = dev.hfm_x.read()["hfm_x"]["value"]
# pitch = dev.hfm_pitch.read()["hfm_pitch"]["value"]
@@ -1072,8 +1051,6 @@ from epics import caget, caput
import time
# def get_new_beamsizes(mirror):
# filename = f"luts/{mirror}_size.csv"
# if mirror == "hfm":
@@ -1300,7 +1277,6 @@ def mirror_pitch(mirror="vfm"):
results = np.array(results)
def vfm_yaw_check():
slit_size = 0.02
slit_point1 = -1.2
@@ -1370,6 +1346,7 @@ def vfm_yaw_check():
plt.tight_layout()
plt.show()
# @dataclass
# class SlitAxis:
# name: str
@@ -1495,11 +1472,6 @@ def vfm_yaw_check():
# print(f"=== Finished {slit.name} ===")
# def define_slits():
# slits = [
# Slit(
@@ -1728,9 +1700,9 @@ import matplotlib.pyplot as plt
from epics import caget
def scan_scin_focus(planner, type="fine"):
def scan_scin_focus(type="fine"):
dev.bcu_shutter.put(0)
planner.move_to(BeamlineState.BEAM_VISUALISATION)
# bl.planner.move_to(BeamlineState.BEAM_VISUALISATION)
dev.bcu_shutter.put(1)
z0 = dev.diag_z.position
@@ -1746,11 +1718,12 @@ def scan_scin_focus(planner, type="fine"):
umv(dev.scam_zoom, 1000)
dev.transm.put(0.1)
time.sleep(3)
auto_exposure("samcam", target=200, max_iter=20)
time.sleep(0.2)
go_to_peak(dev.diag_z, dev.scam, min, max, steps, relative=True, gomax=True, confirm=False)
go_to_peak(dev.diag_z, dev.scam, min, max, steps, relative=False, gomax=True, confirm=False)
dev.bcu_shutter.put(0)
@@ -1829,7 +1802,6 @@ def centre_bcu_bpm_y():
return centre
def centre_ss_bpm_x():
positions = []
errors = []
@@ -1866,7 +1838,7 @@ def centre_ss_bpm_y():
ch1 = get_value(dev.ss_bpm1)
ch2 = get_value(dev.ss_bpm2)
ch3 = get_value(dev.ss_bpm3)
ch4 = get_value(dev.ss_bpm4)
ch4 = get_value(dev.ss_bpm4)
_, y_error = bpm_errors(ch1, ch2, ch3, ch4)
print(f"BPM_Y = {float(y)}, Yerror = {y_error}")
@@ -1879,6 +1851,7 @@ def centre_ss_bpm_y():
umv(dev.ss_bpm_y, centre)
return centre
def centre_lu_bpm_x():
positions = []
errors = []
@@ -1915,7 +1888,7 @@ def centre_lu_bpm_y():
ch1 = get_value(dev.lu_bpm1)
ch2 = get_value(dev.lu_bpm2)
ch3 = get_value(dev.lu_bpm3)
ch4 = get_value(dev.lu_bpm4)
ch4 = get_value(dev.lu_bpm4)
_, y_error = bpm_errors(ch1, ch2, ch3, ch4)
print(f"BPM_Y = {float(y)}, Yerror = {y_error}")
@@ -1928,6 +1901,8 @@ def centre_lu_bpm_y():
umv(dev.lu_bpm_y, centre)
return centre
def read_i1():
return dev.i1.read()["i1"]["value"]
@@ -2159,10 +2134,11 @@ def measure_omega_runout(
# def dcm_pitch_calibration(start_energy=6000, end_energy=31000, step=3000):
def dcm_pitch_calibration():
"""Quick calibration of DCM Pitch against energy"""
energies = [5500, 8000, 10000, 12000, 14000, 16000, 18000,
20000, 22000, 24000, 27000, 30000]
# energies = [5500, 8000, 10000, 12000, 14000, 16000, 18000,
# 20000, 22000, 24000, 27000, 30000]
energies = [16000, 18000, 20000, 22000, 24000, 27000, 30000]
filename = bl.env.lut_dir / f"dcm_pitch_{filenow(date_only=True)}.csv"
with open(filename, "w", encoding="utf-8") as f:
with open(filename, "a", encoding="utf-8") as f:
f.write("energy,bragg,pitch\n")
# for energy in range(start_energy, end_energy, step):
try:
@@ -2199,9 +2175,7 @@ def dcm_pitch_calibration():
# import matplotlib.pyplot as plt
def analyse_roll_calibration(
energies=(6000, 9000, 12000, 15000, 18000, 21000),
):
def analyse_roll_calibration(energies=(6000, 9000, 12000, 15000, 18000, 21000)):
"""
Analyse roll calibration scans at several energies.
@@ -2237,20 +2211,13 @@ def analyse_roll_calibration(
fits[energy] = (m, b)
# Plot measured points
ax.scatter(
roll,
y,
label=f"{energy / 1000:g} keV",
)
ax.scatter(roll, y, label=f"{energy / 1000:g} keV")
# Plot fitted line
xx = np.linspace(roll.min(), roll.max(), 100)
ax.plot(xx, m * xx + b, "--")
print(
f"{energy / 1000:4.1f} keV: "
f"Y = {m:8.3f} × roll + {b:8.3f}"
)
print(f"{energy / 1000:4.1f} keV: " f"Y = {m:8.3f} × roll + {b:8.3f}")
# ------------------------------------------------------------
# Find roll where fitted lines have minimum spread
@@ -2265,12 +2232,7 @@ def analyse_roll_calibration(
best_roll = -np.sum(dm * db) / np.sum(dm**2)
# Predicted Y position at best roll
predicted_y = np.array(
[
m * best_roll + b
for m, b in fits.values()
]
)
predicted_y = np.array([m * best_roll + b for m, b in fits.values()])
mean_y = predicted_y.mean()
std_y = predicted_y.std()
@@ -2285,33 +2247,17 @@ def analyse_roll_calibration(
print("\nPredicted Y at best roll:")
for energy, y in zip(energies, predicted_y):
print(
f" {energy / 1000:4.1f} keV : "
f"{y:8.2f} px"
)
print(f" {energy / 1000:4.1f} keV : " f"{y:8.2f} px")
print(
f"\nMean Y = {mean_y:.2f} px"
f"\nSpread = {std_y:.2f} px"
)
print(f"\nMean Y = {mean_y:.2f} px" f"\nSpread = {std_y:.2f} px")
# ------------------------------------------------------------
# Plot optimum
# ------------------------------------------------------------
ax.axvline(
best_roll,
color="black",
linestyle=":",
label=f"Best roll = {best_roll:.4f}",
)
ax.axvline(best_roll, color="black", linestyle=":", label=f"Best roll = {best_roll:.4f}")
ax.axhline(
mean_y,
color="grey",
linestyle=":",
alpha=0.5,
)
ax.axhline(mean_y, color="grey", linestyle=":", alpha=0.5)
ax.set_xlabel("Roll")
ax.set_ylabel("Scintillator Y position (px)")
@@ -2329,6 +2275,7 @@ def analyse_roll_calibration(
"predicted_y": dict(zip(energies, predicted_y)),
}
def analyse_roll(low=6000, high=25000, low_file=None, high_file=None):
"""Find the DCM fine-roll value where low/high-energy beam positions intersect."""
@@ -2417,36 +2364,79 @@ def scan_roll(energy, start, stop, steps=9, settle=0.2, filename=None, overwrite
results = read_scintXY()
if not results["ok"]:
raise RuntimeError(
f"Beam position measurement failed at "
f"{energy / 1000:.1f} keV, roll={roll:.3f}: "
f"{results['message']}"
f"Beam position measurement failed at "
f"{energy / 1000:.1f} keV, roll={roll:.3f}: "
f"{results['message']}"
)
print(
f"{energy / 1000:.1f} keV | "
f"roll = {roll:.3f} | "
f"Y = {results['Y_pixels']:.2f}"
)
print(f"{energy / 1000:.1f} keV | " f"roll = {roll:.3f} | " f"Y = {results["Y_pixels"]:.2f}")
f.write(f"{roll:.6f},{results["Y_pixels"]}\n")
f.write(f"{roll:.6f},{results['Y_pixels']}\n")
finally:
finish_scint_measurement()
return filename
def measure_rolls():
def scan_roll_bcu(energy, start, stop, steps=9, settle=0.2, filename=None, overwrite=False):
"""Measure scintillator Y position as a function of DCM fine roll."""
scanpoints = np.linspace(start, stop, steps)
print(f"Scanning roll from {start} to {stop}")
if filename is None:
filename = bl.env.lut_dir / f"roll_{filenow(date_only=True)}_{energy:.0f}.csv"
if filename.exists() and not overwrite:
print(f"{filename.name} already exists.")
print("Use overwrite=True if you want to repeat this energy.")
return filename
bl_energy(energy)
with open(filename, "w", encoding="utf-8") as f:
f.write("Roll,Y pos\n")
for roll in scanpoints:
umv(dev.dcm_froll, roll)
time.sleep(settle)
result = get_bcu_beampos()
print(
f"{energy / 1000:.1f} keV | "
f"roll = {roll:.3f} | "
f"Y = {result['bcu_y_um']:.2f}"
)
f.write(f"{roll:.6f},{result['bcu_y_um']}\n")
return filename
def measure_rolls(monitor="scint"):
energies = [6000, 9000, 12000, 15000, 18000, 21000]
try:
for energy in energies:
scan_roll(energy, start=2, stop=5, overwrite=True)
finally:
print("done")
# bl_energy(12400)
if monitor == "scint":
try:
for energy in energies:
scan_roll(energy, start=2.0, stop=4, overwrite=True)
finally:
print("done")
# bl_energy(12400)
else:
try:
for energy in energies:
scan_roll_bcu(energy, start=2.8, stop=4, overwrite=True)
finally:
print("done")
# bl_energy(12400)
def find_best_roll(
low=6000,
high=25000,
low_range=(2.0, 4.0),
high_range=(1.0, 8.0),
steps=9,
):
def find_best_roll(low=6000, high=25000, low_range=(2.0, 4.0), high_range=(1.0, 8.0), steps=9):
"""Collect missing roll scans and calculate optimum roll."""
filename_prefix = filenow(date_only=True)
@@ -2454,17 +2444,12 @@ def find_best_roll(
high_file = bl.env.lut_dir / f"roll_{filename_prefix}_{high:.0f}.csv"
# Define the two scans
scans = [
(low, low_range, low_file),
(high, high_range, high_file),
]
scans = [(low, low_range, low_file), (high, high_range, high_file)]
# Start with whichever energy is closest to the current energy
current_energy = get_current_energy()
scans.sort(
key=lambda scan: abs(current_energy - scan[0])
)
scans.sort(key=lambda scan: abs(current_energy - scan[0]))
print(f"Current energy: {current_energy:.0f} eV")
print(f"Scanning {scans[0][0]:.0f} eV first")
@@ -2472,12 +2457,7 @@ def find_best_roll(
# Run scans in the chosen order
for energy, scan_range, filename in scans:
if not filename.exists():
scan_roll(
energy,
*scan_range,
steps=steps,
filename=filename,
)
scan_roll(energy, *scan_range, steps=steps, filename=filename)
else:
print(f"Using existing {filename.name}")
@@ -2577,6 +2557,7 @@ def test_vfm3(dmove, offset=0.002):
f"Position error = {(final-target)*1000:+.3f} µm"
)
def beamstop_check_energies():
bs_in = 0.904
@@ -2588,9 +2569,9 @@ def beamstop_check_energies():
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):
for energy in range(start_energy, end_energy, -2000):
print(f"Moving to energy {energy}")
bl_energy(energy)
dev.transm.put(0.1)
@@ -2614,6 +2595,7 @@ def beamstop_check_energies():
umv(dev.bs_y, bs_in)
bl_energy(12400)
def check_perp():
Xref = -0.0059957
Yref = -0.000993
@@ -2622,9 +2604,9 @@ def check_perp():
start_perp = get_value(dev.dcm_perp)
perps = [start_perp, start_perp-0.1, start_perp, start_perp+0.1, start_perp]
perps = [start_perp, start_perp - 0.1, start_perp, start_perp + 0.1, start_perp]
filename = bl.env.lut_dir / f"perp_check.csv"
with open(filename, 'w', encoding="utf-8") as f:
with open(filename, "w", encoding="utf-8") as f:
f.write("Delta perp, Pitch, Xnorm, Delta X")
for perp in perps:
@@ -2633,9 +2615,9 @@ def check_perp():
mono_pitch_scan()
pitch = get_value(dev.dcm_pitch)
Xnorm, Ynorm = compute_norm_ss()
Xdelta = -0.247*(Xnorm-Xref)
Ydelta = 0.0420*(Ynorm-Yref)
with open(filename, 'a', encoding="utf-8") as f:
Xdelta = -0.247 * (Xnorm - Xref)
Ydelta = 0.0420 * (Ynorm - Yref)
with open(filename, "a", encoding="utf-8") as f:
f.write(f"{delta_perp},{pitch:.5g},{Xnorm:.5g},{Xdelta:.5g}")
print(
f"Delta perp: {delta_perp}, "
@@ -2644,6 +2626,7 @@ def check_perp():
f"delta X: {Xdelta:.5g}"
)
def calibrate_dcm_perp():
"""
Determine the DCM perp position required to maintain a fixed beam
@@ -2679,16 +2662,7 @@ def calibrate_dcm_perp():
# beam displacement (mm) / perp displacement (mm)
beam_per_perp = 1.46
energies = [
6000,
8000,
10000,
12000,
16000,
20000,
24000,
30000,
]
energies = [6000, 8000, 10000, 12000, 16000, 20000, 24000, 30000]
filename = bl.env.lut_dir / "dcm_perp_calibration.csv"
@@ -2800,6 +2774,8 @@ def calibrate_dcm_perp():
print(filename)
finally:
bl_energy(12400)
from datetime import datetime
@@ -2833,10 +2809,7 @@ def scan_dcm_exit_vs_energy():
30000,
]
filename = (
bl.env.lut_dir
/ f"dcm_exit_scan_{datetime.now():%Y%m%d_%H%M%S}.csv"
)
filename = bl.env.lut_dir / f"dcm_exit_scan_{datetime.now():%Y%m%d_%H%M%S}.csv"
# --------------------------------------------------------------
# Create CSV
@@ -2869,7 +2842,6 @@ def scan_dcm_exit_vs_energy():
# Move to energy using current calibration
bl_energy(energy)
# Read actual DCM positions after optimisation
perp = get_value(dev.dcm_perp)
pitch = get_value(dev.dcm_pitch)
@@ -2923,13 +2895,16 @@ def scan_dcm_exit_vs_energy():
finally:
bl_energy(12400)
def perp_again():
energies = [8000, 10000, 12400, 15000, 18000, 20000, 22000, 25000]
perp_offsets = [-0.05, -0.025, 0.0, 0.025, 0.05, 0.0]
filename = bl.env.lut_dir / f"perp_{filenow()}.csv"
with open(filename, "w", encoding="utf-8") as f:
f.write("energy,dcm_pitch,dcm_perp,lu_bpmsum,lu_Xnorm,lu_Ynorm,bcu_bpmsum,bcu_x,bcu_y,samcam_x,samcam_y\n")
f.write(
"energy,dcm_pitch,dcm_perp,lu_bpmsum,lu_Xnorm,lu_Ynorm,bcu_bpmsum,bcu_x,bcu_y,samcam_x,samcam_y\n"
)
try:
for energy in energies:
bl_energy(energy)
@@ -2941,13 +2916,12 @@ def perp_again():
for offset in perp_offsets:
perp = perp_value + offset
umv(dev.dcm_perp,perp)
umv(dev.dcm_perp, perp)
mono_pitch_scan()
dcm_pitch = get_value(dev.dcm_pitch)
dcm_perp = get_value(dev.dcm_perp)
lu_bpmsum = get_value(dev.lu_bpmsum)
lu = compute_norm_lu()
lu_Xnorm = lu["lu_Xnorm"]
@@ -2961,7 +2935,7 @@ def perp_again():
scint = read_scintXY(image=False)
samcam_x = scint["X_diff_microns"]
samcam_y = scint["Y_diff_microns"]
with open(filename, "a", encoding="utf-8") as f:
f.write(
f"{energy:.0f},{dcm_pitch:.5f},{dcm_perp:.5f},{lu_bpmsum:.1f},{lu_Xnorm:.3f},{lu_Ynorm:.3f},"
@@ -2971,6 +2945,7 @@ def perp_again():
finish_scint_measurement()
bl_energy(12400)
def perp_ss_xi():
# energies = [8000, 10000, 12400, 15000, 18000, 20000, 22000, 25000]
energies = [25000]
@@ -2978,11 +2953,11 @@ def perp_ss_xi():
filename = bl.env.lut_dir / f"perp_xi_{filenow()}.csv"
with open(filename, "w", encoding="utf-8") as f:
f.write("energy,dcm_pitch,dcm_perp,lu_bpms,lu_Xnorm,lu_Ynorm,xi_cam_x,xi_cam_y\n")
f.write("energy,dcm_pitch,dcm_perp,lu_bpms,lu_Xnorm,lu_Ynorm,xi_cam_x,xi_cam_y\n")
try:
for energy in energies:
bl_energy(energy)
prepare_ss_xi_measurement()
perp_value = get_value(dev.dcm_perp)
@@ -2990,23 +2965,21 @@ def perp_ss_xi():
for offset in perp_offsets:
perp = perp_value + offset
umv(dev.dcm_perp,perp)
umv(dev.dcm_perp, perp)
mono_pitch_scan()
dcm_pitch = get_value(dev.dcm_pitch)
dcm_perp = get_value(dev.dcm_perp)
lu_bpmsum = get_value(dev.lu_bpmsum)
lu = compute_norm_lu()
lu_Xnorm = lu["lu_Xnorm"]
lu_Ynorm = lu["lu_Ynorm"]
xi_read = read_ss_xicamXY()
xicam_x = xi_read["X_pixels"]
xicam_y = xi_read["Y_pixels"]
with open(filename, "a", encoding="utf-8") as f:
f.write(
f"{energy:.0f},{dcm_pitch:.5f},{dcm_perp:.5f},{lu_bpmsum:.1f},{lu_Xnorm:.3f},{lu_Ynorm:.3f},"
@@ -3016,6 +2989,7 @@ def perp_ss_xi():
finish_ss_xi_measurement()
# bl_energy(12400)
def gap_beam():
# energies = [8000, 10000, 12400, 15000, 18000, 20000, 22000, 25000]
energies = [22000]
@@ -3023,13 +2997,13 @@ def gap_beam():
filename = bl.env.lut_dir / f"gap_xi_{filenow()}.csv"
filename2 = bl.env.lut_dir / f"gap_scint_{filenow()}.csv"
with open(filename, "w", encoding="utf-8") as f:
f.write("energy,gap,lu_bpmsum,lu_Xnorm,lu_Ynorm,xi_cam_x,xi_cam_y\n")
f.write("energy,gap,lu_bpmsum,lu_Xnorm,lu_Ynorm,xi_cam_x,xi_cam_y\n")
with open(filename2, "w", encoding="utf-8") as f:
f.write("energy,gap,lu_bpmsum,lu_Xnorm,lu_Ynorm,samcam_x,samcam_y\n")
f.write("energy,gap,lu_bpmsum,lu_Xnorm,lu_Ynorm,samcam_x,samcam_y\n")
try:
# for energy in energies:
# bl_energy(energy)
# prepare_ss_xi_measurement()
# scanpoints=np.linspace(4.535,4.555,20)
@@ -3047,7 +3021,7 @@ def gap_beam():
# xi_read = read_ss_xicamXY()
# xicam_x = xi_read["X_pixels"]
# xicam_y = xi_read["Y_pixels"]
# with open(filename, "a", encoding="utf-8") as f:
# f.write(
# f"{energy:.0f},{gap:.4f},{lu_bpmsum:.1f},{lu_Xnorm:.3f},{lu_Ynorm:.3f},"
@@ -3059,7 +3033,7 @@ def gap_beam():
finish_ss_xi_measurement()
prepare_scint_measurement()
scanpoints=np.linspace(4.535,4.555,20)
scanpoints = np.linspace(4.535, 4.555, 20)
for i in scanpoints:
umv(dev.id_gap, i)
time.sleep(0.2)
@@ -3071,33 +3045,30 @@ def gap_beam():
lu_Xnorm = lu["lu_Xnorm"]
lu_Ynorm = lu["lu_Ynorm"]
scint_read = read_scintXY(image=False)
samcam_x = scint_read["X_diff_microns"]
samcam_y = scint_read["Y_diff_microns"]
with open(filename2, "a", encoding="utf-8") as f:
f.write(
f"{energy:.0f},{gap:.4f},{lu_bpmsum:.1f},{lu_Xnorm:.3f},{lu_Ynorm:.3f},"
f"{samcam_x:.1f},{samcam_y:.1f}\n"
)
)
finally:
finish_ss_xi_measurement()
finish_scint_measurement()
# bl_energy(12400)
def gap_test():
for energy in range(6000, 30001, 500):
try:
harmonic, gap = Gap.calculate_gap(energy)
print(
f"{energy:5d} eV "
f"H{harmonic:2d} "
f"{gap:.4f} mm"
)
print(f"{energy:5d} eV " f"H{harmonic:2d} " f"{gap:.4f} mm")
except ValueError as exc:
print(energy, exc)
def energy_beam_position():
"""
Measure the beam trajectory as a function of energy.
@@ -3112,17 +3083,7 @@ def energy_beam_position():
No DCM perp correction or beam steering is applied.
"""
energies = [
6000,
8000,
10000,
12400,
15000,
18000,
20000,
22000,
25000,
]
energies = [6000, 8000, 10000, 12400, 15000, 18000, 20000, 22000, 25000]
filename = bl.env.lut_dir / f"energy_beam_position_{filenow()}.csv"
@@ -3208,7 +3169,7 @@ def energy_beam_position():
samcam_y = scint["Y_diff_microns"]
# Flux
_,flux = bpm2flux()
_, flux = bpm2flux()
# ----------------------------------------------------------
# Save immediately
@@ -3255,4 +3216,4 @@ def energy_beam_position():
print("\nReturning to 12400 eV")
bl_energy(12400)
print(f"\nResults saved to:\n{filename}")
print(f"\nResults saved to:\n{filename}")
pxii_bec/macros/pxii_parameters.py
+3 -4
View File
@@ -20,7 +20,7 @@ class EnergyDefaults:
mono_fpitch = dev.dcm_fpitch
mono_perp = dev.dcm_perp
mono_roll = dev.dcm_froll
mono_roll_value = 3.88
mono_roll_value = 3.186
stripe_thresholds = {"silicon": 9000, "rhodium": 20000, "platinum": 40000}
pitch_scan = {"halfwidth": 0.075, "steps": 20}
@@ -29,9 +29,8 @@ class EnergyDefaults:
class Calibration:
"""Calibration parameters for PXII optics"""
# pitch_calib = [-7.60082220e-19, 6.09893342e-14, -1.81702221e-09, 2.45457520e-05, -5.61275367e00]
pitch_calib = [-1.84031410e-07,-6.76777668e-05, -5.59684124]
# pitch_calib = [-1.84031410e-07,-6.76777668e-05, -5.59684124]
pitch_calib = [-1.75794156e-07, -6.68437530e-05, -5.66961048e+00]
# roll = np.array([2.28291039e-03, -2.41928101e01])
pxii_bec/macros/pxii_setup.py
+1 -1
View File
@@ -263,7 +263,7 @@ def scan_ge(edge=11103.1, mode="coarse"):
if mode == "coarse":
scan_range, scanpoints = 100, 41
if mode == "medium":
scan_range, scanpoints = 40, 41
scan_range, scanpoints = 40, 61
elif mode == "fine":
scan_range, scanpoints = 15, 61
else:
pxii_bec/macros/weekly.py
+36 -8
View File
@@ -105,15 +105,43 @@ def save_weekly_results(results):
"energy_offset_eV": format_value(edge_result.get("energy_offset"), ".2f")
}
exists = filename.exists()
with open(filename, "a", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=list(row))
if not exists:
writer.writeheader()
writer.writerow(row)
print(f"\nWeekly results saved to:\n{filename}")
return filename
fieldnames = list(row.keys())
file_exists = filename.exists()
# Check whether the existing CSV has the current schema
if file_exists:
with open(filename, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
existing_header = next(reader, None)
if existing_header != fieldnames:
# Archive the old CSV rather than writing incompatible rows to it
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_filename = filename.with_name(
f"{filename.stem}_archive_{timestamp}{filename.suffix}"
)
filename.rename(archive_filename)
print(
"⚠ Weekly beamline CSV schema has changed.\n"
f" Previous history archived as: {archive_filename.name}\n"
f" Starting new history file: {filename.name}"
)
file_exists = False
with open(filename, "a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
if not file_exists:
writer.writeheader()
writer.writerow(row)
print(f"✓ Weekly beamline history saved to {filename}")
return filename
def send_week_to_scilog(filename=None):
if filename is None: