129 edge detection

src/dodal/beamlines/i23.py

@@ -1,10 +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_image_recognition import PinTipDetection
-from dodal.log import LOGGER
 from dodal.log import set_beamline as set_log_beamline
-from dodal.utils import get_beamline_name
+from dodal.utils import get_beamline_name, get_hostname, skip_device
 
 BL = get_beamline_name("i23")
 set_log_beamline(BL)
@@ -22,18 +21,13 @@
     )
 
 
+@skip_device(lambda: not get_hostname().startswith("i23-ws"))
 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"""
 
-    if get_beamline_name("") != "i23":
-        LOGGER.warning(
-            "Not running on i23 - forcing pin tip detection into simulation mode"
-        )
-        fake_with_ophyd_sim = True
-
     return device_instantiation(
         PinTipDetection,
         "pin_tip_detection",
         "-DI-OAV-01:",

src/dodal/devices/oav/pin_image_recognition/__init__.py

@@ -35,8 +35,9 @@
 
     Used for pin tip centring workflow.
 
-    Note that if the sample is off-screen, this class will return the centre as the "edge"
-    of the image.
+    Note that if the tip of the sample is off-screen, this class will return the centre as the "edge"
+    of the image. If the entire sample if off-screen (i.e. no suitable edges were detected at all)
+    then it will return (None, None).
     """
 
     def __init__(self, prefix: str, name: str = ""):
@@ -176,7 +177,7 @@
         )
 
     async def describe(self) -> dict[str, Descriptor]:
         return OrderedDict(
             [
                 (
                     self._name,

src/dodal/devices/oav/pin_image_recognition/manual_test.py

@@ -5,29 +5,30 @@
 
 It is otherwise unused.
 """
+# pragma: no cover
 import asyncio
 
 from dodal.devices.oav.pin_image_recognition import PinTipDetection
 
 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")

tests/devices/unit_tests/oav/image_recognition/test_pin_tip_detect.py

@@ -1,5 +1,5 @@
 import asyncio
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import numpy as np
 import pytest
@@ -54,6 +54,9 @@ async def test_numeric_soft_parameters_can_be_changed():
     await device.close_iterations.set(20)
     await device.min_tip_height.set(25)
     await device.scan_direction.set(-1)
+    await device.preprocess.set(2)
+    await device.preprocess_ksize.set(3)
+    await device.preprocess_iterations.set(4)
 
     assert await device.timeout.get_value() == 100.0
     assert await device.canny_lower.get_value() == 5
@@ -62,6 +65,31 @@ async def test_numeric_soft_parameters_can_be_changed():
     assert await device.close_iterations.get_value() == 20
     assert await device.min_tip_height.get_value() == 25
     assert await device.scan_direction.get_value() == -1
+    assert await device.preprocess.get_value() == 2
+    assert await device.preprocess_ksize.get_value() == 3
+    assert await device.preprocess_iterations.get_value() == 4
+
+
+@pytest.mark.asyncio
+async def test_invalid_processing_func_uses_identity_function():
+    device = await _get_pin_tip_detection_device()
+
+    set_sim_value(device.preprocess, 50)  # Invalid index
+
+    with patch.object(
+        MxSampleDetect, "__init__", return_value=None
+    ) as mock_init, patch.object(
+        MxSampleDetect, "processArray", return_value=((None, None), None)
+    ):
+        await device.read()
+
+        mock_init.assert_called_once()
+
+        captured_func = mock_init.call_args[1]["preprocess"]
+
+    # Assert captured preprocess function is the identitiy function
+    arg = object()
+    assert arg == captured_func(arg)
 
 
 @pytest.mark.parametrize(