rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'background-detector-things' into 'v3'

Browse source 
Background detector things

See merge request openflexure/openflexure-microscope-server!461
This commit is contained in:
Julian Stirling 2026-01-14 16:51:49 +00:00
commit 5762fc5947
16 changed files with 376 additions and 419 deletions
Download patch file Download diff file Expand all files Collapse all files

4
ofm_config_full.json
View file

@ -16,7 +16,9 @@
"scans_folder": "/var/openflexure/scans/"
}
},
"stage_measure":"openflexure_microscope_server.things.stage_measure:RangeofMotionThing"
"stage_measure": "openflexure_microscope_server.things.stage_measure:RangeofMotionThing",
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"
},
"settings_folder": "/var/openflexure/settings/",
"log_folder": "/var/openflexure/logs/"

4
ofm_config_simulation.json
View file

@ -10,7 +10,9 @@
"kwargs": {
"scans_folder": "./openflexure/scans/"
}
}
},
"bg_color_channels_luv": "openflexure_microscope_server.things.background_detect:ColourChannelDetectLUV",
"bg_channel_deviations_luv": "openflexure_microscope_server.things.background_detect:ChannelDeviationLUV"
},
"settings_folder": "./openflexure/settings/",
"log_folder": "./openflexure/logs/"

BIN
picamera_coverage.zip
View file

Binary file not shown.

275
src/openflexure_microscope_server/background_detect.py → src/openflexure_microscope_server/things/background_detect.py
View file

@ -5,13 +5,15 @@ for analysis. Information from these images is used to detect whether an image f
current camera field of view contains sample.
"""
from typing import Any, Generic, Optional, TypeVar
from typing import Optional, TypeVar
import cv2
import numpy as np
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel
from labthings_fastapi.thing_description import type_to_dataschema
import labthings_fastapi as lt
from openflexure_microscope_server.ui import PropertyControl, property_control_for
SettingsType = TypeVar("SettingsType", bound=BaseModel)
BackgroundType = TypeVar("BackgroundType", bound=BaseModel)
@ -29,110 +31,27 @@ class ChannelBlankError(RuntimeError):
"""
class BackgroundDetectorStatus(BaseModel):
"""The status information about a background detector instance needed for the GUI.
Each BackgroundDetectAlgorithm must be able to return one of these models when
``status`` is called.
"""
ready: bool
"""True if ready to be used, if False this detector isn't initialised for use.
This could be called ``has_background_data`` or similar, but the more generic
``ready`` is used in case more complex methods are added in the future, which
need different initialisation.
"""
settings: dict[str, Any]
"""The settings for the current background detect Algorithm. These are a dictionary
dumped from the base model."""
# Setting schema is a dict until LabThings FastAPI issue #154 is fixed and
# DataSchema can be used directly. For now `model_dump()` must be used to dump schema
# to a dict.
settings_schema: dict[str, Any]
"""The schema for the settings for the current background detect Algorithm.
This is reported so that the UI can dynamically create a UI for any background detector
algorithm.
"""
class BackgroundDetectAlgorithm(Generic[SettingsType, BackgroundType]):
class BackgroundDetectAlgorithm(lt.Thing):
"""The base class for defining background detect algorithms."""
background_data_model: type[BackgroundType]
"""The data model of the background data. This must be set by child classes"""
settings_data_model: type[SettingsType]
"""The data model of algorithm settings. This must be set by child classes"""
display_name: str = lt.property(default="Base Detector", readonly=True)
def __init__(self) -> None:
"""Initialise the algorithm settings."""
if not hasattr(self, "background_data_model") or not hasattr(
self, "settings_data_model"
):
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise and create the lock."""
if self.display_name == "Base Detector":
raise NotImplementedError(
"All BackgroundDetectAlgorithm subclesses must set their own settings "
"and background data models."
"Do not try to use the BackgroungDetectAlgorithm directly. "
" Use a subclass"
)
self._settings: SettingsType = self.settings_data_model()
super().__init__(thing_server_interface)
@property
def status(self) -> BackgroundDetectorStatus:
"""The status information needed for the GUI. Read only."""
return BackgroundDetectorStatus(
ready=self.background_data is not None,
settings=self.settings.model_dump(),
# Dump model with `model_dump()` for reason explained when defining
# BackgroundDetectorStatus
settings_schema=type_to_dataschema(self.settings_data_model).model_dump(),
@lt.property
def ready(self) -> bool:
"""Whether the background detector is ready."""
raise NotImplementedError(
"Each background detect algorithm must implement a ready property."
)
# Requires a getter and a setter to support being a BaseModel but being
# saved to file as a dict
_background_data: Optional[BackgroundType] = None
@property
def background_data(self) -> Optional[BackgroundType]:
"""The statistics of the background image."""
bd = self._background_data
if bd is None:
return None
return bd
@background_data.setter
def background_data(self, value: Optional[BackgroundType | dict]) -> None:
"""Set the statistics for the background image.
This should be None, of no data is available. It can be set from either
a dictionary or a base model of the type specified in
``self.background_data_model``.
"""
if value is None:
self._background_data = None
elif isinstance(value, self.background_data_model):
self._background_data = value
elif isinstance(value, dict):
self._background_data = self.background_data_model(**value)
else:
raise TypeError(
f"Cannot set background_data with an object of type {type(value)}"
)
@property
def settings(self) -> SettingsType:
"""The statistics of the background image."""
return self._settings
@settings.setter
def settings(self, value: SettingsType | dict) -> None:
if isinstance(value, self.settings_data_model):
self._settings = value
elif isinstance(value, dict):
self._settings = self.settings_data_model(**value)
else:
raise TypeError(f"Cannot set settings with an object of type {type(value)}")
def image_is_sample(self, image: np.ndarray) -> tuple[bool, str]:
"""Label the current image as either background or sample.
@ -153,44 +72,15 @@ class BackgroundDetectAlgorithm(Generic[SettingsType, BackgroundType]):
"Each background detect algorithm must implement an set_background method."
)
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."""
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
raise NotImplementedError(
"Each background detect algorithm must implement an settings_ui method."
)
class ColourChannelDetectSettings(BaseModel):
"""A BaseModel for storing the settings for colour channel detectors."""
model_config = ConfigDict(extra="forbid")
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.
"""
# Use Field to set Title reported to UI. By default Pydantic will convert the name
# from snake_case to Title Case.
min_sample_coverage: float = Field(25, title="Sample Coverage Required (%)")
"""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[
ColourChannelDetectSettings,
ChannelDistributions,
]
):
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
@ -198,13 +88,46 @@ class ColourChannelDetectLUV(
intuitive way.
"""
background_data_model: type[ChannelDistributions] = ChannelDistributions
settings_data_model: type[ColourChannelDetectSettings] = ColourChannelDetectSettings
display_name: str = lt.property(default="Colour Channel (LUV)", readonly=True)
background_means: Optional[list[float]] = lt.setting(default=None, readonly=True)
"""The mean of each channel in the colourspace."""
background_stds: Optional[list[float]] = lt.setting(default=None, readonly=True)
"""The standard deviation of each channel in the colourspace."""
# These are the same as those used for ChannelDeviationLUV. More detail is
# provided there.
min_stds = [0.5, 0.3, 0.5]
channel_tolerance: float = lt.setting(default=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 = lt.setting(default=25)
"""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.
"""
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
return [
property_control_for(self, "channel_tolerance", label="Channel Tolerance"),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
@lt.property
def ready(self) -> bool:
"""Whether the background detector is ready."""
return self.background_means is not None and self.background_stds is not None
def background_mask(self, image: np.ndarray) -> np.ndarray:
"""Calculate a binary image, showing whether each pixel is background.
@ -213,7 +136,7 @@ class ColourChannelDetectLUV(
The image should be in LUV format, the output will be binary with the
same shape in the first two dimensions.
"""
if self.background_data is None:
if self.background_means is None or self.background_stds is None:
raise RuntimeError(
"Cannot calculated background mask if no background is set."
)
@ -222,13 +145,13 @@ class ColourChannelDetectLUV(
# the height of the sample changes.
# Wrapping in ``[[ ]]`` forces the colour channels to the numpy axis 2
# (3rd axis) so they are compared to the colour channel of each pixel.
means = np.array([[self.background_data.means[1:]]])
stds = np.array([[self.background_data.standard_deviations[1:]]])
means = np.array([[self.background_means[1:]]])
stds = np.array([[self.background_stds[1:]]])
# Compare each image in the pixel with the mean and standard deviation along
# axis to (the colour channels).
return np.all(
np.abs(image[:, :, 1:] - means) < stds * self.settings.channel_tolerance,
np.abs(image[:, :, 1:] - means) < stds * self.channel_tolerance,
axis=2,
)
@ -241,7 +164,7 @@ class ColourChannelDetectLUV(
:returns: A value (between 0 and 100) is the percentage of the image that is
sample.
"""
if not self.background_data:
if self.background_means is None or self.background_stds is None:
raise MissingBackgroundDataError(
"Background is not set: you need to calibrate background detection."
)
@ -260,7 +183,7 @@ class ColourChannelDetectLUV(
sample_coverage = self.get_sample_coverage(image)
# Use bool otherwise get numpy variants of True and False.
is_sample = bool(sample_coverage > self.settings.min_sample_coverage)
is_sample = bool(sample_coverage > self.min_sample_coverage)
message = f"{sample_coverage:0.1f}% sample"
if not is_sample:
message = "only " + message
@ -277,17 +200,11 @@ class ColourChannelDetectLUV(
raise ChannelBlankError("Some LUV channels have no standard deviation.")
std = np.maximum(std, self.min_stds)
self.background_data = ChannelDistributions(
means=mu.tolist(), standard_deviations=std.tolist()
)
self.background_means = mu.tolist()
self.background_stds = std.tolist()
class ChannelDeviationLUV(
BackgroundDetectAlgorithm[
ColourChannelDetectSettings,
ChannelDistributions,
]
):
class ChannelDeviationLUV(BackgroundDetectAlgorithm):
"""Compare the standard deviations of the LUV channels in a grid to background data.
Using an LUV colour space, each image is divided into an 8x8 grid of images.
@ -295,10 +212,10 @@ class ChannelDeviationLUV(
to the median standard deviation for a grid of background images.
"""
# Note we don't use the means in this algorithm but we use the same channel
# distributions model
background_data_model: type[ChannelDistributions] = ChannelDistributions
settings_data_model: type[ColourChannelDetectSettings] = ColourChannelDetectSettings
display_name: str = lt.property(default="Channel Deviation (LUV)", readonly=True)
background_stds: Optional[list[float]] = lt.setting(default=None, readonly=True)
"""The standard deviation of each channel in the colourspace."""
# Empirically, 0.5 seems to be approximate the standard deviation for a good image
# in L and V. U appears to be about 60% of this value. U is about 65% of V when
@ -306,6 +223,35 @@ class ChannelDeviationLUV(
# LUV colour space not converting to the CIELUV numbers)
min_stds = [0.5, 0.3, 0.5]
channel_tolerance: float = lt.setting(default=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 = lt.setting(default=25)
"""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.
"""
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
return [
property_control_for(self, "channel_tolerance", label="Channel Tolerance"),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
@lt.property
def ready(self) -> bool:
"""Whether the background detector is ready."""
return self.background_stds is not None
def get_sample_coverage(self, image: np.ndarray) -> float:
"""Return the percentage of the input image that is background.
@ -316,17 +262,17 @@ class ChannelDeviationLUV(
:returns: A value (between 0 and 100) that is the percentage of the image that is
sample.
"""
if not self.background_data:
if self.background_stds is None:
raise MissingBackgroundDataError(
"Background is not set: you need to calibrate background detection."
)
image_luv = cv2.cvtColor(image, cv2.COLOR_RGB2LUV)
stds = _chunked_stds(image_luv, 8, 8)
bg_stds = self.background_data.standard_deviations
l_cut = bg_stds[0] * self.settings.channel_tolerance
u_cut = bg_stds[1] * self.settings.channel_tolerance
v_cut = bg_stds[2] * self.settings.channel_tolerance
bg_stds = self.background_stds
l_cut = bg_stds[0] * self.channel_tolerance
u_cut = bg_stds[1] * self.channel_tolerance
v_cut = bg_stds[2] * self.channel_tolerance
populated_regions = (
(stds[:, :, 0] > l_cut) | (stds[:, :, 1] > u_cut) | (stds[:, :, 2] > v_cut)
@ -344,7 +290,7 @@ class ChannelDeviationLUV(
sample_coverage = self.get_sample_coverage(image)
# Use bool otherwise get numpy variants of True and False.
is_sample = bool(sample_coverage > self.settings.min_sample_coverage)
is_sample = bool(sample_coverage > self.min_sample_coverage)
message = f"{sample_coverage:0.1f}% sample"
if not is_sample:
message = "only " + message
@ -353,7 +299,6 @@ class ChannelDeviationLUV(
def set_background(self, image: np.ndarray) -> None:
"""Use the input image to update the background distributions."""
image_luv = cv2.cvtColor(image, cv2.COLOR_RGB2LUV)
mu = np.zeros(3)
c_stds = _chunked_stds(image_luv, 8, 8)
channel_blank = np.all(c_stds == 0, axis=(0, 1))
if np.any(channel_blank):
@ -361,9 +306,7 @@ class ChannelDeviationLUV(
std = np.median(c_stds, axis=(0, 1))
std = np.maximum(std, self.min_stds)
self.background_data = ChannelDistributions(
means=mu.tolist(), standard_deviations=std.tolist()
)
self.background_stds = std.tolist()
def _chunked_stds(img: np.ndarray, n_rows: int = 8, n_cols: int = 8) -> np.ndarray:

111
src/openflexure_microscope_server/things/camera/__init__.py
View file

@ -24,13 +24,11 @@ from PIL import Image
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from openflexure_microscope_server.background_detect import (
from openflexure_microscope_server.things.background_detect import (
BackgroundDetectAlgorithm,
BackgroundDetectorStatus,
ChannelDeviationLUV,
ColourChannelDetectLUV,
)
from openflexure_microscope_server.ui import ActionButton, PropertyControl
from openflexure_microscope_server.utilities import coerce_thing_selector
class JPEGBlob(lt.blob.Blob):
@ -161,6 +159,8 @@ class BaseCamera(lt.Thing):
``__init__`` method of the subclass.
"""
_all_background_detectors: Mapping[str, BackgroundDetectAlgorithm] = lt.thing_slot()
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
@ -174,15 +174,20 @@ class BaseCamera(lt.Thing):
dictionary in this function. Configuration will be added at a later date.
"""
super().__init__(thing_server_interface)
self.background_detectors = {
"Colour Channels (LUV)": ColourChannelDetectLUV(),
"Channel Deviations (LUV)": ChannelDeviationLUV(),
}
self._detector_name = "Channel Deviations (LUV)"
# Default is never updated but is used if the value set from settings is
# incorrect. In the future a better way to set defaults for thing slot mappings
# would be ideal.
self._default_background_detector = "bg_channel_deviations_luv"
self._background_detector_name: Optional[str] = None
def __enter__(self) -> Self:
"""Open hardware connection when the Thing context manager is opened."""
raise NotImplementedError("CameraThings must define their own __enter__ method")
self._background_detector_name = coerce_thing_selector(
thing_mapping=self._all_background_detectors,
selected=self._background_detector_name,
default=self._default_background_detector,
)
return self
def __exit__(
self,
@ -191,7 +196,7 @@ class BaseCamera(lt.Thing):
_traceback: Optional[TracebackType],
) -> None:
"""Close hardware connection when the Thing context manager is closed."""
raise NotImplementedError("CameraThings must define their own __exit__ method")
pass
@lt.property
def calibration_required(self) -> bool:
@ -582,81 +587,31 @@ class BaseCamera(lt.Thing):
# Note that the default detector name is set at init. This is over written if
# setting is loaded from disk.
@lt.setting
def detector_name(self) -> str:
def background_detector_name(self) -> Optional[str]:
"""The name of the active background selector."""
return self._detector_name
return self._background_detector_name
@detector_name.setter
def _set_detector_name(self, name: str) -> None:
"""Validate and set detector_name."""
if name not in self.background_detectors:
@background_detector_name.setter
def _set_background_detector_name(self, name: str) -> None:
"""Validate and set background_detector_name."""
if name not in self._all_background_detectors:
self.logger.warning(f"{name} is not a valid background detector name.")
self._detector_name = name
self._background_detector_name = name
@property
def active_detector(self) -> BackgroundDetectAlgorithm:
def background_detector(self) -> Optional[BackgroundDetectAlgorithm]:
"""The active background detector instance."""
return self.background_detectors[self.detector_name]
@lt.property
def background_detector_status(self) -> BackgroundDetectorStatus:
"""The status of the active detector for the UI."""
return self.active_detector.status
@lt.action
def update_detector_settings(self, data: dict[str, Any]) -> None:
"""Update the settings of the current detector.
This is an action not a setting/property as the data model depends on the
selected detector. As such, it cannot be specified with the necessary precision
to be included in a ThingDescription as a setting/property, while retaining
enough useful information to communicate to the UI how it is set and read.
The information on how to read the settings is exposed in
``background_detector_status``.
"""
self.active_detector.settings = data
# Manually save settings as the setter is not called.
self.save_settings()
@lt.setting
def background_detector_data(self) -> dict:
"""The data for each background detector, used to save to disk."""
data = {}
for name, obj in self.background_detectors.items():
bg_data = (
None
if obj.background_data is None
else obj.background_data.model_dump()
)
data[name] = {
"settings": obj.settings.model_dump(),
"background_data": bg_data,
}
return data
@background_detector_data.setter
def _set_background_detector_data(self, data: dict) -> None:
"""Set the data for each detector. Only to be used as settings are loaded from disk.
Do not call over HTTP. This needs to be updated once LbaThings Settings can be
read-only over HTTP (#484).
"""
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:
self.logger.warning(
f"No background detector named {name}, settings will be discarded."
)
if self.background_detector_name is None:
return None
return self._all_background_detectors[self.background_detector_name]
@lt.action
def image_is_sample(self) -> tuple[bool, str]:
"""Label the current image as either background or sample."""
if self.background_detector is None:
raise RuntimeError("No background detectors available.")
current_image = self.grab_as_array(stream_name="lores")
return self.active_detector.image_is_sample(current_image)
return self.background_detector.image_is_sample(current_image)
@lt.action
def set_background(self) -> None:
@ -668,10 +623,10 @@ class BaseCamera(lt.Thing):
future images to the distribution, to determine if each pixel is
foreground or background.
"""
if self.background_detector is None:
raise RuntimeError("No background detectors available.")
background = self.grab_as_array(stream_name="lores")
self.active_detector.set_background(background)
# Manually save settings as the setter is not called.
self.save_settings()
self.background_detector.set_background(background)
@property
def thing_state(self) -> Mapping[str, Any]:

8
src/openflexure_microscope_server/things/camera/opencv.py
View file

@ -41,6 +41,7 @@ class OpenCVCamera(BaseCamera):
def __enter__(self) -> Self:
"""Start the capture thread when the Thing context manager is opened."""
super().__enter__()
self.cap = cv2.VideoCapture(self.camera_index)
self._capture_enabled = True
self._capture_thread = Thread(target=self._capture_frames)
@ -49,9 +50,9 @@ class OpenCVCamera(BaseCamera):
def __exit__(
self,
_exc_type: type[BaseException],
_exc_value: Optional[BaseException],
_traceback: Optional[TracebackType],
exc_type: type[BaseException],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Release the camera when the Thing context manager is closed.
@ -62,6 +63,7 @@ class OpenCVCamera(BaseCamera):
if self._capture_thread is not None:
self._capture_thread.join()
self.cap.release()
super().__exit__(exc_type, exc_value, traceback)
@lt.property
def stream_active(self) -> bool:

29
src/openflexure_microscope_server/things/camera/picamera.py
View file

@ -38,7 +38,7 @@ import labthings_fastapi as lt
from labthings_fastapi.exceptions import ServerNotRunningError
from labthings_fastapi.types.numpy import NDArray
from openflexure_microscope_server.background_detect import ChannelBlankError
from openflexure_microscope_server.things.background_detect import ChannelBlankError
from openflexure_microscope_server.ui import (
ActionButton,
PropertyControl,
@ -380,6 +380,7 @@ class StreamingPiCamera2(BaseCamera):
This opens the picamera connection, initialises the camera, sets the
sensor_modes property, and then starts the streams.
"""
super().__enter__()
self._initialise_picamera(check_sensor_model=True)
# Sensor modes is a cached property read it once after initialising the camera
_modes = self.sensor_modes
@ -417,15 +418,16 @@ class StreamingPiCamera2(BaseCamera):
def __exit__(
self,
_exc_type: type[BaseException],
_exc_value: Optional[BaseException],
_traceback: Optional[TracebackType],
exc_type: type[BaseException],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Close the picamera connection when the Thing context manager is closed."""
self.stop_streaming()
with self._streaming_picamera() as cam:
cam.close()
del self._picamera
super().__exit__(exc_type, exc_value, traceback)
@lt.action
def start_streaming(
@ -759,15 +761,16 @@ class StreamingPiCamera2(BaseCamera):
self.set_static_green_equalisation()
self.set_ce_enable_to_off()
self.calibrate_lens_shading()
for _i in range(3):
try:
time.sleep(self._sensor_info.long_pause)
self.set_background()
# Return if background is set
return
except ChannelBlankError:
# If channel is blank, sleep a second and try again.
pass
if self.background_detector is not None:
for _i in range(3):
try:
time.sleep(self._sensor_info.long_pause)
self.set_background()
# Return if background is set
return
except ChannelBlankError:
# If channel is blank, sleep a second and try again.
pass
raise RuntimeError("Couldn't set background")
@lt.property

15
src/openflexure_microscope_server/things/camera/simulation.py
View file

@ -181,7 +181,9 @@ class SimulatedCamera(BaseCamera):
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating."""
return not self.background_detector_status.ready
if self.background_detector is None:
return True
return not self.background_detector.ready
def generate_sprites(self) -> None:
"""Generate sprites to populate the image."""
@ -344,20 +346,22 @@ class SimulatedCamera(BaseCamera):
def __enter__(self) -> Self:
"""Start the capture thread when the Thing context manager is opened."""
super().__enter__()
self.generate_canvas()
self.start_streaming()
return self
def __exit__(
self,
_exc_type: type[BaseException],
_exc_value: Optional[BaseException],
_traceback: Optional[TracebackType],
exc_type: type[BaseException],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Close the capture thread when the Thing context manager is closed."""
if self._capture_thread is not None and self._capture_thread.is_alive():
self._capture_enabled = False
self._capture_thread.join()
super().__exit__(exc_type, exc_value, traceback)
@lt.action
def start_streaming(
@ -463,7 +467,8 @@ class SimulatedCamera(BaseCamera):
"""
self.remove_sample()
time.sleep(0.2)
self.set_background()
if self.background_detector is not None:
self.set_background()
time.sleep(0.2)
self.load_sample()

5
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -234,7 +234,10 @@ class SmartScanThing(lt.Thing):
self._csm.assert_calibration()
if self.skip_background:
if not self._cam.background_detector_status.ready:
if (
self._cam.background_detector is None
or not self._cam.background_detector.ready
):
raise RuntimeError(
"Background is not set: you need to calibrate background detection."
)

44
src/openflexure_microscope_server/utilities.py
View file

@ -13,6 +13,8 @@ from typing import (
Callable,
Concatenate,
Literal,
Mapping,
Optional,
ParamSpec,
TypeAlias,
TypeVar,
@ -21,6 +23,8 @@ from typing import (
from pydantic import BaseModel
import labthings_fastapi as lt
T = TypeVar("T")
P = ParamSpec("P")
LockableClass = TypeVar("LockableClass")
@ -388,3 +392,43 @@ def resolve_path_from_dir(path: str, directory: str) -> str:
if not os.path.isabs(path):
path = os.path.join(directory, path)
return os.path.normpath(path)
def coerce_thing_selector(
thing_mapping: Mapping[str, lt.Thing],
selected: Optional[str],
default: Optional[str] = None,
) -> Optional[str]:
"""Return a valid selector for a mapping of things or None if empty.
:param thing_mapping: A mapping of str -> ``Thing``
:param selected: The key of the selected thing (or ``None``)
:param default: The default key to fallback to if selected is ``None`` or the key is
missing
:return: ``None`` if the mapping is empty, else a key that is in the mapping. Order
of preference is: selected, default, the first key.
"""
if not thing_mapping:
# Empty mapping
if selected is not None:
LOGGER.warning(
f"Could not select {selected} from Thing mapping as the mapping is empty."
)
# The return is always None if the mapping is empty
return None
if selected in thing_mapping:
# Selected exists, return it.
return selected
if selected is not None:
LOGGER.warning(f"Could not select '{selected}' from Thing mapping")
if default in thing_mapping:
return default
if default is not None:
LOGGER.warning(f"Could not select default key '{default}' from Thing mapping")
# Final option is to return the first key
return list(thing_mapping)[0]

6
tests/hardware_specific_tests/picamera2/cam_test_utils.py
View file

@ -3,6 +3,7 @@
from contextlib import contextmanager
from typing import Optional
from openflexure_microscope_server.things.background_detect import ChannelDeviationLUV
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
from ...shared_utils.lt_test_utils import LabThingsTestEnv
@ -18,7 +19,10 @@ def camera_test_env(settings_folder: Optional[str] = None):
:param settings_folder: The settings folder for the camera, if none is supplied, new
temporary directory will be used as the settings folder.
"""
thing_conf = {"camera": StreamingPiCamera2}
thing_conf = {
"camera": StreamingPiCamera2,
"bg_channel_deviations_luv": ChannelDeviationLUV,
}
with LabThingsTestEnv(things=thing_conf, settings_folder=settings_folder) as env:
yield env

4
tests/hardware_specific_tests/picamera2/test_calibration.py
View file

@ -19,7 +19,7 @@ def test_calibration(picamera_test_env):
# Tuning should start the same as the server is loading with no settings.
assert picamera_thing.default_tuning == picamera_thing.tuning
# The background detector isn't ready as there is no background image.
assert not picamera_thing.background_detector_status.ready
assert not picamera_thing.background_detector.ready
# Run full auto calibrate
picamera_client.full_auto_calibrate()
@ -31,7 +31,7 @@ def test_calibration(picamera_test_env):
# The default should be unchanged
assert picamera_thing.default_tuning == original_default
assert picamera_thing.background_detector_status.ready
assert picamera_thing.background_detector.ready
def test_tuning_is_persistent():

162
tests/unit_tests/test_background_detectors.py
View file

@ -7,18 +7,19 @@ import re
import numpy as np
import pytest
from pydantic import BaseModel
from openflexure_microscope_server.background_detect import (
import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.background_detect import (
BackgroundDetectAlgorithm,
ChannelBlankError,
ChannelDeviationLUV,
ChannelDistributions,
ColourChannelDetectLUV,
ColourChannelDetectSettings,
MissingBackgroundDataError,
_chunked_stds,
)
from openflexure_microscope_server.ui import PropertyControl
RNG = np.random.default_rng()
IMG_SHAPE = (820, 616, 3)
@ -55,56 +56,45 @@ def test_bg_detect_base_class():
If initialised as is it should raise not implemented error.
"""
with pytest.raises(NotImplementedError):
BackgroundDetectAlgorithm()
create_thing_without_server(BackgroundDetectAlgorithm)
def test_partial_base_classes():
"""Create a partial classes and check they raise the correct errors."""
"""Create a partial class and check it raises the correct errors."""
class BadAlgo1(BackgroundDetectAlgorithm):
"""Only has a settings model so it cannot initialise."""
class BadAlgo(BackgroundDetectAlgorithm):
"""Can initialise. Other properties and methods error."""
settings_data_model = ColourChannelDetectSettings
display_name: str = lt.property(default="Bad Algorithm", readonly=True)
bad_algo = create_thing_without_server(BadAlgo)
with pytest.raises(NotImplementedError):
BadAlgo1()
class BadAlgo2(BackgroundDetectAlgorithm):
"""Only has a background model so it cannot initialise."""
background_data_model = ChannelDistributions
bad_algo.ready
with pytest.raises(NotImplementedError):
BadAlgo2()
class BadAlgo3(BackgroundDetectAlgorithm):
"""Has both models, intalises by cannot run set_background or image_is_sample."""
settings_data_model = ColourChannelDetectSettings
background_data_model = ChannelDistributions
bad_algo3 = BadAlgo3()
bad_algo.settings_ui
with pytest.raises(NotImplementedError):
bad_algo3.set_background(background_image)
bad_algo.set_background(background_image)
with pytest.raises(NotImplementedError):
bad_algo3.image_is_sample(background_image)
bad_algo.image_is_sample(background_image)
def test_colour_channel_luv(background_image, sample_image):
"""Test measuring if a sample is background."""
cc_luv = ColourChannelDetectLUV()
cc_luv = create_thing_without_server(ColourChannelDetectLUV)
# No background data so it is not ready and will error if image_is_sample is called.
assert not cc_luv.status.ready
assert not cc_luv.ready
with pytest.raises(MissingBackgroundDataError):
cc_luv.image_is_sample(background_image)
# Set the background
cc_luv.set_background(background_image)
# Now it is ready
assert cc_luv.status.ready
assert cc_luv.ready
sample, message = cc_luv.image_is_sample(background_image)
assert not sample
assert "0.0%" in message
@ -116,7 +106,7 @@ def test_colour_channel_luv(background_image, sample_image):
assert 49.8 < float(match.group(1)) < 50.2
# Require 75% coverage
cc_luv.settings = ColourChannelDetectSettings(min_sample_coverage=75.0)
cc_luv.min_sample_coverage = 75.0
sample, message = cc_luv.image_is_sample(sample_image)
# No longer detected as a sample.
@ -127,69 +117,6 @@ def test_colour_channel_luv(background_image, sample_image):
assert 49.8 < float(match.group(1)) < 50.2
def test_colour_channel_luv_save_load(background_image, sample_image):
"""Get settings and data as dicts, and creating new instance using these dicts.
This is how the camera will load/save settings from/to disk.
"""
cc_luv = ColourChannelDetectLUV()
cc_luv.set_background(background_image)
# Check types for background data
assert cc_luv.background_data_model is ChannelDistributions
assert isinstance(cc_luv.background_data, cc_luv.background_data_model)
# Change Settings
cc_luv.settings = ColourChannelDetectSettings(min_sample_coverage=10.0)
setting_dict = cc_luv.settings.model_dump()
data_dict = cc_luv.background_data.model_dump()
# Remove the old detector so we don't accidentally use it!
del cc_luv
# Create a new instance
cc_luv2 = ColourChannelDetectLUV()
assert not cc_luv2.status.ready
# Load settings and channels from dictionary as the camera will do on init.
cc_luv2.settings = setting_dict
cc_luv2.background_data = data_dict
# Now should be ready to use
assert cc_luv2.status.ready
assert cc_luv2.settings.min_sample_coverage == 10.0
sample, _ = cc_luv2.image_is_sample(sample_image)
assert sample
# Remove the 2nd detector so we don't accidentally use it!
del cc_luv2
# One final test that None can be set explicitly to background data as this will
# happen if loading with background detect not saved.
cc_luv3 = ColourChannelDetectLUV()
assert not cc_luv3.status.ready
# Load settings and channels from dictionary as the camera will do on init.
cc_luv3.settings = setting_dict
cc_luv3.background_data = None
# Still not ready
assert not cc_luv3.status.ready
def test_colour_channel_luv_load_bad_data():
"""Check a type error is thrown on bad data input."""
class WrongModel(BaseModel):
"""Using a different BaseModel as this is most likely to cause confusion."""
prop1: int = 8
prop2: str = "foo"
cc_luv = ColourChannelDetectLUV()
with pytest.raises(TypeError):
cc_luv.settings = WrongModel()
with pytest.raises(TypeError):
cc_luv.background_data = WrongModel()
def create_patchwork_image(magnitude=3, blank_channels=None):
"""Create a patchwork image, with known stds per chunk.
@ -239,7 +166,7 @@ def test_chunked_stds_with_precomputed_chunk_stds():
def test_channel_deviation_luv_set_background(mocker):
"""Test set_background takes the median of each channel, and errors for blank channels."""
cd_luv = ChannelDeviationLUV()
cd_luv = create_thing_without_server(ChannelDeviationLUV)
# Patch RGB to LUV so or we don't know what the STDs should be
mocker.patch("cv2.cvtColor", side_effect=lambda img, _method: img)
@ -252,7 +179,7 @@ def test_channel_deviation_luv_set_background(mocker):
# Do a somewhat verbose checking for clarity
for channel in range(3):
# Saved std
channel_std = cd_luv.background_data.standard_deviations[channel]
channel_std = cd_luv.background_stds[channel]
# Expected median
channel_median = np.median(expected_stds[:, :, channel])
# If the median is above the minimum allowed then it should be returned
@ -270,21 +197,21 @@ def test_channel_deviation_luv_set_background(mocker):
def test_channel_deviation_luv_image_is_sample(background_image, mocker):
"""Check image_is_sample reports the result from get_sample_coverage."""
cd_luv = ChannelDeviationLUV()
cd_luv = create_thing_without_server(ChannelDeviationLUV)
# No background data so it is not ready and will error if image_is_sample is called.
assert not cd_luv.status.ready
assert not cd_luv.ready
with pytest.raises(MissingBackgroundDataError):
cd_luv.image_is_sample(background_image)
cd_luv.settings.min_sample_coverage = 20
cd_luv.min_sample_coverage = 20
cd_luv.get_sample_coverage = mocker.Mock(return_value=10)
is_sample, message = cd_luv.image_is_sample(background_image)
assert not is_sample
assert message == r"only 10.0% sample"
# Reduce the min coverage
cd_luv.settings.min_sample_coverage = 9
cd_luv.min_sample_coverage = 9
is_sample, message = cd_luv.image_is_sample(background_image)
assert is_sample
@ -293,35 +220,48 @@ def test_channel_deviation_luv_image_is_sample(background_image, mocker):
def test_channel_deviation_luv_get_sample_coverage(background_image, mocker):
"""Check _get_sample_coverage returns the values expected."""
cd_luv = ChannelDeviationLUV()
cd_luv = create_thing_without_server(ChannelDeviationLUV)
# Create fake chunked STD data where each channel is the numbers 0 -> 31.5 in 0.5
# steps
grid = np.arange(0, 32, 0.5).reshape(8, 8)
fake_stds = np.stack([grid, grid, grid], axis=-1)
mocker.patch(
"openflexure_microscope_server.background_detect._chunked_stds",
"openflexure_microscope_server.things.background_detect._chunked_stds",
return_value=fake_stds,
)
# Create fake background
cd_luv.background_data = ChannelDistributions(
means=[0, 0, 0], standard_deviations=[1.1, 1.1, 1.1]
)
cd_luv.background_stds = [1.1, 1.1, 1.1]
# Get sample coverage with channel tolerance of 7. Checking each channel for the
# numbers below 7.7. There are 16 out of 64. So 75% should be sample
cd_luv.settings.channel_tolerance = 7
cd_luv.channel_tolerance = 7
assert cd_luv.get_sample_coverage(background_image) == 75
# This is unchanged if two channels have larger background values.
cd_luv.background_data = ChannelDistributions(
means=[0, 0, 0], standard_deviations=[1.6, 1.6, 1.1]
)
cd_luv.background_stds = [1.6, 1.6, 1.1]
assert cd_luv.get_sample_coverage(background_image) == 75
# But coverage increases if any channels has a lower background value.
cd_luv.background_data = ChannelDistributions(
means=[0, 0, 0], standard_deviations=[1.6, 0.6, 1.1]
)
cd_luv.background_stds = [1.6, 0.6, 1.1]
assert cd_luv.get_sample_coverage(background_image) == 85.9375
# Returns to 75% if that channel is empty
fake_stds[:, :, 1] = 0
assert cd_luv.get_sample_coverage(background_image) == 75
@pytest.mark.parametrize("detector_cls", [ColourChannelDetectLUV, ChannelDeviationLUV])
def test_background_detect_settings_ui(detector_cls):
"""Check that both background detectors provide the expected UI to the webapp.
As both have identical settings they can be tested together
"""
detector = create_thing_without_server(detector_cls)
ui = detector.settings_ui
assert len(ui) == 2
assert isinstance(ui[0], PropertyControl)
assert ui[0].property_name == "channel_tolerance"
assert ui[0].label == "Channel Tolerance"
assert isinstance(ui[1], PropertyControl)
assert ui[1].property_name == "min_sample_coverage"
assert ui[1].label == "Sample Coverage Required (%)"

9
tests/unit_tests/test_simulated_camera.py
View file

@ -10,6 +10,7 @@ from hypothesis import strategies as st
import labthings_fastapi as lt
from openflexure_microscope_server.things.background_detect import ChannelDeviationLUV
from openflexure_microscope_server.things.camera import simulation
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
from openflexure_microscope_server.things.stage.dummy import DummyStage
@ -20,7 +21,11 @@ from ..shared_utils.lt_test_utils import LabThingsTestEnv
@pytest.fixture
def test_env() -> LabThingsTestEnv:
"""Yield a test environment with the Simulated Camera and Dummy Stage."""
thing_conf = {"camera": SimulatedCamera, "stage": DummyStage}
thing_conf = {
"camera": SimulatedCamera,
"stage": DummyStage,
"bg_channel_deviations_luv": ChannelDeviationLUV,
}
with LabThingsTestEnv(things=thing_conf) as env:
yield env
@ -181,4 +186,4 @@ def test_simulation_cam_calibration(camera):
assert camera.calibration_required
camera.full_auto_calibrate()
assert not camera.calibration_required
assert camera.background_detector_status.ready
assert camera.background_detector.ready

59
tests/unit_tests/test_thing_selector.py Normal file
View file

@ -0,0 +1,59 @@
"""Test selector coercion logic for Thing mappings."""
import logging
import pytest
from openflexure_microscope_server.utilities import coerce_thing_selector
@pytest.mark.parametrize(
("selected", "default", "warning_count"),
[
("a", "b", 1),
(None, "b", 0),
("a", None, 1),
(None, None, 0),
],
)
def test_coerce_with_empty_mapping(selected, default, warning_count, caplog):
"""Test None always returned for empty mappings, check warnings when appropriate."""
with caplog.at_level(logging.WARNING):
output = coerce_thing_selector(
thing_mapping={}, selected=selected, default=default
)
assert output is None
assert len(caplog.records) == warning_count
@pytest.mark.parametrize(
("selected", "default", "returned", "warning_count"),
[
("b", "b", "b", 0),
("b", "a", "b", 0),
("b", None, "b", 0),
("b", "z", "b", 0),
(None, "b", "b", 0),
(None, "a", "a", 0),
(None, None, "a", 0),
(None, "z", "a", 1),
("z", "b", "b", 1),
("z", "a", "a", 1),
("z", None, "a", 1),
("z", "z", "a", 2),
],
)
def test_coerce_with_populated_mapping(
selected, default, returned, warning_count, caplog
):
"""Test coercion of selected key for a populated thing mapping.
Check that warnings are raised when appropriate.
"""
thing_mapping = {"a": "Foo", "b": "Bar", "c": "FooBar"}
with caplog.at_level(logging.WARNING):
output = coerce_thing_selector(
thing_mapping=thing_mapping, selected=selected, default=default
)
assert output == returned
assert len(caplog.records) == warning_count

60
webapp/src/components/tabContentComponents/backgroundDetectComponents/paneBackgroundDetect.vue
View file

@ -5,18 +5,13 @@
<li>
<a class="uk-accordion-title" href="#">Configure</a>
<div class="uk-accordion-content">
<h4 v-if="backgroundDetectorName" class="detector-name">
{{ backgroundDetectorName }}
<h4 v-if="backgroundDetectorDisplayName" class="detector-name">
{{ backgroundDetectorDisplayName }}
</h4>
<input-from-schema
v-if="backgroundDetectorStatus"
v-model="backgroundDetectorStatus.settings"
:data-schema="backgroundDetectorStatus.settings_schema"
label=""
:animate="animate"
@requestUpdate="readSettings"
@sendValue="writeSettings"
@animationShown="resetAnimate"
<server-specified-property-control
v-for="(setting, index) in backgroundDetectorSettings"
:key="'detector_setting' + index"
:property-data="setting"
/>
</div>
</li>
@ -51,33 +46,26 @@
<script>
import ActionButton from "../../labThingsComponents/actionButton.vue";
import InputFromSchema from "../../labThingsComponents/inputFromSchema.vue";
import ServerSpecifiedPropertyControl from "../../labThingsComponents/serverSpecifiedPropertyControl.vue";
export default {
components: {
ActionButton,
InputFromSchema,
ServerSpecifiedPropertyControl,
},
data() {
return {
backgroundDetectorStatus: undefined,
ready: false,
backgroundDetectorName: undefined,
backgroundDetectorDisplayName: undefined,
backgroundDetectorSettings: [],
animate: false,
};
},
computed: {
ready() {
const status = this.backgroundDetectorStatus;
return status && status.ready === true;
},
},
async created() {
this.backgroundDetectorStatus = await this.readThingProperty(
"camera",
"background_detector_status",
);
this.readSettings();
},
methods: {
@ -94,19 +82,21 @@ export default {
this.modalNotify(`Current image is ${label} (${r.output[1]})`);
},
readSettings: async function () {
this.backgroundDetectorName = await this.readThingProperty("camera", "detector_name");
this.backgroundDetectorStatus = await this.readThingProperty(
this.backgroundDetectorName = await this.readThingProperty(
"camera",
"background_detector_status",
"background_detector_name",
);
},
writeSettings: async function (requestedValue) {
await this.invokeAction("camera", "update_detector_settings", { data: requestedValue });
this.animate = true;
this.readSettings();
},
resetAnimate: function () {
this.animate = false;
if (this.backgroundDetectorName) {
this.ready = await this.readThingProperty(this.backgroundDetectorName, "ready");
this.backgroundDetectorSettings = await this.readThingProperty(
this.backgroundDetectorName,
"settings_ui",
);
this.backgroundDetectorDisplayName = await this.readThingProperty(
this.backgroundDetectorName,
"display_name",
);
}
},
},
};