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

A fond farewell to the last of the InvocationLoggers

Browse source
This commit is contained in:
Julian Stirling 2025-12-17 15:41:22 +00:00
parent f6d4b368d0
commit be5f9630ca
5 changed files with 64 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/openflexure_microscope_server/stitching.py
View file

@ -244,7 +244,7 @@ class FinalStitcher(BaseStitcher):
First the scan_data_dict is inspected for values to allow ``overlap`` and First the scan_data_dict is inspected for values to allow ``overlap`` and
``correlation_resize`` to be set correctly, if these values are not available ``correlation_resize`` to be set correctly, if these values are not available
then default values are used, and a warning is logged to the invocation logger. then default values are used, and a warning is logged to the thing logger.
:param overlap: overlap as input to __init__ :param overlap: overlap as input to __init__
:param correlation_resize: correlation_resize as input to __init__ :param correlation_resize: correlation_resize as input to __init__

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

@ -80,13 +80,13 @@ def _scan_running(
This decorator is used by all methods in SmartScanThing that are using This decorator is used by all methods in SmartScanThing that are using
the variables set for the scan. It will throw a runtime error if the variables set for the scan. It will throw a runtime error if
self._scan_logger is not set, as all scan variables are set at self._scan_lock is not locked, as all scan variables are set at
the same time and released with the lock the same time and released with the lock
""" """
def scan_running_wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T: def scan_running_wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T:
# Only start the method is the scan logger is set """Only start the requested method if the scan is running."""
if self._scan_logger is not None: if self._scan_lock.locked():
return method(self, *args, **kwargs) return method(self, *args, **kwargs)
raise ScanNotRunningError( raise ScanNotRunningError(
"Calling a @scan_running method can only be done while a scan is running!" "Calling a @scan_running method can only be done while a scan is running!"
@ -119,13 +119,6 @@ class SmartScanThing(lt.Thing):
# Variables set by the scan # Variables set by the scan
self._latest_scan_name: Optional[str] = None self._latest_scan_name: Optional[str] = None
# Scan logger is the invocation logger labthings-fastapi creates
# when the `sample_scan` lt.action is called. It is saved as
# private class variable along with many others here.
# Access to these variables requires a scan to be running,
# any method that calls these should be decorated with
# @_scan_running
self._scan_logger: Optional[lt.deps.InvocationLogger] = None
self._autofocus: Optional[AutofocusDep] = None self._autofocus: Optional[AutofocusDep] = None
self._stage: Optional[StageDep] = None self._stage: Optional[StageDep] = None
self._cam: Optional[CameraClient] = None self._cam: Optional[CameraClient] = None
@ -138,7 +131,6 @@ class SmartScanThing(lt.Thing):
@lt.action @lt.action
def sample_scan( def sample_scan(
self, self,
logger: lt.deps.InvocationLogger,
autofocus: AutofocusDep, autofocus: AutofocusDep,
stage: StageDep, stage: StageDep,
cam: CameraClient, cam: CameraClient,
@ -156,7 +148,6 @@ class SmartScanThing(lt.Thing):
raise RuntimeError("Trying to run scan while scan is already running!") raise RuntimeError("Trying to run scan while scan is already running!")
# Set private variables for this scan # Set private variables for this scan
self._scan_logger = logger
self._autofocus = autofocus self._autofocus = autofocus
self._stage = stage self._stage = stage
self._cam = cam self._cam = cam
@ -177,7 +168,7 @@ class SmartScanThing(lt.Thing):
self._return_to_starting_position() self._return_to_starting_position()
if not isinstance(e, scan_directories.NotEnoughFreeSpaceError): if not isinstance(e, scan_directories.NotEnoughFreeSpaceError):
# Don't stitch if drive is full (already logged) # Don't stitch if drive is full (already logged)
self._scan_logger.info( self.logger.info(
"Attempting to stitch and archive the images acquired so far." "Attempting to stitch and archive the images acquired so far."
) )
self._perform_final_stitch() self._perform_final_stitch()
@ -185,7 +176,6 @@ class SmartScanThing(lt.Thing):
raise e raise e
finally: finally:
# However the scan finishes, unset all variables and release lock # However the scan finishes, unset all variables and release lock
self._scan_logger = None
self._autofocus = None self._autofocus = None
self._stage = None self._stage = None
self._cam = None self._cam = None
@ -198,7 +188,7 @@ class SmartScanThing(lt.Thing):
self._stack_params = None self._stack_params = None
# Remove any scan folders containing zero images. # Remove any scan folders containing zero images.
self.purge_empty_scans(logger=logger) self.purge_empty_scans()
@_scan_running @_scan_running
def _check_background_and_csm_set(self) -> None: def _check_background_and_csm_set(self) -> None:
@ -222,7 +212,7 @@ class SmartScanThing(lt.Thing):
"Background is not set: you need to calibrate background detection." "Background is not set: you need to calibrate background detection."
) )
else: else:
self._scan_logger.warning( self.logger.warning(
"This scan will run in a spiral from the starting point " "This scan will run in a spiral from the starting point "
f"until you cancel it, or until it has moved by {self.max_range} steps " f"until you cancel it, or until it has moved by {self.max_range} steps "
"in every direction. Make sure you watch it run to stop it leaving " "in every direction. Make sure you watch it run to stop it leaving "
@ -250,7 +240,7 @@ class SmartScanThing(lt.Thing):
if z_estimate is None: if z_estimate is None:
z_estimate = self._stage.position["z"] z_estimate = self._stage.position["z"]
self._scan_logger.info(f"Moving to {next_point}") self.logger.info(f"Moving to {next_point}")
self._stage.move_absolute( self._stage.move_absolute(
x=next_point[0], x=next_point[0],
y=next_point[1], y=next_point[1],
@ -305,19 +295,19 @@ class SmartScanThing(lt.Thing):
dx, dy = self._calc_displacement_from_test_image(overlap) dx, dy = self._calc_displacement_from_test_image(overlap)
correlation_resize = stitching.STITCHING_RESOLUTION[0] / self.save_resolution[0] correlation_resize = stitching.STITCHING_RESOLUTION[0] / self.save_resolution[0]
self._scan_logger.debug( self.logger.debug(
f"Resizing images when correlating by a factor of {correlation_resize}" f"Resizing images when correlating by a factor of {correlation_resize}"
) )
self._scan_logger.info( self.logger.info(
f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}" f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}"
) )
autofocus_dz = self.autofocus_dz autofocus_dz = self.autofocus_dz
if autofocus_dz == 0: if autofocus_dz == 0:
self._scan_logger.info("Running scan without autofocus") self.logger.info("Running scan without autofocus")
elif autofocus_dz <= 200: elif autofocus_dz <= 200:
self._scan_logger.warning( self.logger.warning(
f"Your autofocus range is {autofocus_dz} steps, which is too short to " f"Your autofocus range is {autofocus_dz} steps, which is too short to "
"attempt to focus. Running without autofocus" "attempt to focus. Running without autofocus"
) )
@ -386,13 +376,13 @@ class SmartScanThing(lt.Thing):
self._save_final_scan_data(scan_result="success") self._save_final_scan_data(scan_result="success")
except lt.exceptions.InvocationCancelledError: except lt.exceptions.InvocationCancelledError:
self._scan_logger.info("Stopping scan because it was cancelled.") self.logger.info("Stopping scan because it was cancelled.")
self._save_final_scan_data(scan_result="cancelled by user") self._save_final_scan_data(scan_result="cancelled by user")
except Exception as e: except Exception as e:
err_name = type(e).__name__ err_name = type(e).__name__
if self._scan_data is not None: if self._scan_data is not None:
self._save_final_scan_data(scan_result=f"{err_name}: {e}") self._save_final_scan_data(scan_result=f"{err_name}: {e}")
self._scan_logger.error( self.logger.error(
f"The scan stopped because of an error: {e}", f"The scan stopped because of an error: {e}",
exc_info=e, exc_info=e,
) )
@ -457,7 +447,7 @@ class SmartScanThing(lt.Thing):
new_pos_xyz, imaged=False, focused=False new_pos_xyz, imaged=False, focused=False
) )
msg = f"Skipping {new_pos_xyz} as it is {bg_message}." msg = f"Skipping {new_pos_xyz} as it is {bg_message}."
self._scan_logger.info(msg) self.logger.info(msg)
continue continue
focused, focused_height = self._autofocus.run_smart_stack( focused, focused_height = self._autofocus.run_smart_stack(
@ -482,7 +472,7 @@ class SmartScanThing(lt.Thing):
@_scan_running @_scan_running
def _return_to_starting_position(self) -> None: def _return_to_starting_position(self) -> None:
"""Return to the initial scan position, if set.""" """Return to the initial scan position, if set."""
self._scan_logger.info("Returning to starting position.") self.logger.info("Returning to starting position.")
if self._scan_data is not None: if self._scan_data is not None:
self._stage.move_absolute( self._stage.move_absolute(
**self._scan_data.starting_position, block_cancellation=True **self._scan_data.starting_position, block_cancellation=True
@ -499,20 +489,19 @@ class SmartScanThing(lt.Thing):
def _perform_final_stitch(self) -> None: def _perform_final_stitch(self) -> None:
"""Update the scan zip and perform final stitch of the data.""" """Update the scan zip and perform final stitch of the data."""
if self._scan_data.image_count <= 3: if self._scan_data.image_count <= 3:
self._scan_logger.info("Not performing a stitch as 3 or fewer images taken") self.logger.info("Not performing a stitch as 3 or fewer images taken")
return return
self._ongoing_scan.zip_files() self._ongoing_scan.zip_files()
self._scan_logger.info("Waiting for background processes to finish...") self.logger.info("Waiting for background processes to finish...")
if self._preview_stitcher is not None: if self._preview_stitcher is not None:
self._preview_stitcher.wait() self._preview_stitcher.wait()
if self._scan_data.stitch_automatically: if self._scan_data.stitch_automatically:
self._scan_logger.info("Stitching final image (may take some time)...") self.logger.info("Stitching final image (may take some time)...")
self.stitch_scan( self.stitch_scan(
logger=self._scan_logger,
scan_name=self._ongoing_scan.name, scan_name=self._ongoing_scan.name,
correlation_resize=self._scan_data.correlation_resize, correlation_resize=self._scan_data.correlation_resize,
overlap=self._scan_data.overlap, overlap=self._scan_data.overlap,
@ -621,17 +610,17 @@ class SmartScanThing(lt.Thing):
400: {"description": "An error occurred while trying to delete scan"}, 400: {"description": "An error occurred while trying to delete scan"},
}, },
) )
def delete_scan(self, scan_name: str, logger: lt.deps.InvocationLogger) -> None: def delete_scan(self, scan_name: str) -> None:
"""Delete the folder for the specified scan. """Delete the folder for the specified scan.
This endpoint allows scans to be deleted from disk. This endpoint allows scans to be deleted from disk.
Takes the scan name to delete, and the Invocation Logger :param scan_name: The name of the scan to delete
""" """
if not self._scan_dir_manager.exists(scan_name): if not self._scan_dir_manager.exists(scan_name):
logger.warning(f"Cannot find a scan of name {scan_name}") self.logger.warning(f"Cannot find a scan of name {scan_name}")
raise HTTPException(400, "Scan not found") raise HTTPException(400, "Scan not found")
deleted_scan_success = self._delete_scan(scan_name, logger) deleted_scan_success = self._delete_scan(scan_name)
if not deleted_scan_success: if not deleted_scan_success:
raise HTTPException(400, "Couldn't delete scan, check log for details") raise HTTPException(400, "Couldn't delete scan, check log for details")
@ -639,7 +628,7 @@ class SmartScanThing(lt.Thing):
"delete", "delete",
"scans", "scans",
) )
def delete_all_scans(self, logger: lt.deps.InvocationLogger) -> None: def delete_all_scans(self) -> None:
"""Delete all the scans on the microscope. """Delete all the scans on the microscope.
**This will irreversibly remove all scanned data from the **This will irreversibly remove all scanned data from the
@ -647,30 +636,30 @@ class SmartScanThing(lt.Thing):
Use with extreme caution. Use with extreme caution.
""" """
for scan_name in self._scan_dir_manager.all_scans: for scan_name in self._scan_dir_manager.all_scans:
self._delete_scan(scan_name, logger) self._delete_scan(scan_name)
@lt.action @lt.action
def purge_empty_scans(self, logger: lt.deps.InvocationLogger) -> None: def purge_empty_scans(self) -> None:
"""Delete all scan folders containing no images at the top level.""" """Delete all scan folders containing no images at the top level."""
# JSON is ignored as it's created before any images are captured # JSON is ignored as it's created before any images are captured
for scan_info in self._get_all_scan_info(): for scan_info in self._get_all_scan_info():
if scan_info.number_of_images == 0: if scan_info.number_of_images == 0:
self._delete_scan(scan_info.name, logger) self._delete_scan(scan_info.name)
def _delete_scan(self, scan_name: str, logger: lt.deps.InvocationLogger) -> bool: def _delete_scan(self, scan_name: str) -> bool:
"""Delete a scan. """Delete a scan.
This is a wrapper around scan manager's delete_scan that logs to the This is a wrapper around scan manager's delete_scan that logs to the
invocation logger id there is a problem. things logger if there is a problem.
""" """
if self._ongoing_scan is not None and scan_name == self._ongoing_scan.name: if self._ongoing_scan is not None and scan_name == self._ongoing_scan.name:
logger.error("Attempted to delete ongoing scan.") self.logger.error("Attempted to delete ongoing scan.")
return False return False
try: try:
self._scan_dir_manager.delete_scan(scan_name) self._scan_dir_manager.delete_scan(scan_name)
return True return True
except Exception as e: except Exception as e:
logger.warning( self.logger.warning(
"Attempted to delete scan " + scan_name + ", which failed." "Attempted to delete scan " + scan_name + ", which failed."
" Server sent response" + str(e) " Server sent response" + str(e)
) )
@ -723,22 +712,19 @@ class SmartScanThing(lt.Thing):
@lt.action @lt.action
def stitch_scan( def stitch_scan(
self, self,
logger: lt.deps.InvocationLogger,
scan_name: str, scan_name: str,
correlation_resize: Optional[float] = None, correlation_resize: Optional[float] = None,
overlap: Optional[float] = None, overlap: Optional[float] = None,
) -> None: ) -> None:
"""Generate a stitched image based on stage position metadata. """Generate a stitched image based on stage position metadata."""
Note that as this is a lt.action it needs the logger passed as
a variable if called from another thing action
"""
scan_data_dict = self._scan_dir_manager.get_scan_data_dict(scan_name) scan_data_dict = self._scan_dir_manager.get_scan_data_dict(scan_name)
if scan_data_dict is None: if scan_data_dict is None:
logger.warning("Couldn't read scan data - it may be missing or corrupt.") self.logger.warning(
"Couldn't read scan data - it may be missing or corrupt."
)
final_stitcher = stitching.FinalStitcher( final_stitcher = stitching.FinalStitcher(
self._scan_dir_manager.img_dir_for(scan_name), self._scan_dir_manager.img_dir_for(scan_name),
logger=logger, logger=self.logger,
overlap=overlap, overlap=overlap,
correlation_resize=correlation_resize, correlation_resize=correlation_resize,
stitch_tiff=self.stitch_tiff, stitch_tiff=self.stitch_tiff,
@ -750,7 +736,7 @@ class SmartScanThing(lt.Thing):
# Sleep for 1 second just to allow invocation logs to pass to user. # Sleep for 1 second just to allow invocation logs to pass to user.
time.sleep(1) time.sleep(1)
except SubprocessError as e: except SubprocessError as e:
self._scan_logger.error(f"Stitching failed: {e}", exc_info=e) self.logger.error(f"Stitching failed: {e}", exc_info=e)
@lt.action @lt.action
def download_zip( def download_zip(
@ -765,14 +751,13 @@ class SmartScanThing(lt.Thing):
return ZipBlob.from_file(zip_fname) return ZipBlob.from_file(zip_fname)
@lt.action @lt.action
def stitch_all_scans(self, logger: lt.deps.InvocationLogger) -> None: def stitch_all_scans(self) -> None:
"""Check the list of scans, and stitch any that don't have a DZI associated with it. """Check the list of scans, and stitch any that don't have a DZI associated with it.
:raises RuntimeError: if the microscope is currently running a scan :raises RuntimeError: if the microscope is currently running a scan
""" """
if self._scan_logger is not None: if self._scan_lock.locked():
raise RuntimeError("Can't stitch previous scans while a scan is ongoing") raise RuntimeError("Can't stitch previous scans while a scan is ongoing")
for scan in self._get_all_scan_info(): for scan in self._get_all_scan_info():
if scan.dzi is None: if scan.dzi is None:
self.stitch_scan(logger=logger, scan_name=scan.name) self.stitch_scan(scan_name=scan.name)

3
tests/test_scan_directories.py
View file

@ -24,9 +24,6 @@ from openflexure_microscope_server.scan_directories import (
from .test_scan_data import _fake_scan_data from .test_scan_data import _fake_scan_data
from .utilities import assert_unique_of_length from .utilities import assert_unique_of_length
# A global logger to pass in as an Invocation Logger
LOGGER = logging.getLogger("mock-invocation_logger")
# Use our own dir in the root temp dir not a dynamically generated one so we # Use our own dir in the root temp dir not a dynamically generated one so we
# have some control of when it is deleted # have some control of when it is deleted
BASE_SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") BASE_SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans")

52
tests/test_smart_scan.py
View file

@ -40,9 +40,6 @@ from .mock_things.mock_camera import MockCameraThing
from .mock_things.mock_csm import MockCSMThing from .mock_things.mock_csm import MockCSMThing
from .mock_things.mock_stage import MockStageThing from .mock_things.mock_stage import MockStageThing
# A global logger to pass in as an Invocation Logger
LOGGER = logging.getLogger("mock-invocation_logger")
# Use our own dir in the root temp dir not a dynamically generated one so we # Use our own dir in the root temp dir not a dynamically generated one so we
# have some control of when it is deleted # have some control of when it is deleted
SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans")
@ -92,16 +89,16 @@ def test_private_delete_scan(smart_scan_thing, caplog):
# Make the outer scan dir, but not the one to delete # Make the outer scan dir, but not the one to delete
os.makedirs(SCAN_DIR) os.makedirs(SCAN_DIR)
# Attempt to delete the fake scan. Expect it to fail and provide a warning # Attempt to delete the fake scan. Expect it to fail and provide a warning
deleted = smart_scan_thing._delete_scan(fake_scan_name, LOGGER) deleted = smart_scan_thing._delete_scan(fake_scan_name)
assert not deleted assert not deleted
assert len(caplog.records) == 1 assert len(caplog.records) == 1
assert caplog.records[0].levelname == "WARNING" assert caplog.records[0].levelname == "WARNING"
assert caplog.records[0].name == "mock-invocation_logger" assert caplog.records[0].name == "labthings_fastapi.things.smartscanthing"
# Make a dir for the fake scan and delete it. # Make a dir for the fake scan and delete it.
os.makedirs(fake_scan_path) os.makedirs(fake_scan_path)
assert os.path.exists(fake_scan_path) assert os.path.exists(fake_scan_path)
deleted = smart_scan_thing._delete_scan(fake_scan_name, LOGGER) deleted = smart_scan_thing._delete_scan(fake_scan_name)
assert not os.path.exists(fake_scan_path) assert not os.path.exists(fake_scan_path)
assert deleted assert deleted
# Check no extra logs generated # Check no extra logs generated
@ -120,18 +117,18 @@ def test_public_delete_scan(smart_scan_thing, caplog):
# Attempt to delete the fake scan. Expect it to fail # Attempt to delete the fake scan. Expect it to fail
with pytest.raises(HTTPException) as exc_info: with pytest.raises(HTTPException) as exc_info:
smart_scan_thing.delete_scan(fake_scan_name, LOGGER) smart_scan_thing.delete_scan(fake_scan_name)
# Should raise a 400 error if the scan doesn't exist, not a 404 as the server # Should raise a 400 error if the scan doesn't exist, not a 404 as the server
# was not expecting to receive the scan files # was not expecting to receive the scan files
assert exc_info.value.status_code == 400 assert exc_info.value.status_code == 400
assert len(caplog.records) == 1 assert len(caplog.records) == 1
assert caplog.records[0].levelname == "WARNING" assert caplog.records[0].levelname == "WARNING"
assert caplog.records[0].name == "mock-invocation_logger" assert caplog.records[0].name == "labthings_fastapi.things.smartscanthing"
# Make a dir for the fake scan and delete it. # Make a dir for the fake scan and delete it.
os.makedirs(fake_scan_path) os.makedirs(fake_scan_path)
assert os.path.exists(fake_scan_path) assert os.path.exists(fake_scan_path)
smart_scan_thing.delete_scan(fake_scan_name, LOGGER) smart_scan_thing.delete_scan(fake_scan_name)
assert not os.path.exists(fake_scan_path) assert not os.path.exists(fake_scan_path)
# Check no extra logs generated # Check no extra logs generated
assert len(caplog.records) == 1 assert len(caplog.records) == 1
@ -151,7 +148,7 @@ def test_delete_all_scans(smart_scan_thing, caplog):
fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name)
os.makedirs(fake_scan_path) os.makedirs(fake_scan_path)
assert os.path.exists(fake_scan_path) assert os.path.exists(fake_scan_path)
smart_scan_thing.delete_all_scans(LOGGER) smart_scan_thing.delete_all_scans()
for fake_scan_name in fake_scan_names: for fake_scan_name in fake_scan_names:
fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name)
assert not os.path.exists(fake_scan_path) assert not os.path.exists(fake_scan_path)
@ -190,7 +187,6 @@ def _run_only_outer_scan(adjust_initial_state: Optional[Callable] = None):
"""Check scan vars are set up as expected""" """Check scan vars are set up as expected"""
assert not self._scan_lock.acquire(timeout=0.1) assert not self._scan_lock.acquire(timeout=0.1)
assert self._scan_logger is LOGGER
assert self._autofocus is af_mock assert self._autofocus is af_mock
assert self._stage is stage_mock assert self._stage is stage_mock
assert self._cam is cam_mock assert self._cam is cam_mock
@ -207,7 +203,6 @@ def _run_only_outer_scan(adjust_initial_state: Optional[Callable] = None):
exec_info = None exec_info = None
try: try:
mock_ss_thing.sample_scan( mock_ss_thing.sample_scan(
logger=LOGGER,
autofocus=af_mock, autofocus=af_mock,
stage=stage_mock, stage=stage_mock,
cam=cam_mock, cam=cam_mock,
@ -219,7 +214,6 @@ def _run_only_outer_scan(adjust_initial_state: Optional[Callable] = None):
assert mock_ss_thing._scan_lock.acquire(timeout=0.1) assert mock_ss_thing._scan_lock.acquire(timeout=0.1)
mock_ss_thing._scan_lock.release() mock_ss_thing._scan_lock.release()
assert mock_ss_thing._scan_logger is None
assert mock_ss_thing._autofocus is None assert mock_ss_thing._autofocus is None
assert mock_ss_thing._stage is None assert mock_ss_thing._stage is None
assert mock_ss_thing._cam is None assert mock_ss_thing._cam is None
@ -280,23 +274,25 @@ def _expected_scan_data():
@pytest.fixture @pytest.fixture
def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker): def scan_thing_mocked_for_scan_data(smart_scan_thing, mocker):
"""Return a scan thing that is mocked so that _collect_scan_data will run.""" """Return a scan thing that is mocked so that _collect_scan_data will run."""
# Give the scan thing a scan invocation logger so it thinks a scan is running. # Set the lock so it thinks the scan is running
smart_scan_thing._scan_logger = LOGGER with smart_scan_thing._scan_lock:
mocker.patch.object( mocker.patch.object(
smart_scan_thing, "_calc_displacement_from_test_image", return_value=[100, 100] smart_scan_thing,
) "_calc_displacement_from_test_image",
mock_ongoing_scan = mocker.Mock() return_value=[100, 100],
type(mock_ongoing_scan).name = mocker.PropertyMock(return_value=MOCK_SCAN_NAME) )
type(mock_ongoing_scan).images_dir = mocker.PropertyMock(return_value=MOCK_SCAN_DIR) mock_ongoing_scan = mocker.Mock()
mock_stage = mocker.Mock() mock_ongoing_scan.name = MOCK_SCAN_NAME
type(mock_stage).position = mocker.PropertyMock(return_value=MOCK_START_POS) mock_ongoing_scan.images_dir = MOCK_SCAN_DIR
mock_stage = mocker.Mock()
mock_stage.position = MOCK_START_POS
mock_autofocus = mocker.Mock() mock_autofocus = mocker.Mock()
smart_scan_thing._ongoing_scan = mock_ongoing_scan smart_scan_thing._ongoing_scan = mock_ongoing_scan
smart_scan_thing._stage = mock_stage smart_scan_thing._stage = mock_stage
smart_scan_thing._autofocus = mock_autofocus smart_scan_thing._autofocus = mock_autofocus
return smart_scan_thing yield smart_scan_thing
def test_collect_scan_data(scan_thing_mocked_for_scan_data): def test_collect_scan_data(scan_thing_mocked_for_scan_data):

2
tests/test_stage_measure.py
View file

@ -14,8 +14,6 @@ from openflexure_microscope_server.things.camera_stage_mapping import (
csm_stage_to_img, csm_stage_to_img,
) )
LOGGER = logging.getLogger("mock-invocation_logger")
# Useful generators # Useful generators
def increasing_xy_dict_generator(*_args, **_kwargs): def increasing_xy_dict_generator(*_args, **_kwargs):