From 573fc20dec435cb18b29c80a8db3f11f9f93b4b1 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sun, 3 Aug 2025 16:42:38 +0100 Subject: [PATCH] Add new tests new scan directory function and for scan data serialisation --- .../scan_directories.py | 14 +- .../things/smart_scan.py | 2 +- tests/test_scan_data.py | 120 ++++++++++++++++++ tests/test_scan_directories.py | 79 ++++++++++++ 4 files changed, 209 insertions(+), 6 deletions(-) create mode 100644 tests/test_scan_data.py diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 6533c589..122ff2ce 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -48,6 +48,8 @@ class ScanData(BaseModel): Properties that are not known until the end have ``None`` serialised as "Unknown" """ + model_config = {"extra": "forbid"} + # TODO: Docs for each variable # TODO: Make note about timestamp format. Is it too ambiguous? scan_name: str @@ -58,8 +60,8 @@ class ScanData(BaseModel): autofocus_range: int autofocus_on: bool start_time: datetime - skip_background: str - stitch_automatically: str + skip_background: bool + stitch_automatically: bool # TODO: Think about changing stitch_resize name as it is NOT the resize for # just for correlation stitching stitch_resize: float @@ -113,7 +115,7 @@ class ScanData(BaseModel): hrs = int(total_secs // 3600) mins = int((total_secs % 3600) // 60) secs = int((total_secs % 60)) - return f"{hrs}:{mins}:{secs}" + return f"{hrs}:{mins:02}:{secs:02}" @field_validator("final_image_count", "scan_result", mode="before") @classmethod @@ -124,9 +126,9 @@ class ScanData(BaseModel): return value @field_serializer("final_image_count", "scan_result") - def serialize_none_as_unknown(self, value: Optional[str | int]) -> str: + def serialize_none_as_unknown(self, value: Optional[str | int]) -> str | int: """Serialise None as "Unknown" for a more human readable result.""" - return "Unknown" if value is None else str(value) + return "Unknown" if value is None else value class ScanDirectoryManager: @@ -214,6 +216,8 @@ class ScanDirectoryManager: If no scan data JSON file is found, return None """ scan_data_path = ScanDirectory(scan_name, self.base_dir).scan_data_path + if scan_data_path is None: + return None if not os.path.isfile(scan_data_path): return None return scan_data_path diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index cb1ef83d..dc391731 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -324,7 +324,7 @@ class SmartScanThing(lt.Thing): Takes scan_result, a string that is either "success", "cancelled by user", or the error that ended the scan. """ - self.scan_data.set_final_data( + self._scan_data.set_final_data( result=scan_result, final_image_count=self._scan_images_taken, ) diff --git a/tests/test_scan_data.py b/tests/test_scan_data.py new file mode 100644 index 00000000..8c2ae113 --- /dev/null +++ b/tests/test_scan_data.py @@ -0,0 +1,120 @@ +"""Test the functionality in the scan_directories module.""" + +from datetime import datetime, timedelta +from copy import copy +import json +from math import floor + +from openflexure_microscope_server.scan_directories import ScanData + +MOCK_START_TIME = datetime( + year=2024, + month=12, + day=25, + hour=11, + minute=0, + second=0, + microsecond=1234, +) + +MOCK_END_TIME = datetime( + year=2024, + month=12, + day=25, + hour=12, + minute=29, + second=3, + microsecond=5678, +) + + +def _fake_scan_data(**kwargs) -> ScanData: + """Make fake scan data, the start time is now. Final properties are not added. + + :param **kwargs: Key word arguments can be used to override other values. + """ + data_dict = { + "scan_name": "fake_scan_0001", + "overlap": 0.1, + "max_dist": 100000, + "dx": 100, + "dy": 100, + "autofocus_range": 1000, + "autofocus_on": True, + "start_time": copy(MOCK_START_TIME), + "skip_background": True, + "stitch_automatically": True, + "stitch_resize": 0.25, + "save_resolution": (1000, 1000), + } + for key, value in kwargs.items(): + data_dict[key] = value + return ScanData(**data_dict) + + +def test_set_final_data(): + """Check that adding final data to a ScanData object works as expected.""" + scan_data = _fake_scan_data() + + assert scan_data.final_image_count is None + assert scan_data.duration is None + assert scan_data.scan_result is None + + # Should set duration based of finishing at datetime.now() + scan_data.set_final_data(result="Success", final_image_count=123) + expected_duration = datetime.now() - MOCK_START_TIME + expected_duration_s = expected_duration.total_seconds() + scan_duration_s = scan_data.duration.total_seconds() + + assert scan_data.final_image_count == 123 + assert expected_duration_s - 1 < scan_duration_s < expected_duration_s + 1 + assert scan_data.scan_result == "Success" + + +def test_custom_serialisation(): + """Check that the custom serialisation in ScanData works as expected.""" + scan_data = _fake_scan_data() + # Serialise to string then load directly as json + scan_data_dict = json.loads(scan_data.model_dump_json()) + assert scan_data_dict["start_time"] == "11_00_00-25_12_2024" + assert scan_data_dict["final_image_count"] == "Unknown" + assert scan_data_dict["duration"] == "Unknown" + assert scan_data_dict["scan_result"] == "Unknown" + + scan_data.set_final_data(result="Success", final_image_count=123) + # Can't mock datetime.now as datetime is immutable. So just replace duraction + scan_data.duration = MOCK_END_TIME - scan_data.start_time + scan_data_dict = json.loads(scan_data.model_dump_json()) + assert scan_data_dict["final_image_count"] == 123 + assert scan_data_dict["duration"] == "1:29:03" + assert scan_data_dict["scan_result"] == "Success" + + +def test_round_trip_not_finalised(): + """Check that ScanData without final data can be serialised and deserialised.""" + scan_data = _fake_scan_data() + scan_data_dict = json.loads(scan_data.model_dump_json()) + scan_data_reloaded = ScanData(**scan_data_dict) + + # For the round trip to be equal we must remove microseconds from the start + # time as they are not saved + scan_data.start_time = scan_data.start_time.replace(microsecond=0) + assert scan_data == scan_data_reloaded + + +def test_round_trip_finalised(): + """Check that finalised ScanData can be serialised and deserialised.""" + scan_data = _fake_scan_data() + # Finalise the data. + scan_data.set_final_data(result="Success", final_image_count=123) + + scan_data_dict = json.loads(scan_data.model_dump_json()) + scan_data_reloaded = ScanData(**scan_data_dict) + + # For the round trip to be equal we must remove microseconds from the start + # time and duration as they are not saved + scan_data.model_copy(update={}) + scan_data.start_time = scan_data.start_time.replace(microsecond=0) + scan_data.duration = timedelta(seconds=floor(scan_data.duration.total_seconds())) + + assert scan_data == scan_data_reloaded diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py index 1f6c0609..3fb317cc 100644 --- a/tests/test_scan_directories.py +++ b/tests/test_scan_directories.py @@ -7,6 +7,7 @@ import shutil import logging import random import time +import json from collections import namedtuple import pytest @@ -17,8 +18,10 @@ from openflexure_microscope_server.scan_directories import ( ScanInfo, get_files_in_zip, NotEnoughFreeSpaceError, + SCAN_DATA_FILENAME, ) +from .test_scan_data import _fake_scan_data from .utilities import assert_unique_of_length # A global logger to pass in as an Invocation Logger @@ -281,6 +284,46 @@ def test_get_final_stitch(): assert scan_dir.get_final_stitch_name() == fake_scan_name +def test_get_scan_data_path(): + """Check that a scan data path behaves as expected.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_name = scan_dir.name + expected_path = os.path.join(scan_dir.images_dir, SCAN_DATA_FILENAME) + # Asking the scan manager for the path will return None as there is no file. + assert scan_dir_manager.get_scan_data_path(scan_name) is None + # Asking the scan directly will return the location the file should be located + # this is so it can be created. + assert scan_dir.scan_data_path == expected_path + + # Remove the images directory. + shutil.rmtree(scan_dir.images_dir) + assert scan_dir_manager.get_scan_data_path(scan_name) is None + + +def test_get_scan_data_dict(): + """Check that the dictionary for the scan data is returned, or None if doesn't exist.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_name = scan_dir.name + # Doesn't yet exist + assert scan_dir_manager.get_scan_data_dict(scan_name) is None + + fake_data = {"foo": 1, "bar": "foobar"} + with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file: + json.dump(fake_data, json_file) + + # Should now be able to load this fake data from disk + assert scan_dir_manager.get_scan_data_dict(scan_name) == fake_data + + # Check None is returned if the data cannot be read. + with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file: + json_file.write("this is not json") + assert scan_dir_manager.get_scan_data_dict(scan_name) is None + + def test_empty_scan_info(): """Test the scan info is correct even if the scan is empty.""" _clear_scan_dir() @@ -339,6 +382,42 @@ def test_zipping_scan_data(): assert not file.endswith(".dzi") +def test_saving_and_loading_scan_data(): + """Test that scan data is saved and loaded as expected.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_name = scan_dir.name + + # Should start without a scan data file. + assert not os.path.isfile(scan_dir.scan_data_path) + # Create + scan_data_obj = _fake_scan_data() + scan_dir.save_scan_data(scan_data_obj) + # File should now exist + assert os.path.isfile(scan_dir.scan_data_path) + + # Dump the scan json to a string an reload it + # Note that more detailed checking of the dumping and loading of ScanData is in + # tests/test_scan_data.py + scan_data_dict = json.loads(scan_data_obj.model_dump_json()) + # What is loaded from file should be the same as from dumping and loading. + assert scan_dir_manager.get_scan_data_dict(scan_name) == scan_data_dict + + +def test_saving_scan_data_error(): + """Test that saving scan data if there is no images directory raises FileNotFoundError.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + + # Remove the images directory. + shutil.rmtree(scan_dir.images_dir) + # Should raise FileNotFoundError. + with pytest.raises(FileNotFoundError): + scan_dir.save_scan_data(_fake_scan_data()) + + def test_all_files(): """Test all_files returns the path, and respects skipped directories.""" _clear_scan_dir()