From 2eff41bf6411fc44ed182767a8739f4bd1bb0fa6 Mon Sep 17 00:00:00 2001 From: x06da Date: Sun, 4 Oct 2026 16:59:55 +0200 Subject: [PATCH] updates to beamline initialisation --- pxiii_bec/macros/init_beamline.py | 143 -------------------------- pxiii_bec/scripts/beamline_context.py | 23 +++++ pxiii_bec/scripts/init_beamline.py | 114 ++++++++++++++++++++ 3 files changed, 137 insertions(+), 143 deletions(-) delete mode 100644 pxiii_bec/macros/init_beamline.py create mode 100644 pxiii_bec/scripts/beamline_context.py create mode 100755 pxiii_bec/scripts/init_beamline.py diff --git a/pxiii_bec/macros/init_beamline.py b/pxiii_bec/macros/init_beamline.py deleted file mode 100644 index 78330d6..0000000 --- a/pxiii_bec/macros/init_beamline.py +++ /dev/null @@ -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 diff --git a/pxiii_bec/scripts/beamline_context.py b/pxiii_bec/scripts/beamline_context.py new file mode 100644 index 0000000..3e0de47 --- /dev/null +++ b/pxiii_bec/scripts/beamline_context.py @@ -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() diff --git a/pxiii_bec/scripts/init_beamline.py b/pxiii_bec/scripts/init_beamline.py new file mode 100755 index 0000000..39e25d7 --- /dev/null +++ b/pxiii_bec/scripts/init_beamline.py @@ -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