Remove device_factory and supporting methods

Author: tpoliaw

src/dodal/cli.py

@@ -1,17 +1,15 @@
 import importlib
 import os
 from collections.abc import Mapping
-from pathlib import Path
 
 import click
 from bluesky.run_engine import RunEngine
-from ophyd_async.core import NotConnectedError, StaticPathProvider, UUIDFilenameProvider
-from ophyd_async.plan_stubs import ensure_connected
+from click.exceptions import ClickException
+from ophyd_async.core import NotConnectedError
 
 from dodal.beamlines import all_beamline_names, module_name_for_beamline
-from dodal.common.beamlines.beamline_utils import set_path_provider
 from dodal.device_manager import DeviceManager
-from dodal.utils import AnyDevice, filter_ophyd_devices, make_all_devices
+from dodal.utils import AnyDevice
 
 from . import __version__
 
@@ -103,7 +101,7 @@ def connect(
 
     # We need to make a RunEngine to allow ophyd-async devices to connect.
     # See https://blueskyproject.io/ophyd-async/main/explanations/event-loop-choice.html
-    run_engine = RunEngine(call_returns_result=True)
+    _run_engine = RunEngine(call_returns_result=True)
 
     print(f"Attempting connection to {beamline} (using {full_module_path})")
 
@@ -121,15 +119,9 @@ def connect(
             timeout=timeout,
         )
     else:
-        print(f"No device manager named '{device_manager}' found in {mod}")
-        _spoof_path_provider()
-        devices, instance_exceptions = make_all_devices(
-            full_module_path,
-            include_skipped=all,
-            fake_with_ophyd_sim=sim_backend,
-            wait_for_connection=False,
+        raise ClickException(
+            f"No device manager named '{device_manager}' found in {mod}"
         )
-        devices, connect_exceptions = _connect_devices(run_engine, devices, sim_backend)
 
     # Inform user of successful connections
     _report_successful_devices(devices, sim_backend)
@@ -151,35 +143,3 @@ def _report_successful_devices(
 
     print(f"{len(devices)} devices connected{sim_statement}:")
     print(connected_devices)
-
-
-def _connect_devices(
-    run_engine: RunEngine,
-    devices: Mapping[str, AnyDevice],
-    sim_backend: bool,
-) -> tuple[Mapping[str, AnyDevice], Mapping[str, Exception]]:
-    ophyd_devices, ophyd_async_devices = filter_ophyd_devices(devices)
-    exceptions = {}
-
-    # Connect ophyd devices
-    for name, device in ophyd_devices.items():
-        try:
-            device.wait_for_connection()
-        except Exception as ex:
-            exceptions[name] = ex
-
-    # Connect ophyd-async devices
-    try:
-        run_engine(ensure_connected(*ophyd_async_devices.values(), mock=sim_backend))
-    except NotConnectedError as ex:
-        exceptions = {**exceptions, **ex.sub_errors}
-
-    # Only return the subset of devices that haven't raised an exception
-    successful_devices = {
-        name: device for name, device in devices.items() if name not in exceptions
-    }
-    return successful_devices, exceptions
-
-
-def _spoof_path_provider() -> None:
-    set_path_provider(StaticPathProvider(UUIDFilenameProvider(), Path("/tmp")))

src/dodal/common/beamlines/beamline_utils.py

@@ -1,30 +1,17 @@
 import inspect
 from collections.abc import Callable
-from typing import Annotated, Final, TypeVar, cast
+from typing import TypeVar
 
-from bluesky.run_engine import call_in_bluesky_event_loop
 from daq_config_server import ConfigClient
-from ophyd import Device as OphydV1Device
-from ophyd.sim import make_fake_device
 from ophyd_async.core import (
-    DEFAULT_TIMEOUT,
     PathProvider,
 )
-from ophyd_async.core import Device as OphydV2Device
-from ophyd_async.core import wait_for_connection as v2_device_wait_for_connection
 
 from dodal.log import LOGGER
 from dodal.utils import (
     AnyDevice,
-    BeamlinePrefix,
-    DeviceInitializationController,
-    SkipType,
-    skip_device,
 )
 
-DEFAULT_CONNECTION_TIMEOUT: Final[float] = 5.0
-
-ACTIVE_DEVICES: dict[str, AnyDevice] = {}
 BL = ""
 
 
@@ -33,134 +20,15 @@ def set_beamline(beamline: str):
     BL = beamline
 
 
-def clear_devices():
-    global ACTIVE_DEVICES
-    for name in list(ACTIVE_DEVICES):
-        clear_device(name)
-
-
-def clear_device(name: str):
-    global ACTIVE_DEVICES
-    device = ACTIVE_DEVICES[name]
-    if isinstance(device, OphydV1Device):
-        device.destroy()
-    del ACTIVE_DEVICES[name]
-
-
-def list_active_devices() -> list[str]:
-    global ACTIVE_DEVICES
-    return list(ACTIVE_DEVICES.keys())
-
-
 def active_device_is_same_type(
     active_device: AnyDevice, device: Callable[..., AnyDevice]
 ) -> bool:
     return inspect.isclass(device) and isinstance(active_device, device)
 
 
-def wait_for_connection(
-    device: AnyDevice,
-    timeout: float = DEFAULT_CONNECTION_TIMEOUT,
-    mock: bool = False,
-) -> None:
-    if isinstance(device, OphydV1Device):
-        device.wait_for_connection(timeout=timeout)
-    elif isinstance(device, OphydV2Device):
-        call_in_bluesky_event_loop(
-            v2_device_wait_for_connection(
-                coros=device.connect(mock=mock, timeout=timeout)
-            ),
-        )
-    else:
-        raise TypeError(
-            f"Invalid type {device.__class__.__name__} in _wait_for_connection"
-        )
-
-
 T = TypeVar("T", bound=AnyDevice)
 
 
-@skip_device()
-def device_instantiation(
-    device_factory: Callable[..., T],
-    name: str,
-    prefix: str,
-    wait: bool,
-    fake: bool,
-    post_create: Callable[[T], None] | None = None,
-    bl_prefix: bool = True,
-    **kwargs,
-) -> T:
-    """Method to allow generic creation of singleton devices. Meant to be used to easily
-    define lists of devices in beamline files. Additional keyword arguments are passed
-    directly to the device constructor.
-
-    Args:
-        device_factory (Callable): The device class.
-        name (str): The name for ophyd.
-        prefix (str): The PV prefix for the most (usually all) components.
-        wait (bool): Whether to run .wait_for_connection().
-        fake (bool): Whether to fake with ophyd.sim.
-        post_create (Callable): (optional) a function to be run on the device after
-            creation.
-        bl_prefix (bool): If true, add the beamline prefix when instantiating.
-        **kwargs: Arguments passed on to every device factory.
-
-    Returns:
-        The instance of the device.
-    """
-    already_existing_device: AnyDevice | None = ACTIVE_DEVICES.get(name)
-    if fake:
-        device_factory = cast(Callable[..., T], make_fake_device(device_factory))
-    if already_existing_device is None:
-        device_instance = device_factory(
-            name=name,
-            prefix=(
-                f"{(BeamlinePrefix(BL).beamline_prefix)}{prefix}"
-                if bl_prefix
-                else prefix
-            ),
-            **kwargs,
-        )
-        ACTIVE_DEVICES[name] = device_instance
-        if wait:
-            wait_for_connection(device_instance, mock=fake)
-
-    else:
-        if not active_device_is_same_type(already_existing_device, device_factory):
-            raise TypeError(
-                f"Can't instantiate device of type {device_factory} with the same "
-                f"name as an existing device. Device name '{name}' already used for "
-                f"a(n) {type(already_existing_device)}."
-            )
-        device_instance = cast(T, already_existing_device)
-    if post_create:
-        post_create(device_instance)
-    return device_instance
-
-
-def device_factory(
-    *,
-    use_factory_name: Annotated[bool, "Use factory name as name of device"] = True,
-    timeout: Annotated[float, "Timeout for connecting to the device"] = DEFAULT_TIMEOUT,
-    mock: Annotated[bool, "Use Signals with mock backends for device"] = False,
-    skip: Annotated[
-        SkipType,
-        "mark the factory to be (conditionally) skipped when beamline is imported by external program",
-    ] = False,
-) -> Callable[[Callable[[], T]], DeviceInitializationController[T]]:
-    def decorator(factory: Callable[[], T]) -> DeviceInitializationController[T]:
-        return DeviceInitializationController(
-            factory,
-            use_factory_name,
-            timeout,
-            mock,
-            skip,
-        )
-
-    return decorator
-
-
 def set_path_provider(provider: PathProvider):
     global PATH_PROVIDER
 

src/dodal/device_manager.py

@@ -14,28 +14,26 @@
     NamedTuple,
     ParamSpec,
     Self,
+    TypeAlias,
     TypeVar,
 )
 
 from bluesky.run_engine import get_bluesky_event_loop
+from ophyd.device import Device as OphydV1Device
 from ophyd.sim import make_fake_device
-
-from dodal.common.beamlines.beamline_utils import wait_for_connection
-from dodal.utils import (
-    AnyDevice,
-    OphydV1Device,
-    OphydV2Device,
-    SkipType,
-)
+from ophyd_async.core import Device as OphydV2Device
 
 DEFAULT_TIMEOUT = 30
 NO_DOCS = "No documentation available."
 
 T = TypeVar("T")
 Args = ParamSpec("Args")
 
+SkipType = bool | Callable[[], bool]
+
 V1 = TypeVar("V1", bound=OphydV1Device)
 V2 = TypeVar("V2", bound=OphydV2Device)
+AnyDevice: TypeAlias = OphydV1Device | OphydV2Device
 
 DeviceFactoryDecorator = Callable[[Callable[Args, V2]], "DeviceFactory[Args, V2]"]
 OphydInitialiser = Callable[Concatenate[V1, Args], V1 | None]
@@ -249,7 +247,7 @@ def __call__(self, dev: V1, *args: Args.args, **kwargs: Args.kwargs):
     def create(self, *args: Args.args, **kwargs: Args.kwargs) -> V1:
         device = self.factory(name=self.name, prefix=self.prefix)
         if self.wait:
-            wait_for_connection(device, timeout=self.timeout)
+            device.wait_for_connection(timeout=self.timeout)
         self.post_create(device, *args, **kwargs)
         return device
 

src/dodal/plans/save_panda.py

@@ -1,4 +1,5 @@
 import argparse
+import importlib
 import os
 import sys
 from argparse import ArgumentParser
@@ -12,7 +13,7 @@
 )
 
 from dodal.beamlines import module_name_for_beamline
-from dodal.utils import make_device
+from dodal.device_manager import DeviceFactory, DeviceManager
 
 
 def main(argv: list[str] | None = None):
@@ -70,19 +71,25 @@ def main(argv: list[str] | None = None):
     return 0
 
 
+def _build_panda(beamline, device_name) -> Device:
+    print(f"Building {device_name} for beamline {beamline}")
+    module_name = module_name_for_beamline(beamline)
+    mod = importlib.import_module("dodal.beamlines." + module_name)
+    device_manager: DeviceManager = mod.devices
+    return cast(DeviceFactory, device_manager[device_name]).build(
+        connect_immediately=True
+    )
+
+
 def _save_panda(beamline, device_name, output_directory, file_name):
     run_engine = RunEngine()
-    print("Creating devices...")
-    module_name = module_name_for_beamline(beamline)
+
     try:
-        devices = make_device(
-            f"dodal.beamlines.{module_name}", device_name, connect_immediately=True
-        )
+        panda = _build_panda(beamline, device_name)
     except Exception as error:
         sys.stderr.write(f"Couldn't create device {device_name}: {error}\n")
         sys.exit(1)
 
-    panda = devices[device_name]
     print(
         f"Saving to {output_directory}/{file_name} from {device_name} on {beamline}..."
     )