From 65250459e8fe22647598ada438a5b0b8feaac423 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Tue, 14 Oct 2025 17:16:07 +0100 Subject: [PATCH 1/9] Add load_centre_collect, robot_unload and udc_default_state blueapi plans --- src/mx_bluesky/hyperion/blueapi_config.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/src/mx_bluesky/hyperion/blueapi_config.yaml b/src/mx_bluesky/hyperion/blueapi_config.yaml index da8bfcf231..30c5b50317 100644 --- a/src/mx_bluesky/hyperion/blueapi_config.yaml +++ b/src/mx_bluesky/hyperion/blueapi_config.yaml @@ -16,3 +16,4 @@ logging: graylog: url: "tcp://graylog-log-target.diamond.ac.uk:12232" enabled: true + \ No newline at end of file From 1f4e828a6d7e70c39f60c354038a1c6fdc2c97a3 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Fri, 5 Dec 2025 12:26:09 +0000 Subject: [PATCH 2/9] Updates to work with DeviceManager, BartRobot changes --- src/mx_bluesky/hyperion/blueapi_config.yaml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/mx_bluesky/hyperion/blueapi_config.yaml b/src/mx_bluesky/hyperion/blueapi_config.yaml index 30c5b50317..680fce96bb 100644 --- a/src/mx_bluesky/hyperion/blueapi_config.yaml +++ b/src/mx_bluesky/hyperion/blueapi_config.yaml @@ -15,5 +15,4 @@ api: logging: graylog: url: "tcp://graylog-log-target.diamond.ac.uk:12232" - enabled: true - \ No newline at end of file + enabled: true \ No newline at end of file From ae530e1b5607691e88f67fb728df7b91743e9cde Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Fri, 5 Dec 2025 13:16:36 +0000 Subject: [PATCH 3/9] Pin dodal --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index f22c5df38b..1f6a976739 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,7 @@ dependencies = [ "ophyd >= 1.10.5", "ophyd-async >= 0.14.0", "bluesky >= 1.14.6", - "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@main", + "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@a946a46ffd418ea25fbae9134954659cfa223d6f", ] From 75c9b468ff56fcdec555678c63211f29f8a3788f Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Tue, 9 Dec 2025 09:38:36 +0000 Subject: [PATCH 4/9] Refactor UDC-related plans into the PlanRunner --- .../common/parameters/components.py | 4 + src/mx_bluesky/hyperion/__main__.py | 50 ++++---- src/mx_bluesky/hyperion/baton_handler.py | 88 ++++---------- .../external_interaction/agamemnon.py | 11 +- src/mx_bluesky/hyperion/parameters/cli.py | 1 + .../hyperion/parameters/components.py | 13 ++ src/mx_bluesky/hyperion/plan_runner.py | 114 ++++++++++++++++-- tests/system_tests/hyperion/test_main_udc.py | 2 +- tests/unit_tests/conftest.py | 4 +- .../unit_tests/hyperion/test_baton_handler.py | 78 ++++++------ tests/unit_tests/hyperion/test_plan_runner.py | 12 +- .../hyperion/test_plan_runner_api.py | 4 +- 12 files changed, 226 insertions(+), 155 deletions(-) diff --git a/src/mx_bluesky/common/parameters/components.py b/src/mx_bluesky/common/parameters/components.py index 49eef01db7..9fe2761e49 100644 --- a/src/mx_bluesky/common/parameters/components.py +++ b/src/mx_bluesky/common/parameters/components.py @@ -32,6 +32,10 @@ PARAMETER_VERSION = Version.parse("5.3.0") +def get_param_version() -> SemanticVersion: + return SemanticVersion.validate_from_str(str(PARAMETER_VERSION)) + + class RotationAxis(StrEnum): OMEGA = "omega" PHI = "phi" diff --git a/src/mx_bluesky/hyperion/__main__.py b/src/mx_bluesky/hyperion/__main__.py index cf57f1fd92..b4b926e4fa 100755 --- a/src/mx_bluesky/hyperion/__main__.py +++ b/src/mx_bluesky/hyperion/__main__.py @@ -35,7 +35,7 @@ ) from mx_bluesky.hyperion.parameters.constants import CONST, HyperionConstants from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect -from mx_bluesky.hyperion.plan_runner import PlanRunner +from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanRunner from mx_bluesky.hyperion.plan_runner_api import create_server_for_udc from mx_bluesky.hyperion.runner import ( GDARunner, @@ -163,27 +163,33 @@ def main(): args = parse_cli_args() initialise_globals(args) hyperion_port = HyperionConstants.HYPERION_PORT - context = setup_context(dev_mode=args.dev_mode) - - if args.mode == HyperionMode.GDA: - runner = GDARunner(context=context) - app = create_app(runner) - flask_thread = threading.Thread( - target=lambda: app.run( - host="0.0.0.0", port=hyperion_port, debug=True, use_reloader=False - ), - daemon=True, - ) - flask_thread.start() - LOGGER.info( - f"Hyperion now listening on {hyperion_port} ({'IN DEV' if args.dev_mode else ''})" - ) - runner.wait_on_queue() - else: - plan_runner = PlanRunner(context, args.dev_mode) - create_server_for_udc(plan_runner) - _register_sigterm_handler(plan_runner) - run_forever(plan_runner) + + match args.mode: + case HyperionMode.GDA: + context = setup_context(dev_mode=args.dev_mode) + runner = GDARunner(context=context) + app = create_app(runner) + flask_thread = threading.Thread( + target=lambda: app.run( + host="0.0.0.0", port=hyperion_port, debug=True, use_reloader=False + ), + daemon=True, + ) + flask_thread.start() + LOGGER.info( + f"Hyperion now listening on {hyperion_port} ({'IN DEV' if args.dev_mode else ''})" + ) + runner.wait_on_queue() + case HyperionMode.UDC: + context = setup_context(dev_mode=args.dev_mode) + plan_runner = InProcessRunner(context, args.dev_mode) + create_server_for_udc(plan_runner) + _register_sigterm_handler(plan_runner) + run_forever(plan_runner) + case HyperionMode.SUPERVISOR: + raise RuntimeError( + "Supervisor mode not supported yet see https://github.com/DiamondLightSource/mx-bluesky/issues/1365" + ) def _register_sigterm_handler(runner: PlanRunner): diff --git a/src/mx_bluesky/hyperion/baton_handler.py b/src/mx_bluesky/hyperion/baton_handler.py index 339d4f23a4..2014929df3 100644 --- a/src/mx_bluesky/hyperion/baton_handler.py +++ b/src/mx_bluesky/hyperion/baton_handler.py @@ -1,45 +1,30 @@ from collections.abc import Sequence -from functools import partial -from typing import Any from blueapi.core.context import BlueskyContext from bluesky import plan_stubs as bps from bluesky import preprocessors as bpp from bluesky.utils import MsgGenerator, RunEngineInterrupted from dodal.common.beamlines.commissioning_mode import set_commissioning_signal -from dodal.devices.aperturescatterguard import ApertureScatterguard from dodal.devices.baton import Baton -from dodal.devices.detector.detector_motion import DetectorMotion, ShutterState -from dodal.devices.motors import XYZStage -from dodal.devices.robot import BartRobot -from dodal.devices.smargon import Smargon -from mx_bluesky.common.device_setup_plans.robot_load_unload import robot_unload from mx_bluesky.common.external_interaction.alerting import ( AlertService, get_alerting_service, ) -from mx_bluesky.common.parameters.components import MxBlueskyParameters +from mx_bluesky.common.parameters.components import ( + MxBlueskyParameters, + get_param_version, +) from mx_bluesky.common.utils.context import ( - device_composite_from_context, find_device_in_context, ) from mx_bluesky.common.utils.exceptions import BeamlineCheckFailureError from mx_bluesky.common.utils.log import LOGGER -from mx_bluesky.hyperion.experiment_plans.load_centre_collect_full_plan import ( - create_devices, - load_centre_collect_full, -) -from mx_bluesky.hyperion.experiment_plans.udc_default_state import ( - UDCDefaultDevices, - move_to_udc_default_state, -) from mx_bluesky.hyperion.external_interaction.agamemnon import ( create_parameters_from_agamemnon, ) from mx_bluesky.hyperion.external_interaction.alerting.constants import Subjects -from mx_bluesky.hyperion.parameters.components import Wait -from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect +from mx_bluesky.hyperion.parameters.components import UDCCleanup, UDCDefaultState from mx_bluesky.hyperion.plan_runner import PlanError, PlanRunner from mx_bluesky.hyperion.utils.context import ( clear_all_device_caches, @@ -102,7 +87,14 @@ def collect() -> MsgGenerator: """ _raise_udc_start_alert(get_alerting_service()) yield from bpp.contingency_wrapper( - _move_to_udc_default_state(context), + runner.decode_and_execute( + None, + [ + UDCDefaultState.model_validate( + {"parameter_model_version": get_param_version()} + ) + ], + ), except_plan=trap_default_state_exception, auto_raise=False, ) @@ -115,7 +107,14 @@ def collect() -> MsgGenerator: baton, runner, current_visit ) if current_visit: - yield from _clean_up_udc(runner.context, current_visit) + yield from runner.decode_and_execute( + current_visit, + [ + UDCCleanup.model_validate( + {"parameter_model_version": get_param_version()} + ) + ], + ) def release_baton() -> MsgGenerator: # If hyperion has given up the baton itself we need to also release requested @@ -177,23 +176,9 @@ def _fetch_and_process_agamemnon_instruction( ) -> MsgGenerator[str | None]: parameter_list: Sequence[MxBlueskyParameters] = create_parameters_from_agamemnon() if parameter_list: - for parameters in parameter_list: - LOGGER.info( - f"Executing plan with parameters: {parameters.model_dump_json(indent=2)}" - ) - match parameters: - case LoadCentreCollect(): - current_visit = parameters.visit - devices: Any = create_devices(runner.context) - yield from runner.execute_plan( - partial(load_centre_collect_full, devices, parameters) - ) - case Wait(): - yield from runner.execute_plan(partial(_runner_sleep, parameters)) - case _: - raise AssertionError( - f"Unsupported instruction decoded from agamemnon {type(parameters)}" - ) + current_visit = yield from runner.decode_and_execute( + current_visit, parameter_list + ) else: _raise_udc_completed_alert(get_alerting_service()) # Release the baton for orderly exit from the instruction loop @@ -224,20 +209,11 @@ def _raise_udc_completed_alert(alert_service: AlertService): ) -def _runner_sleep(parameters: Wait) -> MsgGenerator: - yield from bps.sleep(parameters.duration_s) - - def _is_requesting_baton(baton: Baton) -> MsgGenerator: requested_user = yield from bps.rd(baton.requested_user) return requested_user == HYPERION_USER -def _move_to_udc_default_state(context: BlueskyContext): - udc_default_devices = device_composite_from_context(context, UDCDefaultDevices) - yield from move_to_udc_default_state(udc_default_devices) - - def _get_baton(context: BlueskyContext) -> Baton: return find_device_in_context(context, "baton", Baton) @@ -255,19 +231,3 @@ def _unrequest_baton(baton: Baton) -> MsgGenerator[str]: yield from bps.abs_set(baton.requested_user, NO_USER) return NO_USER return requested_user - - -def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: - cleanup_group = "cleanup" - robot = find_device_in_context(context, "robot", BartRobot) - smargon = find_device_in_context(context, "smargon", Smargon) - aperture_scatterguard = find_device_in_context( - context, "aperture_scatterguard", ApertureScatterguard - ) - lower_gonio = find_device_in_context(context, "lower_gonio", XYZStage) - detector_motion = find_device_in_context(context, "detector_motion", DetectorMotion) - yield from bps.abs_set( - detector_motion.shutter, ShutterState.CLOSED, group=cleanup_group - ) - yield from robot_unload(robot, smargon, aperture_scatterguard, lower_gonio, visit) - yield from bps.wait(cleanup_group) diff --git a/src/mx_bluesky/hyperion/external_interaction/agamemnon.py b/src/mx_bluesky/hyperion/external_interaction/agamemnon.py index 5752e60b62..3fe676be40 100644 --- a/src/mx_bluesky/hyperion/external_interaction/agamemnon.py +++ b/src/mx_bluesky/hyperion/external_interaction/agamemnon.py @@ -12,12 +12,11 @@ from deepdiff.diff import DeepDiff from dodal.utils import get_beamline_name from jsonschema import ValidationError -from pydantic_extra_types.semantic_version import SemanticVersion from mx_bluesky.common.parameters.components import ( - PARAMETER_VERSION, MxBlueskyParameters, WithVisit, + get_param_version, ) from mx_bluesky.common.parameters.constants import ( GridscanParamConstants, @@ -88,7 +87,7 @@ def create_parameters_from_agamemnon() -> Sequence[MxBlueskyParameters]: Wait.model_validate( { "duration_s": data, - "parameter_model_version": _get_param_version(), + "parameter_model_version": get_param_version(), } ) ] @@ -228,10 +227,6 @@ def _get_withenergy_parameters_from_agamemnon(parameters: dict) -> dict[str, Any return {"demand_energy_ev": None} -def _get_param_version() -> SemanticVersion: - return SemanticVersion.validate_from_str(str(PARAMETER_VERSION)) - - def _populate_parameters_from_agamemnon( agamemnon_params, ) -> Sequence[LoadCentreCollect]: @@ -250,7 +245,7 @@ def _populate_parameters_from_agamemnon( return [ LoadCentreCollect.model_validate( { - "parameter_model_version": _get_param_version(), + "parameter_model_version": get_param_version(), "visit": visit, "detector_distance_mm": detector_distance, "sample_id": agamemnon_params["sample"]["id"], diff --git a/src/mx_bluesky/hyperion/parameters/cli.py b/src/mx_bluesky/hyperion/parameters/cli.py index fd4dc62235..18c9bf572c 100644 --- a/src/mx_bluesky/hyperion/parameters/cli.py +++ b/src/mx_bluesky/hyperion/parameters/cli.py @@ -9,6 +9,7 @@ class HyperionMode(StrEnum): GDA = "gda" UDC = "udc" + SUPERVISOR = "supervisor" @dataclass diff --git a/src/mx_bluesky/hyperion/parameters/components.py b/src/mx_bluesky/hyperion/parameters/components.py index 537ed7a8b0..894e3c24dd 100644 --- a/src/mx_bluesky/hyperion/parameters/components.py +++ b/src/mx_bluesky/hyperion/parameters/components.py @@ -8,3 +8,16 @@ class Wait(MxBlueskyParameters): """ duration_s: float + + +class UDCDefaultState(MxBlueskyParameters): + """Represents an instruction to execute the UDC default state plan.""" + + pass + + +class UDCCleanup(MxBlueskyParameters): + """Represents an instruction to perform UDC Cleanup, + in which the detector shutter is closed and a robot unload is performed.""" + + pass diff --git a/src/mx_bluesky/hyperion/plan_runner.py b/src/mx_bluesky/hyperion/plan_runner.py index e4def72eb3..cf493deb4b 100644 --- a/src/mx_bluesky/hyperion/plan_runner.py +++ b/src/mx_bluesky/hyperion/plan_runner.py @@ -1,14 +1,36 @@ import threading import time -from collections.abc import Callable +from abc import abstractmethod +from collections.abc import Callable, Sequence +from functools import partial +from typing import Any from blueapi.core import BlueskyContext from bluesky import plan_stubs as bps from bluesky.utils import MsgGenerator, RequestAbort - +from dodal.devices.aperturescatterguard import ApertureScatterguard +from dodal.devices.detector.detector_motion import DetectorMotion, ShutterState +from dodal.devices.motors import XYZStage +from dodal.devices.robot import BartRobot +from dodal.devices.smargon import Smargon + +from mx_bluesky.common.device_setup_plans.robot_load_unload import robot_unload +from mx_bluesky.common.parameters.components import MxBlueskyParameters from mx_bluesky.common.parameters.constants import Status +from mx_bluesky.common.utils.context import ( + device_composite_from_context, + find_device_in_context, +) from mx_bluesky.common.utils.exceptions import WarningError from mx_bluesky.common.utils.log import LOGGER +from mx_bluesky.hyperion.blueapi_plans import move_to_udc_default_state +from mx_bluesky.hyperion.experiment_plans.load_centre_collect_full_plan import ( + create_devices, + load_centre_collect_full, +) +from mx_bluesky.hyperion.experiment_plans.udc_default_state import UDCDefaultDevices +from mx_bluesky.hyperion.parameters.components import UDCCleanup, UDCDefaultState, Wait +from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect from mx_bluesky.hyperion.runner import BaseRunner @@ -19,18 +41,67 @@ class PlanError(Exception): class PlanRunner(BaseRunner): + def __init__(self, context: BlueskyContext, dev_mode: bool): + super().__init__(context) + self.is_dev_mode = dev_mode + + @abstractmethod + def decode_and_execute( + self, current_visit: str | None, parameters: Sequence[MxBlueskyParameters] + ) -> MsgGenerator: + pass + + @abstractmethod + def reset_callback_watchdog_timer(self): + pass + + @property + @abstractmethod + def current_status(self): + pass + + +class InProcessRunner(PlanRunner): """Runner that executes experiments from inside a running Bluesky plan""" EXTERNAL_CALLBACK_WATCHDOG_TIMER_S = 60 EXTERNAL_CALLBACK_POLL_INTERVAL_S = 1 def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: - super().__init__(context) - self.current_status: Status = Status.IDLE - self.is_dev_mode = dev_mode + super().__init__(context, dev_mode) + self._current_status: Status = Status.IDLE self._callbacks_started = False self._callback_watchdog_expiry = time.monotonic() + def decode_and_execute( + self, current_visit: str, parameter_list: Sequence[MxBlueskyParameters] + ) -> MsgGenerator: + for parameters in parameter_list: + LOGGER.info( + f"Executing plan with parameters: {parameters.model_dump_json(indent=2)}" + ) + match parameters: + case LoadCentreCollect(): + current_visit = parameters.visit + devices: Any = create_devices(self.context) + yield from self.execute_plan( + partial(load_centre_collect_full, devices, parameters) + ) + case Wait(): + yield from self.execute_plan(partial(_runner_sleep, parameters)) + case UDCDefaultState(): + udc_default_devices: UDCDefaultDevices = ( + device_composite_from_context(self.context, UDCDefaultDevices) + ) + yield from move_to_udc_default_state(udc_default_devices) + case UDCCleanup(): + yield from _clean_up_udc(self.context, current_visit) + case _: + raise AssertionError( + f"Unsupported instruction decoded from agamemnon {type(parameters)}" + ) + return current_visit + def execute_plan( self, experiment: Callable[[], MsgGenerator], @@ -42,7 +113,7 @@ def execute_plan( PlanError: If the plan raised an exception RequestAbort: If the RunEngine aborted during execution""" - self.current_status = Status.BUSY + self._current_status = Status.BUSY try: callback_expiry = time.monotonic() + self.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S @@ -60,17 +131,17 @@ def execute_plan( "External callback watchdog timer expired, check external callbacks are running." ) yield from experiment() - self.current_status = Status.IDLE + self._current_status = Status.IDLE except WarningError as e: LOGGER.warning("Plan failed with warning", exc_info=e) - self.current_status = Status.FAILED + self._current_status = Status.FAILED except RequestAbort: # This will occur when the run engine processes an abort when we shut down LOGGER.info("UDC Runner aborting") raise except Exception as e: LOGGER.error("Plan failed with exception", exc_info=e) - self.current_status = Status.FAILED + self._current_status = Status.FAILED raise PlanError("Exception thrown in plan execution") from e def shutdown(self): @@ -91,7 +162,7 @@ def issue_abort(): LOGGER.info("Shutting down: Stopping the run engine gracefully") if self.current_status != Status.ABORTING: - self.current_status = Status.ABORTING + self._current_status = Status.ABORTING stopping_thread = threading.Thread(target=issue_abort) stopping_thread.start() return @@ -105,3 +176,26 @@ def reset_callback_watchdog_timer(self): def _external_callbacks_are_alive(self) -> bool: return time.monotonic() < self._callback_watchdog_expiry + + def current_status(self): + return self._current_status + + +def _runner_sleep(parameters: Wait) -> MsgGenerator: + yield from bps.sleep(parameters.duration_s) + + +def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: + cleanup_group = "cleanup" + robot = find_device_in_context(context, "robot", BartRobot) + smargon = find_device_in_context(context, "smargon", Smargon) + aperture_scatterguard = find_device_in_context( + context, "aperture_scatterguard", ApertureScatterguard + ) + lower_gonio = find_device_in_context(context, "lower_gonio", XYZStage) + detector_motion = find_device_in_context(context, "detector_motion", DetectorMotion) + yield from bps.abs_set( + detector_motion.shutter, ShutterState.CLOSED, group=cleanup_group + ) + yield from robot_unload(robot, smargon, aperture_scatterguard, lower_gonio, visit) + yield from bps.wait(cleanup_group) diff --git a/tests/system_tests/hyperion/test_main_udc.py b/tests/system_tests/hyperion/test_main_udc.py index 7d6f15068a..c4b699c01e 100644 --- a/tests/system_tests/hyperion/test_main_udc.py +++ b/tests/system_tests/hyperion/test_main_udc.py @@ -35,7 +35,7 @@ def patched_setup_devices(context: BlueskyContext, dev_mode: bool): @pytest.fixture(autouse=True) def patch_udc_default_state(): - with patch("mx_bluesky.hyperion.baton_handler.move_to_udc_default_state"): + with patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"): yield diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index 29cc19a30f..eb5d9cb1f5 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -197,8 +197,8 @@ def use_beamline_t01(): with ( patch.dict(sys.modules, {"dodal.beamlines.t01": tests.unit_tests.t01}), - patch("mx_bluesky.hyperion.baton_handler.move_to_udc_default_state"), - patch("mx_bluesky.hyperion.baton_handler.device_composite_from_context"), + patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), + patch("mx_bluesky.hyperion.plan_runner.device_composite_from_context"), ): yield tests.unit_tests.t01 diff --git a/tests/unit_tests/hyperion/test_baton_handler.py b/tests/unit_tests/hyperion/test_baton_handler.py index a086d7f454..52a8fe7f7e 100644 --- a/tests/unit_tests/hyperion/test_baton_handler.py +++ b/tests/unit_tests/hyperion/test_baton_handler.py @@ -46,7 +46,7 @@ from mx_bluesky.hyperion.external_interaction.alerting.constants import Subjects from mx_bluesky.hyperion.parameters.components import Wait from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect -from mx_bluesky.hyperion.plan_runner import PlanError, PlanRunner +from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanError, PlanRunner from mx_bluesky.hyperion.utils.context import setup_context from .conftest import AGAMEMNON_WAIT_INSTRUCTION, launch_test_in_runner_event_loop @@ -193,7 +193,7 @@ def single_collection_agamemnon_request( "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", side_effect=[[load_centre_collect_params], []], ), - patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state"), + patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), ): yield mock_load_centre_collect @@ -210,7 +210,7 @@ def single_collection_agamemnon_request_then_wait_forever( [AGAMEMNON_WAIT_INSTRUCTION] * 1000, ], ), - patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state"), + patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), ): yield mock_load_centre_collect @@ -225,7 +225,7 @@ def baton_with_requested_user( @pytest.fixture() def udc_runner(bluesky_context: BlueskyContext) -> PlanRunner: - runner = PlanRunner(bluesky_context, True) + runner = InProcessRunner(bluesky_context, True) runner.reset_callback_watchdog_timer() return runner @@ -233,10 +233,8 @@ def udc_runner(bluesky_context: BlueskyContext) -> PlanRunner: @pytest.fixture def mock_load_centre_collect(): with ( - patch("mx_bluesky.hyperion.baton_handler.create_devices"), - patch( - "mx_bluesky.hyperion.baton_handler.load_centre_collect_full" - ) as mock_plan, + patch("mx_bluesky.hyperion.plan_runner.create_devices"), + patch("mx_bluesky.hyperion.plan_runner.load_centre_collect_full") as mock_plan, ): yield mock_plan @@ -250,7 +248,7 @@ def mock_create_params_from_agamemnon(): yield patched_func -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.bps.sleep") @pytest.mark.timeout(10) def test_loop_until_hyperion_requested( @@ -275,7 +273,7 @@ def set_hyperion_requested(*args): assert mock_sleep.call_count == number_of_sleep_calls -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) def test_when_hyperion_requested_then_hyperion_set_to_current_user( bluesky_context: BlueskyContext, udc_runner: PlanRunner, @@ -290,7 +288,7 @@ def test_when_hyperion_requested_then_hyperion_set_to_current_user( ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state") +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") def test_when_hyperion_requested_then_default_state_and_collection_run( default_state: MagicMock, bluesky_context: BlueskyContext, @@ -303,7 +301,7 @@ def test_when_hyperion_requested_then_default_state_and_collection_run( run_udc_when_requested(bluesky_context, udc_runner) assert parent_mock.method_calls == [ - call.default_state(bluesky_context), + call.default_state(ANY), call.main_loop(), ] @@ -314,7 +312,7 @@ async def _assert_baton_released(baton: Baton): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_when_exception_raised_in_collection_then_loop_stops_and_baton_released( agamemnon: MagicMock, mock_load_centre_collect: MagicMock, @@ -335,7 +333,7 @@ async def test_when_exception_raised_in_collection_then_loop_stops_and_baton_rel @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_when_warning_exception_raised_in_collection_then_loop_continues( agamemnon: MagicMock, mock_load_centre_collect: MagicMock, @@ -365,7 +363,7 @@ def raise_else_return(value): return value -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state") +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") async def test_when_exception_raised_in_default_state_then_baton_released( default_state: MagicMock, bluesky_context: BlueskyContext, @@ -380,7 +378,7 @@ async def test_when_exception_raised_in_default_state_then_baton_released( @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_when_exception_raised_in_getting_agamemnon_instruction_then_loop_stops_and_baton_released( agamemnon: MagicMock, udc_runner: PlanRunner, @@ -395,8 +393,8 @@ async def test_when_exception_raised_in_getting_agamemnon_instruction_then_loop_ @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler.load_centre_collect_full") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.load_centre_collect_full") +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_when_no_agamemnon_instructions_left_then_loop_stops_and_baton_released( collection: MagicMock, agamemnon: MagicMock, @@ -412,7 +410,7 @@ async def test_when_no_agamemnon_instructions_left_then_loop_stops_and_baton_rel @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_when_other_user_requested_collection_finished_then_baton_released( agamemnon: MagicMock, bluesky_context: BlueskyContext, @@ -443,7 +441,7 @@ def fake_collection_with_baton_request_part_way_through(*args): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state") +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") async def test_when_multiple_agamemnon_instructions_then_default_state_only_run_once( default_state: MagicMock, agamemnon: MagicMock, @@ -490,13 +488,13 @@ def test_initialise_udc_reloads_all_devices(dont_patch_clear_devices): ] ), ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", MagicMock()) def test_baton_handler_loop_waits_if_wait_instruction_received( bluesky_context_with_sim_run_engine: tuple[list[Msg], BlueskyContext], sim_run_engine: RunEngineSimulator, ): msgs, context = bluesky_context_with_sim_run_engine - udc_runner = PlanRunner(context, True) + udc_runner = InProcessRunner(context, True) udc_runner.reset_callback_watchdog_timer() run_udc_when_requested(context, udc_runner) @@ -505,7 +503,7 @@ def test_baton_handler_loop_waits_if_wait_instruction_received( ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", MagicMock()) def test_main_loop_rejects_unrecognised_instruction_when_received( bluesky_context_with_sim_run_engine: tuple[list[Msg], BlueskyContext], sim_run_engine: RunEngineSimulator, @@ -527,10 +525,10 @@ def test_main_loop_rejects_unrecognised_instruction_when_received( ), ), ): - run_udc_when_requested(context, PlanRunner(context, True)) + run_udc_when_requested(context, InProcessRunner(context, True)) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_shutdown_releases_the_baton( mock_create_params_from_agamemnon: MagicMock, @@ -564,7 +562,7 @@ async def wait_and_then_shutdown(): ), ], ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_run_forever_resumes_collection_when_baton_taken_away( mock_create_parameters_from_agamemnon: MagicMock, @@ -608,7 +606,7 @@ async def take_requested_baton_away_then_wait_for_release_then_re_request(): [AGAMEMNON_WAIT_INSTRUCTION], ], ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_run_forever_resumes_collection_when_normal_completion_and_baton_returned( mock_create_parameters_from_agamemnon: MagicMock, @@ -642,7 +640,7 @@ async def wait_for_baton_release_then_re_request(): "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", side_effect=AssertionError("Runner started command processing without baton"), ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_run_forever_handles_shutdown_while_waiting_for_baton( mock_create_parameters_from_agamemnon: MagicMock, udc_runner: PlanRunner, @@ -674,7 +672,7 @@ async def issue_shutdown_without_baton(): [], ], ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) def test_run_forever_clears_error_status_on_resume( mock_create_parameters_from_agamemnon: MagicMock, udc_runner: PlanRunner, @@ -685,7 +683,7 @@ def test_run_forever_clears_error_status_on_resume( async def error_with_command_then_resume(): try: with patch( - "mx_bluesky.hyperion.baton_handler._runner_sleep", + "mx_bluesky.hyperion.plan_runner._runner_sleep", side_effect=RuntimeError("Simulated plan exception"), ): function_is_patched.set() @@ -718,7 +716,7 @@ async def error_with_command_then_resume(): ], ) @patch("mx_bluesky.hyperion.baton_handler.set_commissioning_signal") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) async def test_commissioning_signal_set_on_baton_acquire( mock_set_commissioning_signal: MagicMock, mock_create_parameters_from_agamemnon: MagicMock, @@ -769,7 +767,7 @@ async def release_baton_and_check_commissioning_signal_set(): "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", MagicMock(side_effect=[[AGAMEMNON_WAIT_INSTRUCTION], []]), ) -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_udc_start_alert_when_baton_acquired( mock_get_alerting_service: MagicMock, @@ -785,7 +783,7 @@ def test_run_udc_when_requested_raises_udc_start_alert_when_baton_acquired( @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.clear_all_device_caches", MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_baton_release_udc_completed_event_when_hyperion_releases_the_baton( @@ -828,7 +826,7 @@ def wait_then_nothing(): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.baton_handler._move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.clear_all_device_caches", MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_baton_release_event_when_baton_requested_from_hyperion( @@ -868,7 +866,7 @@ def wait_then_nothing(): ) -@patch("mx_bluesky.hyperion.baton_handler.robot_unload") +@patch("mx_bluesky.hyperion.plan_runner.robot_unload") def test_robot_unload_performed_when_no_more_agamemnon_instructions( mock_robot_unload, bluesky_context: BlueskyContext, @@ -902,7 +900,7 @@ def request_baton_away_from_hyperion(*args): mock_load_centre_collect.side_effect = request_baton_away_from_hyperion -@patch("mx_bluesky.hyperion.baton_handler.robot_unload") +@patch("mx_bluesky.hyperion.plan_runner.robot_unload") def test_robot_unload_performed_when_baton_requested_away_from_hyperion( mock_robot_unload, bluesky_context: BlueskyContext, @@ -923,7 +921,7 @@ def test_robot_unload_performed_when_baton_requested_away_from_hyperion( ) -@patch("mx_bluesky.hyperion.baton_handler.robot_unload") +@patch("mx_bluesky.hyperion.plan_runner.robot_unload") def test_robot_unload_not_performed_when_beamline_error( mock_robot_unload, bluesky_context: BlueskyContext, @@ -939,7 +937,7 @@ def test_robot_unload_not_performed_when_beamline_error( mock_robot_unload.assert_not_called() -@patch("mx_bluesky.hyperion.baton_handler.robot_unload") +@patch("mx_bluesky.hyperion.plan_runner.robot_unload") def test_robot_unload_still_performed_when_sample_exception( mock_robot_unload, bluesky_context: BlueskyContext, @@ -963,7 +961,7 @@ def test_robot_unload_still_performed_when_sample_exception( ) -@patch("mx_bluesky.hyperion.baton_handler.robot_unload") +@patch("mx_bluesky.hyperion.plan_runner.robot_unload") def test_detector_shutter_closed_when_baton_requested_away_from_hyperion( mock_robot_unload, bluesky_context: BlueskyContext, @@ -983,7 +981,7 @@ def test_detector_shutter_closed_when_baton_requested_away_from_hyperion( get_mock_put(detector_motion.shutter).assert_called_once() -@patch("mx_bluesky.hyperion.baton_handler.move_to_udc_default_state") +@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") def test_hyperion_doesnt_exit_if_udc_default_state_fails_a_check( mock_move_to_udc_default_state, bluesky_context: BlueskyContext, diff --git a/tests/unit_tests/hyperion/test_plan_runner.py b/tests/unit_tests/hyperion/test_plan_runner.py index dde6e1692b..10299d5046 100644 --- a/tests/unit_tests/hyperion/test_plan_runner.py +++ b/tests/unit_tests/hyperion/test_plan_runner.py @@ -8,13 +8,13 @@ from bluesky import RunEngine from bluesky import plan_stubs as bps -from mx_bluesky.hyperion.plan_runner import PlanError, PlanRunner +from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanError @pytest.fixture(autouse=True) def patch_timer_poll_interval(): with patch( - "mx_bluesky.hyperion.plan_runner.PlanRunner.EXTERNAL_CALLBACK_POLL_INTERVAL_S", + "mx_bluesky.hyperion.plan_runner.InProcessRunner.EXTERNAL_CALLBACK_POLL_INTERVAL_S", 0.01, ): yield @@ -23,14 +23,14 @@ def patch_timer_poll_interval(): @pytest.fixture() def patch_timer_expiry(): with patch( - "mx_bluesky.hyperion.plan_runner.PlanRunner.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S", + "mx_bluesky.hyperion.plan_runner.InProcessRunner.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S", 0.1, ): yield def test_external_callbacks_waits_for_external_callback_ping(run_engine: RunEngine): - runner = PlanRunner(BlueskyContext(run_engine=run_engine), True) + runner = InProcessRunner(BlueskyContext(run_engine=run_engine), True) plan_started = Event() def execute_test(): @@ -52,7 +52,7 @@ def test_plan(): def test_external_callbacks_raises_if_never_started( run_engine: RunEngine, patch_timer_expiry ): - runner = PlanRunner(BlueskyContext(run_engine=run_engine), True) + runner = InProcessRunner(BlueskyContext(run_engine=run_engine), True) plan_started = Event() def execute_test(): @@ -79,7 +79,7 @@ def test_external_callbacks_not_running_raises_exception_for_plan_execution( run_engine: RunEngine, patch_timer_expiry, ): - runner = PlanRunner(BlueskyContext(run_engine=run_engine), True) + runner = InProcessRunner(BlueskyContext(run_engine=run_engine), True) def execute_test(): runner.reset_callback_watchdog_timer() diff --git a/tests/unit_tests/hyperion/test_plan_runner_api.py b/tests/unit_tests/hyperion/test_plan_runner_api.py index 13f58e5adb..535ab72a84 100644 --- a/tests/unit_tests/hyperion/test_plan_runner_api.py +++ b/tests/unit_tests/hyperion/test_plan_runner_api.py @@ -5,7 +5,7 @@ from flask.testing import FlaskClient from mx_bluesky.common.parameters.constants import Status -from mx_bluesky.hyperion.plan_runner import PlanRunner +from mx_bluesky.hyperion.plan_runner import InProcessRunner from mx_bluesky.hyperion.plan_runner_api import ( create_app_for_udc, ) @@ -13,7 +13,7 @@ @pytest.fixture() def mock_runner(): - return MagicMock(spec=PlanRunner) + return MagicMock(spec=InProcessRunner) @pytest.fixture() From 32cf06522bd0b8e37cf72a4578d342a4cc0c1308 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Tue, 9 Dec 2025 18:10:50 +0000 Subject: [PATCH 5/9] Missing changes from plan runner refactor --- .../hyperion/blueapi_plans/__init__.py | 19 ++++++++++++++ src/mx_bluesky/hyperion/plan_runner.py | 26 ++++++++++--------- 2 files changed, 33 insertions(+), 12 deletions(-) diff --git a/src/mx_bluesky/hyperion/blueapi_plans/__init__.py b/src/mx_bluesky/hyperion/blueapi_plans/__init__.py index cef1bab1d0..ce45acd9fc 100644 --- a/src/mx_bluesky/hyperion/blueapi_plans/__init__.py +++ b/src/mx_bluesky/hyperion/blueapi_plans/__init__.py @@ -5,9 +5,11 @@ process. """ +from bluesky import plan_stubs as bps from bluesky.utils import MsgGenerator from dodal.common import inject from dodal.devices.aperturescatterguard import ApertureScatterguard +from dodal.devices.detector.detector_motion import DetectorMotion, ShutterState from dodal.devices.motors import XYZStage from dodal.devices.robot import BartRobot from dodal.devices.smargon import Smargon @@ -33,6 +35,7 @@ "LoadCentreCollectComposite", "LoadCentreCollect", "UDCDefaultDevices", + "clean_up_udc", "load_centre_collect", "move_to_udc_default_state", "robot_unload", @@ -68,6 +71,22 @@ def robot_unload( yield from _robot_unload(robot, smargon, aperture_scatterguard, lower_gonio, visit) +def clean_up_udc( + visit: str, + cleanup_group="cleanup", + robot: BartRobot = inject("robot"), + smargon: Smargon = inject("smargon"), + aperture_scatterguard: ApertureScatterguard = inject("aperture_scatterguard"), + lower_gonio: XYZStage = inject("lower_gonio"), + detector_motion: DetectorMotion = inject("detector_motion"), +): + yield from bps.abs_set( + detector_motion.shutter, ShutterState.CLOSED, group=cleanup_group + ) + yield from _robot_unload(robot, smargon, aperture_scatterguard, lower_gonio, visit) + yield from bps.wait(cleanup_group) + + def move_to_udc_default_state( composite: UDCDefaultDevices = inject(), ) -> MsgGenerator: diff --git a/src/mx_bluesky/hyperion/plan_runner.py b/src/mx_bluesky/hyperion/plan_runner.py index cf493deb4b..7a59d4f9f8 100644 --- a/src/mx_bluesky/hyperion/plan_runner.py +++ b/src/mx_bluesky/hyperion/plan_runner.py @@ -9,12 +9,11 @@ from bluesky import plan_stubs as bps from bluesky.utils import MsgGenerator, RequestAbort from dodal.devices.aperturescatterguard import ApertureScatterguard -from dodal.devices.detector.detector_motion import DetectorMotion, ShutterState +from dodal.devices.detector.detector_motion import DetectorMotion from dodal.devices.motors import XYZStage from dodal.devices.robot import BartRobot from dodal.devices.smargon import Smargon -from mx_bluesky.common.device_setup_plans.robot_load_unload import robot_unload from mx_bluesky.common.parameters.components import MxBlueskyParameters from mx_bluesky.common.parameters.constants import Status from mx_bluesky.common.utils.context import ( @@ -23,7 +22,7 @@ ) from mx_bluesky.common.utils.exceptions import WarningError from mx_bluesky.common.utils.log import LOGGER -from mx_bluesky.hyperion.blueapi_plans import move_to_udc_default_state +from mx_bluesky.hyperion.blueapi_plans import clean_up_udc, move_to_udc_default_state from mx_bluesky.hyperion.experiment_plans.load_centre_collect_full_plan import ( create_devices, load_centre_collect_full, @@ -47,7 +46,7 @@ def __init__(self, context: BlueskyContext, dev_mode: bool): @abstractmethod def decode_and_execute( - self, current_visit: str | None, parameters: Sequence[MxBlueskyParameters] + self, current_visit: str | None, parameter_list: Sequence[MxBlueskyParameters] ) -> MsgGenerator: pass @@ -57,7 +56,7 @@ def reset_callback_watchdog_timer(self): @property @abstractmethod - def current_status(self): + def current_status(self) -> Status: pass @@ -74,7 +73,7 @@ def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: self._callback_watchdog_expiry = time.monotonic() def decode_and_execute( - self, current_visit: str, parameter_list: Sequence[MxBlueskyParameters] + self, current_visit: str | None, parameter_list: Sequence[MxBlueskyParameters] ) -> MsgGenerator: for parameters in parameter_list: LOGGER.info( @@ -177,7 +176,8 @@ def reset_callback_watchdog_timer(self): def _external_callbacks_are_alive(self) -> bool: return time.monotonic() < self._callback_watchdog_expiry - def current_status(self): + @property + def current_status(self) -> Status: return self._current_status @@ -186,7 +186,6 @@ def _runner_sleep(parameters: Wait) -> MsgGenerator: def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: - cleanup_group = "cleanup" robot = find_device_in_context(context, "robot", BartRobot) smargon = find_device_in_context(context, "smargon", Smargon) aperture_scatterguard = find_device_in_context( @@ -194,8 +193,11 @@ def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: ) lower_gonio = find_device_in_context(context, "lower_gonio", XYZStage) detector_motion = find_device_in_context(context, "detector_motion", DetectorMotion) - yield from bps.abs_set( - detector_motion.shutter, ShutterState.CLOSED, group=cleanup_group + yield from clean_up_udc( + visit, + robot=robot, + smargon=smargon, + aperture_scatterguard=aperture_scatterguard, + lower_gonio=lower_gonio, + detector_motion=detector_motion, ) - yield from robot_unload(robot, smargon, aperture_scatterguard, lower_gonio, visit) - yield from bps.wait(cleanup_group) From 30682b707b6a0c78551667d7ebdcdea43f459f88 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Thu, 18 Dec 2025 16:23:44 +0000 Subject: [PATCH 6/9] Move InProcessRunner to separate file Fix unit tests --- src/mx_bluesky/hyperion/__main__.py | 3 +- src/mx_bluesky/hyperion/in_process_runner.py | 132 ++++++++++++++++++ src/mx_bluesky/hyperion/plan_runner.py | 95 +++++++------ tests/unit_tests/conftest.py | 6 +- .../unit_tests/hyperion/test_baton_handler.py | 103 +++++++++----- tests/unit_tests/hyperion/test_plan_runner.py | 7 +- .../hyperion/test_plan_runner_api.py | 2 +- 7 files changed, 262 insertions(+), 86 deletions(-) create mode 100644 src/mx_bluesky/hyperion/in_process_runner.py diff --git a/src/mx_bluesky/hyperion/__main__.py b/src/mx_bluesky/hyperion/__main__.py index b4b926e4fa..0277d352ef 100755 --- a/src/mx_bluesky/hyperion/__main__.py +++ b/src/mx_bluesky/hyperion/__main__.py @@ -28,6 +28,7 @@ compare_params, update_params_from_agamemnon, ) +from mx_bluesky.hyperion.in_process_runner import InProcessRunner from mx_bluesky.hyperion.parameters.cli import ( HyperionArgs, HyperionMode, @@ -35,7 +36,7 @@ ) from mx_bluesky.hyperion.parameters.constants import CONST, HyperionConstants from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect -from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanRunner +from mx_bluesky.hyperion.plan_runner import PlanRunner from mx_bluesky.hyperion.plan_runner_api import create_server_for_udc from mx_bluesky.hyperion.runner import ( GDARunner, diff --git a/src/mx_bluesky/hyperion/in_process_runner.py b/src/mx_bluesky/hyperion/in_process_runner.py new file mode 100644 index 0000000000..4fcef1991e --- /dev/null +++ b/src/mx_bluesky/hyperion/in_process_runner.py @@ -0,0 +1,132 @@ +import time +from collections.abc import Callable, Sequence +from functools import partial +from typing import Any + +from blueapi.core import BlueskyContext +from bluesky import plan_stubs as bps +from bluesky.utils import MsgGenerator, RequestAbort +from dodal.devices.aperturescatterguard import ApertureScatterguard +from dodal.devices.detector.detector_motion import DetectorMotion +from dodal.devices.motors import XYZStage +from dodal.devices.robot import BartRobot +from dodal.devices.smargon import Smargon + +from mx_bluesky.common.parameters.components import MxBlueskyParameters +from mx_bluesky.common.parameters.constants import Status +from mx_bluesky.common.utils.context import ( + device_composite_from_context, + find_device_in_context, +) +from mx_bluesky.common.utils.exceptions import WarningError +from mx_bluesky.common.utils.log import LOGGER +from mx_bluesky.hyperion.blueapi_plans import clean_up_udc, move_to_udc_default_state +from mx_bluesky.hyperion.experiment_plans import load_centre_collect_full +from mx_bluesky.hyperion.experiment_plans.load_centre_collect_full_plan import ( + create_devices, +) +from mx_bluesky.hyperion.experiment_plans.udc_default_state import UDCDefaultDevices +from mx_bluesky.hyperion.parameters.components import UDCCleanup, UDCDefaultState, Wait +from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect +from mx_bluesky.hyperion.plan_runner import PlanError, PlanRunner + + +class InProcessRunner(PlanRunner): + """Runner that executes experiments from inside a running Bluesky plan""" + def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: + super().__init__(context, dev_mode) + self._current_status: Status = Status.IDLE + + def decode_and_execute( + self, current_visit: str | None, parameter_list: Sequence[MxBlueskyParameters] + ) -> MsgGenerator: + for parameters in parameter_list: + LOGGER.info( + f"Executing plan with parameters: {parameters.model_dump_json(indent=2)}" + ) + match parameters: + case LoadCentreCollect(): + current_visit = parameters.visit + devices: Any = create_devices(self.context) + yield from self.execute_plan( + partial(load_centre_collect_full, devices, parameters) + ) + case Wait(): + yield from self.execute_plan(partial(_runner_sleep, parameters)) + case UDCDefaultState(): + udc_default_devices: UDCDefaultDevices = ( + device_composite_from_context(self.context, UDCDefaultDevices) + ) + yield from move_to_udc_default_state(udc_default_devices) + case UDCCleanup(): + yield from _clean_up_udc(self.context, current_visit) + case _: + raise AssertionError( + f"Unsupported instruction decoded from agamemnon {type(parameters)}" + ) + return current_visit + + def execute_plan( + self, + experiment: Callable[[], MsgGenerator], + ) -> MsgGenerator: + """Execute the specified experiment plan. + Args: + experiment: The experiment to run + Raises: + PlanError: If the plan raised an exception + RequestAbort: If the RunEngine aborted during execution""" + + self._current_status = Status.BUSY + + try: + yield from self.check_external_callbacks_are_alive() + yield from experiment() + self._current_status = Status.IDLE + except WarningError as e: + LOGGER.warning("Plan failed with warning", exc_info=e) + self._current_status = Status.FAILED + except RequestAbort: + # This will occur when the run engine processes an abort when we shut down + LOGGER.info("UDC Runner aborting") + raise + except Exception as e: + LOGGER.error("Plan failed with exception", exc_info=e) + self._current_status = Status.FAILED + raise PlanError("Exception thrown in plan execution") from e + + def shutdown(self): + """Performs a prompt shutdown. Aborts the run engine and terminates the loop + waiting for messages.""" + + LOGGER.info("Shutting down: Stopping the run engine gracefully") + if self.current_status != Status.ABORTING: + self._current_status = Status.ABORTING + self.request_run_engine_abort() + return + + @property + def current_status(self) -> Status: + return self._current_status + + +def _runner_sleep(parameters: Wait) -> MsgGenerator: + yield from bps.sleep(parameters.duration_s) + + +def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: + robot = find_device_in_context(context, "robot", BartRobot) + smargon = find_device_in_context(context, "smargon", Smargon) + aperture_scatterguard = find_device_in_context( + context, "aperture_scatterguard", ApertureScatterguard + ) + lower_gonio = find_device_in_context(context, "lower_gonio", XYZStage) + detector_motion = find_device_in_context(context, "detector_motion", DetectorMotion) + yield from clean_up_udc( + visit, + robot=robot, + smargon=smargon, + aperture_scatterguard=aperture_scatterguard, + lower_gonio=lower_gonio, + detector_motion=detector_motion, + ) diff --git a/src/mx_bluesky/hyperion/plan_runner.py b/src/mx_bluesky/hyperion/plan_runner.py index 7a59d4f9f8..b91cca24cf 100644 --- a/src/mx_bluesky/hyperion/plan_runner.py +++ b/src/mx_bluesky/hyperion/plan_runner.py @@ -40,8 +40,13 @@ class PlanError(Exception): class PlanRunner(BaseRunner): + EXTERNAL_CALLBACK_POLL_INTERVAL_S = 1 + EXTERNAL_CALLBACK_WATCHDOG_TIMER_S = 60 + def __init__(self, context: BlueskyContext, dev_mode: bool): super().__init__(context) + self._callbacks_started = False + self._callback_watchdog_expiry = time.monotonic() self.is_dev_mode = dev_mode @abstractmethod @@ -50,27 +55,63 @@ def decode_and_execute( ) -> MsgGenerator: pass - @abstractmethod def reset_callback_watchdog_timer(self): - pass + """Called periodically to reset the watchdog timer when the external callbacks ping us.""" + self._callbacks_started = True + self._callback_watchdog_expiry = ( + time.monotonic() + self.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S + ) @property @abstractmethod def current_status(self) -> Status: pass + def check_external_callbacks_are_alive(self): + callback_expiry = time.monotonic() + self.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S + while time.monotonic() < callback_expiry: + if self._callbacks_started: + break + # If on first launch the external callbacks aren't started yet, wait until they are + LOGGER.info("Waiting for external callbacks to start") + yield from bps.sleep(self.EXTERNAL_CALLBACK_POLL_INTERVAL_S) + else: + raise RuntimeError("External callbacks not running - try restarting") + + if not self._external_callbacks_are_alive(): + raise RuntimeError( + "External callback watchdog timer expired, check external callbacks are running." + ) + + def request_run_engine_abort(self): + """Asynchronously request an abort from the run engine. This cannot be done from + inside the main thread.""" + + def issue_abort(): + try: + # abort() causes the run engine to throw a RequestAbort exception + # inside the plan, which will propagate through the contingency wrappers. + # When the plan returns, the run engine will raise RunEngineInterrupted + self.run_engine.abort() + except Exception as e: + LOGGER.warning( + "Exception encountered when issuing abort() to RunEngine:", + exc_info=e, + ) + + stopping_thread = threading.Thread(target=issue_abort) + stopping_thread.start() + + def _external_callbacks_are_alive(self) -> bool: + return time.monotonic() < self._callback_watchdog_expiry + class InProcessRunner(PlanRunner): """Runner that executes experiments from inside a running Bluesky plan""" - EXTERNAL_CALLBACK_WATCHDOG_TIMER_S = 60 - EXTERNAL_CALLBACK_POLL_INTERVAL_S = 1 - def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: super().__init__(context, dev_mode) self._current_status: Status = Status.IDLE - self._callbacks_started = False - self._callback_watchdog_expiry = time.monotonic() def decode_and_execute( self, current_visit: str | None, parameter_list: Sequence[MxBlueskyParameters] @@ -115,20 +156,7 @@ def execute_plan( self._current_status = Status.BUSY try: - callback_expiry = time.monotonic() + self.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S - while time.monotonic() < callback_expiry: - if self._callbacks_started: - break - # If on first launch the external callbacks aren't started yet, wait until they are - LOGGER.info("Waiting for external callbacks to start") - yield from bps.sleep(self.EXTERNAL_CALLBACK_POLL_INTERVAL_S) - else: - raise RuntimeError("External callbacks not running - try restarting") - - if not self._external_callbacks_are_alive(): - raise RuntimeError( - "External callback watchdog timer expired, check external callbacks are running." - ) + yield from self.check_external_callbacks_are_alive() yield from experiment() self._current_status = Status.IDLE except WarningError as e: @@ -147,35 +175,12 @@ def shutdown(self): """Performs a prompt shutdown. Aborts the run engine and terminates the loop waiting for messages.""" - def issue_abort(): - try: - # abort() causes the run engine to throw a RequestAbort exception - # inside the plan, which will propagate through the contingency wrappers. - # When the plan returns, the run engine will raise RunEngineInterrupted - self.run_engine.abort() - except Exception as e: - LOGGER.warning( - "Exception encountered when issuing abort() to RunEngine:", - exc_info=e, - ) - LOGGER.info("Shutting down: Stopping the run engine gracefully") if self.current_status != Status.ABORTING: self._current_status = Status.ABORTING - stopping_thread = threading.Thread(target=issue_abort) - stopping_thread.start() + self.request_run_engine_abort() return - def reset_callback_watchdog_timer(self): - """Called periodically to reset the watchdog timer when the external callbacks ping us.""" - self._callbacks_started = True - self._callback_watchdog_expiry = ( - time.monotonic() + self.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S - ) - - def _external_callbacks_are_alive(self) -> bool: - return time.monotonic() < self._callback_watchdog_expiry - @property def current_status(self) -> Status: return self._current_status diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index eb5d9cb1f5..0275cd2bf4 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -197,8 +197,10 @@ def use_beamline_t01(): with ( patch.dict(sys.modules, {"dodal.beamlines.t01": tests.unit_tests.t01}), - patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), - patch("mx_bluesky.hyperion.plan_runner.device_composite_from_context"), + patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state"), + patch( + "mx_bluesky.hyperion.in_process_runner.device_composite_from_context" + ), ): yield tests.unit_tests.t01 diff --git a/tests/unit_tests/hyperion/test_baton_handler.py b/tests/unit_tests/hyperion/test_baton_handler.py index 52a8fe7f7e..31520ea779 100644 --- a/tests/unit_tests/hyperion/test_baton_handler.py +++ b/tests/unit_tests/hyperion/test_baton_handler.py @@ -44,9 +44,10 @@ LoadCentreCollectComposite, ) from mx_bluesky.hyperion.external_interaction.alerting.constants import Subjects +from mx_bluesky.hyperion.in_process_runner import InProcessRunner from mx_bluesky.hyperion.parameters.components import Wait from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect -from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanError, PlanRunner +from mx_bluesky.hyperion.plan_runner import PlanError, PlanRunner from mx_bluesky.hyperion.utils.context import setup_context from .conftest import AGAMEMNON_WAIT_INSTRUCTION, launch_test_in_runner_event_loop @@ -193,7 +194,7 @@ def single_collection_agamemnon_request( "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", side_effect=[[load_centre_collect_params], []], ), - patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), + patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state"), ): yield mock_load_centre_collect @@ -210,7 +211,7 @@ def single_collection_agamemnon_request_then_wait_forever( [AGAMEMNON_WAIT_INSTRUCTION] * 1000, ], ), - patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state"), + patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state"), ): yield mock_load_centre_collect @@ -233,8 +234,10 @@ def udc_runner(bluesky_context: BlueskyContext) -> PlanRunner: @pytest.fixture def mock_load_centre_collect(): with ( - patch("mx_bluesky.hyperion.plan_runner.create_devices"), - patch("mx_bluesky.hyperion.plan_runner.load_centre_collect_full") as mock_plan, + patch("mx_bluesky.hyperion.in_process_runner.create_devices"), + patch( + "mx_bluesky.hyperion.in_process_runner.load_centre_collect_full" + ) as mock_plan, ): yield mock_plan @@ -248,7 +251,9 @@ def mock_create_params_from_agamemnon(): yield patched_func -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @patch("mx_bluesky.hyperion.baton_handler.bps.sleep") @pytest.mark.timeout(10) def test_loop_until_hyperion_requested( @@ -273,7 +278,9 @@ def set_hyperion_requested(*args): assert mock_sleep.call_count == number_of_sleep_calls -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) def test_when_hyperion_requested_then_hyperion_set_to_current_user( bluesky_context: BlueskyContext, udc_runner: PlanRunner, @@ -288,7 +295,7 @@ def test_when_hyperion_requested_then_hyperion_set_to_current_user( ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state") def test_when_hyperion_requested_then_default_state_and_collection_run( default_state: MagicMock, bluesky_context: BlueskyContext, @@ -312,7 +319,9 @@ async def _assert_baton_released(baton: Baton): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_when_exception_raised_in_collection_then_loop_stops_and_baton_released( agamemnon: MagicMock, mock_load_centre_collect: MagicMock, @@ -333,7 +342,9 @@ async def test_when_exception_raised_in_collection_then_loop_stops_and_baton_rel @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_when_warning_exception_raised_in_collection_then_loop_continues( agamemnon: MagicMock, mock_load_centre_collect: MagicMock, @@ -363,7 +374,7 @@ def raise_else_return(value): return value -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state") async def test_when_exception_raised_in_default_state_then_baton_released( default_state: MagicMock, bluesky_context: BlueskyContext, @@ -378,7 +389,9 @@ async def test_when_exception_raised_in_default_state_then_baton_released( @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_when_exception_raised_in_getting_agamemnon_instruction_then_loop_stops_and_baton_released( agamemnon: MagicMock, udc_runner: PlanRunner, @@ -393,8 +406,10 @@ async def test_when_exception_raised_in_getting_agamemnon_instruction_then_loop_ @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.load_centre_collect_full") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch("mx_bluesky.hyperion.in_process_runner.load_centre_collect_full") +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_when_no_agamemnon_instructions_left_then_loop_stops_and_baton_released( collection: MagicMock, agamemnon: MagicMock, @@ -410,7 +425,9 @@ async def test_when_no_agamemnon_instructions_left_then_loop_stops_and_baton_rel @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_when_other_user_requested_collection_finished_then_baton_released( agamemnon: MagicMock, bluesky_context: BlueskyContext, @@ -441,7 +458,7 @@ def fake_collection_with_baton_request_part_way_through(*args): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state") async def test_when_multiple_agamemnon_instructions_then_default_state_only_run_once( default_state: MagicMock, agamemnon: MagicMock, @@ -488,7 +505,7 @@ def test_initialise_udc_reloads_all_devices(dont_patch_clear_devices): ] ), ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", MagicMock()) +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", MagicMock()) def test_baton_handler_loop_waits_if_wait_instruction_received( bluesky_context_with_sim_run_engine: tuple[list[Msg], BlueskyContext], sim_run_engine: RunEngineSimulator, @@ -503,7 +520,7 @@ def test_baton_handler_loop_waits_if_wait_instruction_received( ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", MagicMock()) +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", MagicMock()) def test_main_loop_rejects_unrecognised_instruction_when_received( bluesky_context_with_sim_run_engine: tuple[list[Msg], BlueskyContext], sim_run_engine: RunEngineSimulator, @@ -528,7 +545,9 @@ def test_main_loop_rejects_unrecognised_instruction_when_received( run_udc_when_requested(context, InProcessRunner(context, True)) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_shutdown_releases_the_baton( mock_create_params_from_agamemnon: MagicMock, @@ -562,7 +581,9 @@ async def wait_and_then_shutdown(): ), ], ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_run_forever_resumes_collection_when_baton_taken_away( mock_create_parameters_from_agamemnon: MagicMock, @@ -606,7 +627,9 @@ async def take_requested_baton_away_then_wait_for_release_then_re_request(): [AGAMEMNON_WAIT_INSTRUCTION], ], ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @pytest.mark.timeout(PYTEST_TEST_TIMEOUT_S) async def test_run_forever_resumes_collection_when_normal_completion_and_baton_returned( mock_create_parameters_from_agamemnon: MagicMock, @@ -640,7 +663,9 @@ async def wait_for_baton_release_then_re_request(): "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", side_effect=AssertionError("Runner started command processing without baton"), ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_run_forever_handles_shutdown_while_waiting_for_baton( mock_create_parameters_from_agamemnon: MagicMock, udc_runner: PlanRunner, @@ -672,7 +697,9 @@ async def issue_shutdown_without_baton(): [], ], ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) def test_run_forever_clears_error_status_on_resume( mock_create_parameters_from_agamemnon: MagicMock, udc_runner: PlanRunner, @@ -683,7 +710,7 @@ def test_run_forever_clears_error_status_on_resume( async def error_with_command_then_resume(): try: with patch( - "mx_bluesky.hyperion.plan_runner._runner_sleep", + "mx_bluesky.hyperion.in_process_runner._runner_sleep", side_effect=RuntimeError("Simulated plan exception"), ): function_is_patched.set() @@ -716,7 +743,9 @@ async def error_with_command_then_resume(): ], ) @patch("mx_bluesky.hyperion.baton_handler.set_commissioning_signal") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) async def test_commissioning_signal_set_on_baton_acquire( mock_set_commissioning_signal: MagicMock, mock_create_parameters_from_agamemnon: MagicMock, @@ -767,7 +796,9 @@ async def release_baton_and_check_commissioning_signal_set(): "mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon", MagicMock(side_effect=[[AGAMEMNON_WAIT_INSTRUCTION], []]), ) -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_udc_start_alert_when_baton_acquired( mock_get_alerting_service: MagicMock, @@ -783,7 +814,9 @@ def test_run_udc_when_requested_raises_udc_start_alert_when_baton_acquired( @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @patch("mx_bluesky.hyperion.baton_handler.clear_all_device_caches", MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_baton_release_udc_completed_event_when_hyperion_releases_the_baton( @@ -826,7 +859,9 @@ def wait_then_nothing(): @patch("mx_bluesky.hyperion.baton_handler.create_parameters_from_agamemnon") -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state", new=MagicMock()) +@patch( + "mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state", new=MagicMock() +) @patch("mx_bluesky.hyperion.baton_handler.clear_all_device_caches", MagicMock()) @patch("mx_bluesky.hyperion.baton_handler.get_alerting_service") def test_run_udc_when_requested_raises_baton_release_event_when_baton_requested_from_hyperion( @@ -866,7 +901,7 @@ def wait_then_nothing(): ) -@patch("mx_bluesky.hyperion.plan_runner.robot_unload") +@patch("mx_bluesky.hyperion.blueapi_plans._robot_unload") def test_robot_unload_performed_when_no_more_agamemnon_instructions( mock_robot_unload, bluesky_context: BlueskyContext, @@ -900,7 +935,7 @@ def request_baton_away_from_hyperion(*args): mock_load_centre_collect.side_effect = request_baton_away_from_hyperion -@patch("mx_bluesky.hyperion.plan_runner.robot_unload") +@patch("mx_bluesky.hyperion.blueapi_plans._robot_unload") def test_robot_unload_performed_when_baton_requested_away_from_hyperion( mock_robot_unload, bluesky_context: BlueskyContext, @@ -921,7 +956,7 @@ def test_robot_unload_performed_when_baton_requested_away_from_hyperion( ) -@patch("mx_bluesky.hyperion.plan_runner.robot_unload") +@patch("mx_bluesky.hyperion.blueapi_plans._robot_unload") def test_robot_unload_not_performed_when_beamline_error( mock_robot_unload, bluesky_context: BlueskyContext, @@ -937,7 +972,7 @@ def test_robot_unload_not_performed_when_beamline_error( mock_robot_unload.assert_not_called() -@patch("mx_bluesky.hyperion.plan_runner.robot_unload") +@patch("mx_bluesky.hyperion.blueapi_plans._robot_unload") def test_robot_unload_still_performed_when_sample_exception( mock_robot_unload, bluesky_context: BlueskyContext, @@ -961,7 +996,7 @@ def test_robot_unload_still_performed_when_sample_exception( ) -@patch("mx_bluesky.hyperion.plan_runner.robot_unload") +@patch("mx_bluesky.hyperion.blueapi_plans._robot_unload") def test_detector_shutter_closed_when_baton_requested_away_from_hyperion( mock_robot_unload, bluesky_context: BlueskyContext, @@ -981,7 +1016,7 @@ def test_detector_shutter_closed_when_baton_requested_away_from_hyperion( get_mock_put(detector_motion.shutter).assert_called_once() -@patch("mx_bluesky.hyperion.plan_runner.move_to_udc_default_state") +@patch("mx_bluesky.hyperion.in_process_runner.move_to_udc_default_state") def test_hyperion_doesnt_exit_if_udc_default_state_fails_a_check( mock_move_to_udc_default_state, bluesky_context: BlueskyContext, diff --git a/tests/unit_tests/hyperion/test_plan_runner.py b/tests/unit_tests/hyperion/test_plan_runner.py index 10299d5046..caef41310c 100644 --- a/tests/unit_tests/hyperion/test_plan_runner.py +++ b/tests/unit_tests/hyperion/test_plan_runner.py @@ -8,13 +8,14 @@ from bluesky import RunEngine from bluesky import plan_stubs as bps -from mx_bluesky.hyperion.plan_runner import InProcessRunner, PlanError +from mx_bluesky.hyperion.in_process_runner import InProcessRunner +from mx_bluesky.hyperion.plan_runner import PlanError @pytest.fixture(autouse=True) def patch_timer_poll_interval(): with patch( - "mx_bluesky.hyperion.plan_runner.InProcessRunner.EXTERNAL_CALLBACK_POLL_INTERVAL_S", + "mx_bluesky.hyperion.plan_runner.PlanRunner.EXTERNAL_CALLBACK_POLL_INTERVAL_S", 0.01, ): yield @@ -23,7 +24,7 @@ def patch_timer_poll_interval(): @pytest.fixture() def patch_timer_expiry(): with patch( - "mx_bluesky.hyperion.plan_runner.InProcessRunner.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S", + "mx_bluesky.hyperion.plan_runner.PlanRunner.EXTERNAL_CALLBACK_WATCHDOG_TIMER_S", 0.1, ): yield diff --git a/tests/unit_tests/hyperion/test_plan_runner_api.py b/tests/unit_tests/hyperion/test_plan_runner_api.py index 535ab72a84..e9c08ab4a7 100644 --- a/tests/unit_tests/hyperion/test_plan_runner_api.py +++ b/tests/unit_tests/hyperion/test_plan_runner_api.py @@ -5,7 +5,7 @@ from flask.testing import FlaskClient from mx_bluesky.common.parameters.constants import Status -from mx_bluesky.hyperion.plan_runner import InProcessRunner +from mx_bluesky.hyperion.in_process_runner import InProcessRunner from mx_bluesky.hyperion.plan_runner_api import ( create_app_for_udc, ) From a7aff93d7aff5fc0f801aa4bccbfa791bbe5633e Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Thu, 18 Dec 2025 17:00:30 +0000 Subject: [PATCH 7/9] Remove InProcessRunner duplicate implementation rebase artifact --- src/mx_bluesky/hyperion/plan_runner.py | 126 +------------------------ 1 file changed, 2 insertions(+), 124 deletions(-) diff --git a/src/mx_bluesky/hyperion/plan_runner.py b/src/mx_bluesky/hyperion/plan_runner.py index b91cca24cf..561c13e479 100644 --- a/src/mx_bluesky/hyperion/plan_runner.py +++ b/src/mx_bluesky/hyperion/plan_runner.py @@ -1,35 +1,15 @@ import threading import time from abc import abstractmethod -from collections.abc import Callable, Sequence -from functools import partial -from typing import Any +from collections.abc import Sequence from blueapi.core import BlueskyContext from bluesky import plan_stubs as bps -from bluesky.utils import MsgGenerator, RequestAbort -from dodal.devices.aperturescatterguard import ApertureScatterguard -from dodal.devices.detector.detector_motion import DetectorMotion -from dodal.devices.motors import XYZStage -from dodal.devices.robot import BartRobot -from dodal.devices.smargon import Smargon +from bluesky.utils import MsgGenerator from mx_bluesky.common.parameters.components import MxBlueskyParameters from mx_bluesky.common.parameters.constants import Status -from mx_bluesky.common.utils.context import ( - device_composite_from_context, - find_device_in_context, -) -from mx_bluesky.common.utils.exceptions import WarningError from mx_bluesky.common.utils.log import LOGGER -from mx_bluesky.hyperion.blueapi_plans import clean_up_udc, move_to_udc_default_state -from mx_bluesky.hyperion.experiment_plans.load_centre_collect_full_plan import ( - create_devices, - load_centre_collect_full, -) -from mx_bluesky.hyperion.experiment_plans.udc_default_state import UDCDefaultDevices -from mx_bluesky.hyperion.parameters.components import UDCCleanup, UDCDefaultState, Wait -from mx_bluesky.hyperion.parameters.load_centre_collect import LoadCentreCollect from mx_bluesky.hyperion.runner import BaseRunner @@ -104,105 +84,3 @@ def issue_abort(): def _external_callbacks_are_alive(self) -> bool: return time.monotonic() < self._callback_watchdog_expiry - - -class InProcessRunner(PlanRunner): - """Runner that executes experiments from inside a running Bluesky plan""" - - def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: - super().__init__(context, dev_mode) - self._current_status: Status = Status.IDLE - - def decode_and_execute( - self, current_visit: str | None, parameter_list: Sequence[MxBlueskyParameters] - ) -> MsgGenerator: - for parameters in parameter_list: - LOGGER.info( - f"Executing plan with parameters: {parameters.model_dump_json(indent=2)}" - ) - match parameters: - case LoadCentreCollect(): - current_visit = parameters.visit - devices: Any = create_devices(self.context) - yield from self.execute_plan( - partial(load_centre_collect_full, devices, parameters) - ) - case Wait(): - yield from self.execute_plan(partial(_runner_sleep, parameters)) - case UDCDefaultState(): - udc_default_devices: UDCDefaultDevices = ( - device_composite_from_context(self.context, UDCDefaultDevices) - ) - yield from move_to_udc_default_state(udc_default_devices) - case UDCCleanup(): - yield from _clean_up_udc(self.context, current_visit) - case _: - raise AssertionError( - f"Unsupported instruction decoded from agamemnon {type(parameters)}" - ) - return current_visit - - def execute_plan( - self, - experiment: Callable[[], MsgGenerator], - ) -> MsgGenerator: - """Execute the specified experiment plan. - Args: - experiment: The experiment to run - Raises: - PlanError: If the plan raised an exception - RequestAbort: If the RunEngine aborted during execution""" - - self._current_status = Status.BUSY - - try: - yield from self.check_external_callbacks_are_alive() - yield from experiment() - self._current_status = Status.IDLE - except WarningError as e: - LOGGER.warning("Plan failed with warning", exc_info=e) - self._current_status = Status.FAILED - except RequestAbort: - # This will occur when the run engine processes an abort when we shut down - LOGGER.info("UDC Runner aborting") - raise - except Exception as e: - LOGGER.error("Plan failed with exception", exc_info=e) - self._current_status = Status.FAILED - raise PlanError("Exception thrown in plan execution") from e - - def shutdown(self): - """Performs a prompt shutdown. Aborts the run engine and terminates the loop - waiting for messages.""" - - LOGGER.info("Shutting down: Stopping the run engine gracefully") - if self.current_status != Status.ABORTING: - self._current_status = Status.ABORTING - self.request_run_engine_abort() - return - - @property - def current_status(self) -> Status: - return self._current_status - - -def _runner_sleep(parameters: Wait) -> MsgGenerator: - yield from bps.sleep(parameters.duration_s) - - -def _clean_up_udc(context: BlueskyContext, visit: str) -> MsgGenerator: - robot = find_device_in_context(context, "robot", BartRobot) - smargon = find_device_in_context(context, "smargon", Smargon) - aperture_scatterguard = find_device_in_context( - context, "aperture_scatterguard", ApertureScatterguard - ) - lower_gonio = find_device_in_context(context, "lower_gonio", XYZStage) - detector_motion = find_device_in_context(context, "detector_motion", DetectorMotion) - yield from clean_up_udc( - visit, - robot=robot, - smargon=smargon, - aperture_scatterguard=aperture_scatterguard, - lower_gonio=lower_gonio, - detector_motion=detector_motion, - ) From da524f425bd7a9f2e8d45c8b100990f4047a0df9 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Fri, 19 Dec 2025 09:48:27 +0000 Subject: [PATCH 8/9] Make ruff happy --- src/mx_bluesky/hyperion/in_process_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mx_bluesky/hyperion/in_process_runner.py b/src/mx_bluesky/hyperion/in_process_runner.py index 4fcef1991e..6a86bb2532 100644 --- a/src/mx_bluesky/hyperion/in_process_runner.py +++ b/src/mx_bluesky/hyperion/in_process_runner.py @@ -1,4 +1,3 @@ -import time from collections.abc import Callable, Sequence from functools import partial from typing import Any @@ -33,6 +32,7 @@ class InProcessRunner(PlanRunner): """Runner that executes experiments from inside a running Bluesky plan""" + def __init__(self, context: BlueskyContext, dev_mode: bool) -> None: super().__init__(context, dev_mode) self._current_status: Status = Status.IDLE From 5b392ce546f960d96320d75c2351d4e46004d171 Mon Sep 17 00:00:00 2001 From: Robert Tuck Date: Thu, 15 Jan 2026 15:08:18 +0000 Subject: [PATCH 9/9] Unpin dodal --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 1f6a976739..f22c5df38b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,7 @@ dependencies = [ "ophyd >= 1.10.5", "ophyd-async >= 0.14.0", "bluesky >= 1.14.6", - "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@a946a46ffd418ea25fbae9134954659cfa223d6f", + "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@main", ]