187 lines
6.4 KiB
Python
187 lines
6.4 KiB
Python
"""Test the functionality in the scan_directories module."""
|
|
|
|
import json
|
|
from copy import copy
|
|
from datetime import datetime, timedelta
|
|
from math import floor
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from openflexure_microscope_server.scan_directories import HistoricScanData
|
|
from openflexure_microscope_server.stitching import StitchingSettings
|
|
from openflexure_microscope_server.things.smart_scan import ActiveScanData
|
|
|
|
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_legacy_scan_data(**kwargs) -> HistoricScanData:
|
|
"""Make fake legacy scan data.
|
|
|
|
:param **kwargs: Key word arguments can be used to override other values.
|
|
"""
|
|
data_dict = {
|
|
"scan_name": "fake_scan_0001",
|
|
"starting_position": {"x": 123, "y": 456, "z": 789},
|
|
"overlap": 0.1,
|
|
"max_dist": 100000,
|
|
"dx": 100,
|
|
"dy": 100,
|
|
"autofocus_dz": 1000,
|
|
"autofocus_on": True,
|
|
"start_time": copy(MOCK_START_TIME),
|
|
"skip_background": True,
|
|
"stitch_automatically": True,
|
|
"correlation_resize": 0.25,
|
|
"save_resolution": (1000, 1000),
|
|
}
|
|
for key, value in kwargs.items():
|
|
data_dict[key] = value
|
|
return HistoricScanData(**data_dict)
|
|
|
|
|
|
class MockWorkflowSettingModel(BaseModel):
|
|
"""A mock model to check that ActiveScanData can hold arbitrary models."""
|
|
|
|
setting_1: int
|
|
setting_2: int
|
|
setting_3: str
|
|
|
|
|
|
def fake_active_scan_data():
|
|
"""Fake scan data for and active scan.
|
|
|
|
The start time is now. Final properties are not added.
|
|
"""
|
|
return ActiveScanData(
|
|
schema_version=2,
|
|
scan_name="fake_scan_0001",
|
|
starting_position={"x": 123, "y": 456, "z": 789},
|
|
start_time=copy(MOCK_START_TIME),
|
|
stitch_automatically=True,
|
|
save_resolution=(1000, 1000),
|
|
stitching_settings=StitchingSettings(correlation_resize=0.25, overlap=0.1),
|
|
workflow="MockWorkflow",
|
|
workflow_settings=MockWorkflowSettingModel(
|
|
setting_1=1,
|
|
setting_2=2,
|
|
setting_3="three",
|
|
),
|
|
)
|
|
|
|
|
|
def assert_active_and_historic_data_equivalent(active_data, historic_data):
|
|
"""Raise and error if active and historic scan data is not equivalent."""
|
|
assert isinstance(active_data, ActiveScanData)
|
|
assert isinstance(historic_data, HistoricScanData)
|
|
# For the round trip to be equal we must remove microseconds from the start
|
|
# time as they are not saved
|
|
active_data.start_time = active_data.start_time.replace(microsecond=0)
|
|
for key in active_data.model_fields:
|
|
if key == "workflow_settings":
|
|
# For workflow_settings check the base model serialises to the historic
|
|
# data.
|
|
active_wf_setting_dict = active_data.workflow_settings.model_dump()
|
|
assert historic_data.workflow_settings == active_wf_setting_dict
|
|
continue
|
|
|
|
assert getattr(active_data, key) == getattr(historic_data, key)
|
|
|
|
|
|
def test_legacy_data_validates():
|
|
"""Check that legacy scan data validates."""
|
|
scan_data = _fake_legacy_scan_data()
|
|
assert isinstance(scan_data, HistoricScanData)
|
|
assert scan_data.image_count == 0
|
|
assert scan_data.duration is None
|
|
assert scan_data.scan_result is None
|
|
# Most importantly legacy stitching data should now be in the StitchingSettings
|
|
# model
|
|
assert scan_data.stitching_settings.correlation_resize == 0.25
|
|
assert scan_data.stitching_settings.overlap == 0.1
|
|
|
|
|
|
def test_set_final_data():
|
|
"""Check that adding final data to a ActiveScanData object works as expected."""
|
|
scan_data = fake_active_scan_data()
|
|
|
|
assert scan_data.image_count == 0
|
|
assert scan_data.duration is None
|
|
assert scan_data.scan_result is None
|
|
|
|
# Quickly take 123 images!
|
|
scan_data.image_count += 123
|
|
# Should set duration based on finishing at datetime.now()
|
|
scan_data.set_final_data(result="Success")
|
|
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.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 ActiveScanData works as expected."""
|
|
scan_data = fake_active_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"] == "2024-12-25_11:00:00"
|
|
assert scan_data_dict["image_count"] == 0
|
|
assert scan_data_dict["duration"] == "Unknown"
|
|
assert scan_data_dict["scan_result"] == "Unknown"
|
|
|
|
scan_data.image_count += 123
|
|
scan_data.set_final_data(result="Success")
|
|
# Can't mock datetime.now as datetime is immutable. So just replace duration
|
|
scan_data.duration = MOCK_END_TIME - scan_data.start_time
|
|
scan_data_dict = json.loads(scan_data.model_dump_json())
|
|
assert scan_data_dict["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 ActiveScanData without final data can be serialised and deserialised."""
|
|
scan_data = fake_active_scan_data()
|
|
scan_data_dict = json.loads(scan_data.model_dump_json())
|
|
scan_data_reloaded = HistoricScanData(**scan_data_dict)
|
|
|
|
assert_active_and_historic_data_equivalent(scan_data, scan_data_reloaded)
|
|
|
|
|
|
def test_round_trip_finalised():
|
|
"""Check that finalised HistoricScanData can be serialised and deserialised."""
|
|
scan_data = fake_active_scan_data()
|
|
# Finalise the data.
|
|
scan_data.image_count += 123
|
|
scan_data.set_final_data(result="Success")
|
|
|
|
scan_data_dict = json.loads(scan_data.model_dump_json())
|
|
scan_data_reloaded = HistoricScanData(**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_active_and_historic_data_equivalent(scan_data, scan_data_reloaded)
|