fix all lint and pytest

src/dodal/beamlines/example_beamline.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Callable, Dict, Optional
+from typing import Callable, Optional
 
 from bluesky.run_engine import call_in_bluesky_event_loop
 from ophyd_async.core import DEFAULT_TIMEOUT
@@ -12,17 +12,15 @@
 
 
 class XYZDetector(OphydV2Device):
-    def __init__(self, *args, **kwargs):
+    def __init__(self, prefix: str, *args, **kwargs):
+        self.prefix = prefix
         super().__init__(*args, **kwargs)
 
     @property
     def hints(self):
         raise NotImplementedError
 
 
-LAZY_DEVICES: Dict[str, AnyDevice] = {}
-
-
 class ConnectionMode(Enum):
     LAZY = 0
     EAGER = 1
@@ -62,17 +60,12 @@ def __init__(self, config: DeviceInitializationConfig) -> None:
     # TODO right now the cache is in a global variable ACTIVE_DEVICES, that should change
     def see_if_device_is_in_cache(self, name: str) -> Optional[AnyDevice]:
         d = ACTIVE_DEVICES.get(name)
-        assert isinstance(d, OphydV2Device) or d is None
+        assert isinstance(d, AnyDevice) or d is None
         return d
 
     def add_device_to_cache(self, device: AnyDevice) -> None:
         ACTIVE_DEVICES[device.name] = device
 
-    # TODO right now the lazy devices are in a global variable LAZY_DEVICES, that should change
-    def add_device_to_lazy_cache(self, device: AnyDevice) -> None:
-        print(f"Device {device.name} is lazy, not initializing now")
-        LAZY_DEVICES[device.name] = device
-
     def __call__(
         self,
         factory: Callable[[], AnyDevice],
@@ -96,8 +89,7 @@ def __call__(
                     mock=self.config.default_use_mock_at_connection,
                 )
             )
-        else:
-            self.add_device_to_lazy_cache(device)
+
         return device
 
 

src/dodal/beamlines/i22.py

@@ -4,7 +4,6 @@
 from ophyd_async.panda import HDFPanda
 
 from dodal.common.beamlines.beamline_utils import (
-    device_factory,
     device_instantiation,
     get_directory_provider,
     set_directory_provider,
@@ -42,7 +41,6 @@
 )
 
 
-@device_factory(lazy=True)
 def saxs() -> PilatusDetector:
     return NXSasPilatus(
         "-EA-PILAT-01:",
@@ -374,4 +372,3 @@ def linkam(
         wait_for_connection,
         fake_with_ophyd_sim,
     )
-

src/dodal/common/beamlines/beamline_utils.py

@@ -1,5 +1,4 @@
 import inspect
-from functools import wraps
 from typing import (
     Callable,
     Dict,
@@ -89,9 +88,11 @@ def __call__(
 
 _device_is_lazy: Dict[DeviceFactory, bool] = {}
 
+
 def get_device_factories() -> Dict[DeviceFactory, bool]:
     return _device_is_lazy.copy()
 
+
 @skip_device()
 def device_instantiation(
     device_factory: Callable[..., T],
@@ -162,63 +163,3 @@ def get_directory_provider() -> UpdatingDirectoryProvider:
             "DirectoryProvider has not been set! Ophyd-async StandardDetectors will not be able to write!"
         )
     return DIRECTORY_PROVIDER
-
-
-@skip_device()
-def decorator_name(
-    name: str,
-    prefix: str,
-    not_lazy: bool,
-    fake: bool,
-    post_create: Optional[Callable[[T], None]] = None,
-    timeout: float = DEFAULT_CONNECTION_TIMEOUT,
-    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.
-
-    Arguments:
-        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, if
-                                    false the complete PV prefix must be supplied.
-    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 not_lazy:
-            call_in_bluesky_event_loop(device_instance.connect(timeout=timeout))
-
-    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
-
-