From 2245d9357d0d8456e4932ca0463c4e129c25bdb1 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Thu, 17 Jul 2025 18:00:11 +0100 Subject: [PATCH] Continue to rebase background detect. --- .../background_detect.py | 166 +++++++++++++++--- .../things/background_detect.py | 163 ----------------- .../things/camera/__init__.py | 89 ++++++---- .../things/smart_scan.py | 6 +- 4 files changed, 193 insertions(+), 231 deletions(-) delete mode 100644 src/openflexure_microscope_server/things/background_detect.py diff --git a/src/openflexure_microscope_server/background_detect.py b/src/openflexure_microscope_server/background_detect.py index b3818f1f..71a1d8a9 100644 --- a/src/openflexure_microscope_server/background_detect.py +++ b/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 - min_sample_coverage: float = 25.0 + + background_data_model: BaseModel = ChannelDistributions + 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 - statistics for a background image on a per-pixel basis - * Returned value (between 0 and 100) is the percentage of the image that is - background. + Evaluate whether it is foreground or background by comparing it to the saved + statistics for a background image on a per-pixel basis + + :returns: A value (between 0 and 100) is the percentage of the image that is + 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: - """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}.`` + """ sample_coverage = self.get_sample_coverage(image) - fraction_threshold = self.fraction - return sample_coverage > self.min_sample_coverage - - - def set_background(self, image: np.ndarray) -> np.ndarray: + is_sample = sample_coverage > self.min_sample_coverage + message = f"{sample_coverage}% sample" + if not is_sample: + 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( - means=mu.tolist(), - standard_deviations=std.tolist(), - colorspace="LUV", + self.background_data = ChannelDistributions( + means=mu.tolist(), standard_deviations=std.tolist() ) diff --git a/src/openflexure_microscope_server/things/background_detect.py b/src/openflexure_microscope_server/things/background_detect.py deleted file mode 100644 index 1c147093..00000000 --- a/src/openflexure_microscope_server/things/background_detect.py +++ /dev/null @@ -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, - } diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py index 734c01bd..7a26335c 100644 --- a/src/openflexure_microscope_server/things/camera/__init__.py +++ b/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( - initial_value=7.0, - model=float, - ) - """How many standard deviations to allow for the background.""" + # Note that the default detector name is set at init. This is over written if + # setting is loaded from disk. + @lt.thing_setting + def detector_name(self) -> str: + """The name of the active background selector.""" + return self._detector_name - min_sample_coverage = lt.ThingSetting( - initial_value=25.0, - model=float, - ) - """The percentage of the image that needs to be not be background to label as sample.""" + @detector_name.setter + def detector_name(self, name: str) -> None: + """Validate and set detector_name.""" + if name not in self.background_detectors: + 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 - # saved to file as a dict - _background_distributions: Optional[ChannelDistributions] = None + @property + def active_detector(self): + """The active background detector instance.""" + return self.background_detectors[self.detector_name] @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) + def background_detector_data(self) -> str: + """The name of the active background selector.""" + data = {} + for name, obj in self.background_detectors.items(): + data[name] = { + "settings": obj.settings.model_dump(), + "background_data": obj.background_data.model_dump(), + } - @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)}" - ) + @detector_name.setter + def detector_name(self, data: str) -> None: + """Validate and set detector_name.""" + for name, instance_data in data.items(): + if name in self.background_detectors: + obj = self.background_detectors[name] + obj.settings = instance_data["settings"] + obj.background_data = instance_data["background_data"] + else: + LOGGER.warning( + f"No background detector named {name}, settings will be discarded." + ) @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/") diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index f7e5d675..3db02813 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/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 - back_perc = round(self._background_detect.background_fraction(), 0) - msg = f"Skipping {new_pos_xyz} as it is {back_perc}% background." + msg = f"Skipping {new_pos_xyz} as it is {bg_message}." self._scan_logger.info(msg) continue