129 edge detection

pyproject.toml

@@ -14,14 +14,17 @@ classifiers = [
 ]
 description = "Ophyd devices and other utils that could be used across DLS beamlines"
 dependencies = [
-    "ophyd @ git+https://github.com/bluesky/ophyd@58d6fadb632fa2e49273e9097c7deb57f6b34c68",
+    "ophyd@git+https://github.com/Tom-Willemsen/ophyd@40e4b72c582e65e63d13a1650f76de709e5c70bb",  # Switch back to just "ophyd" once <relevant PR> merged.
     "bluesky",
     "pyepics",
     "dataclasses-json",
     "pillow",
     "requests",
     "graypy",
     "pydantic<2.0",
+    "opencv-python-headless",  # For pin-tip detection.
+    "aioca",  # Required for CA support with Ophyd V2.
+    "p4p",  # Required for PVA support with Ophyd V2.
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -38,6 +41,7 @@ dev = [
     "pipdeptree",
     "pre-commit",
     "pydata-sphinx-theme>=0.12",
+    "pytest-asyncio",
     "pytest-cov",
     "pytest-random-order",
     "sphinx-autobuild",

src/dodal/beamlines/beamline_utils.py

@@ -1,11 +1,17 @@
-from typing import Callable, Dict, List, Optional
+import asyncio
+from typing import Callable, Dict, Final, List, Optional, TypeVar, cast
 
-from ophyd import Device
+from bluesky.run_engine import call_in_bluesky_event_loop
+from ophyd import Device as OphydV1Device
 from ophyd.sim import make_fake_device
+from ophyd.v2.core import Device as OphydV2Device
+from ophyd.v2.core import wait_for_connection as v2_device_wait_for_connection
 
-from dodal.utils import BeamlinePrefix, skip_device
+from dodal.utils import AnyDevice, BeamlinePrefix, skip_device
 
-ACTIVE_DEVICES: Dict[str, Device] = {}
+DEFAULT_CONNECTION_TIMEOUT: Final[float] = 5.0
+
+ACTIVE_DEVICES: Dict[str, AnyDevice] = {}
 BL = ""
 
 
@@ -30,21 +36,44 @@ def list_active_devices() -> List[str]:
     return list(ACTIVE_DEVICES.keys())
 
 
-def active_device_is_same_type(active_device, device):
+def active_device_is_same_type(
+    active_device: AnyDevice, device: Callable[..., AnyDevice]
+) -> bool:
     return type(active_device) == device
 
 
+def _wait_for_connection(
+    device: AnyDevice, timeout: float = DEFAULT_CONNECTION_TIMEOUT
+) -> None:
+    if isinstance(device, OphydV1Device):
+        device.wait_for_connection(timeout=timeout)
+    elif isinstance(device, OphydV2Device):
+        call_in_bluesky_event_loop(
+            asyncio.wait_for(
+                v2_device_wait_for_connection(coros=device.connect()),
+                timeout=timeout,
+            )
+        )
+    else:
+        raise TypeError(
+            "Invalid type {} in _wait_for_connection".format(device.__class__.__name__)
+        )
+
+
+T = TypeVar("T", bound=AnyDevice)
+
+
 @skip_device()
 def device_instantiation(
-    device: Callable,
+    device: Callable[..., T],
     name: str,
     prefix: str,
     wait: bool,
     fake: bool,
-    post_create: Optional[Callable] = None,
+    post_create: Optional[Callable[[T], None]] = None,
     bl_prefix: bool = True,
     **kwargs,
-) -> Device:
+) -> 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.
@@ -62,26 +91,32 @@ def device_instantiation(
     Returns:
         The instance of the device.
     """
-    active_device = ACTIVE_DEVICES.get(name)
+    device_instance: T
+    active_device: Optional[AnyDevice] = ACTIVE_DEVICES.get(name)
     if fake:
         device = make_fake_device(device)
     if active_device is None:
-        ACTIVE_DEVICES[name] = device(
+        device_instance = device(
             name=name,
             prefix=f"{(BeamlinePrefix(BL).beamline_prefix)}{prefix}"
             if bl_prefix
             else prefix,
             **kwargs,
         )
+        ACTIVE_DEVICES[name] = device_instance
         if wait:
-            ACTIVE_DEVICES[name].wait_for_connection()
+            _wait_for_connection(device_instance)
+
     else:
         if not active_device_is_same_type(active_device, device):
             raise TypeError(
                 f"Can't instantiate device of type {type(active_device)} with the same "
                 f"name as an existing device. Device name '{name}' already used for "
                 f"a(n) {device}."
             )
+        else:
+            # We have manually checked types
+            device_instance = cast(T, ACTIVE_DEVICES[name])
     if post_create:
-        post_create(ACTIVE_DEVICES[name])
-    return ACTIVE_DEVICES[name]
+        post_create(device_instance)
+    return device_instance

src/dodal/beamlines/i23.py

@@ -1,6 +1,9 @@
 from dodal.beamlines.beamline_utils import device_instantiation
 from dodal.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.devices.i23.gonio import Gonio
+from dodal.devices.oav.pin_tip_detection.oav_with_pin_tip_detection import (
+    PinTipDetection,
+)
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import get_beamline_name
 
@@ -18,3 +21,16 @@ def gonio(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -
         wait_for_connection,
         fake_with_ophyd_sim,
     )
+
+
+def oav_pin_tip_detection(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> PinTipDetection:
+    """Get the i23 OAV pin-tip detection device"""
+    return device_instantiation(
+        PinTipDetection,
+        "PinTipDetection",
+        "-DI-OAV-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )

src/dodal/devices/oav/pin_tip_detection/oav_with_pin_tip_detection.py

@@ -0,0 +1,208 @@
+import asyncio
+import time
+from collections import OrderedDict
+from typing import Callable, Optional, Tuple, Type, TypeVar
+
+import numpy as np
+from bluesky.protocols import Descriptor
+from numpy.typing import NDArray
+from ophyd.v2.core import Device, Readable, Reading, SimConverter, SimSignalBackend
+from ophyd.v2.epics import SignalR, SignalRW, epics_signal_r
+
+from dodal.devices.oav.pin_tip_detection.pin_tip_detect_utils import (
+    ArrayProcessingFunctions,
+    MxSampleDetect,
+)
+from dodal.log import LOGGER
+
+T = TypeVar("T")
+
+
+def _create_soft_signal(
+    name: str, datatype: T | Type[T], initial_value: T
+) -> SignalRW[T]:
+    class _Converter(SimConverter):
+        def make_initial_value(self, *args, **kwargs) -> T:
+            return initial_value
+
+    class _SimSignalBackend(SimSignalBackend):
+        async def connect(self) -> None:
+            self.converter = _Converter()
+            self._initial_value = self.converter.make_initial_value()
+            self._severity = 0
+
+            await self.put(None)
+
+    backend = _SimSignalBackend(
+        datatype, f"sim://oav_with_pin_tip_detection_soft_params:{name}"
+    )
+    signal: SignalRW[T] = SignalRW(backend)
+    return signal
+
+
+class PinTipDetection(Readable, Device):
+    def __init__(self, prefix: str, name: str = ""):
+        self._prefix: str = prefix
+        self._name = name
+
+        self.array_data: SignalR[NDArray[np.uint8]] = epics_signal_r(
+            NDArray[np.uint8], f"pva://{prefix}PVA:ARRAY"
+        )
+
+        self.oav_width: SignalR[int] = epics_signal_r(
+            int, f"{prefix}PVA:ArraySize1_RBV"
+        )
+        self.oav_height: SignalR[int] = epics_signal_r(
+            int, f"{prefix}PVA:ArraySize2_RBV"
+        )
+
+        # Soft parameters for pin-tip detection.
+        self.timeout: SignalRW[float] = _create_soft_signal("timeout", float, 10.0)
+        self.preprocess: SignalRW[
+            Callable[[np.ndarray], np.ndarray]
+        ] = _create_soft_signal(
+            "preprocess",
+            Callable[[np.ndarray], np.ndarray],
+            ArrayProcessingFunctions.identity(),
+        )
+        self.canny_upper: SignalRW[int] = _create_soft_signal("canny_upper", int, 100)
+        self.canny_lower: SignalRW[int] = _create_soft_signal("canny_lower", int, 50)
+        self.close_ksize: SignalRW[int] = _create_soft_signal("close_ksize", int, 5)
+        self.close_iterations: SignalRW[int] = _create_soft_signal(
+            "close_iterations", int, 5
+        )
+        self.scan_direction: SignalRW[int] = _create_soft_signal(
+            "scan_direction", int, 1
+        )
+        self.min_tip_height: SignalRW[int] = _create_soft_signal(
+            "min_tip_height", int, 5
+        )
+
+        super().__init__(name=name)
+
+    async def connect(self, sim: bool = False):
+        for signal in [self.array_data, self.oav_width, self.oav_height]:
+            await signal.connect(sim=sim)
+
+        for soft_signal in [
+            self.timeout,
+            self.preprocess,
+            self.canny_upper,
+            self.canny_lower,
+            self.close_ksize,
+            self.close_iterations,
+            self.scan_direction,
+            self.min_tip_height,
+        ]:
+            await soft_signal.connect(sim=False)  # Soft signals don't need simulating.
+
+    async def _get_tip_position(
+        self,
+    ) -> Tuple[Tuple[Optional[int], Optional[int]], float]:
+        """
+        Gets the location of the pin tip.
+
+        Returns tuple of:
+            ((tip_x, tip_y), timestamp)
+        """
+        sample_detection = MxSampleDetect(
+            preprocess=await self.preprocess.get_value(),
+            canny_lower=await self.canny_lower.get_value(),
+            canny_upper=await self.canny_upper.get_value(),
+            close_ksize=await self.close_ksize.get_value(),
+            close_iterations=await self.close_iterations.get_value(),
+            scan_direction=await self.scan_direction.get_value(),
+            min_tip_height=await self.min_tip_height.get_value(),
+        )
+
+        array_reading: dict[str, Reading] = await self.array_data.read()
+        array_data: NDArray[np.uint8] = array_reading[""]["value"]
+        timestamp: float = array_reading[""]["timestamp"]
+
+        height: int = await self.oav_height.get_value()
+        width: int = await self.oav_width.get_value()
+
+        num_pixels: int = height * width  # type: ignore
+        value_len = array_data.shape[0]
+
+        try:
+            if value_len == num_pixels * 3:
+                # RGB data
+                value = array_data.reshape(height, width, 3)
+            elif value_len == num_pixels:
+                # Grayscale data
+                value = array_data.reshape(height, width)
+            else:
+                # Something else?
+                raise ValueError(
+                    "Unexpected data array size: expected {} (grayscale data) or {} (rgb data), got {}",
+                    num_pixels,
+                    num_pixels * 3,
+                    value_len,
+                )
+
+            start_time = time.time()
+            location = sample_detection.processArray(value)
+            end_time = time.time()
+            LOGGER.debug(
+                "Sample location detection took {}ms".format(
+                    (end_time - start_time) * 1000.0
+                )
+            )
+            tip_x = location.tip_x
+            tip_y = location.tip_y
+        except Exception as e:
+            LOGGER.error("Failed to detect pin-tip location due to exception: ", e)
+            tip_x = None
+            tip_y = None
+
+        return (tip_x, tip_y), timestamp
+
+    async def read(self) -> dict[str, Reading]:
+        tip_pos, timestamp = await asyncio.wait_for(
+            self._get_tip_position(), timeout=await self.timeout.get_value()
+        )
+
+        return OrderedDict(
+            [
+                (self._name, {"value": tip_pos, "timestamp": timestamp}),
+            ]
+        )
+
+    async def describe(self) -> dict[str, Descriptor]:
+        return OrderedDict(
+            [
+                (
+                    self._name,
+                    {
+                        "source": "pva://{}PVA:ARRAY".format(self._prefix),
+                        "dtype": "number",
+                        "shape": [2],  # Tuple of (x, y) tip position
+                    },
+                )
+            ],
+        )
+
+
+if __name__ == "__main__":
+    x = PinTipDetection(prefix="BL03I-DI-OAV-01:", name="edgeDetect")
+
+    async def acquire():
+        await x.connect()
+        img = await x.array_data.read()
+        tip = await x.read()
+        return img, tip
+
+    img, tip = asyncio.get_event_loop().run_until_complete(
+        asyncio.wait_for(acquire(), timeout=10)
+    )
+    print(tip)
+    print("Tip: {}".format(tip["edgeDetect"]["value"]))
+
+    try:
+        import matplotlib.pyplot as plt
+
+        plt.imshow(img[""]["value"].reshape(768, 1024, 3))
+        plt.show()
+    except ImportError:
+        print("matplotlib not available; cannot show acquired image")