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

Rename "active_detector" for background detecor. Improve thing selector coercion

Browse source
This commit is contained in:
Julian Stirling 2026-01-14 15:09:18 +00:00
parent e46836ffe1
commit ca46439269
10 changed files with 182 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -28,6 +28,7 @@ from openflexure_microscope_server.things.background_detect import (
BackgroundDetectAlgorithm,
)
from openflexure_microscope_server.ui import ActionButton, PropertyControl
from openflexure_microscope_server.utilities import coerce_thing_selector
class JPEGBlob(lt.blob.Blob):
@ -158,7 +159,7 @@ class BaseCamera(lt.Thing):
``__init__`` method of the subclass.
"""
background_detectors: Mapping[str, BackgroundDetectAlgorithm] = lt.thing_slot()
_all_background_detectors: Mapping[str, BackgroundDetectAlgorithm] = lt.thing_slot()
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
@ -176,12 +177,17 @@ class BaseCamera(lt.Thing):
# 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_detector = "bg_channel_deviations_luv"
self._detector_name = "bg_channel_deviations_luv"
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,
@ -190,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:
@ -581,30 +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."""
if self.detector_name not in self.background_detectors:
self.logger.warning(f"No detector named {self.detector_name}")
self.detector_name = self._default_detector
return self.background_detectors[self.detector_name]
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:
@ -616,8 +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)
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:

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

@ -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.active_detector.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.active_detector.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]