Add new tests new scan directory function and for scan data serialisation

This commit is contained in:
Julian Stirling 2025-08-03 16:42:38 +01:00
parent 6ceb1a9845
commit 573fc20dec
4 changed files with 209 additions and 6 deletions

View file

@ -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

View file

@ -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,
)

120
tests/test_scan_data.py Normal file
View file

@ -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

View file

@ -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()