Continue to rebase background detect.

src/openflexure_microscope_server/background_detect.py

@@ -1,7 +1,104 @@
+"""Provide functionality to detect if the camera is imaging sample or background.
 
+An example background image is captured by the camera and sent to classes in the module
+for analysis. Information from this images is used to detect whether an image from the
+current camera field of view contains sample.
+"""
+
+from typing import Optional
 import cv2
 import numpy as np
 from pydantic import BaseModel
+from pydantic.errors import PydanticUserError
+from scipy.stats import norm
+
+
+class BackgroundDetectAlgorithm:
+    """The base class for defining background detect algorithms."""
+
+    background_data_model: BaseModel
+    """The data model of the background data. This must be set by child classes"""
+    settings_data_model: BaseModel
+    """The data model of algorithm settings. This must be set by child classes"""
+
+    def __init__(self):
+        """Initialise the algorithm settings."""
+        try:
+            _settings: BaseModel = self.settings_data_model()
+        except PydanticUserError as e:
+            raise NotImplementedError(
+                "BackgroundDetectAlgorithms must set their own settings data model."
+            ) from e
+
+    # Requires a getter and a setter to support being a BaseModel but being
+    # saved to file as a dict
+    _background_data: Optional[BaseModel] = None
+
+    @property
+    def background_data(self) -> Optional[BaseModel]:
+        """The statistics of the background image."""
+        bd = self._background_data
+        if bd is None:
+            return None
+        try:
+            return self.background_data_model(**bd)
+        except PydanticUserError as e:
+            raise NotImplementedError(
+                "BackgroundDetectAlgorithms must set their own background data model."
+            ) from e
+
+    @background_data.setter
+    def background_data(self, value: Optional[BaseModel | dict]) -> None:
+        if value is None:
+            self._background_data = None
+        elif isinstance(value, self.background_data_model):
+            self._background_data = value.model_dump()
+        elif isinstance(value, dict):
+            self._background_data = value
+        else:
+            raise TypeError(
+                f"Cannot set background_data with an object of type {type(value)}"
+            )
+
+    @property
+    def settings(self) -> BaseModel:
+        """The statistics of the background image."""
+        bd = self._background_data
+        if bd is None:
+            return None
+        return self.settings_data_model(**bd)
+
+    @settings.setter
+    def settings(self, value: Optional[BaseModel | dict]) -> None:
+        if value is None:
+            self._settings = None
+        elif isinstance(value, self.settings_data_model):
+            self._settings = value.model_dump()
+        elif isinstance(value, dict):
+            self._settings = value
+        else:
+            raise TypeError(f"Cannot set settings with an object of type {type(value)}")
+
+    def image_is_sample(self, image: np.ndarrayl) -> tuple[bool, str]:
+        """Label the current image as either background or sample.
+
+        :returns: A tuple of the result (boolean), and explanation string. The
+            explanation string is formatted so it can be added into a sentence such as
+            ``An action was taken because the image is {message}.``
+        """
+        raise NotImplementedError(
+            "Each background detect algorithm must implement an image_is_sample method."
+        )
+
+    def set_background(self, image: np.ndarray) -> BaseModel:
+        """Use the input image to update the background data.
+
+        Background data must be a Pydantic BaseModel.
+        """
+        raise NotImplementedError(
+            "Each background detect algorithm must implement an set_background method."
+        )
+
 
 class ChannelDistributions(BaseModel):
     """A BaseModel for storing the channel distribution of a background image."""
@@ -10,19 +107,34 @@ class ChannelDistributions(BaseModel):
     """The mean of each channel in the colourspace."""
     standard_deviations: list[float]
     """The standard deviation of each channel in the colourspace."""
-    colorspace: str = "LUV"
-    """The colourspace used."""
 
 
-class BackgroundDetectLUV:
+class ColourChannelDetectSettings(BaseModel):
+    """A BaseModel for storing the settings for colour channel detectors."""
+
+    channel_tolerance: float = 7.0
+    """Channel Tolerance
+
+    The number of standard deviations a pixel value must be from the background mean
+    to be considered sample.
+    """
+    min_sample_coverage: float = 25.0
+    """Sample Coverage Required (%)
+
+    The minimum percentage of the image that needs to be identified as sample for the
+    image to be labeled as containing sample.
+    """
+
+
+class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
     """Compare images with a known background in LUV colourspace.
 
     This uses an LUV colour space checking only the mean and standard deviation of the
     U and V channels. The LUV colourspace as it collect colours together in a
     """
-    background_distributions: Optional[ChannelDistributions] = None
+
-    background_tolerance: float = 7.0
+    background_data_model: BaseModel = ChannelDistributions
-    min_sample_coverage: float = 25.0
+    settings_data_model: BaseModel = ColourChannelDetectSettings
 
     def background_mask(self, image: np.ndarray) -> np.ndarray:
         """Calculate a binary image, showing whether each pixel is background.
@@ -32,7 +144,7 @@ class BackgroundDetectLUV:
 
         # human-intuitive way
         """
-        d = self.background_distributions
+        d = self.background_data
         if not d:
             raise RuntimeError(
                 "Background is not set: you need to calibrate background detection."
@@ -43,34 +155,40 @@ class BackgroundDetectLUV:
         return np.all(
             np.abs(image[:, :, 1:] - np.array(d.means[1:])[np.newaxis, np.newaxis, :])
             < np.array(d.standard_deviations[1:])[np.newaxis, np.newaxis, :]
-            * self.background_tolerance,
+            * self.channel_tolerance,
             axis=2,
         )
 
     def get_sample_coverage(self, image: np.ndarray) -> float:
-        """Measure what percentage of the current image is background.
+        """Return the percentage of the input image that is background.
 
-        * Acquire a new image from the preview stream,
+        Evaluate whether it is foreground or background by comparing it to the saved
-        * Evaluate whether it is foreground or background, by comparing it to the saved
+        statistics for a background image on a per-pixel basis
-            statistics for a background image on a per-pixel basis
+
-        * Returned value (between 0 and 100) is the percentage of the image that is
+        :returns: A value (between 0 and 100) is the percentage of the image that is
-            background.
+            sample.
         """
         image_luv = cv2.cvtColor(image, cv2.COLOR_RGB2LUV)
         mask = self.background_mask(image_luv)
         return (1 - np.count_nonzero(mask) / np.prod(mask.shape)) * 100
 
+    def image_is_sample(self, image: np.ndarrayl) -> tuple[bool, str]:
+        """Label the current image as either background or sample.
 
-    def image_is_sample(self, image: np.ndarrayl) -> bool:
+        :returns: A tuple of the result (boolean), and explanation string. The
-        """Label the current image as either background or sample."""
+            explanation string is formatted so it can be added into a sentence such as
+            ``An action was taken because the image is {message}.``
+        """
         sample_coverage = self.get_sample_coverage(image)
-        fraction_threshold = self.fraction
 
-        return sample_coverage > self.min_sample_coverage
+        is_sample = sample_coverage > self.min_sample_coverage
-
+        message = f"{sample_coverage}% sample"
-
+        if not is_sample:
-    def set_background(self, image: np.ndarray) -> np.ndarray:
+            message = "only " + message
+        return is_sample, message
 
+    def set_background(self, image: np.ndarray) -> ChannelDistributions:
+        """Use the input image to update the background distributions."""
         image_luv = cv2.cvtColor(image, cv2.COLOR_RGB2LUV)
 
         ch1 = (image_luv.T[0]).flatten()
@@ -82,8 +200,6 @@ class BackgroundDetectLUV:
         # Get the mean and standard deviation of values in each channel
         mu, std = np.apply_along_axis(norm.fit, 0, points)
 
-        self.background_distributions = ChannelDistributions(
+        self.background_data = ChannelDistributions(
-            means=mu.tolist(),
+            means=mu.tolist(), standard_deviations=std.tolist()
-            standard_deviations=std.tolist(),
-            colorspace="LUV",
         )

src/openflexure_microscope_server/things/background_detect.py

@@ -1,163 +0,0 @@
-"""Provide functionality to detect if the camera is imaging sample or background.
-
-An example background image must be captured and analysed by BackgroundDetectThing,
-information from this images is used to detect whether the current camera field of
-view contains sample.
-"""
-
-from typing import Mapping, Optional
-import cv2
-import numpy as np
-from PIL import Image
-from pydantic import BaseModel
-from scipy.stats import norm
-
-import labthings_fastapi as lt
-from .camera import CameraDependency as CamDep
-
-
-class ChannelDistributions(BaseModel):
-    """A BaseModel for storing the channel distribution of a background image."""
-
-    means: list[float]
-    """The mean of each channel in the colourspace."""
-    standard_deviations: list[float]
-    """The standard deviation of each channel in the colourspace."""
-    colorspace: str = "LUV"
-    """The colourspace used."""
-
-
-class BackgroundDetectThing(lt.Thing):
-    """Thing for setting a background image and detecting sample in the field of view.
-
-    This uses an LUV colour space checking only the mean and standard deviation of the
-    UV channels. Over time different, selectable, background detection methods will be
-    added.
-    """
-
-    # Requires a getter and a setter to support being a BaseModel but being
-    # saved to file as a dict
-    _background_distributions: Optional[ChannelDistributions] = None
-
-    @lt.thing_setting
-    def background_distributions(self) -> Optional[ChannelDistributions]:
-        """The statistics of the background image."""
-        bd = self._background_distributions
-        if bd is None:
-            return None
-        return ChannelDistributions(**bd)
-
-    @background_distributions.setter
-    def background_distributions(
-        self, value: Optional[ChannelDistributions | dict]
-    ) -> None:
-        if value is None:
-            self._background_distributions = None
-        elif isinstance(value, ChannelDistributions):
-            self._background_distributions = value.model_dump()
-        elif isinstance(value, dict):
-            self._background_distributions = value
-        else:
-            raise TypeError(
-                f"Cannot set background_distributions with an object of type {type(value)}"
-            )
-
-    tolerance = lt.ThingSetting(
-        initial_value=7.0,
-        model=float,
-    )
-    """How many standard deviations to allow for the background."""
-
-    fraction = lt.ThingSetting(
-        initial_value=25.0,
-        model=float,
-    )
-    """How much of the image needs to be not background to label as sample"""
-
-    def background_mask(self, image: np.ndarray) -> np.ndarray:
-        """Calculate a binary image, showing whether each pixel is background.
-
-        The image should be in LUV format, the output will be binary with the
-        same shape in the first two dimensions.
-        """
-        d = self.background_distributions
-        if not d:
-            raise RuntimeError(
-                "Background is not set: you need to calibrate background detection."
-            )
-        # This image is in LUV space. But the brightness (L) often changes as the
-        # height of the sample changes. Hence in the line below we are only using
-        # the UV (colour) channels.
-        return np.all(
-            np.abs(image[:, :, 1:] - np.array(d.means[1:])[np.newaxis, np.newaxis, :])
-            < np.array(d.standard_deviations[1:])[np.newaxis, np.newaxis, :]
-            * self.tolerance,
-            axis=2,
-        )
-
-    @lt.thing_action
-    def background_fraction(self, cam: CamDep) -> float:
-        """Determine what fraction of the current image is background.
-
-        This action will acquire a new image from the preview stream, then
-        evaluate whether it is foreground or background, by comparing it
-        too the saved statistics. This is done on a per-pixel basis, and
-        the returned value (between 0 and 100) is the fraction of the image
-        that is background.
-        """
-        current_image = cam.grab_jpeg()
-        current_image = np.array(Image.open(current_image.open()))
-
-        # we're working in the LUV colourspace as it collect colours together in a human-intuitive way
-        current_image_luv = cv2.cvtColor(current_image, cv2.COLOR_RGB2LUV)
-        mask = self.background_mask(current_image_luv)
-        return np.count_nonzero(mask) / np.prod(mask.shape) * 100
-
-    @lt.thing_action
-    def image_is_sample(self, cam: CamDep) -> bool:
-        """Label the current image as either background or sample."""
-        b_fraction = self.background_fraction(cam)
-        fraction_threshold = self.fraction
-
-        return (100 - b_fraction) > fraction_threshold
-
-    @lt.thing_action
-    def set_background(self, cam: CamDep):
-        """Grab an image, and use its statistics to set the background.
-
-        This should be run when the microscope is looking at an empty region,
-        and will calculate the mean and standard deviation of the pixel values
-        in the LUV colourspace. These values will then be used to compare
-        future images to the distribution, to determine if each pixel is
-        foreground or background.
-        """
-        background = cam.grab_jpeg()
-        background = np.array(Image.open(background.open()))
-
-        # we're working in the LUV colourspace as it collect colours together in a human-intuitive way
-        background_luv = cv2.cvtColor(background, cv2.COLOR_RGB2LUV)
-
-        ch1 = (background_luv.T[0]).flatten()
-        ch2 = (background_luv.T[1]).flatten()
-        ch3 = (background_luv.T[2]).flatten()
-
-        points = np.array([np.asarray(ch1), np.asarray(ch2), np.asarray(ch3)]).T
-
-        # we get the mean and standard deviation of values in each channel
-        mu, std = np.apply_along_axis(norm.fit, 0, points)
-
-        self.background_distributions = ChannelDistributions(
-            means=mu.tolist(),
-            standard_deviations=std.tolist(),
-            colorspace="LUV",
-        )
-
-    @property
-    def thing_state(self) -> Mapping:
-        """Summary metadata describing the current state of the Thing."""
-        bd = self.background_distributions
-        return {
-            "background_distributions": bd.model_dump() if bd else None,
-            "tolerance": self.tolerance,
-            "fraction": self.fraction,
-        }

src/openflexure_microscope_server/things/camera/__init__.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 from typing import Literal, Optional, Tuple, Any
 import json
 import time
+import logging
 
 import numpy as np
 from pydantic import RootModel
@@ -19,7 +20,10 @@ import piexif
 import labthings_fastapi as lt
 from labthings_fastapi.types.numpy import NDArray
 
-from openflexure_microscope_server.background_detect import ChannelDistributions, BackgroundDetectLUV
+from openflexure_microscope_server.background_detect import BackgroundDetectLUV
+
+LOGGER = logging.getLogger(__name__)
+
 
 class JPEGBlob(lt.blob.Blob):
     """A class representing a JPEG image as a LabThings FastAPI Blob."""
@@ -157,8 +161,13 @@ class BaseCamera(lt.Thing):
     _memory_buffer = CameraMemoryBuffer()
 
     def __init__(self):
+        """Initialise the base camera, this creates the background detectors.
+
+        This must be run by all child camera classes.
+        """
         super().__init__()
-        self.background_detector = BackgroundDetectLUV()
+        self.background_detectors = {"Colour Channels (LUV)": BackgroundDetectLUV()}
+        self._detector_name = "Colour Channels (LUV)"
 
     def __enter__(self) -> None:
         """Open hardware connection when the Thing context manager is opened."""
@@ -460,51 +469,54 @@ class BaseCamera(lt.Thing):
         time.sleep(self.settling_time)
         self.discard_frames()
 
-    background_tolerance = lt.ThingSetting(
+    # Note that the default detector name is set at init. This is over written if
-        initial_value=7.0,
+    # setting is loaded from disk.
-        model=float,
+    @lt.thing_setting
-    )
+    def detector_name(self) -> str:
-    """How many standard deviations to allow for the background."""
+        """The name of the active background selector."""
+        return self._detector_name
 
-    min_sample_coverage = lt.ThingSetting(
+    @detector_name.setter
-        initial_value=25.0,
+    def detector_name(self, name: str) -> None:
-        model=float,
+        """Validate and set detector_name."""
-    )
+        if name not in self.background_detectors:
-    """The percentage of the image that needs to be not be background to label as sample."""
+            raise ValueError(f"{name} is not a valid background detector name")
+        self._detector_name = name
 
-    # Requires a getter and a setter to support being a BaseModel but being
+    @property
-    # saved to file as a dict
+    def active_detector(self):
-    _background_distributions: Optional[ChannelDistributions] = None
+        """The active background detector instance."""
+        return self.background_detectors[self.detector_name]
 
     @lt.thing_setting
-    def background_distributions(self) -> Optional[ChannelDistributions]:
+    def background_detector_data(self) -> str:
-        """The statistics of the background image."""
+        """The name of the active background selector."""
-        bd = self._background_distributions
+        data = {}
-        if bd is None:
+        for name, obj in self.background_detectors.items():
-            return None
+            data[name] = {
-        return ChannelDistributions(**bd)
+                "settings": obj.settings.model_dump(),
+                "background_data": obj.background_data.model_dump(),
+            }
 
-    @background_distributions.setter
+    @detector_name.setter
-    def background_distributions(
+    def detector_name(self, data: str) -> None:
-        self, value: Optional[ChannelDistributions | dict]
+        """Validate and set detector_name."""
-    ) -> None:
+        for name, instance_data in data.items():
-        if value is None:
+            if name in self.background_detectors:
-            self._background_distributions = None
+                obj = self.background_detectors[name]
-        elif isinstance(value, ChannelDistributions):
+                obj.settings = instance_data["settings"]
-            self._background_distributions = value.model_dump()
+                obj.background_data = instance_data["background_data"]
-        elif isinstance(value, dict):
+            else:
-            self._background_distributions = value
+                LOGGER.warning(
-        else:
+                    f"No background detector named {name}, settings will be discarded."
-            raise TypeError(
+                )
-                f"Cannot set background_distributions with an object of type {type(value)}"
-            )
 
     @lt.thing_action
-    def image_is_sample(self, portal: lt.deps.BlockingPortal) -> bool:
+    def image_is_sample(self, portal: lt.deps.BlockingPortal) -> tuple[bool, str]:
         """Label the current image as either background or sample."""
         current_image = self.grab_jpeg(portal)
         current_image = np.array(Image.open(current_image.open()))
-        return self.background_detector.image_is_sample(current_image)
+        return self.active_detector.image_is_sample(current_image)
 
     @lt.thing_action
     def set_background(self, portal: lt.deps.BlockingPortal):
@@ -518,8 +530,7 @@ class BaseCamera(lt.Thing):
         """
         background = self.grab_jpeg(portal)
         background = np.array(Image.open(background.open()))
-        self.background_detector.set_background(current_image)
+        self.active_detector.set_background(background)
-
 
 
 CameraDependency = lt.deps.direct_thing_client_dependency(BaseCamera, "/camera/")

src/openflexure_microscope_server/things/smart_scan.py

@@ -511,15 +511,13 @@ class SmartScanThing(lt.Thing):
             capture_image = True
             # If skipping background, take an image to check if current field of view is background
             if self._scan_data["skip_background"]:
-                capture_image = self._background_detect.image_is_sample()
+                capture_image, bg_message = self._background_detect.image_is_sample()
 
             if not capture_image:
                 route_planner.mark_location_visited(
                     new_pos_xyz, imaged=False, focused=False
                 )
-                # Background fraction is actually a percentage
+                msg = f"Skipping {new_pos_xyz} as it is {bg_message}."
-                back_perc = round(self._background_detect.background_fraction(), 0)
-                msg = f"Skipping {new_pos_xyz} as it is {back_perc}% background."
                 self._scan_logger.info(msg)
                 continue