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

Merge branch 'scanning-stability' into 'v3'

Browse source 
Improve the stability of scanning

Closes #599 and #600

See merge request openflexure/openflexure-microscope-server!434
This commit is contained in:
Joe Knapper 2025-12-02 15:47:03 +00:00
commit 99afe89c10
10 changed files with 874 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

97
tests/test_autofocus.py Normal file
View file

@ -0,0 +1,97 @@
"""Tests for the autofoucs logic.
This doesn't check the behaviour of the JPEG shaprness monitor.
"""
import pytest
import numpy as np
from openflexure_microscope_server.things.autofocus import (
AutofocusThing,
NoFocusFoundError,
)
def fake_sharpness_data(
dz: int, start_z: int, max_loc: int, length: int = 41
) -> tuple[list[float], np.ndarray, np.ndarray]:
"""Create some fake data for the shapeness.
The highest returned sharpness is closest to max_loc
"""
# Some fake timestamps
times = [i / 10 + 100000 for i in range(length)]
img_dz = dz / (length - 1)
heights = [round(start_z + i * img_dz) for i in range(length)]
# Sharpnesses fall off linearly in this model.
sharpnesses = [10 * dz - abs(max_loc - h) for h in heights]
return times, np.array(heights), np.array(sharpnesses)
@pytest.mark.parametrize(
("start_z", "max_loc", "centre", "attempts_expected", "passes"),
[
# To complete, the max must be in the central 1200, so -600 to 600 when looping
# from -1000 to 1000
(0, 550, True, 1, True), # Found in loop1 from -1000 to 1000
(0, 650, True, 2, True), # Just outside the limit in loop1
(0, 1300, True, 2, True), # Found in loop2 from 0 to 2000
(0, 1300, False, 1, True), # Found in loop1 from 0 to 2000 (as start="base")
(0, -1300, True, 2, True), # Found in loop2 from 0 to 2000
(0, 7300, True, 8, True), # Found in loop8 from 6000 to 8000
(0, 9300, True, 10, True), # Found in loop10 from 8000 to 10000
(0, 9900, True, 10, False), # Still not central in loop 10, doesn't pass
],
)
def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes, mocker):
"""Test the high level looping autofocus algorithm."""
dz = 2000
# Make a mock stage where move_absolute abs and relative updates the position counter.
stage = mocker.Mock()
stage.position = {"x": 0, "y": 0, "z": start_z}
def set_pos(**kwargs: int) -> None:
"""Move absolute should update position. So make a side effect for the mock."""
for axis, value in kwargs.items():
stage.position[axis] = value
def adjust_pos(**kwargs: int) -> None:
"""Move relative should update position. So make a side effect for the mock."""
for axis, value in kwargs.items():
stage.position[axis] += value
stage.move_absolute.side_effect = set_pos
stage.move_relative.side_effect = adjust_pos
# Make a mock sharpness monitor that can generate sharpness data.
sharpness_monitor = mocker.MagicMock()
sharpness_monitor.focus_rel.return_value = (0, 0)
def return_sharpness(*_args) -> tuple[list[float], np.ndarray, np.ndarray]:
"""Generate sharpnesses based on parameterised input, and mock stage position."""
return fake_sharpness_data(
dz=dz,
start_z=stage.position["z"],
max_loc=max_loc,
)
sharpness_monitor.move_data.side_effect = return_sharpness
autofocus_thing = AutofocusThing()
if passes:
autofocus_thing.looping_autofocus(
stage=stage,
sharpness_monitor=sharpness_monitor,
dz=dz,
start="centre" if centre else "base",
)
else:
with pytest.raises(NoFocusFoundError):
autofocus_thing.looping_autofocus(
stage=stage,
sharpness_monitor=sharpness_monitor,
dz=dz,
start="centre" if centre else "base",
)
assert sharpness_monitor.focus_rel.call_count == attempts_expected
assert abs(max_loc - stage.position["z"]) < dz / 40

140
tests/test_background_detectors.py
View file

@ -15,6 +15,9 @@ from openflexure_microscope_server.background_detect import (
ChannelDistributions,
ColourChannelDetectSettings,
ColourChannelDetectLUV,
_chunked_stds,
ChannelDeviationLUV,
ChannelBlankError,
)
RNG = np.random.default_rng()
@ -181,3 +184,140 @@ def test_colour_channel_luv_load_bad_data():
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.
:return: The image, and the (8,8,3) array of stds
"""
if blank_channels is None:
blank_channels = []
n_rows = 8
n_cols = 8
chunk_h = 10
chunk_w = 10
# Precompute chunk stds and construct the final image
expected_stds = np.zeros((n_rows, n_cols, 3), dtype=float)
img = np.zeros((n_rows * chunk_h, n_cols * chunk_w, 3), dtype=np.uint8)
for i in range(n_rows):
for j in range(n_cols):
for channel in range(3):
if channel in blank_channels:
chunk = np.zeros((chunk_h, chunk_w), dtype=np.uint8)
else:
chunk = 9.8 * np.ones((chunk_h, chunk_w))
chunk += RNG.normal(scale=magnitude, size=(chunk_h, chunk_w))
chunk = chunk.astype(np.uint8)
# Store its std
expected_stds[i, j, channel] = np.std(chunk)
# Insert chunk into the final large image
y0, y1 = i * chunk_h, (i + 1) * chunk_h
x0, x1 = j * chunk_w, (j + 1) * chunk_w
img[y0:y1, x0:x1, channel] = chunk
return img, expected_stds
def test_chunked_stds_with_precomputed_chunk_stds():
"""Test _chunked_stds returns the answer calculated when making a patchwork image."""
img, expected_stds = create_patchwork_image()
# Run the function under test
result = _chunked_stds(img, n_rows=8, n_cols=8)
# Compare
np.testing.assert_allclose(result, expected_stds, rtol=1e-6, atol=1e-12)
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()
# 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)
for magnitude in [0.2, 1, 10]:
# Create an image and set it as background
img, expected_stds = create_patchwork_image(magnitude=magnitude)
cd_luv.set_background(img)
# Do a somewhat verbose checking for clarity
for channel in range(3):
# Saved std
channel_std = cd_luv.background_data.standard_deviations[channel]
# Expected median
channel_median = np.median(expected_stds[:, :, channel])
# If the median is above the minimum allowed then it should be returned
if channel_median > cd_luv.min_stds[channel]:
assert np.isclose(channel_std, channel_median, rtol=1e-6, atol=1e-12)
else:
# If not the minimum is returned.
assert channel_std == cd_luv.min_stds[channel]
# Also check that if channels are blank then an error is thrown
img, _expected_stds = create_patchwork_image(magnitude=1, blank_channels=[1, 2])
with pytest.raises(ChannelBlankError):
cd_luv.set_background(img)
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()
# No background data so it is not ready and will error if image_is_sample is called.
assert not cd_luv.status.ready
with pytest.raises(MissingBackgroundDataError):
cd_luv.image_is_sample(background_image)
cd_luv.settings.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
is_sample, message = cd_luv.image_is_sample(background_image)
assert is_sample
assert message == r"10.0% sample"
def test_channel_deviation_luv_get_sample_coverage(background_image, mocker):
"""Check _get_sample_coverage returns the values expected."""
cd_luv = 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",
return_value=fake_stds,
)
# Create fake background
cd_luv.background_data = ChannelDistributions(
means=[0, 0, 0], standard_deviations=[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
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]
)
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]
)
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

344
tests/test_stack.py
View file

@ -1,7 +1,4 @@
"""Tests for the smart/fast stacking.
Currently these tests don't test the Thing itself, just surrounding functionality
"""
"""Tests for the smart and fast stacking."""
from typing import Optional
import tempfile
@ -23,6 +20,10 @@ from openflexure_microscope_server.things.autofocus import (
MAX_TEST_IMAGE_COUNT,
_get_capture_by_id,
_get_capture_index_by_id,
EXTRA_STACK_CAPTURES,
NotAPeakError,
_get_peak_turning_point,
_count_turning_points,
)
from openflexure_microscope_server.scan_directories import IMAGE_REGEX
@ -352,3 +353,338 @@ def test_coercing_stack_save_ims(
assert stack_params.images_to_save == coerced_save_ims
# Check that the setting in the Thing was updated to the coerced value
assert stack_params.images_to_save == autofocus_thing.stack_images_to_save
@pytest.mark.parametrize("pass_on", [1, 2, 3, 4])
def test_run_smart_stack(pass_on, autofocus_thing, mocker):
"""Test Running smart stack with the stack passing on different attempts."""
cam = mocker.Mock()
stage = mocker.Mock()
sharpness_monitor = mocker.MagicMock()
stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
)
assert stack_params.max_attempts == 3
# Set up returns from z-stack
fake_captures = [
CaptureInfo(
buffer_id="first", position={"x": 0, "y": 0, "z": -99}, sharpness=123
),
CaptureInfo(
buffer_id="pick_me", position={"x": 0, "y": 0, "z": 555}, sharpness=456
),
CaptureInfo(
buffer_id="last", position={"x": 0, "y": 0, "z": 999}, sharpness=123
),
]
successful_return = (True, fake_captures, "pick_me")
failed_return = (False, fake_captures, "pick_me")
return_list = [failed_return] * (pass_on - 1) + [successful_return]
# Mock z_stack and looping_autofocus
autofocus_thing.z_stack = mocker.Mock(side_effect=return_list)
autofocus_thing.looping_autofocus = mocker.Mock()
# Run it
success, final_z = autofocus_thing.run_smart_stack(
cam=cam,
stage=stage,
sharpness_monitor=sharpness_monitor,
stack_parameters=stack_params,
save_on_failure=False,
check_turning_points=True,
)
# Only passes if the attempt it passes on is less than max attempts
assert success == (pass_on <= stack_params.max_attempts)
# Final z is the one from the id returned by the stack "pick_me"
assert final_z == 555
# z_stack should run up until the time it passes. Running no more than max_attempts
n_stacks = min(pass_on, stack_params.max_attempts)
assert autofocus_thing.z_stack.call_count == n_stacks
# Move absolute should be 1 less time that the number of times z_stack_run
assert stage.move_absolute.call_count == n_stacks - 1
# As should looping autofocus
assert autofocus_thing.looping_autofocus.call_count == n_stacks - 1
# Check rest stack is moving to the first image in the stack.
if n_stacks > 1:
assert stage.move_absolute.call_args.kwargs["z"] == -99
# Mock called to save image
assert cam.save_from_memory.call_count == (1 if success else 0)
def setup_and_run_z_stack(check_returns, check_turning_points, autofocus_thing, mocker):
"""Set up a z_stack, run it, and return the result.
:param check_returns: The return values from check_stack_result. Note that if this
is a list, it will be set as a side effect (and should be a list of tuples of
results). If it a tuple (or anything else), it is set as a return value.
"""
stack_params = autofocus_thing.create_stack_params(
autofocus_dz=2000,
images_dir="/this/is/fake",
save_resolution=(1640, 1232),
logger=LOGGER,
)
stack_params.settling_time = 0 # Don't settle or tests take forever.
stage = mocker.Mock()
cam = mocker.Mock()
autofocus_thing.capture_stack_image = mocker.Mock()
if isinstance(check_returns, list):
autofocus_thing.check_stack_result = mocker.Mock(side_effect=check_returns)
else:
autofocus_thing.check_stack_result = mocker.Mock(return_value=check_returns)
return autofocus_thing.z_stack(
stack_parameters=stack_params,
check_turning_points=check_turning_points,
cam=cam,
stage=stage,
)
def test_z_stack_turning_toggle_passed(autofocus_thing, mocker):
"""Check that the toggling of turning points is passed to the check."""
check_returns = ("success", "mock_id")
for check_turning in [True, False]:
setup_and_run_z_stack(check_returns, check_turning, autofocus_thing, mocker)
check_kwargs = autofocus_thing.check_stack_result.call_args.kwargs
assert check_kwargs["check_turning_points"] == check_turning
def test_z_stack_returns_on_success_and_restart(autofocus_thing, mocker):
"""Check that if the check returns success or restart then the stack exits with correct return value."""
for result in ["success", "restart"]:
check_returns = (result, "mock_id")
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker)
assert autofocus_thing.check_stack_result.call_count == 1
# Check the number of images taken is exactly the call count.
ims_taken = autofocus_thing.capture_stack_image.call_count
assert ims_taken == autofocus_thing.stack_min_images_to_test
# And the result is as expected.
assert ret[0] == (result == "success")
def test_z_stack_exits_if_focus_never_found(autofocus_thing, mocker):
"""Check that if the check returns continue the stack exits eventually with a failure."""
check_returns = ("continue", "mock_id")
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker)
assert autofocus_thing.check_stack_result.call_count == EXTRA_STACK_CAPTURES + 1
# Check the number of images taken is the maximum possible, set by the min images to
# test and the number of extra images that can be taken
ims_taken = autofocus_thing.capture_stack_image.call_count
max_ims = autofocus_thing.stack_min_images_to_test + EXTRA_STACK_CAPTURES
assert ims_taken == max_ims
# And the result is as expected.
assert not ret[0]
def test_z_stack_return(autofocus_thing, mocker):
"""Check z-stack returns as expected for more complex cases the fixed results above."""
for i in range(2, EXTRA_STACK_CAPTURES):
check_returns = [
("restart" if j == i - 1 else "continue", f"id_{j}") for j in range(i)
]
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker)
# Calculate images taken
images_taken = autofocus_thing.stack_min_images_to_test + i - 1
assert autofocus_thing.capture_stack_image.call_count == images_taken
# Check it reports a failure
assert not ret[0]
# Repeat ending with a success rather than a failure
check_returns = [
("success" if j == i - 1 else "continue", f"id_{j}") for j in range(i)
]
ret = setup_and_run_z_stack(check_returns, True, autofocus_thing, mocker)
# Calculate images taken
assert autofocus_thing.capture_stack_image.call_count == images_taken
# Check it reports a success
assert ret[0]
def test_capture_stack_image(autofocus_thing, mocker):
"""Check that capture stack image calls the expected functions and returns the expected data."""
stage = mocker.Mock()
stage.position = {"x": 123, "y": 456, "z": 789}
cam = mocker.Mock()
cam.capture_to_memory.return_value = "fake_buffer_id"
cam.grab_jpeg_size.return_value = 54321
buffer_max = 11
info = autofocus_thing.capture_stack_image(
cam=cam, stage=stage, buffer_max=buffer_max
)
assert cam.capture_to_memory.call_count == 1
assert cam.grab_jpeg_size.call_count == 1
assert info.buffer_id == "fake_buffer_id"
assert info.position == {"x": 123, "y": 456, "z": 789}
assert info.sharpness == 54321
def mock_capture(buffer_id: int, sharpness: int) -> CaptureInfo:
"""Create a CaptureInfo instance with a dummy position."""
return CaptureInfo(
buffer_id=buffer_id,
position={"x": 0, "y": 0, "z": buffer_id},
sharpness=sharpness,
)
def test_check_stack_single_image_returns_success(autofocus_thing):
"""A single image is always successful."""
captures = [mock_capture("mock-id", 10)]
result, cap_id = autofocus_thing.check_stack_result(
captures, check_turning_points=False
)
assert result == "success"
assert cap_id == "mock-id"
@pytest.mark.parametrize(
("sharpnesses", "expected"),
[
([5, 10, 3], "success"),
([10, 4, 2], "restart"),
([1, 2, 10], "continue"),
],
)
def test_check_stack_three_image_logic(sharpnesses, expected, autofocus_thing):
"""For 3 images, success is the highest one is central."""
captures = [mock_capture(i, s) for i, s in enumerate(sharpnesses)]
result, _ = autofocus_thing.check_stack_result(captures, check_turning_points=False)
assert result == expected
def _run_check_stack_with_good_peak(autofocus_thing, count_turnings=False):
"""Run check stack on a good peak that should pass, and return the result.
This can be used to check how other mocked results of subfunctions affects the
result.
"""
# Create an obvious peak that would normally pass.
sharpnesses = [1, 2, 4, 7, 12, 7, 4, 2, 1]
captures = [mock_capture(i, s) for i, s in enumerate(sharpnesses)]
result, cap_id = autofocus_thing.check_stack_result(
captures, check_turning_points=count_turnings
)
# Nothing a mocked function does should change which is the sharpest image.
assert cap_id == 4
return result
def test_check_stack_continues_if_no_tuning_point(autofocus_thing, mocker):
"""Check that continue is returned if no turning point is found."""
# Mock to simulate not finding a peak
mocker.patch(
"openflexure_microscope_server.things.autofocus._get_peak_turning_point",
side_effect=NotAPeakError,
)
result = _run_check_stack_with_good_peak(autofocus_thing)
# Check that the NotAPeakError causes it to continue instead.
assert result == "continue"
@pytest.mark.parametrize(
("location", "expected"),
[
(-10, "restart"), # Restart if lower than 1.5 (halfway between im 2 and 3)
(-1, "restart"),
(0, "restart"),
(1, "restart"),
(1.49, "restart"),
(1.5, "success"), # Success up to 6.5 (as we have 9 images, final index is 8)
(2.5, "success"),
(4.5, "success"),
(6.5, "success"),
(6.51, "continue"), # Continue if thrung point is after 6.5
(7, "continue"),
(8.1, "continue"),
(123, "continue"),
],
)
def test_check_stack_affected_by_turning_point_location(
location, expected, autofocus_thing, mocker
):
"""Check that the turning point location affects the return as expected."""
# Mock to give the turning point location specified
mocker.patch(
"openflexure_microscope_server.things.autofocus._get_peak_turning_point",
return_value=location,
)
result = _run_check_stack_with_good_peak(autofocus_thing)
assert result == expected
def test_check_stack_affected_by_number_of_turning_points(autofocus_thing, mocker):
"""Check that the turning point location affects the return as expected."""
# Set the turning point to the centre
mocker.patch(
"openflexure_microscope_server.things.autofocus._get_peak_turning_point",
return_value=5,
)
mocker.patch(
"openflexure_microscope_server.things.autofocus._count_turning_points",
return_value=1,
)
result = _run_check_stack_with_good_peak(autofocus_thing, count_turnings=True)
# Successful with 1 peak
assert result == "success"
# Change return to be 2 peaks
mocker.patch(
"openflexure_microscope_server.things.autofocus._count_turning_points",
return_value=2,
)
result = _run_check_stack_with_good_peak(autofocus_thing, count_turnings=True)
# Continue with 2 peaks
assert result == "continue"
# Unless this check is turned off
result = _run_check_stack_with_good_peak(autofocus_thing, count_turnings=False)
assert result == "success"
def test_get_peak_turning_point():
"""Check that the peak fitting returns expected value (or error)."""
with pytest.raises(NotAPeakError):
_get_peak_turning_point(np.ones(9))
linear = np.arange(9)
u_shape = 2 * (linear - 4) ** 2 + 17
peak = -2 * (linear - 4) ** 2 + 55
with pytest.raises(NotAPeakError):
_get_peak_turning_point(linear)
with pytest.raises(NotAPeakError):
_get_peak_turning_point(u_shape)
# Should be 4 to within a fitting error
assert abs(_get_peak_turning_point(peak) - 4) < 1e-7
def test_count_turning_points():
"""Check the turing point count works as expected."""
linear = np.arange(9)
u_shape = 2 * (linear - 4) ** 2 + 17
peak = -2 * (linear - 4) ** 2 + 55
assert _count_turning_points(np.ones(9)) == 0
assert _count_turning_points(linear) == 0
assert _count_turning_points(u_shape) == 1
assert _count_turning_points(peak) == 1
assert _count_turning_points(np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])) == 1
# Double peak is 3 points
assert _count_turning_points(np.array([1, 2, 3, 4, 2, 4, 3, 2, 1])) == 3
# But only one if the dip isn't prominent
assert _count_turning_points(np.array([1, 2, 3, 4, 3.8, 4, 3, 2, 1])) == 1