Remove ability to use internal callbacks

helmchart/templates/deployment.yaml

@@ -119,7 +119,6 @@ spec:
             value: "/dls_sw/apps/zocalo/live/configuration.yaml"
           - name: ISPYB_CONFIG_PATH
             value: "/dls_sw/dasc/mariadb/credentials/ispyb-hyperion-{{ .Values.hyperion.beamline }}.cfg"
-        args: [ "--external-callbacks" ]
           {{- end }}
         readinessProbe:
           exec:

run_hyperion.sh

@@ -5,7 +5,6 @@ START=1
 SKIP_STARTUP_CONNECTION=false
 VERBOSE_EVENT_LOGGING=false
 IN_DEV=false
-EXTERNAL_CALLBACK_SERVICE=false
 
 for option in "$@"; do
     case $option in
@@ -28,9 +27,6 @@ for option in "$@"; do
         --verbose-event-logging)
             VERBOSE_EVENT_LOGGING=true
             ;;
-        --external-callbacks)
-            EXTERNAL_CALLBACK_SERVICE=true
-            ;;
 
         --help|--info|--h)
 
@@ -119,11 +115,9 @@ if [[ $START == 1 ]]; then
 
     #Add future arguments here
     declare -A h_only_args=(        ["SKIP_STARTUP_CONNECTION"]="$SKIP_STARTUP_CONNECTION"
-                                    ["VERBOSE_EVENT_LOGGING"]="$VERBOSE_EVENT_LOGGING"
-                                    ["EXTERNAL_CALLBACK_SERVICE"]="$EXTERNAL_CALLBACK_SERVICE" )
+                                    ["VERBOSE_EVENT_LOGGING"]="$VERBOSE_EVENT_LOGGING" )
     declare -A h_only_arg_strings=( ["SKIP_STARTUP_CONNECTION"]="--skip-startup-connection"
-                                    ["VERBOSE_EVENT_LOGGING"]="--verbose-event-logging"
-                                    ["EXTERNAL_CALLBACK_SERVICE"]="--external-callbacks")
+                                    ["VERBOSE_EVENT_LOGGING"]="--verbose-event-logging" )
 
     declare -A h_and_cb_args=( ["IN_DEV"]="$IN_DEV" )
     declare -A h_and_cb_arg_strings=( ["IN_DEV"]="--dev" )
@@ -146,9 +140,7 @@ if [[ $START == 1 ]]; then
 
     unset PYEPICS_LIBCA
     hyperion `echo $h_commands;`>$start_log_path  2>&1 &
-    if [ $EXTERNAL_CALLBACK_SERVICE == true ]; then
-        hyperion-callbacks `echo $cb_commands;`>$callback_start_log_path 2>&1 &
-    fi
+    hyperion-callbacks `echo $cb_commands;`>$callback_start_log_path 2>&1 &
     echo "$(date) Waiting for Hyperion to start"
 
     for i in {1..30}

src/mx_bluesky/hyperion/__main__.py

@@ -38,12 +38,6 @@
     PLAN_REGISTRY,
     PlanNotFound,
 )
-from mx_bluesky.hyperion.external_interaction.callbacks.__main__ import (
-    setup_logging as setup_callback_logging,
-)
-from mx_bluesky.hyperion.external_interaction.callbacks.common.callback_util import (
-    CallbacksFactory,
-)
 from mx_bluesky.hyperion.parameters.cli import parse_cli_args
 from mx_bluesky.hyperion.parameters.constants import CONST
 from mx_bluesky.hyperion.utils.context import setup_context
@@ -57,7 +51,6 @@ class Command:
     devices: Any | None = None
     experiment: Callable[[Any, Any], MsgGenerator] | None = None
     parameters: MxBlueskyParameters | None = None
-    callbacks: CallbacksFactory | None = None
 
 
 @dataclass
@@ -89,7 +82,6 @@ def __init__(
         RE: RunEngine,
         context: BlueskyContext,
         skip_startup_connection=False,
-        use_external_callbacks: bool = False,
     ) -> None:
         self.command_queue: Queue[Command] = Queue()
         self.current_status: StatusAndMessage = StatusAndMessage(Status.IDLE)
@@ -100,15 +92,12 @@ def __init__(
 
         self.RE = RE
         self.context = context
-        self.subscribed_per_plan_callbacks: list[int] = []
         RE.subscribe(self.aperture_change_callback)
         RE.subscribe(self.logging_uid_tag_callback)
 
-        self.use_external_callbacks = use_external_callbacks
-        if self.use_external_callbacks:
-            LOGGER.info("Connecting to external callback ZMQ proxy...")
-            self.publisher = Publisher(f"localhost:{CONST.CALLBACK_0MQ_PROXY_PORTS[0]}")
-            RE.subscribe(self.publisher)
+        LOGGER.info("Connecting to external callback ZMQ proxy...")
+        self.publisher = Publisher(f"localhost:{CONST.CALLBACK_0MQ_PROXY_PORTS[0]}")
+        RE.subscribe(self.publisher)
 
         if VERBOSE_EVENT_LOGGING:
             RE.subscribe(VerbosePlanExecutionLoggingCallback())
@@ -124,7 +113,6 @@ def start(
         experiment: Callable,
         parameters: MxBlueskyParameters,
         plan_name: str,
-        callbacks: CallbacksFactory | None,
     ) -> StatusAndMessage:
         LOGGER.info(f"Started with parameters: {parameters.model_dump_json(indent=2)}")
 
@@ -143,7 +131,6 @@ def start(
                     devices=devices,
                     experiment=experiment,
                     parameters=parameters,
-                    callbacks=callbacks,
                 )
             )
             return StatusAndMessage(Status.SUCCESS)
@@ -182,17 +169,6 @@ def wait_on_queue(self):
                 if command.experiment is None:
                     raise ValueError("No experiment provided for START")
                 try:
-                    if (
-                        not self.use_external_callbacks
-                        and command.callbacks
-                        and (cbs := command.callbacks())
-                    ):
-                        LOGGER.info(
-                            f"Using callbacks for this plan: {not self.use_external_callbacks} - {cbs}"
-                        )
-                        self.subscribed_per_plan_callbacks += [
-                            self.RE.subscribe(cb) for cb in cbs
-                        ]
                     with TRACER.start_span("do_run"):
                         self.RE(command.experiment(command.devices, command.parameters))
 
@@ -213,11 +189,6 @@ def wait_on_queue(self):
                         self.last_run_aborted = False
                     else:
                         self.current_status = make_error_status_and_message(exception)
-                finally:
-                    [
-                        self.RE.unsubscribe(cb)
-                        for cb in self.subscribed_per_plan_callbacks
-                    ]
 
 
 def compose_start_args(context: BlueskyContext, plan_name: str, action: Actions):
@@ -226,7 +197,6 @@ def compose_start_args(context: BlueskyContext, plan_name: str, action: Actions)
         raise PlanNotFound(f"Experiment plan '{plan_name}' not found in registry.")
 
     experiment_internal_param_type = experiment_registry_entry.get("param_type")
-    callback_type = experiment_registry_entry.get("callback_collection_type")
     plan = context.plan_functions.get(plan_name)
     if experiment_internal_param_type is None:
         raise PlanNotFound(
@@ -244,7 +214,7 @@ def compose_start_args(context: BlueskyContext, plan_name: str, action: Actions)
         raise ValueError(
             f"Supplied parameters don't match the plan for this endpoint {request.data}, for plan {plan_name}"
         ) from e
-    return plan, parameters, plan_name, callback_type
+    return plan, parameters, plan_name
 
 
 class RunExperiment(Resource):
@@ -257,12 +227,10 @@ def put(self, plan_name: str, action: Actions):
         status_and_message = StatusAndMessage(Status.FAILED, f"{action} not understood")
         if action == Actions.START.value:
             try:
-                plan, params, plan_name, callback_type = compose_start_args(
+                plan, params, plan_name = compose_start_args(
                     self.context, plan_name, action
                 )
-                status_and_message = self.runner.start(
-                    plan, params, plan_name, callback_type
-                )
+                status_and_message = self.runner.start(plan, params, plan_name)
             except Exception as e:
                 status_and_message = make_error_status_and_message(e)
                 LOGGER.error(format_exception(e))
@@ -313,15 +281,13 @@ def create_app(
     test_config=None,
     RE: RunEngine = RunEngine({}),
     skip_startup_connection: bool = False,
-    use_external_callbacks: bool = False,
 ) -> tuple[Flask, BlueskyRunner]:
     context = setup_context(
         wait_for_connection=not skip_startup_connection,
     )
     runner = BlueskyRunner(
         RE,
         context=context,
-        use_external_callbacks=use_external_callbacks,
         skip_startup_connection=skip_startup_connection,
     )
     app = Flask(__name__)
@@ -351,12 +317,9 @@ def create_targets():
     do_default_logging_setup(
         CONST.LOG_FILE_NAME, CONST.GRAYLOG_PORT, dev_mode=args.dev_mode
     )
-    if not args.use_external_callbacks:
-        setup_callback_logging(args.dev_mode)
     LOGGER.info(f"Hyperion launched with args:{argv}")
     app, runner = create_app(
         skip_startup_connection=args.skip_startup_connection,
-        use_external_callbacks=args.use_external_callbacks,
     )
     return app, runner, hyperion_port, args.dev_mode
 

src/mx_bluesky/hyperion/experiment_plans/experiment_registry.py

@@ -11,13 +11,6 @@
     pin_centre_then_xray_centre_plan,
     robot_load_then_centre_plan,
 )
-from mx_bluesky.hyperion.external_interaction.callbacks.common.callback_util import (
-    CallbacksFactory,
-    create_gridscan_callbacks,
-    create_load_centre_collect_callbacks,
-    create_robot_load_and_centre_callbacks,
-    create_rotation_callbacks,
-)
 from mx_bluesky.hyperion.parameters.gridscan import (
     GridScanWithEdgeDetect,
     HyperionSpecifiedThreeDGridScan,
@@ -47,44 +40,36 @@ class ExperimentRegistryEntry(TypedDict):
         | LoadCentreCollect
         | RobotLoadThenCentre
     ]
-    callbacks_factory: CallbacksFactory
 
 
 PLAN_REGISTRY: dict[str, ExperimentRegistryEntry] = {
     "flyscan_xray_centre": {
         "setup": flyscan_xray_centre_plan.create_devices,
         "param_type": HyperionSpecifiedThreeDGridScan,
-        "callbacks_factory": create_gridscan_callbacks,
     },
     "grid_detect_then_xray_centre": {
         "setup": grid_detect_then_xray_centre_plan.create_devices,
         "param_type": GridScanWithEdgeDetect,
-        "callbacks_factory": create_gridscan_callbacks,
     },
     "rotation_scan": {
         "setup": rotation_scan_plan.create_devices,
         "param_type": RotationScan,
-        "callbacks_factory": create_rotation_callbacks,
     },
     "pin_tip_centre_then_xray_centre": {
         "setup": pin_centre_then_xray_centre_plan.create_devices,
         "param_type": PinTipCentreThenXrayCentre,
-        "callbacks_factory": create_gridscan_callbacks,
     },
     "robot_load_then_centre": {
         "setup": robot_load_then_centre_plan.create_devices,
         "param_type": RobotLoadThenCentre,
-        "callbacks_factory": create_robot_load_and_centre_callbacks,
     },
     "multi_rotation_scan": {
         "setup": rotation_scan_plan.create_devices,
         "param_type": MultiRotationScan,
-        "callbacks_factory": create_rotation_callbacks,
     },
     "load_centre_collect_full": {
         "setup": load_centre_collect_full_plan.create_devices,
         "param_type": LoadCentreCollect,
-        "callbacks_factory": create_load_centre_collect_callbacks,
     },
 }
 

src/mx_bluesky/hyperion/external_interaction/callbacks/__main__.py

@@ -3,6 +3,7 @@
 from threading import Thread
 from time import sleep
 
+from bluesky.callbacks import CallbackBase
 from bluesky.callbacks.zmq import Proxy, RemoteDispatcher
 from dodal.log import LOGGER as dodal_logger
 from dodal.log import set_up_all_logging_handlers
@@ -48,17 +49,33 @@
 ERROR_LOG_BUFFER_LINES = 5000
 
 
-def setup_callbacks():
-    return [
+def create_gridscan_callbacks() -> tuple[
+    GridscanNexusFileCallback, GridscanISPyBCallback
+]:
+    return (
         GridscanNexusFileCallback(param_type=HyperionSpecifiedThreeDGridScan),
         GridscanISPyBCallback(
             param_type=GridCommonWithHyperionDetectorParams,
             emit=ZocaloCallback(CONST.PLAN.DO_FGS, CONST.ZOCALO_ENV),
         ),
+    )
+
+
+def create_rotation_callbacks() -> tuple[
+    RotationNexusFileCallback, RotationISPyBCallback
+]:
+    return (
         RotationNexusFileCallback(),
         RotationISPyBCallback(
             emit=ZocaloCallback(CONST.PLAN.ROTATION_MAIN, CONST.ZOCALO_ENV)
         ),
+    )
+
+
+def setup_callbacks() -> list[CallbackBase]:
+    return [
+        *create_gridscan_callbacks(),
+        *create_rotation_callbacks(),
         LogUidTaggingCallback(),
         RobotLoadISPyBCallback(),
         SampleHandlingCallback(),