From 0d2f3243244c5edc8a2f247196e50a82d8511516 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 10 Jun 2025 10:56:14 +0100 Subject: [PATCH 1/9] Split creation of scan directory information into class with small testable methods --- .../things/smart_scan.py | 112 ++++++++++++------ 1 file changed, 79 insertions(+), 33 deletions(-) diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index b206058a..1720b52a 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -71,8 +71,8 @@ class ScanInfo(BaseModel): """Summary information about a scan folder""" name: str - created: datetime - modified: datetime + created: float + modified: float number_of_images: int stitch_available: bool dzi: Optional[str] @@ -93,6 +93,77 @@ SCAN_ZERO_PAD_DIGITS = 4 STITCHING_RESOLUTION = (820, 616) +class ScanDirectory: + """A class for handling interactions with scan directories + + Initalisation parameters: + dir_path: the directory of the outer scan directory + """ + + dir_path: str + + def __init__(self, dir_path: str): + if not os.path.isdir(dir_path): + raise FileNotFoundError(f"The scan directory {dir_path} cannot be found") + self.dir_path = dir_path + + @property + def images_dir(self) -> Optional[str]: + """ + The path to the images directory. None is returned is no images directory + was created + """ + im_path = os.path.join(self.dir_path, IMG_DIR_NAME) + if os.path.isdir(im_path): + return im_path + return None + + @property + def created_time(self) -> float: + """The time the directory was created on disk""" + return os.path.getctime(self.dir_path) + + @property + def name(self) -> str: + """The name of the scan directory""" + return os.path.basename(self.dir_path) + + def get_scan_files(self): + """Return a list of the files in the images dir""" + if self.images_dir is None: + return [] + return os.listdir(self.images_dir) + + def get_modified_time(self) -> float: + """Return the modified time of the directory""" + return max(os.stat(root).st_mtime for root, _, _ in os.walk(self.dir_path)) + + def get_scan_info(self): + """Return the inforomation for the scan directory as a ScanInfo object""" + folder_contents = self.get_scan_files() + if folder_contents: + scan_images = [i for i in folder_contents if IMAGE_REGEX.search(i)] + stitches = [i for i in folder_contents if i.endswith("_stitched.jpg")] + dzi_files = [i for i in folder_contents if i.endswith("dzi")] + + number_of_images = len(scan_images) + stitch_available = len(stitches) > 0 + dzi = None if not dzi_files else str(dzi_files[0]) + else: + number_of_images = 0 + stitch_available = False + dzi = None + + return ScanInfo( + name=self.name, + created=self.created_time, + modified=self.get_modified_time(), + number_of_images=number_of_images, + stitch_available=stitch_available, + dzi=dzi, + ) + + def _scan_running(method): """ This decorator is used by all methods in SmartScanThing that are using @@ -812,38 +883,13 @@ class SmartScanThing(Thing): scans: list[ScanInfo] = [] if not os.path.isdir(self.base_scan_dir): return scans - for f in os.listdir(self.base_scan_dir): - path = os.path.join(self.base_scan_dir, f) - if os.path.isdir(path): - images_folder = os.path.join(path, IMG_DIR_NAME) - if os.path.isdir(images_folder): - folder_contents = os.listdir(images_folder) - scan_images = [i for i in folder_contents if IMAGE_REGEX.search(i)] - stitches = [ - i for i in folder_contents if i.endswith("_stitched.jpg") - ] - number_of_images = len(scan_images) - stitch_available = len(stitches) > 0 - dzi = [i for i in folder_contents if i.endswith("dzi")] - if len(dzi) > 0: - dzi = str(dzi[0]) - else: - dzi = None - else: - number_of_images = 0 - stitch_available = False - modified = max(os.stat(root).st_mtime for root, _, _ in os.walk(path)) - scans.append( - ScanInfo( - name=f, - created=os.path.getctime(path), - modified=modified, - number_of_images=number_of_images, - stitch_available=stitch_available, - dzi=dzi, - ) - ) + for scan_dirname in os.listdir(self.base_scan_dir): + scan_path = os.path.join(self.base_scan_dir, scan_dirname) + if os.path.isdir(scan_path): + scan_dir = ScanDirectory(scan_path) + scans.append(scan_dir.get_scan_info()) + return scans @fastapi_endpoint( From 32f78a4cff15de48490050f76786bd7edfd05df8 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sat, 14 Jun 2025 22:24:10 +0100 Subject: [PATCH 2/9] Trying to make a distinction between file directory operations for scans and scanning code This started as a small change and spiralled out of control it needs good unit tests and testing before considering to merge --- .../scan_directories.py | 313 ++++++++++ .../things/smart_scan.py | 537 +++--------------- tests/test_stack.py | 2 +- 3 files changed, 400 insertions(+), 452 deletions(-) create mode 100644 src/openflexure_microscope_server/scan_directories.py diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py new file mode 100644 index 00000000..6fccb837 --- /dev/null +++ b/src/openflexure_microscope_server/scan_directories.py @@ -0,0 +1,313 @@ +""" +This submodule contains functionality for interacting with scan directories. + +Currently it handles scan getting information from a information from a scan directory, +eventually is should handle all the file system operation for smart_scan. +""" + +from typing import Optional +import os +import re +import shutil +import zipfile + +from pydantic import BaseModel + +IMG_DIR_NAME = "images" +IMAGE_REGEX = re.compile(r"-?[0-9]+_-?[0-9]+\.jpe?g$") +SCAN_ZERO_PAD_DIGITS = 4 + + +class NotEnoughFreeSpaceError(IOError): + pass + + +class ScanInfo(BaseModel): + """Summary information about a scan folder""" + + name: str + created: float + modified: float + number_of_images: int + stitch_available: bool + dzi: Optional[str] + + +class ScanDirectoryManager: + """ + A class for managinging interactions with scan directories + """ + + _base_scan_dir: str + + def __init__(self, base_scan_dir: str): + self._base_scan_dir = base_scan_dir + if not os.path.exists(self._base_scan_dir): + os.makedirs(self._base_scan_dir) + + @property + def base_dir(self) -> str: + """The base directory scans are saved to""" + return self._base_scan_dir + + def exists(self, scan_name: str) -> bool: + """True if scan of this name exists on disk""" + return os.path.isdir(self.path_for(scan_name)) + + def path_for(self, scan_name: str) -> str: + """Return the path for a given scan name + + Returns the path even if it doesn't exist) + """ + return os.path.join(self._base_scan_dir, scan_name) + + def img_dir_for(self, scan_name: str) -> str: + """Return the path for the image dir for a given scan name + + Returns the path even if it doesn't exist + """ + return os.path.join(self._base_scan_dir, scan_name, IMG_DIR_NAME) + + def get_file_from( + self, scan_name: str, filename: str, check_exists: bool = False + ) -> Optional[str]: + """Return the file path for the file within a scan directory + + If check_exists is True then a FileNotFoundError will be raised if + the file does not exist. + """ + file_path = os.path.join(self.path_for(scan_name), filename) + if check_exists: + if not os.path.exists: + return None + return file_path + + def get_file_from_img_dir( + self, scan_name: str, filename: str, check_exists: bool = False + ) -> Optional[str]: + """Return the file path for the file within a scan directory + + If check_exists is True, None is returned if the file does not exist. If False + then the path is returned anway + """ + file_path = os.path.join(self.img_dir_for(scan_name), filename) + if check_exists: + if not os.path.exists: + return None + return file_path + + @property + def all_scans(self): + """Return a list of the scan names in the base directory""" + return [f.name for f in os.scandir(self._base_scan_dir) if f.is_dir()] + + def all_scans_info(self) -> list[ScanInfo]: + """Return a lists of ScanInfo objects for each scan""" + + all_info: list[ScanInfo] = [] + for scan_name in self.all_scans: + all_info.append(ScanDirectory(scan_name, self.base_dir).scan_info) + return all_info + + def _unique_scan_name(self, scan_name: str) -> str: + """Get the next unique scan name starting with the given name + + For more explanation on the scan naming see `new_scan_dir` + """ + # A regex with the scan name and a group for the numbers + scan_regex = re.compile( + "^" + scan_name + "_([0-9]{" + str(SCAN_ZERO_PAD_DIGITS) + "})$" + ) + + matching_scans = [scan for scan in self.all_scans if scan_regex.match(scan)] + if not matching_scans: + scan_num = 1 + else: + last_matching_scan = sorted(matching_scans)[-1] + # Get the first group from the regex, turn to int, and add 1 + scan_num = int(scan_regex.match(last_matching_scan)[1]) + 1 + + # Set a sensible limit for the number of scans of one name + # based on our zero padding. + max_scan_no = 10**SCAN_ZERO_PAD_DIGITS - 1 + + if scan_num > max_scan_no: + raise FileExistsError( + "Could not create a new scan folder: all names in use!" + ) + + return f"{scan_name}_{scan_num:0{SCAN_ZERO_PAD_DIGITS}d}" + + def new_scan_dir(self, scan_name: str) -> str: + """Get a unique name for this scan and create a directory for it + + The scan will be named `{scan_name}_0001` where the number is + zero-padded to be 4 digits long (to allow correct sorting if the + scans are ordered alphanumerically). + + Creates a new empty folder, into which scans are saved + + Returns the a ScanDirectory object + """ + + # if no scan name is set set to "scan". This done here as empty strings + # get passed in otherwise. + if not scan_name: + scan_name = "scan" + + full_scan_name = self._unique_scan_name(scan_name) + + os.makedirs(self.path_for(full_scan_name)) + os.makedirs(self.img_dir_for(full_scan_name)) + return ScanDirectory(full_scan_name, self.base_dir) + + def delete_scan(self, scan_name: str): + """Delete a scan""" + shutil.rmtree(self.path_for(scan_name)) + + def zip_scan(self, scan_name: str, final_version: bool = False) -> str: + """Zips any images from the scan not yet zipped, return full path to zip + + `final_version` Set true to stitch all files not just the scan images + this should only be done at the end as it is not possible to update a file + in a zip. + """ + return ScanDirectory(scan_name, self.base_dir).zip_files(final_version) + + def check_free_disk_space(self, min_space: int = 500000000) -> None: + """ + Raise an exception if there is not enough free disk space to continue scanning + + Args: + path = path to a location on the disk you want to check + min_space [int] = the minimum space required in bytes + default = 500,000,000 (500MiB) + + Raises: + NotEnoughFreeSpaceError if the remaining storage is below min_space + """ + disk_usage = shutil.disk_usage(self._base_scan_dir) + if disk_usage.free < min_space: + raise NotEnoughFreeSpaceError( + "There is not enough free disk space to continue. " + f"(Required: {min_space >> 20}MB. Free: {disk_usage.free >> 20}MB)." + ) + + +class ScanDirectory: + """A class for handling interactions with scan directories + + Initalisation parameters: + dir_path: the directory of the outer scan directory + """ + + _name: str + _base_scan_dir: str + + def __init__(self, name: str, base_scan_dir: str): + self._name = name + self._base_scan_dir = base_scan_dir + if not os.path.isdir(self.dir_path): + raise FileNotFoundError( + f"The scan directory {self.dir_path} cannot be found" + ) + + @property + def name(self): + """The name of the scan""" + return self._name + + @property + def dir_path(self) -> str: + """The full path to the scan directory""" + os.path.join(self._base_scan_dir, self._name) + + @property + def images_dir(self) -> Optional[str]: + """ + The path to the images directory. None is returned if no images directory + was created + """ + im_path = os.path.join(self.dir_path, IMG_DIR_NAME) + if os.path.isdir(im_path): + return im_path + return None + + @property + def created_time(self) -> float: + """The time the directory was created on disk""" + return os.path.getctime(self.dir_path) + + def get_scan_files(self): + """Return a list of the files in the images dir""" + if self.images_dir is None: + return [] + return os.listdir(self.images_dir) + + def get_modified_time(self) -> float: + """Return the modified time of the directory""" + return max(os.stat(root).st_mtime for root, _, _ in os.walk(self.dir_path)) + + def scan_info(self): + """Return the inforomation for the scan directory as a ScanInfo object""" + folder_contents = self.get_scan_files() + if folder_contents: + scan_images = [i for i in folder_contents if IMAGE_REGEX.search(i)] + stitches = [i for i in folder_contents if i.endswith("_stitched.jpg")] + dzi_files = [i for i in folder_contents if i.endswith("dzi")] + + number_of_images = len(scan_images) + stitch_available = len(stitches) > 0 + dzi = None if not dzi_files else str(dzi_files[0]) + else: + number_of_images = 0 + stitch_available = False + dzi = None + + return ScanInfo( + name=self.name, + created=self.created_time, + modified=self.get_modified_time(), + number_of_images=number_of_images, + stitch_available=stitch_available, + dzi=dzi, + ) + + def all_files(self): + """Return a list of all files in the scan dir relative to the dir""" + files = [] + for file_root, _, filenames in os.walk(self.dir_path): + for filename in filenames: + full_path = os.path.join(file_root, filename) + files.append(os.path.relpath(full_path, self.dir_path)) + return files + + def zip_files(self, final_version: bool = False) -> str: + """Zips any images from the scan not yet zipped, return full path to zip + + `final_version` Set true to stitch all files not just the scan images + this should only be done at the end as it is not possible to update a file + in a zip. + """ + + zip_fname = os.path.join(self.dir_path, "images.zip") + # get a list of files in the existing zip + zip_files = get_files_in_zip(zip_fname) + + with zipfile.ZipFile(zip_fname, mode="a") as scan_zip: + for file in self.all_files(): + # Don't zip zipfiles, or files in the zip + if file.endswith(".zip") or file in zip_files: + break + # If this is not the final version, then only zip image files. + if not final_version and not IMAGE_REGEX.search(file): + break + + scan_zip.write(os.path.join(self.dir_path, file), arcname=file) + return zip_fname + + +def get_files_in_zip(zip_path): + """List the relative paths of all files and folders in the zip folder specified""" + scan_zip = zipfile.ZipFile(zip_path) + return [os.path.normpath(i) for i in scan_zip.namelist()] diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 1720b52a..bda19bfb 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -1,19 +1,15 @@ -import shutil -import zipfile -import threading from typing import Optional, Mapping +import threading +import os +import time +import json +from datetime import datetime +from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, STDOUT + from fastapi import HTTPException from fastapi.responses import FileResponse import numpy as np -import os -import time from PIL import Image -from pydantic import BaseModel -from datetime import datetime -from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, STDOUT -import glob -import json -import re from labthings_fastapi.thing import Thing from labthings_fastapi.dependencies.metadata import GetThingStates @@ -25,145 +21,33 @@ from labthings_fastapi.dependencies.invocation import ( ) from labthings_fastapi.decorators import thing_action, thing_property, fastapi_endpoint from labthings_fastapi.outputs.blob import blob_type -from .camera import CameraDependency as CamDep -from .stage import StageDependency as StageDep from openflexure_microscope_server.utilities import ErrorCapturingThread -from openflexure_microscope_server.things.autofocus import AutofocusThing -from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper -from openflexure_microscope_server.things.background_detect import BackgroundDetectThing +from openflexure_microscope_server import scan_directories from openflexure_microscope_server import scan_planners +# Things +from .autofocus import AutofocusThing +from .camera_stage_mapping import CameraStageMapper +from .background_detect import BackgroundDetectThing +from .camera import CameraDependency as CamDep +from .stage import StageDependency as StageDep + CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/") AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/") BackgroundDep = direct_thing_client_dependency( BackgroundDetectThing, "/background_detect/" ) -IMAGE_REGEX = re.compile(r"-?[0-9]+_-?[0-9]+\.jpe?g$") - - -class NotEnoughFreeSpaceError(IOError): - pass - - -def ensure_free_disk_space(path: str, min_space: int = 500000000) -> None: - """ - Raise an exception if we are running out of disk space - - Args: - path = path to a location on the disk you want to check - min_space [int] = the minimum space required in bytes - default = 500,000,000 (500MiB) - - Raises: - NotEnoughFreeSpaceError if the remaining storage is below min_space - """ - disk_usage = shutil.disk_usage(path) - if disk_usage.free < min_space: - raise NotEnoughFreeSpaceError( - "There is not enough free disk space to continue. " - f"(Required: {min_space >> 20}MB, free: {disk_usage.free >> 20}MB)." - ) - - -class ScanInfo(BaseModel): - """Summary information about a scan folder""" - - name: str - created: float - modified: float - number_of_images: int - stitch_available: bool - dzi: Optional[str] - - -DOWNLOADABLE_SCAN_FILES = ( - "images.zip", - "stitched_thumbnail.jpg", -) - JPEGBlob = blob_type("image/jpeg") ZipBlob = blob_type("application/zip") -IMG_DIR_NAME = "images" + SCAN_DATA_FILENAME = "scan_data.json" STITCHING_CMD = "openflexure-stitch" -SCAN_ZERO_PAD_DIGITS = 4 STITCHING_RESOLUTION = (820, 616) -class ScanDirectory: - """A class for handling interactions with scan directories - - Initalisation parameters: - dir_path: the directory of the outer scan directory - """ - - dir_path: str - - def __init__(self, dir_path: str): - if not os.path.isdir(dir_path): - raise FileNotFoundError(f"The scan directory {dir_path} cannot be found") - self.dir_path = dir_path - - @property - def images_dir(self) -> Optional[str]: - """ - The path to the images directory. None is returned is no images directory - was created - """ - im_path = os.path.join(self.dir_path, IMG_DIR_NAME) - if os.path.isdir(im_path): - return im_path - return None - - @property - def created_time(self) -> float: - """The time the directory was created on disk""" - return os.path.getctime(self.dir_path) - - @property - def name(self) -> str: - """The name of the scan directory""" - return os.path.basename(self.dir_path) - - def get_scan_files(self): - """Return a list of the files in the images dir""" - if self.images_dir is None: - return [] - return os.listdir(self.images_dir) - - def get_modified_time(self) -> float: - """Return the modified time of the directory""" - return max(os.stat(root).st_mtime for root, _, _ in os.walk(self.dir_path)) - - def get_scan_info(self): - """Return the inforomation for the scan directory as a ScanInfo object""" - folder_contents = self.get_scan_files() - if folder_contents: - scan_images = [i for i in folder_contents if IMAGE_REGEX.search(i)] - stitches = [i for i in folder_contents if i.endswith("_stitched.jpg")] - dzi_files = [i for i in folder_contents if i.endswith("dzi")] - - number_of_images = len(scan_images) - stitch_available = len(stitches) > 0 - dzi = None if not dzi_files else str(dzi_files[0]) - else: - number_of_images = 0 - stitch_available = False - dzi = None - - return ScanInfo( - name=self.name, - created=self.created_time, - modified=self.get_modified_time(), - number_of_images=number_of_images, - stitch_available=stitch_available, - dzi=dzi, - ) - - def _scan_running(method): """ This decorator is used by all methods in SmartScanThing that are using @@ -185,7 +69,7 @@ def _scan_running(method): class SmartScanThing(Thing): def __init__(self, scans_folder): - self._scans_folder = scans_folder + self._scan_dir_manager = scan_directories.ScanDirectoryManager(scans_folder) self._preview_stitch_popen = None self._preview_stitch_popen_lock = threading.Lock() self._scan_lock = threading.Lock() @@ -207,7 +91,7 @@ class SmartScanThing(Thing): self._metadata_getter: Optional[GetThingStates] = None self._csm: Optional[CSMDep] = None self._background_detect: Optional[BackgroundDep] = None - self._ongoing_scan_name: Optional[str] = None + self._ongoing_scan: Optional[scan_directories.ScanDirectory] = None self._starting_position: Optional[Mapping[str, int]] = None self._capture_thread: Optional[ErrorCapturingThread] = None self._scan_images_taken: Optional[int] = None @@ -254,8 +138,8 @@ class SmartScanThing(Thing): try: self._check_background_and_csm_set() - self._ongoing_scan_name = self._get_unique_scan_name_and_dir(scan_name) - self.create_zip_of_scan(scan_name=self._ongoing_scan_name) + self._ongoing_scan = self._scan_dir_manager.new_scan_dir(scan_name) + self._latest_scan_name = self._ongoing_scan.name self._autofocus.looping_autofocus(dz=self.autofocus_dz, start="centre") # record starting position so we can return there self._starting_position = self._stage.position @@ -264,7 +148,7 @@ class SmartScanThing(Thing): # If _scan_data is set then scan started if self._scan_data is not None: self._return_to_starting_position() - if not isinstance(e, NotEnoughFreeSpaceError): + if not isinstance(e, scan_directories.NotEnoughFreeSpaceError): # Don't stich if drive is full (already logged) self._perform_final_stitch() # Error must be raised so UI gives correct output @@ -280,35 +164,11 @@ class SmartScanThing(Thing): self._csm = None self._background_detect = None self._capture_thread = None - self._ongoing_scan_name = None + self._ongoing_scan = None self._scan_images_taken = None self._scan_data = None self._scan_lock.release() - def promote_stitch_files( - self, - logger, - ): - """Copy the stitched image from the scan images folder to the top level""" - - # Search the scan images dir for a file ending in '_stitched.jpg - stitched_image_path = glob.glob( - os.path.join(self._ongoing_scan_images_dir, "*_stitched.jpg") - ) - - if len(stitched_image_path) == 0: - logger.warning("Could't find a stitched image to copy") - else: - for stitched_image in stitched_image_path: - stitch_name = os.path.basename(stitched_image) - - shutil.copy( - stitched_image, - os.path.join( - self.dir_for_scan(self._ongoing_scan_name), stitch_name - ), - ) - @_scan_running def _check_background_and_csm_set(self): """Before starting a scan, check that background and camera-stage-mapping are set @@ -339,88 +199,11 @@ class SmartScanThing(Thing): "of motion." ) - @property - def base_scan_dir(self) -> str: - """The path of the directory where the scans are saved.""" - return self._scans_folder - @thing_property def latest_scan_name(self) -> Optional[str]: """The name of the last scan to be started.""" return self._latest_scan_name - def dir_for_scan(self, scan_name: str): - """The path to the scan directory for scan with input name""" - return os.path.join(self.base_scan_dir, scan_name) - - def images_dir_for_scan(self, scan_name: str) -> str: - """The path to the images directory for scan with input name""" - scan_dir = self.dir_for_scan(scan_name=scan_name) - return os.path.join(scan_dir, IMG_DIR_NAME) - - @property - @_scan_running - def _ongoing_scan_dir(self): - """For the ongoing scan, this returns the scan directory""" - if not self._ongoing_scan_name: - raise RuntimeError("Cannot get ongoing scan name while no scan is running") - return self.dir_for_scan(self._ongoing_scan_name) - - @property - @_scan_running - def _ongoing_scan_images_dir(self): - """For the ongoing scan, this returns the image directory""" - if not self._ongoing_scan_name: - raise RuntimeError("Cannot get ongoing scan name while no scan is running") - return self.images_dir_for_scan(self._ongoing_scan_name) - - @_scan_running - def _get_unique_scan_name_and_dir(self, scan_name: str) -> str: - """Get a unique name for this scan and create a directory for it - - The scan will be named `{scan_name}_0001` where the number is - zero-padded to be 4 digits long (to allow correct sorting if the - scans are ordered alphanumerically). - - Note that if you have discontinuous numbering (e.g. you've got scans - numbered 1 through 10, but you deleted scan 5), then the gaps will - get filled in - so there's no guarantee, for now, that the numbers - will correspond to order of creation. This may change in the future. - - Creates a new empty folder, into which scans are saved - - Returns the scan name. - - The directory can be accessed by self.dir_for_scan(unique_scan_name) - - """ - if not os.path.exists(self.base_scan_dir): - os.makedirs(self.base_scan_dir) - - # if no scan name is set set to "scan". This done here as empty strings - # get passed in otherwise. - if not scan_name: - scan_name = "scan" - - # Set a sensible limit for the number of scans of one name - # based on our zero padding. - max_scan_no = 10**SCAN_ZERO_PAD_DIGITS - 1 - - for j in range(max_scan_no): - trial_unique_scan_name = f"{scan_name}_{j:0{SCAN_ZERO_PAD_DIGITS}d}" - trial_dir = self.dir_for_scan(trial_unique_scan_name) - if not os.path.exists(trial_dir): - os.makedirs(trial_dir) - # If we made the directory this is the scan name - # Save the scan name as the latest scan (this persists as a - # property after the scan finishes) - self._latest_scan_name = trial_unique_scan_name - # Create images directory and - os.mkdir(self.images_dir_for_scan(trial_unique_scan_name)) - # Return the scan name - return trial_unique_scan_name - raise FileExistsError("Could not create a new scan folder: all names in use!") - @_scan_running def _move_to_next_point( self, next_point: tuple[int, int], z_estimate: Optional[int] = None @@ -514,7 +297,7 @@ class SmartScanThing(Thing): # Fix scan parameters in case UI is updated during scan. self._scan_data = { - "scan_name": self._ongoing_scan_name, + "scan_name": self._ongoing_scan.name, "overlap": overlap, "max_dist": self.max_range, "dx": dx, @@ -537,7 +320,7 @@ class SmartScanThing(Thing): # Should this be a method of the scan_data dataclass? data = { - "scan_name": self._ongoing_scan_name, + "scan_name": self._ongoing_scan.name, "overlap": self._scan_data["overlap"], "autofocus range": self._scan_data["autofocus_dz"], "dx": self._scan_data["dx"], @@ -548,7 +331,7 @@ class SmartScanThing(Thing): } scan_inputs_fname = os.path.join( - self._ongoing_scan_images_dir, SCAN_DATA_FILENAME + self._ongoing_scan.images_dir, SCAN_DATA_FILENAME ) with open(scan_inputs_fname, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=4) @@ -577,7 +360,7 @@ class SmartScanThing(Thing): } scan_data_fname = os.path.join( - self._ongoing_scan_images_dir, SCAN_DATA_FILENAME + self._ongoing_scan.images_dir, SCAN_DATA_FILENAME ) with open(scan_data_fname, encoding="utf-8") as f: @@ -629,7 +412,7 @@ class SmartScanThing(Thing): scan_successful = False self._scan_logger.info("Stopping scan because it was cancelled.") self._update_scan_data_json(scan_result="cancelled by user") - except NotEnoughFreeSpaceError as e: + except scan_directories.NotEnoughFreeSpaceError as e: scan_successful = False self._update_scan_data_json(scan_result=str(e)) self._scan_logger.error( @@ -698,7 +481,7 @@ class SmartScanThing(Thing): # captures an image if necessary, updates the scan path and future path, # and updates the zip file with new images. while not route_planner.scan_complete: - ensure_free_disk_space(self._ongoing_scan_dir) + self._scan_dir_manager.check_free_disk_space() self._manage_stitching_threads() next_pos_xy, z_est = route_planner.get_next_location_and_z_estimate() @@ -725,7 +508,7 @@ class SmartScanThing(Thing): continue focused, focused_height = self._autofocus.run_smart_stack( - images_dir=self._ongoing_scan_images_dir, + images_dir=self._ongoing_scan.images_dir, autofocus_dz=self._scan_data["autofocus_dz"], save_resolution=self._scan_data["save_resolution"], ) @@ -736,12 +519,10 @@ class SmartScanThing(Thing): current_pos_xyz, imaged=True, focused=focused ) - # increment capure counter as thread has completed + # increment capture counter as thread has completed self._scan_images_taken += 1 # Add it to the incremental zip - self.update_zip( - scan_name=self._ongoing_scan_name, - ) + self._ongoing_scan.zip_files() @_scan_running def _return_to_starting_position(self): @@ -760,10 +541,8 @@ class SmartScanThing(Thing): self._scan_logger.info("Not performing a stitch as 3 or fewer images taken") return - self.update_zip( - scan_name=self._ongoing_scan_name, - final_version=False, - ) + self._ongoing_scan.zip_files() + self._scan_logger.info("Waiting for background processes to finish...") self._preview_stitch_wait() @@ -772,11 +551,10 @@ class SmartScanThing(Thing): self._scan_logger.info("Stitching final image (may take some time)...") self.stitch_scan( logger=self._scan_logger, - scan_name=self._ongoing_scan_name, + scan_name=self._ongoing_scan.name, stitch_resize=self._scan_data["stitch_resize"], overlap=self._scan_data["overlap"], ) - self.promote_stitch_files(self._scan_logger) except SubprocessError as e: self._scan_logger.error(f"Stitching failed: {e}", exc_info=e) @@ -796,12 +574,12 @@ class SmartScanThing(Thing): This endpoint allows files to be downloaded from a scan. """ - path = os.path.join( - self.images_dir_for_scan(scan_name), "stitched_thumbnail.jpg" + preview_path = self._scan_dir_manager.get_file_from_img_dir( + scan_name=scan_name, filename="stitched_thumbnail.jpg", check_exists=True ) - if not os.path.isfile(path): + if preview_path is None: raise HTTPException(404, "File not found") - return FileResponse(path) + return FileResponse(preview_path) @thing_property def save_resolution(self) -> tuple[int, int]: @@ -871,7 +649,7 @@ class SmartScanThing(Thing): self.thing_settings["stitch_automatically"] = value @thing_property - def scans(self) -> list[ScanInfo]: + def scans(self) -> list[scan_directories.ScanInfo]: """All the available scans Each scan has a name (which can be used to access it), along with @@ -880,45 +658,7 @@ class SmartScanThing(Thing): uses a regular expression, and changes to the naming scheme will break it. """ - scans: list[ScanInfo] = [] - if not os.path.isdir(self.base_scan_dir): - return scans - - for scan_dirname in os.listdir(self.base_scan_dir): - scan_path = os.path.join(self.base_scan_dir, scan_dirname) - if os.path.isdir(scan_path): - scan_dir = ScanDirectory(scan_path) - scans.append(scan_dir.get_scan_info()) - - return scans - - @fastapi_endpoint( - "get", - "scans/{scan_name}/{file}", - responses={ - 200: { - "description": "Successfully downloading file", - "content": {"*/*": {}}, - }, - 403: {"description": "Filename not permitted"}, - 404: {"description": "File not found"}, - }, - ) - def get_scan_file(self, scan_name: str, file: str) -> FileResponse: - """Retrieve a file from a scan. - - This endpoint allows files to be downloaded from a scan. For security - reasons, there is a list of allowable filenames, and paths with additional - slashes are not permitted. - """ - if file not in DOWNLOADABLE_SCAN_FILES: - raise HTTPException( - 403, f"You may only download files named {DOWNLOADABLE_SCAN_FILES}" - ) - path = os.path.join(self.base_scan_dir, scan_name, file) - if not os.path.isfile(path): - raise HTTPException(404, "File not found") - return FileResponse(path) + return self._scan_dir_manager.all_scans_info() @fastapi_endpoint( "delete", @@ -935,11 +675,10 @@ class SmartScanThing(Thing): Takes the scan name to delete, and the Invocation Logger """ - path = os.path.join(self.base_scan_dir, scan_name) - if not os.path.isdir(path): - logger.info(f"can't find {path}") + if not self._scan_dir_manager.exists(scan_name): + logger.warning(f"Cannot find a scan of name {scan_name}") raise HTTPException(400, "Scan not found") - deleted_scan_success = self._delete_scan(path, logger) + deleted_scan_success = self._delete_scan(scan_name, logger) if not deleted_scan_success: raise HTTPException(400, "Couldn't delete scan, check log for details") @@ -954,49 +693,59 @@ class SmartScanThing(Thing): microscope!** Use with extreme caution. """ - for scan in self.scans: - path = os.path.join(self.base_scan_dir, scan.name) - self._delete_scan(path, logger) + for scan_name in self._scan_dir_manager.all_scans: + self._delete_scan(scan_name, logger) @thing_action - def _delete_scan(self, scan_path, logger: InvocationLogger) -> bool: + def purge_empty_scans(self, logger: InvocationLogger) -> None: + """ + Delete all scan folders containing no images at the top level + """ + + # JSON is ignored as it's created before any images are captured + for scan_info in self._scan_dir_manager.all_scans_info(): + if scan_info.number_of_images == 0: + self._delete_scan(scan_info.name, logger) + + def _delete_scan(self, scan_name, logger: InvocationLogger) -> bool: + """ + A wrapper around scan manager's delete_scan that logs to the invocation logger + """ try: - shutil.rmtree(scan_path) + self._scan_dir_manager.delete_scan(scan_name) return True except Exception as e: logger.warning( - "Attempted to delete scan " + scan_path + ", which failed." + "Attempted to delete scan " + scan_name + ", which failed." " Server sent response" + str(e) ) return False @property - def latest_preview_stitch_path(self): - """Return the path of the latest preview stitched image + def latest_preview_stitch_path(self) -> Optional[str]: + """The path of the latest preview stitched image, or None if not available""" - If the image cannot be found (because there is no latest - scan, or because the latest scan has no preview stitch) then - raise FileNotFoundError - """ if not self.latest_scan_name: - raise FileNotFoundError("No latest scan found") + return None - images_dir = self.images_dir_for_scan(self.latest_scan_name) - stitch_path = os.path.join(images_dir, "preview.jpg") - if not os.path.isfile(stitch_path): - raise FileNotFoundError("Latest scan has no preview stitch") - return stitch_path + return self._scan_dir_manager.get_file_from_img_dir( + scan_name=self.latest_scan_name, filename="preview.jpg", check_exists=True + ) @thing_property - def latest_preview_stitch_time(self) -> Optional[datetime]: - """The modification time of the latest preview image + def latest_preview_stitch_time(self) -> Optional[float]: + """The modification time of the latest preview image, to allow live updating - This will return `null` if there is no preview image to return. + This will return None (`null` to JS) if there is no preview image to return. + + This is used for two things reasons: + 1. If all caching was turned off this stitch would be sent over the network + repeatedly + 2. If caching was is on, then the stitch will not update when needed. """ - try: - return os.path.getmtime(self.latest_preview_stitch_path) - except FileNotFoundError: + if self.latest_preview_stitch_path is None: return None + return os.path.getmtime(self.latest_preview_stitch_path) @fastapi_endpoint( "get", @@ -1011,10 +760,10 @@ class SmartScanThing(Thing): ) def get_latest_preview(self) -> FileResponse: """Retrieve the latest preview image.""" - try: - return FileResponse(self.latest_preview_stitch_path) - except FileNotFoundError: + preview_path = self.latest_preview_stitch_path + if preview_path is None: raise HTTPException(404, "File not found") + return FileResponse(preview_path) @_scan_running def _preview_stitch_start(self, overlap: float) -> None: @@ -1042,7 +791,7 @@ class SmartScanThing(Thing): f"{min_overlap}", "--resize", f"{self._scan_data['stitch_resize']}", - self._ongoing_scan_images_dir, + self._ongoing_scan.images_dir, ] ) @@ -1120,8 +869,9 @@ class SmartScanThing(Thing): Note that as this is a thing_action it needs the logger passed as a variable if called from another thing action """ - images_folder = self.images_dir_for_scan(scan_name=scan_name) - json_fpath = os.path.join(images_folder, SCAN_DATA_FILENAME) + json_fpath = self._scan_dir_manager.get_file_from_img_dir( + scan_name=scan_name, filename=SCAN_DATA_FILENAME + ) if self.stitch_tiff: tiff_arg = "--stitch_tiff" @@ -1183,87 +933,10 @@ class SmartScanThing(Thing): f"{round(overlap * 0.9, 2)}", "--resize", f"{stitch_resize}", - images_folder, + self._scan_dir_manager.img_dir_for(scan_name), ], ) - @thing_action - def create_zip_of_scan( - self, - scan_name: str, - ) -> None: - """Generate an empty zip file for the current scan""" - images_folder = self.images_dir_for_scan(scan_name=scan_name) - scan_folder = self.dir_for_scan(scan_name=scan_name) - - if not os.path.isdir(images_folder): - raise FileNotFoundError( - f"Tried to make a zip archive of {images_folder} but it does not exist." - ) - - zip_fname = os.path.join(scan_folder, "images.zip") - - # Create an empty zip file - if not os.path.isfile(zip_fname): - with zipfile.ZipFile(zip_fname, mode="w"): - pass - - def update_zip( - self, - scan_name: str, - final_version: bool = False, - ) -> None: - """Update the zip file with any images added to the folder since the last run, - except for files containing 'files_to_delay', which are files to only include - once the scan is finished or the user wants to download the zip - """ - scan_folder = self.dir_for_scan(scan_name=scan_name) - - zip_fname = os.path.join(scan_folder, "images.zip") - - # get a list of files in the existing zip - current_zip = self.get_files_in_zip(zip_fname) - - # get a list of files in the folder we're zipping - - files = glob.glob(scan_folder + "/**/*", recursive=True) - files = [os.path.relpath(file, scan_folder) for file in files] - - # This is a list of file names that are updated as the scan goes, - # and should only be zipped at the end of the scan - otherwise they'll - # be appended on every loop as we can't overwrite files in the zip - - # More elegant ways would be to only zip the images matching the global regex, - # which will all be images, then append other files at the end. - # Alternatively, use "output_dir" in stitching to separate stitching image files - files_to_delay = [ - "TileConfiguration", - "tiling_cache", - "stitched.jp", - "stitched_from", - "stitched.om", - "stitching_correlations", - "preview.jp", - ] - - with zipfile.ZipFile(zip_fname, mode="a") as scan_zip: - for file in files: - if any(skipped_name in file for skipped_name in files_to_delay): - pass - elif file in current_zip: - pass - elif ".zip" in file: - pass - else: - scan_zip.write(os.path.join(scan_folder, file), arcname=file) - - # Finally, zip the files skipped previously - if final_version: - with zipfile.ZipFile(zip_fname, mode="a") as scan_zip: - for file in files: - if any(delayed_name in file for delayed_name in files_to_delay): - scan_zip.write(os.path.join(scan_folder, file), arcname=file) - @thing_action def download_zip( self, @@ -1271,43 +944,5 @@ class SmartScanThing(Thing): ): """Update the zip to include the files left until the end, then return the zip file as a Blob""" - self.update_zip( - scan_name=scan_name, - final_version=True, - ) - - scan_folder = self.dir_for_scan(scan_name=scan_name) - - zip_fname = os.path.join(scan_folder, "images.zip") + zip_fname = self._scan_dir_manager.zip_scan(scan_name, final_version=True) return ZipBlob.from_file(zip_fname) - - def get_files_in_zip(self, zip_path): - """List the relative paths of all files and folders in the zip folder specified""" - scan_zip = zipfile.ZipFile(zip_path) - return [os.path.normpath(i) for i in scan_zip.namelist()] - - @thing_action - def purge_empty_scans(self, logger: InvocationLogger) -> None: - """ - Delete all scan folders containing no images at the top level - """ - scan_list = self.scans - - # Filter out scans with no top level files, ignoring JSON files - # JSON is ignored as it's created before any images are captured - for scan in scan_list: - scan_folder = os.path.join(self.base_scan_dir, scan.name, IMG_DIR_NAME) - images_found = False - # Check the scan directory exists, and if it does loop through each file - # to check if they are scan captures. - if os.path.isdir(scan_folder): - for fname in os.listdir(scan_folder): - fpath = os.path.join(scan_folder, fname) - if os.path.isfile(fpath) and IMAGE_REGEX.search(fname): - images_found = True - # break as soon as an image is found. - break - - if not images_found: - path = os.path.join(self.base_scan_dir, scan.name) - self._delete_scan(path, logger) diff --git a/tests/test_stack.py b/tests/test_stack.py index c9fad430..e9295601 100644 --- a/tests/test_stack.py +++ b/tests/test_stack.py @@ -17,7 +17,7 @@ from openflexure_microscope_server.things.autofocus import ( _get_capture_by_id, _get_capture_index_by_id, ) -from openflexure_microscope_server.things.smart_scan import IMAGE_REGEX +from openflexure_microscope_server.scan_directories import IMAGE_REGEX RANDOM_GENERATOR = np.random.default_rng() From 1267d881b91648f72ab76fa8df9d5015ee38baf8 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sat, 14 Jun 2025 13:27:20 +0100 Subject: [PATCH 3/9] Start writing tests for smart scan Focusing on methods that don't start the whole scan. --- .../things/smart_scan.py | 8 +- tests/test_smart_scan.py | 137 ++++++++++++++++++ 2 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 tests/test_smart_scan.py diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index bda19bfb..3164f7ba 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -48,6 +48,10 @@ STITCHING_CMD = "openflexure-stitch" STITCHING_RESOLUTION = (820, 616) +class ScanNotRunningError(RuntimeError): + """Method called when scan not running that requires a scan to be running""" + + def _scan_running(method): """ This decorator is used by all methods in SmartScanThing that are using @@ -60,7 +64,7 @@ def _scan_running(method): # Only start the method is the scan logger is set if self._scan_logger is not None: return method(self, *args, **kwargs) - raise RuntimeError( + raise ScanNotRunningError( "Calling a @scan_running method can only be done while a scan is running!" ) @@ -665,7 +669,7 @@ class SmartScanThing(Thing): "scans/{scan_name}", responses={ 200: {"description": "Successfully deleted scan"}, - 404: {"description": "Scan not found"}, + 400: {"description": "Scan not deleted does"}, }, ) def delete_scan(self, scan_name: str, logger: InvocationLogger) -> None: diff --git a/tests/test_smart_scan.py b/tests/test_smart_scan.py new file mode 100644 index 00000000..879f7abf --- /dev/null +++ b/tests/test_smart_scan.py @@ -0,0 +1,137 @@ +""" +Test the SmartScanThing *without* connecting it to a LabThings Server. + +By testing without connecting to the LabThings server it is possible to +directly poll any properties and to start any methods. + +Any methods with LabThings dependency injections will require dependencies +to be passed in manually. Rather than passing in real LabThings clients +it is possible to create test objects that mock these clients. Thus, entirely +isolating one Thing for testing, at the expense of neededing to create detailed +mock objects. + +For these tests to reliably represent real behaviour the mock Things will need to +be tested for matching signatures with dynamically generated clients. +""" + +import tempfile +import os +import shutil +import logging + +from fastapi import HTTPException +import pytest + +from openflexure_microscope_server.things.smart_scan import ( + SmartScanThing, + ScanNotRunningError, +) + +# 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 +# have some control of when it is deleted +SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") + + +def _clear_scan_dir(): + """Delete the scan dir""" + if os.path.exists(SCAN_DIR): + shutil.rmtree(SCAN_DIR) + + +@pytest.fixture +def smart_scan_thing(): + """Return a smart scan thing as a fixture""" + return SmartScanThing(SCAN_DIR) + + +def test_initial_properties(smart_scan_thing): + assert smart_scan_thing._scan_dir_manager.base_dir == SCAN_DIR + assert smart_scan_thing.latest_scan_name is None + + +def test_inaccessible_scan_methods(smart_scan_thing): + """The @_scan_running decorator should make some methods + inaccessible unless a scan is running""" + with pytest.raises(ScanNotRunningError): + smart_scan_thing._run_scan() + with pytest.raises(ScanNotRunningError): + smart_scan_thing._manage_stitching_threads() + + +def test_private_delete_scan(smart_scan_thing, caplog): + """Test the private _delete_scan method deletes directories or warns if it can't""" + + _clear_scan_dir() + with caplog.at_level(logging.INFO): + fake_scan_name = "fake_scan_0001" + fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) + + # Make the outer scan dir, but not the one to delete + os.makedirs(SCAN_DIR) + deleted = smart_scan_thing._delete_scan(fake_scan_name, LOGGER) + assert not deleted + assert len(caplog.records) == 1 + assert caplog.records[0].levelname == "WARNING" + assert caplog.records[0].name == "mock-invocation_logger" + + # Make a dir for the fake scan and delete it. + os.makedirs(fake_scan_path) + assert os.path.exists(fake_scan_path) + deleted = smart_scan_thing._delete_scan(fake_scan_name, LOGGER) + assert not os.path.exists(fake_scan_path) + assert deleted + # Check no extra logs generated + assert len(caplog.records) == 1 + + +def test_public_delete_scan(smart_scan_thing, caplog): + """Test the delete_scan API call deletes directories or warns if it can't""" + + _clear_scan_dir() + with caplog.at_level(logging.INFO): + fake_scan_name = "fake_scan_0001" + fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) + + # Make the outer scan dir, but not the one to delete + os.makedirs(SCAN_DIR) + + with pytest.raises(HTTPException) as exc_info: + smart_scan_thing.delete_scan(fake_scan_name, LOGGER) + # Should raise a 400 error if the scan doesn't exists, not a 404 as the server + # was not expecting to receive the scan files + assert exc_info.value.status_code == 400 + assert len(caplog.records) == 1 + assert caplog.records[0].levelname == "WARNING" + assert caplog.records[0].name == "mock-invocation_logger" + + # Make a dir for the fake scan and delete it. + os.makedirs(fake_scan_path) + assert os.path.exists(fake_scan_path) + smart_scan_thing.delete_scan(fake_scan_name, LOGGER) + assert not os.path.exists(fake_scan_path) + # Check no extra logs generated + assert len(caplog.records) == 1 + + +def test_delete_all_scans(smart_scan_thing, caplog): + _clear_scan_dir() + with caplog.at_level(logging.INFO): + fake_scan_names = [ + "fake_scan_0001", + "fake_scan_0002", + "fake_scan_0003", + "fake_scan_0004", + ] + for fake_scan_name in fake_scan_names: + fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) + os.makedirs(fake_scan_path) + assert os.path.exists(fake_scan_path) + smart_scan_thing.delete_all_scans(LOGGER) + for fake_scan_name in fake_scan_names: + fake_scan_path = os.path.join(SCAN_DIR, fake_scan_name) + assert not os.path.exists(fake_scan_path) + # No logs generated + assert len(caplog.records) == 0 From 3aa9b9100a8cac952135b6bacdbc7588120341ca Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sun, 15 Jun 2025 00:30:12 +0100 Subject: [PATCH 4/9] Add a number of tests for scan directories and fix bugs --- .../scan_directories.py | 20 +- tests/test_scan_directories.py | 228 ++++++++++++++++++ 2 files changed, 240 insertions(+), 8 deletions(-) create mode 100644 tests/test_scan_directories.py diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 6fccb837..d36f11a4 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -78,7 +78,7 @@ class ScanDirectoryManager: """ file_path = os.path.join(self.path_for(scan_name), filename) if check_exists: - if not os.path.exists: + if not os.path.exists(file_path): return None return file_path @@ -92,7 +92,7 @@ class ScanDirectoryManager: """ file_path = os.path.join(self.img_dir_for(scan_name), filename) if check_exists: - if not os.path.exists: + if not os.path.exists(file_path): return None return file_path @@ -106,7 +106,7 @@ class ScanDirectoryManager: all_info: list[ScanInfo] = [] for scan_name in self.all_scans: - all_info.append(ScanDirectory(scan_name, self.base_dir).scan_info) + all_info.append(ScanDirectory(scan_name, self.base_dir).scan_info()) return all_info def _unique_scan_name(self, scan_name: str) -> str: @@ -220,7 +220,7 @@ class ScanDirectory: @property def dir_path(self) -> str: """The full path to the scan directory""" - os.path.join(self._base_scan_dir, self._name) + return os.path.join(self._base_scan_dir, self._name) @property def images_dir(self) -> Optional[str]: @@ -291,17 +291,21 @@ class ScanDirectory: """ zip_fname = os.path.join(self.dir_path, "images.zip") - # get a list of files in the existing zip - zip_files = get_files_in_zip(zip_fname) + + if os.path.isfile(zip_fname): + # get a list of files in the existing zip + zip_files = get_files_in_zip(zip_fname) + else: + zip_files = [] with zipfile.ZipFile(zip_fname, mode="a") as scan_zip: for file in self.all_files(): # Don't zip zipfiles, or files in the zip if file.endswith(".zip") or file in zip_files: - break + continue # If this is not the final version, then only zip image files. if not final_version and not IMAGE_REGEX.search(file): - break + continue scan_zip.write(os.path.join(self.dir_path, file), arcname=file) return zip_fname diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py new file mode 100644 index 00000000..45a22165 --- /dev/null +++ b/tests/test_scan_directories.py @@ -0,0 +1,228 @@ +import tempfile +import os +import shutil +import logging +import random +import time + +from openflexure_microscope_server.scan_directories import ( + ScanDirectoryManager, + ScanDirectory, + ScanInfo, + get_files_in_zip, +) + +# 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 +# have some control of when it is deleted +BASE_SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") + + +def _clear_scan_dir(): + """Delete the scan dir""" + if os.path.exists(BASE_SCAN_DIR): + shutil.rmtree(BASE_SCAN_DIR) + + +def _add_fake_image(scan_dir: ScanDirectory): + """Make a fake image on disk in the scan directory""" + x_pos = random.randint(-100, 100) + y_pos = random.randint(-100, 100) + filename = f"image_{x_pos}_{y_pos}.jpg" + filepath = os.path.join(scan_dir.images_dir, filename) + with open(filepath, "w") as f_obj: + f_obj.write("fake") + + +def _add_fake_file(scan_dir: ScanDirectory, filename: str, in_im_dir=False): + """Make a fake file on disk in the scan directory""" + if in_im_dir: + filepath = os.path.join(scan_dir.images_dir, filename) + else: + filepath = os.path.join(scan_dir.dir_path, filename) + with open(filepath, "w") as f_obj: + f_obj.write("fake") + + +def test_basic_directory_operations(): + """Test some basic operations + + Rather a long test but systematically iterates through some basics. + """ + _clear_scan_dir() + assert not os.path.isdir(BASE_SCAN_DIR) + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + # Check it makes the scan directory + assert os.path.isdir(BASE_SCAN_DIR) + + # Considering a fake scan + scan_name = "fake_scan_0001" + scan_path = os.path.join(BASE_SCAN_DIR, scan_name) + scan_im_dir = os.path.join(BASE_SCAN_DIR, scan_name, "images") + + # It doesn't exitist but we can check its paths + assert not scan_dir_manager.exists(scan_name) + assert not os.path.isdir(scan_path) + assert scan_dir_manager.path_for(scan_name) == scan_path + assert scan_dir_manager.img_dir_for(scan_name) == scan_im_dir + + # or a get the path of a fake file + fake_file = scan_dir_manager.get_file_from(scan_name, "foo.zip") + assert fake_file == os.path.join(scan_path, "foo.zip") + # But this is none if we check it exists + fake_file = scan_dir_manager.get_file_from(scan_name, "foo.zip", check_exists=True) + assert fake_file is None + + # or a get the path of a fake file + fake_file = scan_dir_manager.get_file_from_img_dir(scan_name, "bar.img") + assert fake_file == os.path.join(scan_im_dir, "bar.img") + # But this is none if we check it exists + fake_file = scan_dir_manager.get_file_from_img_dir( + scan_name, "bar.img", check_exists=True + ) + assert fake_file is None + + # Create the dir + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + # This returns a ScanDirectory object + assert isinstance(scan_dir, ScanDirectory) + # The directory now exists as does the image directory + assert scan_dir_manager.exists(scan_name) + assert os.path.isdir(scan_path) + assert os.path.isdir(scan_im_dir) + + scan_dir_manager.delete_scan(scan_name) + assert not scan_dir_manager.exists(scan_name) + assert not os.path.isdir(scan_path) + + +def test_scan_sequence_and_listing(): + """ + Check created scans are added in order and listed correctly + """ + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + # Make 4 scans + scan_dir_manager.new_scan_dir("fake_scan") + scan_dir_manager.new_scan_dir("fake_scan") + scan_dir_manager.new_scan_dir("fake_scan") + scan_dir_manager.new_scan_dir("fake_scan") + + # Check they exist and are numbered sequentially + all_scans = scan_dir_manager.all_scans + assert len(all_scans) == 4 + assert "fake_scan_0001" in all_scans + assert "fake_scan_0002" in all_scans + assert "fake_scan_0003" in all_scans + assert "fake_scan_0004" in all_scans + + # Check scan data can be read for all of them + # (more detailed scan_info tests below) + all_scan_info = scan_dir_manager.all_scans_info() + for scan_info in all_scan_info: + assert isinstance(scan_info, ScanInfo) + assert scan_info.name.startswith("fake_scan_000") + assert scan_info.number_of_images == 0 + + +def test_scan_name_non_sequential(): + """ + Check created scans is the correct name if the directories + are not sequential + """ + _clear_scan_dir() + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001")) + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0002")) + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0003")) + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0005")) + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0007")) + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0011")) + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + assert scan_dir.name == "fake_scan_0012" + + all_scans = scan_dir_manager.all_scans + assert len(all_scans) == 7 + assert "fake_scan_0001" in all_scans + assert "fake_scan_0002" in all_scans + assert "fake_scan_0003" in all_scans + assert "fake_scan_0005" in all_scans + assert "fake_scan_0007" in all_scans + assert "fake_scan_0011" in all_scans + assert "fake_scan_0012" in all_scans + + +def test_scan_info(): + """Test the scan info is correct even using fake scan data""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + + for i in range(17): + _add_fake_image(scan_dir) + info = scan_dir.scan_info() + now = time.time() + + assert info.name == "fake_scan_0001" + # Created and modifies in the last 5 seconds + assert now - 5 < info.created < now + # Created and modifies in the last 5 seconds + assert now - 5 < info.created < now + assert info.number_of_images == 17 + assert not info.stitch_available + assert info.dzi is None + + # Add a fake scan and check this recognised as a stitch not a scan image + _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) + info = scan_dir.scan_info() + assert info.number_of_images == 17 + assert info.stitch_available + + +def test_empty_scan_info(): + """Test the scan info is correct even if the scan is empty""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + + info = scan_dir.scan_info() + now = time.time() + + assert info.name == "fake_scan_0001" + # Created and modifies in the last 5 seconds + assert now - 5 < info.created < now + # Created and modifies in the last 5 seconds + assert now - 5 < info.created < now + assert info.number_of_images == 0 + assert not info.stitch_available + assert info.dzi is None + + +def test_zipping_scan_data(): + """Test zipping the scan images with fake image data""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + + # Create 21 fake scan files, a fake stitch, and a fake zip + for i in range(21): + _add_fake_image(scan_dir) + _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) + _add_fake_file(scan_dir, "zipfile.zip") + + # zip the directory without setting as the final version. It should only + # zip the 21 scan images + zip_fname = scan_dir.zip_files() + zip_files = get_files_in_zip(zip_fname) + assert len(zip_files) == 21 + + # Zip again with final version on and there should be 22 images + scan_dir.zip_files(final_version=True) + zip_files = get_files_in_zip(zip_fname) + assert len(get_files_in_zip(zip_fname)) == 22 + # Check the zips are not in the zip + for file in zip_files: + assert not file.endswith(".zip") From 7193ec34f55fa3d4849c0c7f72a77039ea61e04a Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Mon, 16 Jun 2025 05:03:57 +0100 Subject: [PATCH 5/9] Starting to test more smart scan functionality using mocked Thing Dependencies --- .../things/smart_scan.py | 4 +- tests/mock_things/__init__.py | 0 tests/mock_things/mock_autofocus.py | 17 +++ tests/mock_things/mock_background_detect.py | 20 +++ tests/mock_things/mock_csm.py | 12 ++ tests/mock_things/mock_stage.py | 12 ++ tests/test_smart_scan.py | 120 ++++++++++++++++++ 7 files changed, 184 insertions(+), 1 deletion(-) create mode 100644 tests/mock_things/__init__.py create mode 100644 tests/mock_things/mock_autofocus.py create mode 100644 tests/mock_things/mock_background_detect.py create mode 100644 tests/mock_things/mock_csm.py create mode 100644 tests/mock_things/mock_stage.py diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 3164f7ba..c7dea601 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -138,7 +138,9 @@ class SmartScanThing(Thing): self._capture_thread = None self._scan_images_taken = 0 - # Don't set self._scan_data dictionary. This is done at the start of _run_scan + # Set _scan_data to None. It will for error checking. It will be set to a value + # by _run_scan + self._scan_data = None try: self._check_background_and_csm_set() diff --git a/tests/mock_things/__init__.py b/tests/mock_things/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/mock_things/mock_autofocus.py b/tests/mock_things/mock_autofocus.py new file mode 100644 index 00000000..6db47aa9 --- /dev/null +++ b/tests/mock_things/mock_autofocus.py @@ -0,0 +1,17 @@ +"""This testing submodule contains mock autofocus things. + +These mocks are designed to be inserted as dependencies to give specific +functionality and returns. + +The mocks do not subclass, and instead return very specific defined answers +to functions +""" + + +class MockAutoFocusThing: + # Counter for checking functions were called + mock_call_count = {"looping_autofocus": 0} + + def looping_autofocus(self, dz=2000, start="centre"): # noqa: ARG002 + """This function mocks autofocus with no return""" + self.mock_call_count["looping_autofocus"] += 1 diff --git a/tests/mock_things/mock_background_detect.py b/tests/mock_things/mock_background_detect.py new file mode 100644 index 00000000..a7eb5c5b --- /dev/null +++ b/tests/mock_things/mock_background_detect.py @@ -0,0 +1,20 @@ +"""This testing submodule contains mock autofocus things. + +These mocks are designed to be inserted as dependencies to give specific +functionality and returns. + +The mocks do not subclass, and instead return very specific defined answers +to functions +""" + +from openflexure_microscope_server.things.background_detect import ChannelDistributions + + +class MockBackgoundDetectThing: + # Counter for checking functions were called + + background_distributions = ChannelDistributions( + means=[128.0, 128.0, 128.0], + standard_deviations=[3.0, 3.0, 3.0], + colorspace="LUV", + ) diff --git a/tests/mock_things/mock_csm.py b/tests/mock_things/mock_csm.py new file mode 100644 index 00000000..78930dae --- /dev/null +++ b/tests/mock_things/mock_csm.py @@ -0,0 +1,12 @@ +"""This testing submodule contains mock camera stage mapping things. + +These mocks are designed to be inserted as dependencies to give specific +functionality and returns. + +The mocks do not subclass, and instead return very specific defined answers +to functions +""" + + +class MockCSMThing: + image_resolution = (123, 456) diff --git a/tests/mock_things/mock_stage.py b/tests/mock_things/mock_stage.py new file mode 100644 index 00000000..9b7a466d --- /dev/null +++ b/tests/mock_things/mock_stage.py @@ -0,0 +1,12 @@ +"""This testing submodule contains stage things. + +These mocks are designed to be inserted as dependencies to give specific +functionality and returns. + +The mocks do not subclass, and instead return very specific defined answers +to functions +""" + + +class MockStageThing: + position = (111, 222, 333) diff --git a/tests/test_smart_scan.py b/tests/test_smart_scan.py index 879f7abf..44ffcf70 100644 --- a/tests/test_smart_scan.py +++ b/tests/test_smart_scan.py @@ -14,6 +14,7 @@ For these tests to reliably represent real behaviour the mock Things will need t be tested for matching signatures with dynamically generated clients. """ +from typing import Callable, Optional import tempfile import os import shutil @@ -27,6 +28,11 @@ from openflexure_microscope_server.things.smart_scan import ( ScanNotRunningError, ) +from .mock_things.mock_csm import MockCSMThing +from .mock_things.mock_autofocus import MockAutoFocusThing +from .mock_things.mock_stage import MockStageThing +from .mock_things.mock_background_detect import MockBackgoundDetectThing + # A global logger to pass in as an Invocation Logger LOGGER = logging.getLogger("mock-invocation_logger") @@ -135,3 +141,117 @@ def test_delete_all_scans(smart_scan_thing, caplog): assert not os.path.exists(fake_scan_path) # No logs generated assert len(caplog.records) == 0 + + +def _run_only_outer_scan(adjust_inital_state: Optional[Callable] = None): + """ + Create a subclass of SmartScanThing to mock _run_scan and run sample_scan + + This should do all the set up for a scan, move into the mocked + _run_scan method where this can be tested. Once this is done + the final scan behaviour can be tested too. + + adjust_initial_scan is a callable which accepts the mocked smart scan thing + as the only variable. It can be used to adjust the initial state of the test + + + This seems hard to do with a fixture so it is being done with a private + function + """ + + # cancel handle shouldn't be used. Set to arbitrary value for checking + cancel_mock = 1 # not called + af_mock = MockAutoFocusThing() + stage_mock = MockStageThing() + cam_mock = 4 # not called + meta_mock = 5 # not called + csm_mock = MockCSMThing() + bkgrnd_det_mock = MockBackgoundDetectThing() + + class MockedSmartScanThing(SmartScanThing): + """ + This is a subclass of SmartScanThing with a mocked method and + mocked thing_settings. + """ + + # Counter for checking functions were called + mock_call_count = {"_run_scan": 0} + + # Mock thing settngs as a dictionary + thing_settings = {"skip_background": True} + + def _run_scan(self): + self.mock_call_count["_run_scan"] += 1 + + """Check scan vars are set up as expected""" + assert not self._scan_lock.acquire(timeout=0.1) + assert self._cancel is cancel_mock + assert self._scan_logger is LOGGER + assert self._autofocus is af_mock + assert self._stage is stage_mock + assert self._cam is cam_mock + assert self._metadata_getter is meta_mock + assert self._csm is csm_mock + assert self._background_detect is bkgrnd_det_mock + assert self._capture_thread is None + assert self._scan_images_taken == 0 + + # mock smart scan thing + mock_ss_thing = MockedSmartScanThing(SCAN_DIR) + + if adjust_inital_state is not None: + adjust_inital_state(mock_ss_thing) + + exec_info = None + try: + mock_ss_thing.sample_scan( + cancel=cancel_mock, # Shouldn't be used, can be checked + logger=LOGGER, + autofocus=af_mock, + stage=stage_mock, + cam=cam_mock, + metadata_getter=meta_mock, + csm=csm_mock, + background_detect=bkgrnd_det_mock, + scan_name="FooBar", + ) + except Exception as e: + exec_info = e + + assert mock_ss_thing._scan_lock.acquire(timeout=0.1) + mock_ss_thing._scan_lock.release() + assert mock_ss_thing._cancel is None + assert mock_ss_thing._scan_logger is None + assert mock_ss_thing._autofocus is None + assert mock_ss_thing._stage is None + assert mock_ss_thing._cam is None + assert mock_ss_thing._metadata_getter is None + assert mock_ss_thing._csm is None + assert mock_ss_thing._background_detect is None + assert mock_ss_thing._capture_thread is None + assert mock_ss_thing._scan_images_taken is None + + # Return the mock thing for further state testing, and the + # exec_info of any uncaught exeptions that were raised + return mock_ss_thing, exec_info + + +def test_outer_scan(): + """Test setup and teardown of the scan.""" + mock_ss_thing, exec_info = _run_only_outer_scan() + assert exec_info is None + # Checked the mocked _run_scan was run exactly once + assert mock_ss_thing.mock_call_count["_run_scan"] == 1 + + +def test_outer_scan_wo_scample_skip(): + """Test setup and teardown of the scan.""" + + def _set_skip_background(mock_ss_thing): + mock_ss_thing.thing_settings["skip_background"] = False + + mock_ss_thing, exec_info = _run_only_outer_scan(_set_skip_background) + + assert exec_info is None + # Checked the mocked _run_scan was run exactly once + assert mock_ss_thing.mock_call_count["_run_scan"] == 1 From 7f0745d77213f1283bcedd862f043b124343ab67 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sat, 28 Jun 2025 23:59:35 +0100 Subject: [PATCH 6/9] Add some more scan_directories tests --- pyproject.toml | 1 + .../scan_directories.py | 2 +- tests/test_scan_directories.py | 124 +++++++++++++++--- 3 files changed, 108 insertions(+), 19 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 5e9a985f..4822ec60 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ dev = [ "mypy-gitlab-code-quality", # "pytest-gitlab-code-quality", # pytest version constraint clashes with labthings "pytest", + "pytest-mock==3.14", "matplotlib~=3.10", "hypothesis", ] diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index d36f11a4..a5001fe5 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -181,7 +181,7 @@ class ScanDirectoryManager: Args: path = path to a location on the disk you want to check min_space [int] = the minimum space required in bytes - default = 500,000,000 (500MiB) + default = 500,000,000 (500MB) Raises: NotEnoughFreeSpaceError if the remaining storage is below min_space diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py index 45a22165..1816035e 100644 --- a/tests/test_scan_directories.py +++ b/tests/test_scan_directories.py @@ -4,12 +4,16 @@ import shutil import logging import random import time +from collections import namedtuple + +import pytest from openflexure_microscope_server.scan_directories import ( ScanDirectoryManager, ScanDirectory, ScanInfo, get_files_in_zip, + NotEnoughFreeSpaceError, ) # A global logger to pass in as an Invocation Logger @@ -155,6 +159,27 @@ def test_scan_name_non_sequential(): assert "fake_scan_0012" in all_scans +def test_all_scan_names_taken(): + """ + If the next sequential scan name needs more than 4 digits check error is thrown. + """ + _clear_scan_dir() + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_9999")) + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + with pytest.raises(FileExistsError): + scan_dir_manager.new_scan_dir("fake_scan") + + +def test_no_scan_names_given(): + """ + Check correct default scan name is used if empty string is given + """ + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("") + assert scan_dir.name == "scan_0001" + + def test_scan_info(): """Test the scan info is correct even using fake scan data""" _clear_scan_dir() @@ -203,26 +228,89 @@ def test_empty_scan_info(): def test_zipping_scan_data(): """Test zipping the scan images with fake image data""" + + # Run twice, once calling the ScanDirectory directly, + # Once calling the ScanDirectoryManager + for caller in ["scan_dir", "manager"]: + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + + # Create 21 fake scan files, a fake stitch, and a fake zip + for i in range(21): + _add_fake_image(scan_dir) + _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) + _add_fake_file(scan_dir, "zipfile.zip") + + # zip the directory without setting as the final version. It should only + # zip the 21 scan images + if caller == "scan_dir": + zip_fname = scan_dir.zip_files() + else: + zip_fname = scan_dir_manager.zip_scan("fake_scan_0001") + zip_files = get_files_in_zip(zip_fname) + assert len(zip_files) == 21 + + # Zip again with final version on and there should be 22 images + if caller == "scan_dir": + scan_dir.zip_files(final_version=True) + else: + scan_dir_manager.zip_scan("fake_scan_0001", final_version=True) + zip_files = get_files_in_zip(zip_fname) + assert len(get_files_in_zip(zip_fname)) == 22 + # Check the zips are not in the zip + for file in zip_files: + assert not file.endswith(".zip") + + +def test_creating_scan_dir_for_missing_scan(): + """Check creating ScanDirectory object for a dir that doesn't exist fails.""" _clear_scan_dir() + with pytest.raises(FileNotFoundError): + ScanDirectory(BASE_SCAN_DIR, "not_real_0001") + + +def test_none_returned_for_missing_images_dir(): + """None should be returned for image dir path if images dir never created.""" + _clear_scan_dir() + os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001")) + scan_dir = ScanDirectory(BASE_SCAN_DIR, "fake_scan_0001") + assert scan_dir.images_dir is None + # Also check that get scan files returns and empty list + assert scan_dir.get_scan_files() == [] + + +DiskUsage = namedtuple("DiskUsage", ["total", "used", "free"]) + + +@pytest.mark.parametrize("free_space", [500_000_001, 700_000_000, 5_000_000_000]) +def test_disk_not_full(mocker, free_space): + """Check no error thrown if disk has over 500MB of space""" + total_space = 16_000_000_000 + # Mock the disk_usage + mocker.patch( + "shutil.disk_usage", + return_value=DiskUsage( + total=total_space, used=total_space - free_space, free=free_space + ), + ) + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) - scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_dir_manager.check_free_disk_space() - # Create 21 fake scan files, a fake stitch, and a fake zip - for i in range(21): - _add_fake_image(scan_dir) - _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) - _add_fake_file(scan_dir, "zipfile.zip") - # zip the directory without setting as the final version. It should only - # zip the 21 scan images - zip_fname = scan_dir.zip_files() - zip_files = get_files_in_zip(zip_fname) - assert len(zip_files) == 21 +@pytest.mark.parametrize("free_space", [100_000_000, 499_999_999]) +def test_disk_full(mocker, free_space): + """Check error thrown if disk has under 500MB of space""" + total_space = 16_000_000_000 + # Mock the disk_usage + mocker.patch( + "shutil.disk_usage", + return_value=DiskUsage( + total=total_space, used=total_space - free_space, free=free_space + ), + ) - # Zip again with final version on and there should be 22 images - scan_dir.zip_files(final_version=True) - zip_files = get_files_in_zip(zip_fname) - assert len(get_files_in_zip(zip_fname)) == 22 - # Check the zips are not in the zip - for file in zip_files: - assert not file.endswith(".zip") + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + with pytest.raises(NotEnoughFreeSpaceError): + scan_dir_manager.check_free_disk_space() From a6e48f8882ad4b4f1e76aca2e7fba24f7489164b Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sun, 29 Jun 2025 23:17:17 +0100 Subject: [PATCH 7/9] Update webapp to accept scan time as timestamp not ISO 8601 datestring --- .../src/components/tabContentComponents/scanListContent.vue | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/webapp/src/components/tabContentComponents/scanListContent.vue b/webapp/src/components/tabContentComponents/scanListContent.vue index 293b9d0c..6b342d4b 100644 --- a/webapp/src/components/tabContentComponents/scanListContent.vue +++ b/webapp/src/components/tabContentComponents/scanListContent.vue @@ -240,8 +240,6 @@ export default { this.scans = scans; } scans.forEach(scan => { - scan.modified = Date.parse(scan.modified); - scan.created = Date.parse(scan.created); scan.can_stitch = !scan.stitch_available && scan.number_of_images > 3; }); scans.sort((a, b) => { @@ -255,7 +253,9 @@ export default { } }, formatDate(timestamp) { - let d = new Date(timestamp); + // Multiply by 1000 as JS uses ms not s + let d = new Date(timestamp*1000); + // Convert to a sting in a very javascript way! let yyyy = d.getFullYear(); let mm = d.getMonth() + 1; let dd = d.getDate(); From ad838e86439f8c256b112a5e5f5248ab40f143f6 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Mon, 30 Jun 2025 15:30:05 +0000 Subject: [PATCH 8/9] Apply suggestions from code review of branch split-scan-dir-operations Co-authored-by: Beth Probert --- .../scan_directories.py | 18 +++---- .../things/smart_scan.py | 6 +-- tests/mock_things/mock_autofocus.py | 2 +- tests/mock_things/mock_background_detect.py | 2 +- tests/test_scan_directories.py | 50 +++++++++++-------- tests/test_smart_scan.py | 18 ++++--- .../tabContentComponents/scanListContent.vue | 2 +- 7 files changed, 55 insertions(+), 43 deletions(-) diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index a5001fe5..25cccd2a 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -35,7 +35,7 @@ class ScanInfo(BaseModel): class ScanDirectoryManager: """ - A class for managinging interactions with scan directories + A class for managing interactions with scan directories """ _base_scan_dir: str @@ -57,7 +57,7 @@ class ScanDirectoryManager: def path_for(self, scan_name: str) -> str: """Return the path for a given scan name - Returns the path even if it doesn't exist) + Returns the path even if it doesn't exist """ return os.path.join(self._base_scan_dir, scan_name) @@ -97,7 +97,7 @@ class ScanDirectoryManager: return file_path @property - def all_scans(self): + def all_scans(self) -> list[str]: """Return a list of the scan names in the base directory""" return [f.name for f in os.scandir(self._base_scan_dir) if f.is_dir()] @@ -138,11 +138,11 @@ class ScanDirectoryManager: return f"{scan_name}_{scan_num:0{SCAN_ZERO_PAD_DIGITS}d}" - def new_scan_dir(self, scan_name: str) -> str: + def new_scan_dir(self, scan_name: str) -> "ScanDirectory": """Get a unique name for this scan and create a directory for it The scan will be named `{scan_name}_0001` where the number is - zero-padded to be 4 digits long (to allow correct sorting if the + zero-padded to be SCAN_ZERO_PAD_DIGITS digits long (to allow correct sorting if the scans are ordered alphanumerically). Creates a new empty folder, into which scans are saved @@ -161,11 +161,11 @@ class ScanDirectoryManager: os.makedirs(self.img_dir_for(full_scan_name)) return ScanDirectory(full_scan_name, self.base_dir) - def delete_scan(self, scan_name: str): + def delete_scan(self, scan_name: str) -> None: """Delete a scan""" shutil.rmtree(self.path_for(scan_name)) - def zip_scan(self, scan_name: str, final_version: bool = False) -> str: + def zip_scan(self, scan_name: str, final_version: bool = False) -> "ScanDirectory": """Zips any images from the scan not yet zipped, return full path to zip `final_version` Set true to stitch all files not just the scan images @@ -213,7 +213,7 @@ class ScanDirectory: ) @property - def name(self): + def name(self) -> str: """The name of the scan""" return self._name @@ -249,7 +249,7 @@ class ScanDirectory: return max(os.stat(root).st_mtime for root, _, _ in os.walk(self.dir_path)) def scan_info(self): - """Return the inforomation for the scan directory as a ScanInfo object""" + """Return the information for the scan directory as a ScanInfo object""" folder_contents = self.get_scan_files() if folder_contents: scan_images = [i for i in folder_contents if IMAGE_REGEX.search(i)] diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index c7dea601..9af232e4 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -155,7 +155,7 @@ class SmartScanThing(Thing): if self._scan_data is not None: self._return_to_starting_position() if not isinstance(e, scan_directories.NotEnoughFreeSpaceError): - # Don't stich if drive is full (already logged) + # Don't stitch if drive is full (already logged) self._perform_final_stitch() # Error must be raised so UI gives correct output raise e @@ -671,7 +671,7 @@ class SmartScanThing(Thing): "scans/{scan_name}", responses={ 200: {"description": "Successfully deleted scan"}, - 400: {"description": "Scan not deleted does"}, + 400: {"description": "An error occurred while trying to delete scan"}, }, ) def delete_scan(self, scan_name: str, logger: InvocationLogger) -> None: @@ -744,7 +744,7 @@ class SmartScanThing(Thing): This will return None (`null` to JS) if there is no preview image to return. - This is used for two things reasons: + This is used for two reasons: 1. If all caching was turned off this stitch would be sent over the network repeatedly 2. If caching was is on, then the stitch will not update when needed. diff --git a/tests/mock_things/mock_autofocus.py b/tests/mock_things/mock_autofocus.py index 6db47aa9..d32b5a50 100644 --- a/tests/mock_things/mock_autofocus.py +++ b/tests/mock_things/mock_autofocus.py @@ -12,6 +12,6 @@ class MockAutoFocusThing: # Counter for checking functions were called mock_call_count = {"looping_autofocus": 0} - def looping_autofocus(self, dz=2000, start="centre"): # noqa: ARG002 + def looping_autofocus(self, dz: int = 2000, start: str = "centre") -> None: # noqa: ARG002 """This function mocks autofocus with no return""" self.mock_call_count["looping_autofocus"] += 1 diff --git a/tests/mock_things/mock_background_detect.py b/tests/mock_things/mock_background_detect.py index a7eb5c5b..bd57ee61 100644 --- a/tests/mock_things/mock_background_detect.py +++ b/tests/mock_things/mock_background_detect.py @@ -1,4 +1,4 @@ -"""This testing submodule contains mock autofocus things. +"""This testing submodule contains mock background detect things. These mocks are designed to be inserted as dependencies to give specific functionality and returns. diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py index 1816035e..e9bd63c9 100644 --- a/tests/test_scan_directories.py +++ b/tests/test_scan_directories.py @@ -24,13 +24,13 @@ LOGGER = logging.getLogger("mock-invocation_logger") BASE_SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") -def _clear_scan_dir(): +def _clear_scan_dir() -> None: """Delete the scan dir""" if os.path.exists(BASE_SCAN_DIR): shutil.rmtree(BASE_SCAN_DIR) -def _add_fake_image(scan_dir: ScanDirectory): +def _add_fake_image(scan_dir: ScanDirectory) -> None: """Make a fake image on disk in the scan directory""" x_pos = random.randint(-100, 100) y_pos = random.randint(-100, 100) @@ -40,8 +40,14 @@ def _add_fake_image(scan_dir: ScanDirectory): f_obj.write("fake") -def _add_fake_file(scan_dir: ScanDirectory, filename: str, in_im_dir=False): - """Make a fake file on disk in the scan directory""" +def _add_fake_file( + scan_dir: ScanDirectory, filename: str, in_im_dir: bool = False +) -> None: + """Make a fake file on disk in the scan directory. + + :param in_im_dir: Boolean, if set True the fake file is created in the images + directory of the scan not the root directory. + """ if in_im_dir: filepath = os.path.join(scan_dir.images_dir, filename) else: @@ -53,7 +59,12 @@ def _add_fake_file(scan_dir: ScanDirectory, filename: str, in_im_dir=False): def test_basic_directory_operations(): """Test some basic operations - Rather a long test but systematically iterates through some basics. + Test some basic operations, including: + - ScanDirectoryManager creates a scan directory + - For a fake (dummy) scan, a path can be created and retrieved + - For a fake (dummy) file, a path can be created and retrieved (for both a .zip and .img file) + - When a scan directory is created, the directory exists as does the image directory + - Scan directories can be deleted and scans can be cleared """ _clear_scan_dir() assert not os.path.isdir(BASE_SCAN_DIR) @@ -66,20 +77,20 @@ def test_basic_directory_operations(): scan_path = os.path.join(BASE_SCAN_DIR, scan_name) scan_im_dir = os.path.join(BASE_SCAN_DIR, scan_name, "images") - # It doesn't exitist but we can check its paths + # It doesn't exist but we can check its paths assert not scan_dir_manager.exists(scan_name) assert not os.path.isdir(scan_path) assert scan_dir_manager.path_for(scan_name) == scan_path assert scan_dir_manager.img_dir_for(scan_name) == scan_im_dir - # or a get the path of a fake file + # Get the path of a fake file fake_file = scan_dir_manager.get_file_from(scan_name, "foo.zip") assert fake_file == os.path.join(scan_path, "foo.zip") # But this is none if we check it exists fake_file = scan_dir_manager.get_file_from(scan_name, "foo.zip", check_exists=True) assert fake_file is None - # or a get the path of a fake file + # Get the path of another fake file fake_file = scan_dir_manager.get_file_from_img_dir(scan_name, "bar.img") assert fake_file == os.path.join(scan_im_dir, "bar.img") # But this is none if we check it exists @@ -97,6 +108,7 @@ def test_basic_directory_operations(): assert os.path.isdir(scan_path) assert os.path.isdir(scan_im_dir) + # Test cleanup. Delete the created scan and scan directory scan_dir_manager.delete_scan(scan_name) assert not scan_dir_manager.exists(scan_name) assert not os.path.isdir(scan_path) @@ -116,7 +128,7 @@ def test_scan_sequence_and_listing(): # Check they exist and are numbered sequentially all_scans = scan_dir_manager.all_scans - assert len(all_scans) == 4 + assert len(set(all_scans)) == 4 assert "fake_scan_0001" in all_scans assert "fake_scan_0002" in all_scans assert "fake_scan_0003" in all_scans @@ -149,7 +161,7 @@ def test_scan_name_non_sequential(): assert scan_dir.name == "fake_scan_0012" all_scans = scan_dir_manager.all_scans - assert len(all_scans) == 7 + assert len(set(all_scans)) == 7 assert "fake_scan_0001" in all_scans assert "fake_scan_0002" in all_scans assert "fake_scan_0003" in all_scans @@ -192,15 +204,14 @@ def test_scan_info(): now = time.time() assert info.name == "fake_scan_0001" - # Created and modifies in the last 5 seconds - assert now - 5 < info.created < now - # Created and modifies in the last 5 seconds + # Created and modified in the last 5 seconds assert now - 5 < info.created < now + assert now - 5 < info.modified < now assert info.number_of_images == 17 assert not info.stitch_available assert info.dzi is None - # Add a fake scan and check this recognised as a stitch not a scan image + # Add a fake scan and check this is recognised as a stitch not a scan image _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) info = scan_dir.scan_info() assert info.number_of_images == 17 @@ -217,10 +228,9 @@ def test_empty_scan_info(): now = time.time() assert info.name == "fake_scan_0001" - # Created and modifies in the last 5 seconds - assert now - 5 < info.created < now - # Created and modifies in the last 5 seconds + # Created and modified in the last 5 seconds assert now - 5 < info.created < now + assert now - 5 < info.modified < now assert info.number_of_images == 0 assert not info.stitch_available assert info.dzi is None @@ -236,7 +246,7 @@ def test_zipping_scan_data(): scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) scan_dir = scan_dir_manager.new_scan_dir("fake_scan") - # Create 21 fake scan files, a fake stitch, and a fake zip + # Create 21 fake scan images, a fake stitch, and a fake zip for i in range(21): _add_fake_image(scan_dir) _add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True) @@ -249,7 +259,7 @@ def test_zipping_scan_data(): else: zip_fname = scan_dir_manager.zip_scan("fake_scan_0001") zip_files = get_files_in_zip(zip_fname) - assert len(zip_files) == 21 + assert len(set(zip_files)) == 21 # Zip again with final version on and there should be 22 images if caller == "scan_dir": @@ -257,7 +267,7 @@ def test_zipping_scan_data(): else: scan_dir_manager.zip_scan("fake_scan_0001", final_version=True) zip_files = get_files_in_zip(zip_fname) - assert len(get_files_in_zip(zip_fname)) == 22 + assert len(set(get_files_in_zip(zip_fname))) == 22 # Check the zips are not in the zip for file in zip_files: assert not file.endswith(".zip") diff --git a/tests/test_smart_scan.py b/tests/test_smart_scan.py index 44ffcf70..da23c4b9 100644 --- a/tests/test_smart_scan.py +++ b/tests/test_smart_scan.py @@ -7,7 +7,7 @@ directly poll any properties and to start any methods. Any methods with LabThings dependency injections will require dependencies to be passed in manually. Rather than passing in real LabThings clients it is possible to create test objects that mock these clients. Thus, entirely -isolating one Thing for testing, at the expense of neededing to create detailed +isolating one Thing for testing, at the expense of needing to create detailed mock objects. For these tests to reliably represent real behaviour the mock Things will need to @@ -41,7 +41,7 @@ LOGGER = logging.getLogger("mock-invocation_logger") SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans") -def _clear_scan_dir(): +def _clear_scan_dir() -> None: """Delete the scan dir""" if os.path.exists(SCAN_DIR): shutil.rmtree(SCAN_DIR) @@ -77,6 +77,7 @@ def test_private_delete_scan(smart_scan_thing, caplog): # Make the outer scan dir, but not the one to delete os.makedirs(SCAN_DIR) + # Attempt to delete the fake scan. Expect it to fail and provide a warning deleted = smart_scan_thing._delete_scan(fake_scan_name, LOGGER) assert not deleted assert len(caplog.records) == 1 @@ -104,9 +105,10 @@ def test_public_delete_scan(smart_scan_thing, caplog): # Make the outer scan dir, but not the one to delete os.makedirs(SCAN_DIR) - with pytest.raises(HTTPException) as exc_info: - smart_scan_thing.delete_scan(fake_scan_name, LOGGER) - # Should raise a 400 error if the scan doesn't exists, not a 404 as the server + # Attempt to delete the fake scan. Expect it to fail + with pytest.raises(HTTPException) as exc_info: + smart_scan_thing.delete_scan(fake_scan_name, LOGGER) + # 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 assert exc_info.value.status_code == 400 assert len(caplog.records) == 1 @@ -151,7 +153,7 @@ def _run_only_outer_scan(adjust_inital_state: Optional[Callable] = None): _run_scan method where this can be tested. Once this is done the final scan behaviour can be tested too. - adjust_initial_scan is a callable which accepts the mocked smart scan thing + adjust_initial_state is a callable which accepts the mocked smart scan thing as the only variable. It can be used to adjust the initial state of the test @@ -177,7 +179,7 @@ def _run_only_outer_scan(adjust_inital_state: Optional[Callable] = None): # Counter for checking functions were called mock_call_count = {"_run_scan": 0} - # Mock thing settngs as a dictionary + # Mock thing settings as a dictionary thing_settings = {"skip_background": True} def _run_scan(self): @@ -244,7 +246,7 @@ def test_outer_scan(): assert mock_ss_thing.mock_call_count["_run_scan"] == 1 -def test_outer_scan_wo_scample_skip(): +def test_outer_scan_wo_sample_skip(): """Test setup and teardown of the scan.""" def _set_skip_background(mock_ss_thing): diff --git a/webapp/src/components/tabContentComponents/scanListContent.vue b/webapp/src/components/tabContentComponents/scanListContent.vue index 6b342d4b..79741558 100644 --- a/webapp/src/components/tabContentComponents/scanListContent.vue +++ b/webapp/src/components/tabContentComponents/scanListContent.vue @@ -255,7 +255,7 @@ export default { formatDate(timestamp) { // Multiply by 1000 as JS uses ms not s let d = new Date(timestamp*1000); - // Convert to a sting in a very javascript way! + // Convert to a string in a very javascript way! let yyyy = d.getFullYear(); let mm = d.getMonth() + 1; let dd = d.getDate(); From 51f9df995d05ff99069b13500f69769cc3f38865 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Mon, 30 Jun 2025 17:05:49 +0100 Subject: [PATCH 9/9] More changes suggested in review, generally docstring issues. --- .../scan_directories.py | 16 +++++++--------- .../things/smart_scan.py | 6 +++--- tests/mock_things/mock_background_detect.py | 2 -- tests/test_scan_directories.py | 8 +++++++- tests/test_smart_scan.py | 6 ++++++ 5 files changed, 23 insertions(+), 15 deletions(-) diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 25cccd2a..e2a33fbc 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -73,8 +73,8 @@ class ScanDirectoryManager: ) -> Optional[str]: """Return the file path for the file within a scan directory - If check_exists is True then a FileNotFoundError will be raised if - the file does not exist. + If check_exists is True then None will be returned if the file does + not exist. """ file_path = os.path.join(self.path_for(scan_name), filename) if check_exists: @@ -178,13 +178,10 @@ class ScanDirectoryManager: """ Raise an exception if there is not enough free disk space to continue scanning - Args: - path = path to a location on the disk you want to check - min_space [int] = the minimum space required in bytes - default = 500,000,000 (500MB) + :param min_space: the minimum space required in bytes. + Default = 500,000,000 (500MB) - Raises: - NotEnoughFreeSpaceError if the remaining storage is below min_space + :raises: NotEnoughFreeSpaceError if the remaining storage is below min_space """ disk_usage = shutil.disk_usage(self._base_scan_dir) if disk_usage.free < min_space: @@ -198,7 +195,8 @@ class ScanDirectory: """A class for handling interactions with scan directories Initalisation parameters: - dir_path: the directory of the outer scan directory + name: the name of the scan (the scan directory basename) + base_scan_dir: Path of the directory that holds all scans """ _name: str diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 9af232e4..832f6170 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -49,7 +49,7 @@ STITCHING_RESOLUTION = (820, 616) class ScanNotRunningError(RuntimeError): - """Method called when scan not running that requires a scan to be running""" + """Exception called when scan not running that requires a scan to be running""" def _scan_running(method): @@ -138,8 +138,8 @@ class SmartScanThing(Thing): self._capture_thread = None self._scan_images_taken = 0 - # Set _scan_data to None. It will for error checking. It will be set to a value - # by _run_scan + # Set _scan_data to None. This is needed just in case an exception is raised + # before _run_scan (which sets the real data). As we check this in the `except` self._scan_data = None try: diff --git a/tests/mock_things/mock_background_detect.py b/tests/mock_things/mock_background_detect.py index bd57ee61..f79a1059 100644 --- a/tests/mock_things/mock_background_detect.py +++ b/tests/mock_things/mock_background_detect.py @@ -11,8 +11,6 @@ from openflexure_microscope_server.things.background_detect import ChannelDistri class MockBackgoundDetectThing: - # Counter for checking functions were called - background_distributions = ChannelDistributions( means=[128.0, 128.0, 128.0], standard_deviations=[3.0, 3.0, 3.0], diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py index e9bd63c9..ee76c9ab 100644 --- a/tests/test_scan_directories.py +++ b/tests/test_scan_directories.py @@ -281,7 +281,13 @@ def test_creating_scan_dir_for_missing_scan(): def test_none_returned_for_missing_images_dir(): - """None should be returned for image dir path if images dir never created.""" + """None should be returned for image dir path if images dir does not exist. + + By default images directories are created at the same time the scan directory + is created. However, edge cases such as problems in deletions, microscopes + with older scans on, etc can cause and empty scan directory, so it is handled + explicitly. + """ _clear_scan_dir() os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001")) scan_dir = ScanDirectory(BASE_SCAN_DIR, "fake_scan_0001") diff --git a/tests/test_smart_scan.py b/tests/test_smart_scan.py index da23c4b9..8c8e791e 100644 --- a/tests/test_smart_scan.py +++ b/tests/test_smart_scan.py @@ -54,6 +54,11 @@ def smart_scan_thing(): def test_initial_properties(smart_scan_thing): + """Check the initial values of properties. + + Test properties of SmartScanThing are available without a ThingServer + and return expected default values. + """ assert smart_scan_thing._scan_dir_manager.base_dir == SCAN_DIR assert smart_scan_thing.latest_scan_name is None @@ -125,6 +130,7 @@ def test_public_delete_scan(smart_scan_thing, caplog): def test_delete_all_scans(smart_scan_thing, caplog): + """Check the delete_all_scan API really does delete all the scans.""" _clear_scan_dir() with caplog.at_level(logging.INFO): fake_scan_names = [