Merge branch 'main' into 364-auto-generate-rest-client-1

pyproject.toml

@@ -29,7 +29,7 @@ dependencies = [
     "uvicorn",
     "requests",
     "dls-bluesky-core",  #requires ophyd-async
-    "dls-dodal",
+    "dls-dodal>=1.24.0",
 ]
 dynamic = ["version"]
 license.file = "LICENSE"

src/blueapi/cli/cli.py

@@ -9,7 +9,7 @@
 from requests.exceptions import ConnectionError
 
 from blueapi import __version__
-from blueapi.cli.amq import AmqClient
+from blueapi.cli.event_bus_client import EventBusClient
 from blueapi.config import ApplicationConfig, ConfigLoader
 from blueapi.core import DataEvent
 from blueapi.messaging import MessageContext
@@ -135,7 +135,9 @@ def listen_to_events(obj: dict) -> None:
     """Listen to events output by blueapi"""
     config: ApplicationConfig = obj["config"]
     if config.stomp is not None:
-        amq_client = AmqClient(StompMessagingTemplate.autoconfigured(config.stomp))
+        event_bus_client = EventBusClient(
+            StompMessagingTemplate.autoconfigured(config.stomp)
+        )
     else:
         raise RuntimeError("Message bus needs to be configured")
 
@@ -150,8 +152,8 @@ def on_event(
         "Subscribing to all bluesky events from "
         f"{config.stomp.host}:{config.stomp.port}"
     )
-    with amq_client:
-        amq_client.subscribe_to_all_events(on_event)
+    with event_bus_client:
+        event_bus_client.subscribe_to_all_events(on_event)
         input("Press enter to exit")
 
 
@@ -182,6 +184,14 @@ def run_plan(
     client: BlueapiRestClient = obj["rest_client"]
 
     logger = logging.getLogger(__name__)
+
+    if config.stomp is not None:
+        _message_template = StompMessagingTemplate.autoconfigured(config.stomp)
+    else:
+        raise RuntimeError(
+            "Cannot run plans without Stomp configuration to track progress"
+        )
+    event_bus_client = EventBusClient(_message_template)
     finished_event: deque[WorkerEvent] = deque()
 
     def store_finished_event(event: WorkerEvent) -> None:
@@ -194,23 +204,15 @@ def store_finished_event(event: WorkerEvent) -> None:
     resp = client.create_task(task)
     task_id = resp.task_id
 
-    if eventbus:
-        if config.stomp is not None:
-            _message_template = StompMessagingTemplate.autoconfigured(config.stomp)
-        else:
-            raise RuntimeError(
-                "Cannot run plans without Stomp configuration to track progress"
-            )
-        amq_client = AmqClient(_message_template)
-        with amq_client:
-            amq_client.subscribe_to_topics(task_id, on_event=store_finished_event)
-            updated = client.update_worker_task(WorkerTask(task_id=task_id))
-
-            amq_client.wait_for_complete(timeout=timeout)
-
-            if amq_client.timed_out:
-                logger.error(f"Plan did not complete within {timeout} seconds")
-                return
+    with event_bus_client:
+        event_bus_client.subscribe_to_topics(task_id, on_event=store_finished_event)
+        updated = client.update_worker_task(WorkerTask(task_id=task_id))
+
+        event_bus_client.wait_for_complete(timeout=timeout)
+
+        if event_bus_client.timed_out:
+            logger.error(f"Plan did not complete within {timeout} seconds")
+            return
 
     process_event_after_finished(finished_event.pop(), logger)
     pprint(updated.dict())

src/blueapi/cli/amq.py → src/blueapi/cli/event_bus_client.py

@@ -18,7 +18,7 @@ def __init__(self, message: str) -> None:
 _Event = WorkerEvent | ProgressEvent | DataEvent
 
 
-class AmqClient:
+class EventBusClient:
     app: MessagingTemplate
     complete: threading.Event
     timed_out: bool | None

src/blueapi/cli/rest.py

@@ -15,7 +15,7 @@
 )
 from blueapi.worker import Task, TrackableTask, WorkerState
 
-from .amq import BlueskyRemoteError
+from .event_bus_client import BlueskyRemoteError
 
 T = TypeVar("T")
 

src/blueapi/core/context.py

@@ -6,14 +6,13 @@
 from types import ModuleType, UnionType
 from typing import Any, Generic, TypeVar, Union, get_args, get_origin, get_type_hints
 
-from bluesky.run_engine import RunEngine, call_in_bluesky_event_loop
+from bluesky.run_engine import RunEngine
 from pydantic import create_model
 from pydantic.fields import FieldInfo, ModelField
 
 from blueapi.config import EnvironmentConfig, SourceKind
 from blueapi.utils import (
     BlueapiPlanModelConfig,
-    connect_ophyd_async_devices,
     load_module_all,
 )
 
@@ -45,7 +44,6 @@ class BlueskyContext:
     plans: dict[str, Plan] = field(default_factory=dict)
     devices: dict[str, Device] = field(default_factory=dict)
     plan_functions: dict[str, PlanGenerator] = field(default_factory=dict)
-    sim: bool = field(default=False)
 
     _reference_cache: dict[type, type] = field(default_factory=dict)
 
@@ -78,13 +76,6 @@ def with_config(self, config: EnvironmentConfig) -> None:
             elif source.kind is SourceKind.DODAL:
                 self.with_dodal_module(mod)
 
-        call_in_bluesky_event_loop(
-            connect_ophyd_async_devices(
-                self.devices.values(),
-                self.sim,
-            )
-        )
-
     def with_plan_module(self, module: ModuleType) -> None:
         """
         Register all functions in the module supplied as plans.

src/blueapi/service/handler.py

@@ -154,9 +154,7 @@ def setup_handler(
 
     handler = Handler(
         config,
-        context=BlueskyContext(
-            sim=False,
-        ),
+        context=BlueskyContext(),
     )
     handler.start()
 

src/blueapi/utils/__init__.py

@@ -1,7 +1,6 @@
 from .base_model import BlueapiBaseModel, BlueapiModelConfig, BlueapiPlanModelConfig
 from .invalid_config_error import InvalidConfigError
 from .modules import load_module_all
-from .ophyd_async_connect import connect_ophyd_async_devices
 from .serialization import serialize
 from .thread_exception import handle_all_exceptions
 
@@ -14,5 +13,4 @@
     "BlueapiModelConfig",
     "BlueapiPlanModelConfig",
     "InvalidConfigError",
-    "connect_ophyd_async_devices",
 ]