Create a device that will give the max pixel from an AD camera (#1723)

* first try

* add trigger function which writes to soft signals

* small changes

* get rid of max pixel location

* add tests - need modifying as made with copilot

* make triggerable for blesky plan

* start to add max-pixel device in i04 (WIP because on beamline)

* Finish adding device to i04.py

* Add changes from beamline testing - from github

* add comment for kernal size

* attemt own tests

* add my tests

* add assert called onece tests

* make changes from Dom's comments

* add greyscale test

* remove print statements in test

* Fix merge issue

---------

Co-authored-by: Dominic Oram <[email protected]>

Author: srishtysajeev

src/dodal/beamlines/i04.py

@@ -23,6 +23,7 @@
 from dodal.devices.i03.dcm import DCM
 from dodal.devices.i04.beamsize import Beamsize
 from dodal.devices.i04.constants import RedisConstants
+from dodal.devices.i04.max_pixel import MaxPixel
 from dodal.devices.i04.murko_results import MurkoResultsDevice
 from dodal.devices.i04.transfocator import Transfocator
 from dodal.devices.ipin import IPin
@@ -390,6 +391,16 @@ def scintillator() -> Scintillator:
     )
 
 
+@device_factory()
+def max_pixel() -> MaxPixel:
+    """Get the i04 max pixel device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in i04, it will return the existing object.
+    """
+    return MaxPixel(
+        f"{PREFIX.beamline_prefix}-DI-OAV-01:",
+    )
+
+
 @device_factory()
 def beamsize() -> Beamsize:
     """Get the i04 beamsize device, instantiate it if it hasn't already been.

src/dodal/devices/i04/max_pixel.py

@@ -0,0 +1,38 @@
+import cv2
+import numpy as np
+from bluesky.protocols import Triggerable
+from ophyd_async.core import AsyncStatus, StandardReadable, soft_signal_r_and_setter
+from ophyd_async.epics.core import (
+    epics_signal_r,
+)
+
+# kernal size describes how many of the neigbouring pixels are used for the blur,
+# higher kernal size means more of a blur effect
+KERNAL_SIZE = (7, 7)
+
+
+class MaxPixel(StandardReadable, Triggerable):
+    """Gets the max pixel (brightest pixel) from the image after some image processing."""
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        self.array_data = epics_signal_r(np.ndarray, f"pva://{prefix}PVA:ARRAY")
+        self.max_pixel_val, self._max_val_setter = soft_signal_r_and_setter(float)
+        super().__init__(name)
+
+    async def _convert_to_gray_and_blur(self):
+        """
+        Preprocess the image array data (convert to grayscale and apply a gaussian blur)
+        Image is converted to grayscale (using a weighted mean as green contributes more to brightness)
+        as we aren't interested in data relating to colour. A blur is then applied to mitigate
+        errors due to rogue hot pixels.
+        """
+        data = await self.array_data.get_value()
+        gray_arr = cv2.cvtColor(data, cv2.COLOR_BGR2GRAY)
+        return cv2.GaussianBlur(gray_arr, KERNAL_SIZE, 0)
+
+    @AsyncStatus.wrap
+    async def trigger(self):
+        arr = await self._convert_to_gray_and_blur()
+        max_val = float(np.max(arr))  # np.int64
+        assert isinstance(max_val, float)
+        self._max_val_setter(max_val)

tests/devices/i04/test_max_pixel.py

@@ -0,0 +1,71 @@
+from collections.abc import AsyncGenerator
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
+
+import cv2
+import numpy as np
+import pytest
+from ophyd_async.core import init_devices, set_mock_value
+
+from dodal.devices.i04.max_pixel import KERNAL_SIZE, MaxPixel
+
+
+@pytest.fixture
+async def max_pixel() -> AsyncGenerator[MaxPixel]:
+    async with init_devices(mock=True):
+        max_pixel = MaxPixel("TEST: MAX_PIXEL")
+    yield max_pixel
+
+
+@pytest.mark.parametrize(
+    "preprocessed_data, expected",
+    [
+        ([1, 2, 3], 3),  # test can handle standard input
+        ([[0, 0, 0], [0, 0, 0], [0, 0, 0]], 0),  # test can handle all 0's
+        ([-9, -8, 0], 0),  # test can handle negatives
+        ([6.9, 8.9, 7.5, 6.45], 8.9),  # check can handle floats
+    ],
+)
+@patch("dodal.devices.i04.max_pixel.MaxPixel._convert_to_gray_and_blur")
+async def test_returns_max(
+    mocked_preprocessed_data: AsyncMock,
+    preprocessed_data,
+    expected,
+    max_pixel: MaxPixel,
+):
+    mocked_preprocessed_data.return_value = preprocessed_data
+    await max_pixel.trigger()
+    assert await max_pixel.max_pixel_val.get_value() == expected
+
+
+@patch("dodal.devices.i04.max_pixel.cv2.cvtColor")
+@patch("dodal.devices.i04.max_pixel.cv2.GaussianBlur")
+async def test_preprocessed_data_grayscale_is_called(
+    mocked_cv2_blur: MagicMock, mocked_cv2_grey: MagicMock, max_pixel: MaxPixel
+):
+    data = np.array([1])
+    set_mock_value(max_pixel.array_data, data)
+    await max_pixel._convert_to_gray_and_blur()
+    mocked_cv2_grey.assert_called_once_with(data, cv2.COLOR_BGR2GRAY)
+    mocked_cv2_blur.assert_called_once_with(ANY, KERNAL_SIZE, 0)
+
+
+test_arr = np.array(
+    [
+        [[123, 65, 0], [0, 0, 0], [0, 0, 0]],
+        [[123, 65, 0], [0, 0, 0], [0, 0, 0]],
+        [[123, 65, 0], [0, 0, 0], [0, 0, 0]],
+        [[123, 65, 0], [0, 0, 0], [0, 0, 0]],
+    ],
+    dtype=np.uint8,
+)
+
+
+async def test_greyscale_works(max_pixel: MaxPixel):
+    test_arr_shape = test_arr.shape  # (4, 3, 3)
+    set_mock_value(max_pixel.array_data, test_arr)
+    await max_pixel.array_data.get_value()
+    processed_data = await max_pixel._convert_to_gray_and_blur()
+    processed_data_shape = processed_data.shape  # (4,3)
+
+    assert processed_data_shape[0] == test_arr_shape[0]
+    assert processed_data_shape[1] == test_arr_shape[1]