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

Merge branch 'fom-af' into 'v3'

Browse source 
Record fom as other sharpness metric

Closes #615

See merge request openflexure/openflexure-microscope-server!578
This commit is contained in:
Joe Knapper 2026-05-13 16:53:28 +00:00
commit 7450a0e47c
7 changed files with 358 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.codespellrc
View file

@ -19,3 +19,5 @@ skip = node_modules,
report.xml
regex = [a-zA-Z0-9\-'’]+
ignore-words-list = fom

4
.codespellrc2
View file

@ -23,7 +23,7 @@ skip = node_modules,
# Ignores split by |
# Ignore hexadecimal numbers preceded by #
# Ignore FoV
ignore-regex = #[0-9A-Fa-f]{3,8}\b|\bFoV\b
# Ignore FoV and FoM
ignore-regex = #[0-9A-Fa-f]{3,8}\b|FoV|FoM
regex = [A-Z]?[a-z\']+

BIN
picamera_coverage.zip
View file

Binary file not shown.

183
src/openflexure_microscope_server/things/autofocus.py
View file

@ -34,10 +34,16 @@ class NotStreamingError(RuntimeError):
"""No images captured from stream. The camera is almost certainly not streaming."""
class SharpnessMethod(enum.Enum):
"""The possible SharpnessMethods for autofocus."""
class SharpnessMethod(enum.IntFlag):
"""The possible SharpnessMethods for autofocus.
JPEG = enum.auto()
Use powers of two for the methods so they can be selected bitwise. This allows choosing what to record
as the sum of methods.
"""
JPEG = 1
FOCUS_FOM = 2
# Next method should be 4 not 3 for bitwise selection.
class AutofocusParams(BaseModel):
@ -235,8 +241,9 @@ class SharpnessDataArrays(BaseModel):
jpeg_times: NDArray
jpeg_sizes: NDArray
focus_foms: NDArray
stage_times: NDArray
stage_positions: list[dict[str, int]]
stage_positions: list[Mapping[str, int]]
class JPEGSharpnessMonitor:
@ -253,24 +260,55 @@ class JPEGSharpnessMonitor:
"""
def __init__(self, stage: BaseStage, camera: BaseCamera) -> None:
def __init__(
self,
stage: BaseStage,
camera: BaseCamera,
method: SharpnessMethod = SharpnessMethod.JPEG,
record: Optional[int] = None,
) -> None:
"""Initialise a new JPEGSharpnessMonitor. The args are injected automatically.
:param stage: A direct_thing_client dependency for the the microscope stage.
:param camera: A raw_thing_client depeendency for the camera. This is a raw
dependency as the underlying class needs to be
dependency as required by the underlying class.
:param method: The sharpness metric used when evaluating autofocus.
:param record: Bitmask of sharpness metrics to record while monitoring.
If ``None``, only the metric specified by ``method`` is recorded.
:raises ValueError: If ``method`` is not included in ``record``.
:raises ValueError: If ``SharpnessMethod.FOCUS_FOM`` is requested but
the camera does not support FocusFoM measurements.
"""
self.camera = camera
self.stage = stage
self.method = method
self.record = method if record is None else record
if not self.method & self.record:
raise ValueError(
f"The sharpness metric {self.record} is not being recorded."
)
if self.record & SharpnessMethod.FOCUS_FOM and not camera.supports_focus_fom:
raise ValueError(
"Cannot record focus FOM as this camera doesn't support it."
)
LOGGER.debug(f"Created sharpness monitor with {stage}, {camera}")
self._stage_positions: list[Mapping[str, int]] = []
self._stage_times: list[float] = []
self._jpeg_times: list[float] = []
self._jpeg_sizes: list[int] = []
self._focus_foms: list[float] = []
@property
def stage_positions(self) -> Sequence[Mapping[str, int]]:
"""The positions recorded for the stage."""
"""The positions recorded for the stage.
:raises ValueError: If stage position recording is not enabled.
"""
if not self._stage_positions:
raise ValueError("Stage positions have not been recorded yet.")
return self._stage_positions
@property
@ -285,17 +323,40 @@ class JPEGSharpnessMonitor:
@property
def jpeg_sizes(self) -> Sequence[int]:
"""The recorded JPEG frame sizes used as a sharpness metric."""
"""The recorded JPEG frame sizes used as a sharpness metric.
:raises ValueError: If JPEG sharpness recording is not enabled.
"""
if not self.record & SharpnessMethod.JPEG:
raise ValueError("JPEG sizes are not being recorded.")
return self._jpeg_sizes
@property
def focus_foms(self) -> Sequence[float]:
"""The recorded FocusFoM values.
:raises ValueError: If FocusFoM recording is not enabled.
"""
if not self.record & SharpnessMethod.FOCUS_FOM:
raise ValueError("FocusFoM values are not being recorded.")
return self._focus_foms
running = False
async def monitor_sharpness(self) -> None:
"""Start monitoring the frame sizes."""
"""Start monitoring sharpness metrics."""
self.running = True
async for frame in self.camera.lores_mjpeg_stream.frame_async_generator():
self._jpeg_times.append(time.time())
# JPEG sharpness metric
if self.record & SharpnessMethod.JPEG:
self._jpeg_sizes.append(len(frame))
# FocusFoM metric
if self.record & SharpnessMethod.FOCUS_FOM:
self._focus_foms.append(self.camera.focus_fom)
if not self.running:
break
@ -352,7 +413,12 @@ class JPEGSharpnessMonitor:
if istop is None:
istop = istart + 2
jpeg_times: np.ndarray = np.array(self.jpeg_times)
jpeg_sizes: np.ndarray = np.array(self.jpeg_sizes)
# Two sharpness metrics are measured - this chooses which to use to focus
if self.method == SharpnessMethod.JPEG:
sharpnesses = np.array(self.jpeg_sizes)
elif self.method == SharpnessMethod.FOCUS_FOM:
sharpnesses = np.array(self.focus_foms)
stage_times: np.ndarray = np.array(self.stage_times)[istart:istop]
stage_heights: np.ndarray = np.array(
[p["z"] for p in self.stage_positions[istart:istop]]
@ -373,7 +439,7 @@ class JPEGSharpnessMonitor:
LOGGER.debug("changing stop to %s", (stop))
jpeg_times = jpeg_times[start:stop]
jpeg_heights: np.ndarray = np.interp(jpeg_times, stage_times, stage_heights)
return jpeg_times, jpeg_heights, jpeg_sizes[start:stop]
return jpeg_times, jpeg_heights, sharpnesses[start:stop]
def sharpest_z_on_move(self, data_index: int) -> int:
"""Return the z position of the sharpest image on a given move."""
@ -387,10 +453,21 @@ class JPEGSharpnessMonitor:
def data_to_array(self) -> SharpnessDataArrays:
"""Return the gathered data as SharpnessDataArrays."""
data = {}
for k in ["jpeg_times", "jpeg_sizes", "stage_times", "stage_positions"]:
data[k] = getattr(self, k)
return SharpnessDataArrays(**data)
return SharpnessDataArrays(
jpeg_times=np.array(self._jpeg_times),
jpeg_sizes=(
np.array(self._jpeg_sizes)
if self.record & SharpnessMethod.JPEG
else np.array([])
),
focus_foms=(
np.array(self._focus_foms)
if self.record & SharpnessMethod.FOCUS_FOM
else np.array([])
),
stage_times=np.array(self._stage_times),
stage_positions=self._stage_positions,
)
def data_dict(self) -> dict:
"""Return the gathered data as dict."""
@ -414,14 +491,34 @@ class AutofocusThing(lt.Thing):
self,
dz: int = 2000,
start: Literal["centre", "base"] = "centre",
sharpness_metric: SharpnessMethod = SharpnessMethod.JPEG,
record: Optional[int] = None,
) -> SharpnessDataArrays:
"""Sweep the stage up and down, then move to the sharpest point.
This method will will move down by dz/2, sweep up by dz, and then evaluate
the position where the image was sharpest. We'll then move back down, and
finally up to the sharpest point.
This method will move down by dz/2, sweep up by dz, and then evaluate
the position where the image was sharpest. We'll then move back down,
and finally up to the sharpest point.
:param dz: Total z-range (in steps) used for the autofocus sweep.
:param start: Starting position of the sweep. "centre" begins at -dz/2,
while "base" begins from the current position.
:param sharpness_metric: Sharpness metric used for evaluation. Either
JPEG file size (1) or inbuilt figure of metric (2).
:param record: Optional bitmask of sharpness metrics to record during
acquisition. Sharpness metrics can be selected as the sum of their int
value (see sharpness_metric).
If None, defaults to recording only the selected sharpness_metric.
:returns: SharpnessDataArrays containing all recorded sharpness and
stage position data for the sweep.
"""
with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
with JPEGSharpnessMonitor(
self._stage,
self._cam,
sharpness_metric,
record=record,
) as sharpness_monitor:
# Move to (-dz / 2)
if start == "centre":
sharpness_monitor.focus_rel(-dz // 2)
@ -444,20 +541,30 @@ class AutofocusThing(lt.Thing):
self,
dz: Sequence[int],
wait: float = 0,
sharpness_metric: SharpnessMethod = SharpnessMethod.JPEG,
record: Optional[int] = None,
) -> SharpnessDataArrays:
"""Make a move (or a series of moves) and monitor sharpness.
This method will will make a series of relative moves in z, and
return the sharpness (JPEG size) vs time, along with timestamps
return the sharpness (JPEG size and/or FOM) vs time, along with timestamps
for the moves. This can be used to calibrate autofocus.
Each move is relative to the last one, i.e. we will finish at
``sum(dz)`` relative to the starting position.
If ``wait`` is specified, we will wait for that many seconds
between moves.
:param dz: A list of moves to make in z (in steps).
:param wait: The time to pause between movements, in seconds.
:param sharpness_metric: Sharpness metric used for evaluation. Either
JPEG file size (1) or inbuilt figure of metric (2).
:param record: Optional bitmask of sharpness metrics to record during
acquisition. Sharpness metrics can be selected as the sum of their int
value (see sharpness_metric).
If None, defaults to recording only the selected sharpness_metric.
"""
with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
with JPEGSharpnessMonitor(
self._stage, self._cam, sharpness_metric, record
) as sharpness_monitor:
for move_index, current_dz in enumerate(dz):
if move_index > 0 and wait > 0:
time.sleep(wait)
@ -469,6 +576,8 @@ class AutofocusThing(lt.Thing):
self,
dz: int = 2000,
start: Literal["centre", "base"] = "centre",
sharpness_metric: SharpnessMethod = SharpnessMethod.JPEG,
record: Optional[int] = None,
) -> tuple[list[float], list[float]]:
"""Repeatedly autofocus the stage until it looks focused.
@ -476,10 +585,28 @@ class AutofocusThing(lt.Thing):
in the middle 3/5 of its range. Such logic can be helpful if the microscope
is close to focus, but not quite within ``dz/2``. It will attempt to autofocus
up to 10 times.
:param dz: Total z-range (in steps) used for the autofocus sweep.
:param start: Starting position of the sweep. "centre" begins at -dz/2,
while "base" begins from the current position.
:param sharpness_metric: Sharpness metric used for evaluation. Either
JPEG file size (1) or inbuilt figure of metric (2).
:param record: Optional bitmask of sharpness metrics to record during
acquisition. Sharpness metrics can be selected as the sum of their int
value (see sharpness_metric).
If None, defaults to recording only the selected sharpness_metric.
:returns: SharpnessDataArrays containing all recorded sharpness and
stage position data for the sweep.
"""
attempt = 0
with JPEGSharpnessMonitor(self._stage, self._cam) as sharpness_monitor:
with JPEGSharpnessMonitor(
self._stage,
self._cam,
sharpness_metric,
record=record,
) as sharpness_monitor:
while attempt < 10:
attempt += 1
if start == "centre":
@ -496,9 +623,11 @@ class AutofocusThing(lt.Thing):
focus_data_index, _ = sharpness_monitor.focus_rel(
dz, block_cancellation=True
)
_times, heights, sizes = sharpness_monitor.move_data(focus_data_index)
_times, heights, sharpnesses = sharpness_monitor.move_data(
focus_data_index
)
peak_height = heights[np.argmax(sizes)]
peak_height = heights[np.argmax(sharpnesses)]
target_min = np.min(heights) + dz / 5
target_max = np.max(heights) - dz / 5
@ -510,7 +639,7 @@ class AutofocusThing(lt.Thing):
if target_min < peak_height < target_max:
# If it is within the target range then return
return heights.tolist(), sizes.tolist()
return heights.tolist(), sharpnesses.tolist()
raise NoFocusFoundError(
"Looping autofocus couldn't converge on a focus location."
)

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

@ -179,6 +179,7 @@ class BaseCamera(OFMThing):
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
supports_focus_fom: bool = False
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise the base camera, this creates the background detectors.
@ -213,6 +214,17 @@ class BaseCamera(OFMThing):
"""Close hardware connection when the Thing context manager is closed."""
pass
@property
def focus_fom(self) -> int:
"""Return the focus figure of merit.
This returns a NotImplementedError if not supported by the camera.
To use, requires self.supports_focus_fom to be set to True.
"""
raise NotImplementedError(
"This camera does not support Figure of Merit focusing."
)
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating.

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

@ -25,7 +25,16 @@ import time
from contextlib import contextmanager
from threading import RLock
from types import TracebackType
from typing import Annotated, Any, Iterator, Literal, Mapping, Optional, Self
from typing import (
TYPE_CHECKING,
Annotated,
Any,
Iterator,
Literal,
Mapping,
Optional,
Self,
)
import numpy as np
from picamera2 import Picamera2
@ -50,6 +59,9 @@ from . import BaseCamera
from . import picamera_recalibrate_utils as recalibrate_utils
from . import picamera_tuning_file_utils as tf_utils
if TYPE_CHECKING:
from libcamera import Request
LOGGER = logging.getLogger(__name__)
SUPPORTED_CAMS_SENSOR_INFO = {
@ -123,6 +135,8 @@ class StreamingPiCamera2(BaseCamera):
generalisation.
"""
_focus_fom: int
supports_focus_fom: bool = True
tuning: dict = lt.setting(default_factory=dict, readonly=True)
"""The Raspberry PiCamera Tuning File JSON."""
@ -366,6 +380,9 @@ class StreamingPiCamera2(BaseCamera):
if self._picamera is None:
# Type narrow (error if failure)
raise RuntimeError("Failed to start Picamera")
self._picamera.pre_callback = self._on_frame_complete
if check_sensor_model:
hw_sensor_model = self._picamera.camera_properties["Model"]
if hw_sensor_model != self._sensor_info.sensor_model:
@ -374,6 +391,17 @@ class StreamingPiCamera2(BaseCamera):
f"but found {hw_sensor_model}."
)
@property
def focus_fom(self) -> int:
"""Return the focus figure of merit."""
return self._focus_fom
def _on_frame_complete(self, request: Request) -> None:
md = request.get_metadata()
fom = md.get("FocusFoM")
if fom is not None:
self._focus_fom = fom
def __enter__(self) -> Self:
"""Start streaming when the Thing context manager is opened.

156
tests/unit_tests/test_autofocus.py
View file

@ -10,7 +10,9 @@ from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.autofocus import (
AutofocusThing,
JPEGSharpnessMonitor,
NoFocusFoundError,
SharpnessMethod,
)
@ -98,3 +100,157 @@ def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes,
)
assert sharpness_monitor.focus_rel.call_count == attempts_expected
assert abs(max_loc - autofocus_thing._stage.position["z"]) < dz / 40
@pytest.fixture
def mock_camera(mocker):
"""Mock a camera to return fom properties."""
camera = mocker.MagicMock()
camera.supports_focus_fom = True
camera.focus_fom = 123
return camera
@pytest.fixture
def mock_stage(mocker):
"""Mock a stage to return position."""
stage = mocker.MagicMock()
stage.position = {"x": 0, "y": 0, "z": 0}
return stage
def test_record_defaults_to_method(mock_stage, mock_camera):
"""If record=None, only the selected method is recorded."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.FOCUS_FOM,
)
assert monitor.record == SharpnessMethod.FOCUS_FOM
def test_record_must_include_selected_method(mock_stage, mock_camera):
"""The selected autofocus metric must also be recorded."""
with pytest.raises(ValueError, match="not being recorded"):
JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.FOCUS_FOM,
record=SharpnessMethod.JPEG,
)
def test_focus_fom_requires_camera_support(mock_stage, mock_camera):
"""FocusFoM recording requires camera support."""
mock_camera.supports_focus_fom = False
with pytest.raises(ValueError, match="doesn't support"):
JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.FOCUS_FOM,
)
def test_jpeg_sizes_property_requires_recording(mock_stage, mock_camera):
"""JPEG sizes should error if JPEG recording disabled."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.FOCUS_FOM,
)
with pytest.raises(ValueError, match="JPEG sizes are not being recorded"):
_ = monitor.jpeg_sizes
def test_focus_fom_property_requires_recording(mock_stage, mock_camera):
"""FocusFoM values should error if FOM recording disabled."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.JPEG,
)
with pytest.raises(ValueError, match="FocusFoM values are not being recorded"):
_ = monitor.focus_foms
@pytest.mark.parametrize(
("method", "expected"),
[
(SharpnessMethod.JPEG, [100, 200, 300]),
(SharpnessMethod.FOCUS_FOM, [1, 5, 2]),
],
)
def test_move_data_uses_selected_method(
method,
expected,
mock_stage,
mock_camera,
):
"""move_data should select the correct sharpness metric."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=method,
record=SharpnessMethod.JPEG + SharpnessMethod.FOCUS_FOM,
)
monitor._jpeg_times = [1.0, 2.0, 3.0]
monitor._jpeg_sizes = [100, 200, 300]
monitor._focus_foms = [1, 5, 2]
monitor._stage_times = [0.5, 3.5]
monitor._stage_positions = [
{"z": 0},
{"z": 100},
]
_, _, sharpnesses = monitor.move_data(0)
assert sharpnesses.tolist() == expected
def test_monitor_records_both_metrics(mock_stage, mock_camera):
"""Both JPEG and FocusFoM metrics can be recorded together."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.JPEG,
record=SharpnessMethod.JPEG + SharpnessMethod.FOCUS_FOM,
)
monitor._jpeg_sizes.append(999)
monitor._focus_foms.append(321)
assert monitor.jpeg_sizes == [999]
assert monitor.focus_foms == [321]
def test_move_data_uses_focus_fom(mock_stage, mock_camera):
"""Test that move_data returns the chosen metric."""
monitor = JPEGSharpnessMonitor(
mock_stage,
mock_camera,
method=SharpnessMethod.FOCUS_FOM,
record=SharpnessMethod.JPEG + SharpnessMethod.FOCUS_FOM,
)
# Record both sharpness metrics
monitor._jpeg_times = [1.0, 2.0, 3.0]
monitor._jpeg_sizes = [100, 200, 300]
monitor._focus_foms = [10, 99, 20]
monitor._stage_times = [0.5, 3.5]
monitor._stage_positions = [
{"z": 0},
{"z": 100},
]
# sharpnesses chosen by the chosen method
_, _, sharpnesses = monitor.move_data(0)
# ensure sharpnesses matches focus_foms
assert sharpnesses.tolist() == [10, 99, 20]