This commit is contained in:
x06da
@@ -1,143 +0,0 @@
|
||||
"""Initialise sample environment devices and beamline states"""
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
|
||||
# from devices_manager import DeviceManager
|
||||
# from beamline_state_manager import DefineStatesManager
|
||||
# from dependencies import planner_deps
|
||||
# from beamline_planner import StateChangePlanner
|
||||
|
||||
|
||||
@dataclass
|
||||
class Environment:
|
||||
|
||||
device_mocks: dict[str, bool] = None
|
||||
beamline: str = None
|
||||
|
||||
beamline_configs = {
|
||||
# "x06sa" : {
|
||||
# "devdir": "/sls/x10sa/config/bec/production/pxi_bec/pxi_bec/device_configs/",
|
||||
# "state": "pxi-state-devices.yaml",
|
||||
# "rest": "pxi-rest-devices.yaml",
|
||||
# "states": "pxi-beamline-states.yaml",
|
||||
# },
|
||||
"x10sa": {
|
||||
"devdir": "/sls/x10sa/config/bec/production/pxii_bec/pxii_bec/device_configs/",
|
||||
"devices": "pxii-devices.yaml",
|
||||
"states": "pxii-beamline-states.yaml",
|
||||
},
|
||||
"x06da": {
|
||||
"devdir": "/sls/x06da/config/bec/production/pxiii_bec/pxiii_bec/device_configs/",
|
||||
"devices": "pxiii-devices.yaml",
|
||||
"states": "pxiii-beamline-states.yaml",
|
||||
},
|
||||
}
|
||||
|
||||
def __post_init__(self):
|
||||
|
||||
if self.device_mocks is None:
|
||||
self.device_mocks = {
|
||||
"aerotech_x": "real",
|
||||
"aerotech_y": "real",
|
||||
"aerotech_z": "real",
|
||||
"aerotech_u": "real",
|
||||
"bl_bright": "real",
|
||||
"bl_pos": "real",
|
||||
"bs_pos": "real",
|
||||
"bs_z": "real",
|
||||
"coll_y": "real",
|
||||
"cryo_pos": "real",
|
||||
"det_cov": "real",
|
||||
"diag_y": "real",
|
||||
"fl_bright": "real",
|
||||
"smargon_x": "real",
|
||||
"smargon_y": "real",
|
||||
"smargon_z": "real",
|
||||
"smargon_chi": "real",
|
||||
"smargon_phi": "real",
|
||||
"xrf_pos": "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")
|
||||
|
||||
# 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}'")
|
||||
|
||||
devdir = cfg["devdir"]
|
||||
|
||||
self.devices_file = devdir + cfg["devices"]
|
||||
self.states_file = devdir + cfg["states"]
|
||||
|
||||
mock_start_positions = {
|
||||
"aerotech_x": "in",
|
||||
"aerotech_y": "mount",
|
||||
"aerotech_z": "mount",
|
||||
"aerotech_u": "mount",
|
||||
"bl_bright": "off",
|
||||
"bl_pos": "out",
|
||||
"bs_pos": "in",
|
||||
"bs_z": "safe",
|
||||
"coll_y": "park",
|
||||
"cryo_pos": "in",
|
||||
"det_cov": "close",
|
||||
"diag_y": "park",
|
||||
"fl_bright": "off",
|
||||
"smargon_x": "mount",
|
||||
"smargon_y": "mount",
|
||||
"smargon_z": "mount",
|
||||
"smargon_chi": "mount",
|
||||
"smargon_phi": "mount",
|
||||
"xrf_pos": "out",
|
||||
}
|
||||
|
||||
@property
|
||||
def mock_devices(self):
|
||||
mock_names = set()
|
||||
for name, device in self.device_mocks.items():
|
||||
if device == "mock":
|
||||
mock_names.add(name)
|
||||
return mock_names
|
||||
|
||||
|
||||
def init_beamline_environment():
|
||||
"""
|
||||
Initializes the beamline with real or mock devices.
|
||||
"""
|
||||
|
||||
beamline = os.environ["BEAMLINE"]
|
||||
|
||||
env = Environment(beamline=beamline)
|
||||
|
||||
# Initialize devices
|
||||
device_manager = DeviceManager()
|
||||
devices = device_manager.initialize_devices(env.devices_file, env.mock_devices)
|
||||
|
||||
# Initialize states
|
||||
state_manager = DefineStatesManager()
|
||||
states, allow_modifiers = state_manager.initialize_states(env.states_file)
|
||||
|
||||
# Setup dependencies
|
||||
deps = planner_deps()
|
||||
|
||||
# Setup planner
|
||||
planner = StateChangePlanner(devices, states, allow_modifiers, deps)
|
||||
print("Initializing beamline state planner and devices")
|
||||
|
||||
# Set mock motor starting positions
|
||||
|
||||
for mock_dev, pos in env.mock_start_positions.items():
|
||||
if mock_dev in env.mock_devices:
|
||||
try:
|
||||
devices[mock_dev].set_position(pos)
|
||||
except Exception as e:
|
||||
print(f"{e} error when setting {mock_dev}")
|
||||
|
||||
return devices, planner
|
||||
@@ -0,0 +1,23 @@
|
||||
"""Shared beamline context.
|
||||
Provides access to the current beamline environment,
|
||||
positioned devices, and beamline state planner."""
|
||||
|
||||
from .init_beamline import init_beamline_environment
|
||||
|
||||
env = None
|
||||
d = None
|
||||
planner = None
|
||||
|
||||
|
||||
def initialise():
|
||||
"""Initialise the beamline environment."""
|
||||
global env, d, planner
|
||||
|
||||
env, d, planner = init_beamline_environment()
|
||||
|
||||
print("Beamline environment, devices and planner have been initialised")
|
||||
|
||||
|
||||
def reload():
|
||||
"""Re-initialise the beamline environment"""
|
||||
initialise()
|
||||
Executable
+114
@@ -0,0 +1,114 @@
|
||||
"""Initialise sample environment devices and beamline states."""
|
||||
|
||||
import os
|
||||
import importlib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass
|
||||
class Environment:
|
||||
"""Define configuration files for a beamline and BEC deployment."""
|
||||
|
||||
beamline: str
|
||||
|
||||
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",
|
||||
},
|
||||
}
|
||||
|
||||
valid_deployments = ("test", "production")
|
||||
|
||||
def __post_init__(self):
|
||||
|
||||
if self.beamline not in self.beamline_configs:
|
||||
raise ValueError(
|
||||
f"Unknown beamline '{self.beamline}'. " f"Choose from {list(self.beamline_configs)}"
|
||||
)
|
||||
|
||||
cfg = self.beamline_configs[self.beamline]
|
||||
self.beamline_name = cfg["name"]
|
||||
|
||||
# Import whichever deployment BEC has already made available
|
||||
package = importlib.import_module(cfg["package"])
|
||||
|
||||
self.bec_dir = Path(package.__file__).resolve().parent
|
||||
|
||||
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"]
|
||||
|
||||
# Determine which deployment is being used
|
||||
self.deployment = self._get_deployment()
|
||||
|
||||
if not self.devices_file.exists():
|
||||
raise FileNotFoundError(f"Device configuration not found: {self.devices_file}")
|
||||
|
||||
if not self.states_file.exists():
|
||||
raise FileNotFoundError(f"State configuration not found: {self.states_file}")
|
||||
|
||||
def _get_deployment(self):
|
||||
"""Determine deployment from the loaded package path."""
|
||||
|
||||
for deployment in ("test", "production"):
|
||||
if deployment in self.bec_dir.parts:
|
||||
return deployment
|
||||
|
||||
raise RuntimeError(f"Could not determine BEC deployment from {self.bec_dir}")
|
||||
|
||||
|
||||
def init_beamline_environment():
|
||||
"""
|
||||
Initialise beamline positioned devices and state planner.
|
||||
"""
|
||||
|
||||
beamline = os.environ["BEAMLINE"]
|
||||
|
||||
env = Environment(
|
||||
beamline=beamline,
|
||||
)
|
||||
|
||||
print("=" * 60)
|
||||
print(f"Beamline : {beamline}")
|
||||
print(f"BEC deployment : {env.deployment.upper()}")
|
||||
print(f"Devices file : {env.devices_file}")
|
||||
print(f"States file : {env.states_file}")
|
||||
print("=" * 60)
|
||||
|
||||
# Initialise positioned devices
|
||||
device_manager = DeviceManager()
|
||||
devices = device_manager.initialize_devices(env.devices_file)
|
||||
|
||||
# Initialise beamline states
|
||||
state_manager = DefineStatesManager()
|
||||
states, allow_modifiers = state_manager.initialize_states(env.states_file)
|
||||
|
||||
# Set up dependencies
|
||||
deps = planner_deps()
|
||||
|
||||
# Set up planner
|
||||
planner = StateChangePlanner(devices, states, allow_modifiers, deps)
|
||||
|
||||
return env, devices, planner
|
||||
Reference in New Issue
Block a user