Update unit tests for capture mode.

This commit is contained in:
Julian Stirling 2026-06-02 17:22:38 +01:00
parent 9ed488e37d
commit c1915759b4
5 changed files with 102 additions and 69 deletions

View file

@ -82,6 +82,7 @@ class OFMThing(lt.Thing):
return rel_data_path return rel_data_path
# TODO Write tests for this
class RelativeDataPath(RootModel[str]): class RelativeDataPath(RootModel[str]):
"""A relative path that is validated, and can have a Thing assigned to it. """A relative path that is validated, and can have a Thing assigned to it.

View file

@ -33,8 +33,8 @@ def test_add_and_get_image():
"""Check images can be captured and retrieved.""" """Check images can be captured and retrieved."""
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image = random_image() misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image, random_metadata()) buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
returned_image, _ = mem_buf.get_image(buffer_id) returned_image, _, _ = mem_buf.get_image(buffer_id)
# It is the same image # It is the same image
assert misc_image is returned_image assert misc_image is returned_image
# It is now removed from memory # It is now removed from memory
@ -46,12 +46,12 @@ def test_add_and_get_image_twice():
"""Check images can be retrieved twice if remove flag set false.""" """Check images can be retrieved twice if remove flag set false."""
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image = random_image() misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image, random_metadata()) buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
returned_image, _ = mem_buf.get_image(buffer_id, remove=False) returned_image, _, _ = mem_buf.get_image(buffer_id, remove=False)
# It is the same image # It is the same image
assert misc_image is returned_image assert misc_image is returned_image
# It is still in memory # It is still in memory
returned_image, _ = mem_buf.get_image(buffer_id) returned_image, _, _ = mem_buf.get_image(buffer_id)
assert misc_image is returned_image assert misc_image is returned_image
# It is now removed from memory # It is now removed from memory
with pytest.raises(NoImageInMemoryError): with pytest.raises(NoImageInMemoryError):
@ -62,8 +62,8 @@ def test_get_without_id():
"""Check images can be captured and retrieved without ID.""" """Check images can be captured and retrieved without ID."""
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image = random_image() misc_image = random_image()
mem_buf.add_image(misc_image, random_metadata()) mem_buf.add_image(misc_image, random_metadata(), "standard")
returned_image, _ = mem_buf.get_image() returned_image, _, _ = mem_buf.get_image()
# It is the same image # It is the same image
assert misc_image is returned_image assert misc_image is returned_image
# It is now removed from memory # It is now removed from memory
@ -76,10 +76,20 @@ def test_get_two_images():
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image1 = random_image() misc_image1 = random_image()
misc_image2 = random_image() misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2) buffer_id1 = mem_buf.add_image(
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2) misc_image1, random_metadata(), "standard", buffer_max=2
returned_image1, _ = mem_buf.get_image(buffer_id1) )
returned_image2, _ = mem_buf.get_image(buffer_id2) buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=2
)
buffer_id1 = mem_buf.add_image(
misc_image1, random_metadata(), "standard", buffer_max=2
)
buffer_id2 = mem_buf.add_image(
misc_image2, random_metadata(), "standard", buffer_max=2
)
returned_image1, _, _ = mem_buf.get_image(buffer_id1)
returned_image2, _, _ = mem_buf.get_image(buffer_id2)
# It they the same images # It they the same images
assert misc_image1 is returned_image1 assert misc_image1 is returned_image1
assert misc_image2 is returned_image2 assert misc_image2 is returned_image2
@ -95,11 +105,11 @@ def test_get_two_images_without_setting_buffer_size():
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image1 = random_image() misc_image1 = random_image()
misc_image2 = random_image() misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata()) buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), "standard")
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata()) buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), "standard")
with pytest.raises(NoImageInMemoryError): with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1) mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2) returned_image2, _, _ = mem_buf.get_image(buffer_id2)
# Image 2 the expected image # Image 2 the expected image
assert misc_image2 is returned_image2 assert misc_image2 is returned_image2
@ -110,16 +120,20 @@ def test_buffer_size_changing():
misc_image1 = random_image() misc_image1 = random_image()
misc_image2 = random_image() misc_image2 = random_image()
misc_image3 = random_image() misc_image3 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=3) buffer_id1 = mem_buf.add_image(
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=3) misc_image1, random_metadata(), "standard", buffer_max=3
)
buffer_id2 = mem_buf.add_image(
misc_image2, random_metadata(), "standard", buffer_max=3
)
# Third capture doesn't set buffer size, so it will be reset # Third capture doesn't set buffer size, so it will be reset
buffer_id3 = mem_buf.add_image(misc_image3, random_metadata()) buffer_id3 = mem_buf.add_image(misc_image3, random_metadata(), "standard")
# As buffer size was reset, images 1 and 2 are deleted # As buffer size was reset, images 1 and 2 are deleted
with pytest.raises(NoImageInMemoryError): with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1) mem_buf.get_image(buffer_id1)
with pytest.raises(NoImageInMemoryError): with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id2) mem_buf.get_image(buffer_id2)
returned_image3, _ = mem_buf.get_image(buffer_id3) returned_image3, _, _ = mem_buf.get_image(buffer_id3)
# Image 3 the expected image # Image 3 the expected image
assert misc_image3 is returned_image3 assert misc_image3 is returned_image3
@ -129,9 +143,9 @@ def test_capture_two_images_get_without_id():
mem_buf = CameraMemoryBuffer() mem_buf = CameraMemoryBuffer()
misc_image1 = random_image() misc_image1 = random_image()
misc_image2 = random_image() misc_image2 = random_image()
mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2) mem_buf.add_image(misc_image1, random_metadata(), "standard", buffer_max=2)
mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2) mem_buf.add_image(misc_image2, random_metadata(), "standard", buffer_max=2)
returned_image, _ = mem_buf.get_image() returned_image, _, _ = mem_buf.get_image()
# When buffer_id is not specified, the most recent image (image2) is expected to # When buffer_id is not specified, the most recent image (image2) is expected to
# be retrieved # be retrieved
assert returned_image is misc_image2 assert returned_image is misc_image2
@ -148,7 +162,9 @@ def test_buffer_size_respected():
buffer_ids = [] buffer_ids = []
for _i in range(10): for _i in range(10):
image = random_image() image = random_image()
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=5) buffer_id = mem_buf.add_image(
image, random_metadata(), "standard", buffer_max=5
)
images.append(image) images.append(image)
buffer_ids.append(buffer_id) buffer_ids.append(buffer_id)
@ -157,7 +173,7 @@ def test_buffer_size_respected():
with pytest.raises(NoImageInMemoryError): with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id) mem_buf.get_image(buffer_id)
else: else:
returned_image, _ = mem_buf.get_image(buffer_id) returned_image, _, _ = mem_buf.get_image(buffer_id)
assert image is returned_image assert image is returned_image
@ -169,7 +185,9 @@ def test_clear_buffer():
buffer_ids = [] buffer_ids = []
for _i in range(10): for _i in range(10):
image = random_image() image = random_image()
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=10) buffer_id = mem_buf.add_image(
image, random_metadata(), "standard", buffer_max=10
)
images.append(image) images.append(image)
buffer_ids.append(buffer_id) buffer_ids.append(buffer_id)
@ -192,7 +210,7 @@ def test_get_metadata_too():
for _i in range(10): for _i in range(10):
image = random_image() image = random_image()
metadata = random_metadata() metadata = random_metadata()
buffer_id = mem_buf.add_image(image, metadata, buffer_max=10) buffer_id = mem_buf.add_image(image, metadata, "standard", buffer_max=10)
images.append(image) images.append(image)
metadatas.append(metadata) metadatas.append(metadata)
buffer_ids.append(buffer_id) buffer_ids.append(buffer_id)
@ -202,6 +220,19 @@ def test_get_metadata_too():
# Check both image and metadata # Check both image and metadata
for image, metadata, buffer_id in zipped: for image, metadata, buffer_id in zipped:
returned_image, returned_metadata = mem_buf.get_image(buffer_id) returned_image, returned_metadata, _ = mem_buf.get_image(buffer_id)
assert image is returned_image assert image is returned_image
assert metadata is returned_metadata assert metadata is returned_metadata
def test_mode_is_returned():
"""Capture 10 images with metadata and check metadata is returned as expected."""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "standard")
_, _, mode = mem_buf.get_image(buffer_id)
assert mode == "standard"
buffer_id = mem_buf.add_image(misc_image, random_metadata(), "foobar")
_, _, mode = mem_buf.get_image(buffer_id)
assert mode == "foobar"

View file

@ -4,6 +4,7 @@ import itertools
import logging import logging
import pytest import pytest
from PIL import Image
from pydantic import BaseModel from pydantic import BaseModel
import labthings_fastapi as lt import labthings_fastapi as lt
@ -11,6 +12,7 @@ from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_planners import SmartSpiral from openflexure_microscope_server.scan_planners import SmartSpiral
from openflexure_microscope_server.stitching import StitchingSettings from openflexure_microscope_server.stitching import StitchingSettings
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.autofocus import SmartStackParams from openflexure_microscope_server.things.autofocus import SmartStackParams
from openflexure_microscope_server.things.camera_stage_mapping import csm_img_to_stage from openflexure_microscope_server.things.camera_stage_mapping import csm_img_to_stage
from openflexure_microscope_server.things.scan_workflows import ( from openflexure_microscope_server.things.scan_workflows import (
@ -44,8 +46,6 @@ def test_partial_base_classes():
bad_workflow = create_thing_without_server(BadWorkflow) bad_workflow = create_thing_without_server(BadWorkflow)
settings = MinimalSettings() settings = MinimalSettings()
with pytest.raises(NotImplementedError):
bad_workflow.check_before_start(settings)
with pytest.raises(NotImplementedError): with pytest.raises(NotImplementedError):
bad_workflow.ready bad_workflow.ready
@ -69,7 +69,17 @@ def test_partial_base_classes():
@pytest.fixture @pytest.fixture
def histo_workflow(): def histo_workflow():
"""Return a HistoScanWorkflow thing with slots mocked.""" """Return a HistoScanWorkflow thing with slots mocked."""
return create_thing_without_server(HistoScanWorkflow, mock_all_slots=True) workflow = create_thing_without_server(HistoScanWorkflow, mock_all_slots=True)
workflow._cam.capture_modes = {"standard": "Mock"}
workflow._cam._capture_image.return_value = Image.new("RGB", (1111, 1222))
return workflow
def test_histo_workflow_save_resolution(histo_workflow):
"""Check that the camera is used to get the save resolution."""
width, height = histo_workflow._get_save_resolution()
assert width == 1111
assert height == 1222
# Use itertools to iterate over every true/false permutation # Use itertools to iterate over every true/false permutation
@ -117,7 +127,12 @@ def test_histo_workflow_settings_generation(histo_workflow, mocker):
mocker.patch.object( mocker.patch.object(
histo_workflow, "_calc_displacement_from_overlap", return_value=(123, 456) histo_workflow, "_calc_displacement_from_overlap", return_value=(123, 456)
) )
workflow_settings, stitching_settings = histo_workflow.all_settings("/this/img_dir")
img_dir = RelativeDataPath("this/img_dir")
workflow_settings, stitching_settings, save_res = histo_workflow.all_settings(
img_dir
)
assert save_res == (1111, 1222)
## Check type ## Check type
assert isinstance(workflow_settings, HistoScanSettingsModel) assert isinstance(workflow_settings, HistoScanSettingsModel)
assert isinstance(stitching_settings, StitchingSettings) assert isinstance(stitching_settings, StitchingSettings)
@ -137,7 +152,7 @@ def test_histo_workflow_settings_generation(histo_workflow, mocker):
assert workflow_settings.dx == 123 assert workflow_settings.dx == 123
assert workflow_settings.dy == 456 assert workflow_settings.dy == 456
# And that the input image dir is passed to stack the stack parameter for saving # And that the input image dir is passed to stack the stack parameter for saving
assert workflow_settings.capture_params.images_dir == "/this/img_dir" assert workflow_settings.capture_params.images_dir.root == "this/img_dir"
def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker): def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker):
@ -148,8 +163,9 @@ def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker
# Different when False # Different when False
histo_workflow.equal_distances = False histo_workflow.equal_distances = False
workflow_settings, _stitching_settings = histo_workflow.all_settings( img_dir = RelativeDataPath("this/img_dir")
"/this/img_dir" workflow_settings, _stitching_settings, _save_res = histo_workflow.all_settings(
img_dir
) )
assert workflow_settings.dx == 123 assert workflow_settings.dx == 123
@ -157,8 +173,8 @@ def test_histo_workflow_settings_generation_equal_overlap(histo_workflow, mocker
# Same when set True # Same when set True
histo_workflow.equal_distances = True histo_workflow.equal_distances = True
workflow_settings, _stitching_settings = histo_workflow.all_settings( workflow_settings, _stitching_settings, _save_res = histo_workflow.all_settings(
"/this/img_dir" img_dir
) )
assert workflow_settings.dx == 123 assert workflow_settings.dx == 123
@ -455,9 +471,8 @@ def test_correlation_resize(histo_workflow, save_res, expected_resize):
target area, taking the square root, and rounding to the nearest integer N. Then target area, taking the square root, and rounding to the nearest integer N. Then
correlation_resize = 1 / N. correlation_resize = 1 / N.
""" """
histo_workflow.save_resolution = save_res
histo_workflow.overlap = 0.1 histo_workflow.overlap = 0.1
settings = histo_workflow._get_stitching_settings_model() settings = histo_workflow._get_stitching_settings_model(save_res)
assert isinstance(settings, StitchingSettings) assert isinstance(settings, StitchingSettings)
assert settings.correlation_resize == expected_resize assert settings.correlation_resize == expected_resize

View file

@ -413,13 +413,14 @@ def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker):
smart_scan_thing._workflow.all_settings.return_value = ( smart_scan_thing._workflow.all_settings.return_value = (
MockWorkflowSettingModel(), MockWorkflowSettingModel(),
StitchingSettings(correlation_resize=0.5, overlap=0.45), StitchingSettings(correlation_resize=0.5, overlap=0.45),
(1640, 1232),
) )
smart_scan_thing._workflow.save_resolution = (1640, 1232)
mock_ongoing_scan = mocker.Mock() mock_ongoing_scan = mocker.Mock()
mock_ongoing_scan.name = MOCK_SCAN_NAME mock_ongoing_scan.name = MOCK_SCAN_NAME
mock_ongoing_scan.images_dir = MOCK_SCAN_DIR mock_ongoing_scan.images_dir = MOCK_SCAN_DIR
smart_scan_thing._ongoing_scan = mock_ongoing_scan smart_scan_thing._ongoing_scan = mock_ongoing_scan
smart_scan_thing._data_dir = "scans"
yield smart_scan_thing yield smart_scan_thing

View file

@ -8,10 +8,13 @@ import numpy as np
import pytest import pytest
from hypothesis import given from hypothesis import given
from hypothesis import strategies as st from hypothesis import strategies as st
from PIL import Image
from pydantic import ValidationError
from labthings_fastapi.testing import create_thing_without_server from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import IMAGE_REGEX from openflexure_microscope_server.scan_directories import IMAGE_REGEX
from openflexure_microscope_server.things import RelativeDataPath
from openflexure_microscope_server.things.autofocus import ( from openflexure_microscope_server.things.autofocus import (
EXTRA_STACK_CAPTURES, EXTRA_STACK_CAPTURES,
AutofocusThing, AutofocusThing,
@ -262,6 +265,10 @@ def histo_scan_workflow():
workflow._csm.calibration_required = False workflow._csm.calibration_required = False
workflow._csm.convert_image_to_stage_coordinates = lambda x, y: {"x": x, "y": y} workflow._csm.convert_image_to_stage_coordinates = lambda x, y: {"x": x, "y": y}
# And set up camera
workflow._cam.capture_modes = {"standard": "Mock"}
workflow._cam._capture_image.return_value = Image.new("RGB", (1000, 1000))
return workflow return workflow
@ -341,7 +348,7 @@ def test_coercing_stack_save_ims(
@pytest.mark.parametrize("pass_on", [1, 2, 3, 4]) @pytest.mark.parametrize("pass_on", [1, 2, 3, 4])
def test_run_smart_stack(pass_on, histo_scan_workflow, autofocus_thing, mocker): def test_run_smart_stack(pass_on, histo_scan_workflow, autofocus_thing, mocker):
"""Test Running smart stack with the stack passing on different attempts.""" """Test Running smart stack with the stack passing on different attempts."""
scan_settings, _ = histo_scan_workflow.all_settings(images_dir="dummy") scan_settings, _, _ = histo_scan_workflow.all_settings(RelativeDataPath("dummy"))
assert scan_settings.smart_stack_params.max_attempts == 3 assert scan_settings.smart_stack_params.max_attempts == 3
# Set up returns from z-stack # Set up returns from z-stack
@ -878,37 +885,15 @@ def test_invalid_stack_settling_raises():
@pytest.mark.parametrize( @pytest.mark.parametrize(
("bad_path", "match_err"), ("bad_path", "error_type"),
[ [
("", "String should have at least 1 character"), (None, TypeError),
(None, "Input should be a valid string"), (67, TypeError),
(67, "Input should be a valid string"), ("../dangerous", ValidationError),
("/usr/bin/bash", ValidationError),
], ],
) )
def test_invalid_capture_dir_raises(bad_path, match_err): def test_invalid_capture_dir_raises(bad_path, error_type):
"""Test basic stack raises expected error for bad image dir paths.""" """Test basic stack raises expected error for bad image dir paths."""
with pytest.raises(ValueError, match=match_err): with pytest.raises(error_type):
# TODO set images dir correctly as a RelDataPath CaptureParams(images_dir=bad_path, capture_mode="foo")
CaptureParams(images_dir=bad_path, save_resolution=(20, 20))
@pytest.mark.parametrize(
("bad_res", "match_err"),
[
((-100, 50), "Input should be greater than or equal to 1"),
((20, 0), "Input should be greater than or equal to 1"),
("", "Input should be a valid tuple"),
(None, "Input should be a valid tuple"),
(67, "Input should be a valid tuple"),
(
["path"],
"Input should be a valid integer, unable to parse string as an integer",
),
((20, 20, 20), "Tuple should have at most 2 items"),
],
)
def test_invalid_capture_res_raises(bad_res, match_err):
"""Test basic stack raises expected error for invalid save resolutions."""
with pytest.raises(ValueError, match=match_err):
# TODO set images dir correctly as a RelDataPath
CaptureParams(images_dir="dummy", save_resolution=bad_res)