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

updates from pxii

Browse Source
This commit is contained in:
x06da
2026-08-04 09:54:32 +02:00
committed by perl_d
parent d25bd17f57
commit 946a278d16
16 changed files with 406 additions and 1525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pxiii_bec/device_configs/pxiii-beamline-states.yaml
+1 -1
View File
@@ -109,7 +109,7 @@ states:
bl_bright: 'off'
bs_pos: in
bs_z: safe
coll_y: out
coll_y: in
cryo_pos: out
det_cov: 'close'
diag_y: scint
pxiii_bec/device_configs/pxiii-rest-devices.yaml
-18
View File
@@ -1,18 +0,0 @@
aerotech_x:
userParameter: {"type": continuous, "in": 0.0, "out": -10.0, "safe": -100, "tol": 0.5}
aerotech_y:
userParameter: {"type": continuous, "mount": 0.0, "work": 0.01, "tol": 0.002}
aerotech_z:
userParameter: {"type": continuous, "mount": 0.0, "work": 0.02, "tol": 0.01}
aerotech_u:
userParameter: {"type": continuous, "mount": 0.0}
smargon_x:
userParameter: {"type": continuous, "mount": 0.0}
smargon_y:
userParameter: {"type": continuous, "mount": 0.0}
smargon_z:
userParameter: {"type": continuous, "mount": 0.0}
smargon_chi:
userParameter: {"type": continuous, "mount": 0.0}
smargon_phi:
userParameter: {"type": continuous, "mount": 0.0}
pxiii_bec/device_configs/pxiii-standard-devices.yaml
-1333
View File
File diff suppressed because it is too large Load Diff
pxiii_bec/device_configs/pxiii-state-devices.yaml
+1 -1
View File
@@ -9,7 +9,7 @@ bl_bright:
- state
readOnly: False
softwareTrigger: false
userParameter: {"type": continuous, "on": 1.3, "off": 0, “tol”: 0.01}
userParameter: {"type": continuous, "on": 1.56, "off": 0, “tol”: 0.01}
bl_pos:
description: Backlight Positioner
pxiii_bec/device_configs/x06da_device_config.yaml
+3 -28
View File
@@ -1,30 +1,5 @@
base_config:
- !include ./pxiii-standard-devices.yaml
states_config:
- !include ./pxiii-state-devices.yaml
- !include ./pxiii-devices.yaml
#states_config:
# - !include ./pxiii-state-devices.yaml
smargon:
description: REST-based device which connects to Smargopolo
deviceClass: pxiii_bec.devices.smargopolo_smargon.Smargon
deviceConfig: {prefix: 'http://x06da-smargopolo.psi.ch:3000'}
onFailure: buffer
enabled: True
readoutPriority: baseline
deviceTags:
- smargon
- motors
readOnly: false
softwareTrigger: false
aerotech:
description: REST-based device which connects to AareScan and Aerotech
deviceClass: pxiii_bec.devices.aerotech.Aerotech
deviceConfig: {prefix: 'http://mx-x06da-queue-01:5234'}
onFailure: buffer
enabled: True
readoutPriority: baseline
deviceTags:
- aerotech
- motors
readOnly: false
softwareTrigger: false
pxiii_bec/macros/beamline_planner.py
+29 -9
View File
@@ -1,7 +1,32 @@
"""Planner to move between beamline statesΩ"""
"""Planner to move between beamline states"""
import time
from collections import defaultdict, deque
from enum import Enum
# from enums import BeamlineState, TemperatureMode
# from matcher import DeviceMatcher, nonzero_is_on
class BeamlineState(str, Enum):
"""List of beamline states"""
ROBOT_SAMPLE_EXCHANGE = "robot_sample_exchange"
SAMPLE_ALIGNMENT = "sample_alignment"
DATA_COLLECTION = "data_collection"
DC_XRF = "DC_XRF"
MANUAL_SAMPLE_EXCHANGE = "manual_sample_exchange"
BEAM_VISUALISATION = "beam_visualisation"
FLUX_MEASUREMENT = "flux_measurement"
BEAMSTOP_ALIGNMENT = "beamstop_alignment"
MAINTENANCE = "maintenance"
XTAL_SNAPSHOT = "xtal_snapshot"
class TemperatureMode(str, Enum):
"""List of temperature modes"""
CRYO = "cryo"
ROOM_TEMP = "room_temp"
class StateChangePlanner:
@@ -37,7 +62,6 @@ class StateChangePlanner:
target.update(self.modifiers[modifier])
return target
def move_to(self, state_name, modifier=None):
"""Move devices to the correct positions to achieve a given state"""
if isinstance(state_name, str):
@@ -102,12 +126,8 @@ class StateChangePlanner:
config = self._merged_state(state, modifier)
# Check if all devices match their expected positions
if all(self.devices[d].is_at(p)
for d, p in config.items()
if d not in ignore_keys
):
# print(f"Current state: {state.name}, Modifier: {modifier.name if modifier else 'None'}")
matches.append((state.name, modifier.name if modifier else 'None'))
if all(self.devices[d].is_at(p) for d, p in config.items() if d not in ignore_keys):
matches.append((state.name, modifier.name if modifier else "None"))
return matches if matches else None
pxiii_bec/macros/build_devices.py
+317 -25
View File
@@ -1,14 +1,15 @@
""" Build the sample environment devices"""
"""Build the sample environment devices"""
import os
import yaml
# from position_device import PositionDevice
def motor_resolver(bec_name):
candidates = [
bec_name,
bec_name.replace("_", "."),
]
def motor_resolver(bec_name):
"""Used to handle aerotech and smargon motors"""
candidates = [bec_name, bec_name.replace("_", ".")]
for path in candidates:
try:
@@ -24,9 +25,9 @@ def motor_resolver(bec_name):
raise ValueError(f"Cannot resolve motor for '{bec_name}'")
def build_devices(yaml_file, mock_devices):
""" Build devices from the beamline states yaml"""
def build_devices(yaml_file, mock_devices):
"""Build devices from the beamline states yaml"""
state_devices = {}
@@ -41,30 +42,321 @@ def build_devices(yaml_file, mock_devices):
if not user:
continue
if bec_name in ("aerotech", "smargon"):
tol = user.get("tol", 0.1)
tol = user.get("tol", 0.01)
positions = {
k: v for k, v in user.items()
if k not in ("type", "tol")
}
for axis in ["x", "y", "z", "u", "chi", "phi"]:
allow_arbitrary = (user["type"] == "continuous")
positions = user.get(axis)
pos_dev = PositionDevice(
bec_name=bec_name,
mot_device = motor_resolver(bec_name),
positions=positions,
tol=tol,
allow_arbitrary=allow_arbitrary,
use_mock=bec_name in mock_devices,
)
if positions is None:
continue
component_name = f"{bec_name}_{axis}"
pos_dev = PositionDevice(
bec_name=component_name,
mot_device=motor_resolver(component_name),
positions=positions,
tol=tol,
allow_arbitrary=True,
use_mock=component_name in mock_devices,
)
state_devices[component_name] = pos_dev
else:
tol = user.get("tol", 0.1)
allow_arbitrary = user.get("type") == "continuous"
positions = {k: v for k, v in user.items() if k not in ("type", "tol")}
pos_dev = PositionDevice(
bec_name=bec_name,
mot_device=motor_resolver(bec_name),
positions=positions,
tol=tol,
allow_arbitrary=allow_arbitrary,
use_mock=bec_name in mock_devices,
)
state_devices[bec_name] = pos_dev
state_devices[bec_name] = pos_dev
return state_devices
def update_position(
device,
position,
value,
axis=None,
force=False,
max_delta=0.5,
):
"""
Update a single saved position in memory.
Does NOT save or reload. Use save_and_reload() or
update_positions() afterwards.
"""
params = device.user_parameter
# ------------------------------------------------------------------
# Nested device (e.g. aerotech.x.in)
# ------------------------------------------------------------------
if axis is not None:
if axis not in params:
raise KeyError(
f"Unknown axis '{axis}'. "
f"Valid axes: {list(params.keys())}"
)
axis_params = params[axis].copy()
if position not in axis_params:
raise KeyError(
f"Unknown position '{position}'. "
f"Valid positions: {list(axis_params.keys())}"
)
old_value = axis_params[position]
delta = abs(value - old_value)
if delta > max_delta and not force:
raise ValueError(
f"Refusing to change "
f"{device.name}.{axis}.{position} "
f"from {old_value} to {value} "
f"(Δ={delta:.3f} > {max_delta}). "
f"Use force=True if intentional."
)
axis_params[position] = value
device.update_user_parameter(
{axis: axis_params}
)
print(
f"Updated {device.name}.{axis}.{position}: "
f"{old_value} -> {value}"
)
return
# ------------------------------------------------------------------
# Flat device (e.g. bs_z)
# ------------------------------------------------------------------
nested_keys = [
k for k, v in params.items()
if isinstance(v, dict)
]
if nested_keys:
raise ValueError(
f"{device.name} has nested parameters "
f"({nested_keys}); specify axis=..."
)
if position not in params:
raise KeyError(
f"Unknown position '{position}'. "
f"Valid positions: {list(params.keys())}"
)
old_value = params[position]
delta = abs(value - old_value)
if delta > max_delta and not force:
raise ValueError(
f"Refusing to change "
f"{device.name}.{position} "
f"from {old_value} to {value} "
f"(Δ={delta:.3f} > {max_delta}). "
f"Use force=True if intentional."
)
device.update_user_parameter(
{position: value}
)
print(
f"Updated {device.name}.{position}: "
f"{old_value} -> {value}"
)
def save_and_reload():
"""
Save current configuration to YAML and rebuild
the beamline environment.
"""
beamline = os.environ["BEAMLINE"]
env = Environment(beamline=beamline)
print(f"(saving changes to {env.devices_file})")
bec.config.save_current_session(env.devices_file)
return init_beamline_environment()
def update_positions(updates):
"""
Apply many updates, then save and reload once.
updates = [
{
"device": dev.aerotech,
"axis": "x",
"position": "in",
"value": 0,
},
{
"device": dev.bs_z,
"position": "max_blout",
"value": 68,
},
]
"""
for update in updates:
update_position(
device=update["device"],
position=update["position"],
value=update["value"],
axis=update.get("axis"),
force=update.get("force", False),
max_delta=update.get("max_delta", 0.5),
)
return save_and_reload()
def save_current_position(
device,
position,
axis=None,
force=False,
max_delta=0.5,
):
"""
Save the current motor position as a named position.
"""
print(f"Saving current postion for {device}")
params = device.user_parameter
if axis is not None:
if axis not in params:
raise KeyError(
f"Unknown axis '{axis}'. "
f"Valid axes: {list(params.keys())}"
)
axis_params = params[axis].copy()
if position not in axis_params:
raise KeyError(
f"Unknown position '{position}'. "
f"Valid positions: {list(axis_params.keys())}"
)
old_value = axis_params[position]
print(f"Old value is {old_value}")
current_value = get_current_value(device, axis=None)
delta = abs(current_value - old_value)
if delta > max_delta and not force:
raise ValueError(
f"Current position differs from saved value:\n"
f" saved = {old_value}\n"
f" current = {current_value}\n"
f" delta = {delta:.3f}\n"
f"Use force=True if intentional."
)
axis_params[position] = current_value
device.update_user_parameter(
{axis: axis_params}
)
print(
f"Updated {device.name}.{axis}.{position}: "
f"{old_value} -> {current_value}"
)
return
# ------------------------------------------------------------------
# Flat device (e.g. bs_z)
# ------------------------------------------------------------------
nested_keys = [
k for k, v in params.items()
if isinstance(v, dict)
]
if nested_keys:
raise ValueError(
f"{device.name} has nested parameters "
f"({nested_keys}); specify axis=..."
)
if position not in params:
raise KeyError(
f"Unknown position '{position}'. "
f"Valid positions: {list(params.keys())}"
)
old_value = params[position]
current_value = get_current_value(device, axis=None)
delta = abs(current_value - old_value)
if delta > max_delta and not force:
raise ValueError(
f"Refusing to change "
f"{device.name}.{position} "
f"from {old_value} to {current_value} "
f"(Δ={delta:.3f} > {max_delta}). "
f"Use force=True if intentional."
)
device.update_user_parameter(
{position: current_value}
)
print(
f"Updated {device.name}.{position}: "
f"{old_value} -> {current_value}"
)
def get_current_value(device, axis=None):
"""
Return current motor position.
"""
if axis is not None:
motor = getattr(device, axis)
else:
motor = device
data = motor.read()
signal_name = next(iter(data))
return round(data[signal_name]["value"],3)
pxiii_bec/macros/dependencies.py
+7 -23
View File
@@ -1,30 +1,14 @@
"""Planner dependencies"""
def planner_deps():
"""Define the dependencies between beamline positions"""
return {
("bs_z", "samp"): [
("aerotech_x", "out"),
("diag_y", "out"),
("coll_y", "out"),
],
("aerotech_x", "in"): [
("diag_y", "out"),
("bs_z", "safe"),
],
("aerotech_x", "out"): [
("diag_y", "out"),
("bs_z", "safe"),
],
("diag_y", "scint"): [
("aerotech_x", "out"),
("bs_z", "safe"),
("cryo_pos", "out"),
],
("diag_y", "i1"): [
("aerotech_x", "out"),
("bs_z", "safe"),
("cryo_pos", "out"),
],
("bs_z", "samp"): [("aerotech_x", "out"), ("diag_y", "out"), ("coll_y", "out")],
("aerotech_x", "in"): [("diag_y", "out"), ("bs_z", "safe")],
("aerotech_x", "out"): [("diag_y", "out"), ("bs_z", "safe")],
("diag_y", "scint"): [("aerotech_x", "out"), ("bs_z", "safe"), ("cryo_pos", "out")],
("diag_y", "i1"): [("aerotech_x", "out"), ("bs_z", "safe"), ("cryo_pos", "out")],
("bs_pos", "out"): [("bs_z", "safe")],
("bs_pos", "in"): [("bs_z", "safe")],
("diag_y", "out"): [("bs_z", "safe")],
pxiii_bec/macros/devices_manager.py
+8 -7
View File
@@ -1,22 +1,23 @@
"""
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:
"""Class for building devices and attaching safety guards and policies."""
@staticmethod
def initialize_devices(state_devices_file, rest_devices_file, mock_devices):
# def initialize_devices(state_devices_file, rest_devices_file, mock_devices):
def initialize_devices(devices_file, mock_devices):
"""
Initializes sample environment devices from the specified file.
"""
devices = build_devices(state_devices_file, mock_devices)
rest_devices = build_devices(rest_devices_file, mock_devices)
devices.update(rest_devices)
devices = build_devices(devices_file, mock_devices)
# rest_devices = build_devices(rest_devices_file, mock_devices)
# devices.update(rest_devices)
attach_guards(devices)
attach_policies(devices)
return devices
pxiii_bec/macros/enums.py
-22
View File
@@ -1,22 +0,0 @@
"""Enums for beamline states"""
from enum import Enum
class BeamlineState(str, Enum):
"""List of beamline states"""
ROBOT_SAMPLE_EXCHANGE = "robot_sample_exchange"
SAMPLE_ALIGNMENT = "sample_alignment"
DATA_COLLECTION = "data_collection"
DC_XRF = "DC_XRF"
MANUAL_SAMPLE_EXCHANGE = "manual_sample_exchange"
BEAM_VISUALISATION = "beam_visualisation"
FLUX_MEASUREMENT = "flux_measurement"
BEAMSTOP_ALIGNMENT = "beamstop_alignment"
MAINTENANCE = "maintenance"
XTAL_SNAPSHOT = "xtal_snapshot"
class TemperatureMode(str, Enum):
"""List of temperature modes"""
CRYO = "cryo"
ROOM_TEMP = "room_temp"
pxiii_bec/macros/guards.py
+1 -3
View File
@@ -15,9 +15,7 @@ class AtPositionGuard:
def check(self):
"""Check if the device is in the specified position."""
if self.device.pos != self.pos:
raise GuardViolation(
f"{self.device.bec_name} must be in the '{self.pos}' position"
)
raise GuardViolation(f"{self.device.bec_name} must be in the '{self.pos}' position")
# print("move allowed")
return True
pxiii_bec/macros/init_beamline.py
+19 -39
View File
@@ -8,6 +8,7 @@ from dataclasses import dataclass
# from dependencies import planner_deps
# from beamline_planner import StateChangePlanner
@dataclass
class Environment:
@@ -21,18 +22,16 @@ class Environment:
# "rest": "pxi-rest-devices.yaml",
# "states": "pxi-beamline-states.yaml",
# },
"x10sa" : {
"x10sa": {
"devdir": "/sls/x10sa/config/bec/production/pxii_bec/pxii_bec/device_configs/",
"state": "pxii-state-devices.yaml",
"rest": "pxii-rest-devices.yaml",
"devices": "pxii-devices.yaml",
"states": "pxii-beamline-states.yaml",
},
"x06da" : {
"x06da": {
"devdir": "/sls/x06da/config/bec/production/pxiii_bec/pxiii_bec/device_configs/",
"state": "pxiii-state-devices.yaml",
"rest": "pxiii-rest-devices.yaml",
"devices": "pxiii-devices.yaml",
"states": "pxiii-beamline-states.yaml",
}
},
}
def __post_init__(self):
@@ -60,34 +59,23 @@ class Environment:
"xrf_pos": "mock",
}
mocks = sorted(
name
for name, backend in self.device_mocks.items()
if backend == "mock"
)
mocks = sorted(name for name, backend in self.device_mocks.items() if backend == "mock")
reals = sorted(
name
for name, backend in self.device_mocks.items()
if backend == "real"
)
reals = sorted(name for name, backend in self.device_mocks.items() if backend == "real")
print(f"Mock devices ({len(mocks)}): {mocks}")
print(f"Real devices ({len(reals)}): {reals}")
# print(f"Mock devices ({len(mocks)}): {mocks}")
# print(f"Real devices ({len(reals)}): {reals}")
try:
cfg = self.beamline_configs[self.beamline]
except KeyError:
raise ValueError(
f"Unknown beamline '{self.beamline}'"
)
raise ValueError(f"Unknown beamline '{self.beamline}'")
devdir = cfg["devdir"]
self.state_devices_file = devdir + cfg["state"]
self.rest_devices_file = devdir + cfg["rest"]
self.devices_file = devdir + cfg["devices"]
self.states_file = devdir + cfg["states"]
mock_start_positions = {
"aerotech_x": "in",
"aerotech_y": "mount",
@@ -107,7 +95,7 @@ class Environment:
"smargon_z": "mount",
"smargon_chi": "mount",
"smargon_phi": "mount",
"xrf_pos": "out",
"xrf_pos": "out",
}
@property
@@ -116,7 +104,8 @@ class Environment:
for name, device in self.device_mocks.items():
if device == "mock":
mock_names.add(name)
return mock_names
return mock_names
def init_beamline_environment():
"""
@@ -125,15 +114,11 @@ def init_beamline_environment():
beamline = os.environ["BEAMLINE"]
env = Environment(beamline = beamline)
env = Environment(beamline=beamline)
# Initialize devices
device_manager = DeviceManager()
devices = device_manager.initialize_devices(
env.state_devices_file,
env.rest_devices_file,
env.mock_devices
)
devices = device_manager.initialize_devices(env.devices_file, env.mock_devices)
# Initialize states
state_manager = DefineStatesManager()
@@ -144,7 +129,7 @@ def init_beamline_environment():
# Setup planner
planner = StateChangePlanner(devices, states, allow_modifiers, deps)
print("Initializing beamline state planner")
print("Initializing beamline state planner and devices")
# Set mock motor starting positions
@@ -155,9 +140,4 @@ def init_beamline_environment():
except Exception as e:
print(f"{e} error when setting {mock_dev}")
return devices, planner
pxiii_bec/macros/mx_methods.py
+11
View File
@@ -17,6 +17,17 @@ import numpy as np
# )
@dataclass(frozen=True)
class FitDefaults:
"""Default values for fitting routines"""
# Constants for default models, baselines, and parameters
MODEL = "Voigt"
BASELINE = "Linear"
SETTLE_TIME = 0.1
RELATIVE_MODE = True
# Method functions
def calculate_step_size(start: float, stop: float, steps: int) -> float:
"""
pxiii_bec/macros/policies.py
+3 -2
View File
@@ -58,15 +58,16 @@ def make_diag_y_policy(d):
def diag_y_policy(target):
cfg = d["diag_y"].positions
# Don't move in if the goniometer is in
if d["aerotech_x"].actual >= d['aerotech_x'].positions['in'] and target > cfg["out"]:
if d["aerotech_x"].actual >= d["aerotech_x"].positions["in"] and target > cfg["out"]:
raise GuardViolation(
f"Diagnostic device cannot move beyond {cfg['out']} mm when goniometer is not OUT"
)
# Don't move if cryocooler is in
if d['cryo_pos'].pos == 'in' and target > cfg['out']:
if d["cryo_pos"].pos == "in" and target > cfg["out"]:
raise GuardViolation(
f"Diagnostic device cannot move beyond {cfg['out']} mm when cryocooler is IN"
)
return diag_y_policy
pxiii_bec/macros/pxiii_energy.py
+5 -3
View File
@@ -147,7 +147,7 @@ def mono_pitch_scan(plot=True):
print("Scanning monochromator pitch and moving to peak, with plotting.")
go_to_peak(
EnergyDefaults.mono_pitch,
EnergyDefaults.signals["sig1"],
EnergyDefaults.signals["sig2"],
-EnergyDefaults.pitch_scan["halfwidth"],
EnergyDefaults.pitch_scan["halfwidth"],
steps=EnergyDefaults.pitch_scan["steps"],
@@ -160,7 +160,7 @@ def mono_pitch_scan(plot=True):
print("Scanning monochromator pitch and moving to peak, without plotting.")
go_to_peak(
EnergyDefaults.mono_pitch,
EnergyDefaults.signals["sig1"],
EnergyDefaults.signals["sig2"],
-EnergyDefaults.pitch_scan["halfwidth"],
EnergyDefaults.pitch_scan["halfwidth"],
steps=EnergyDefaults.pitch_scan["steps"],
@@ -189,8 +189,10 @@ def get_dccm_motors_positions(energy_ev):
# dccm_motor_values = get_value_from_lut(energy_ev)
th1_angle = -convert_from_energy(energy_ev, print_result=False)["bragg_angle_deg"]
th2_angle = convert_from_energy(energy_ev, temp=298, print_result=False)["bragg_angle_deg"]
# add fudge factor
th2_angle_fudge = th2_angle + 0.006
# dccm_motor_values.update({"theta1_angle": th1_angle, "theta2_angle": th2_angle})
dccm_motor_values = {"theta1_angle": th1_angle, "theta2_angle": th2_angle}
dccm_motor_values = {"theta1_angle": th1_angle, "theta2_angle": th2_angle_fudge}
return dccm_motor_values
pxiii_bec/macros/pxiii_parameters.py
+1 -11
View File
@@ -4,16 +4,6 @@ from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class FitDefaults:
"""Default values for fitting routines"""
# Constants for default models, baselines, and parameters
MODEL = "Voigt"
BASELINE = "Linear"
SETTLE_TIME = 0.1
RELATIVE_MODE = True
@dataclass(frozen=True)
class EnergyDefaults:
@@ -32,7 +22,7 @@ class EnergyDefaults:
mono_pitch = dev.dccm_theta2
# LUT_table = "luts/energy_lut.csv"
stripe_thresholds = {"silicon": 9000, "rhodium": 40000}
pitch_scan = {"halfwidth": 0.15, "steps": 30}
pitch_scan = {"halfwidth": 0.01, "steps": 25}
@dataclass(frozen=True)