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 new file mode 100644 index 00000000..e2a33fbc --- /dev/null +++ b/src/openflexure_microscope_server/scan_directories.py @@ -0,0 +1,315 @@ +""" +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 managing 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 None will be returned 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(file_path): + 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(file_path): + return None + return file_path + + @property + 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()] + + 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) -> "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 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 + + 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) -> None: + """Delete a scan""" + shutil.rmtree(self.path_for(scan_name)) + + 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 + 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 + + :param min_space: the minimum space required in bytes. + Default = 500,000,000 (500MB) + + :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: + name: the name of the scan (the scan directory basename) + base_scan_dir: Path of the directory that holds all scans + """ + + _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) -> str: + """The name of the scan""" + return self._name + + @property + def dir_path(self) -> str: + """The full path to the scan directory""" + return 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 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)] + 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") + + 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: + continue + # If this is not the final version, then only zip image files. + if not final_version and not IMAGE_REGEX.search(file): + continue + + 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 b206058a..832f6170 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,74 +21,37 @@ 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: datetime - modified: datetime - 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 ScanNotRunningError(RuntimeError): + """Exception 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 @@ -105,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!" ) @@ -114,7 +73,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() @@ -136,7 +95,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 @@ -179,12 +138,14 @@ 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. 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: 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 @@ -193,8 +154,8 @@ 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): - # Don't stich if drive is full (already logged) + if not isinstance(e, scan_directories.NotEnoughFreeSpaceError): + # Don't stitch if drive is full (already logged) self._perform_final_stitch() # Error must be raised so UI gives correct output raise e @@ -209,35 +170,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 @@ -268,88 +205,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 @@ -443,7 +303,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, @@ -466,7 +326,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"], @@ -477,7 +337,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) @@ -506,7 +366,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: @@ -558,7 +418,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( @@ -627,7 +487,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() @@ -654,7 +514,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"], ) @@ -665,12 +525,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): @@ -689,10 +547,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() @@ -701,11 +557,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) @@ -725,12 +580,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]: @@ -800,7 +655,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 @@ -809,77 +664,14 @@ 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 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, - ) - ) - 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", "scans/{scan_name}", responses={ 200: {"description": "Successfully deleted scan"}, - 404: {"description": "Scan not found"}, + 400: {"description": "An error occurred while trying to delete scan"}, }, ) def delete_scan(self, scan_name: str, logger: InvocationLogger) -> None: @@ -889,11 +681,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") @@ -908,49 +699,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 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", @@ -965,10 +766,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: @@ -996,7 +797,7 @@ class SmartScanThing(Thing): f"{min_overlap}", "--resize", f"{self._scan_data['stitch_resize']}", - self._ongoing_scan_images_dir, + self._ongoing_scan.images_dir, ] ) @@ -1074,8 +875,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" @@ -1137,87 +939,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, @@ -1225,43 +950,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/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..d32b5a50 --- /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: 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 new file mode 100644 index 00000000..f79a1059 --- /dev/null +++ b/tests/mock_things/mock_background_detect.py @@ -0,0 +1,18 @@ +"""This testing submodule contains mock background detect 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: + 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_scan_directories.py b/tests/test_scan_directories.py new file mode 100644 index 00000000..ee76c9ab --- /dev/null +++ b/tests/test_scan_directories.py @@ -0,0 +1,332 @@ +import tempfile +import os +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 +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() -> None: + """Delete the scan dir""" + if os.path.exists(BASE_SCAN_DIR): + shutil.rmtree(BASE_SCAN_DIR) + + +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) + 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: 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: + 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 + + 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) + 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 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 + + # 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 + + # 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 + 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) + + # 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) + + +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(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 + 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(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 + 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_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() + 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 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 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 + 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 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 + + +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 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) + _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(set(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(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") + + +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 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") + 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_manager.check_free_disk_space() + + +@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 + ), + ) + + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + with pytest.raises(NotEnoughFreeSpaceError): + scan_dir_manager.check_free_disk_space() diff --git a/tests/test_smart_scan.py b/tests/test_smart_scan.py new file mode 100644 index 00000000..8c8e791e --- /dev/null +++ b/tests/test_smart_scan.py @@ -0,0 +1,265 @@ +""" +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 needing 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. +""" + +from typing import Callable, Optional +import tempfile +import os +import shutil +import logging + +from fastapi import HTTPException +import pytest + +from openflexure_microscope_server.things.smart_scan import ( + SmartScanThing, + 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") + +# 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() -> None: + """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): + """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 + + +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) + # 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 + 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) + + # 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 + 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): + """Check the delete_all_scan API really does delete all the scans.""" + _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 + + +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_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 + + + 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 settings 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_sample_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 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() diff --git a/webapp/src/components/tabContentComponents/scanListContent.vue b/webapp/src/components/tabContentComponents/scanListContent.vue index 293b9d0c..79741558 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 string in a very javascript way! let yyyy = d.getFullYear(); let mm = d.getMonth() + 1; let dd = d.getDate();