90f106c87d
ScanInfo -> ScanGalleryInfo CaptureInfo -> CaptureGalleryInfo - note there were two classes called CaptureInfo, only one has been changed to deduplicate the names
699 lines
25 KiB
Python
699 lines
25 KiB
Python
"""Functionality to manage file system operations for scan directories."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import threading
|
|
import zipfile
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Literal, Mapping, Optional, Self
|
|
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
ValidationError,
|
|
field_serializer,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
import labthings_fastapi as lt
|
|
from labthings_fastapi.exceptions import InvocationError
|
|
|
|
from openflexure_microscope_server.stitching import StitchingSettings
|
|
from openflexure_microscope_server.utilities import (
|
|
make_name_safe,
|
|
requires_lock,
|
|
save_invocation_logs,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
IMG_DIR_NAME = "images"
|
|
SCAN_ZERO_PAD_DIGITS = 4
|
|
|
|
STITCH_REGEX = re.compile(r"stitched\.jpe?g$")
|
|
IMAGE_REGEX = re.compile(r"-?[0-9]+_-?[0-9]+\.jpe?g$")
|
|
|
|
SCAN_DATA_FILENAME = "scan_data.json"
|
|
SCAN_DATA_SCHEMA_VERSION = 2
|
|
|
|
|
|
class NotEnoughFreeSpaceError(InvocationError):
|
|
"""An exception raised if there is not enough free space on disk to scan."""
|
|
|
|
|
|
class ScanGalleryInfo(BaseModel):
|
|
"""Summary information for the UI about a scan folder."""
|
|
|
|
name: str
|
|
created: float
|
|
modified: float
|
|
duration: Optional[float]
|
|
number_of_images: int
|
|
stitch_available: bool
|
|
dzi: Optional[str]
|
|
thing: str = "smart_scan"
|
|
card_type: Literal["Scan"] = "Scan"
|
|
|
|
|
|
class BaseScanData(BaseModel):
|
|
"""Data about a scan not including workflow specific data.
|
|
|
|
For including workflow specific data see also:
|
|
|
|
* ActiveScanData which subclasses this including the BaseModel used by the
|
|
ScanWorkflow
|
|
* HistoricScanData which has the workflow specific data loaded as a dictionary.
|
|
|
|
Separating historic and active data allows workflows to use any BaseModel for its
|
|
settings, but for the data to be reloaded even if that model has updated or is not
|
|
available. Historic scan data loaded from disk is used for stitching and for
|
|
creating a ScanGalleryInfo object for communicating with the UI. These uses are
|
|
clearly typed by this model.
|
|
|
|
This serialises into a human readable format where possible with
|
|
|
|
timestamps in %Y-%m-%d_%H:%M:%S format
|
|
timedeltas in %H:%M:%S format
|
|
|
|
Properties that are not known until the end have ``None`` serialised as "Unknown"
|
|
"""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
schema_version: int = SCAN_DATA_SCHEMA_VERSION
|
|
|
|
scan_name: str
|
|
"""The name of the scan i.e. scan_0001"""
|
|
|
|
starting_position: Mapping[str, int]
|
|
"""The starting position in dictionary format."""
|
|
|
|
start_time: datetime
|
|
"""The time the scan started."""
|
|
|
|
stitch_automatically: bool
|
|
"""Whether the scan is set to automatically stitch when complete."""
|
|
|
|
save_resolution: tuple[int, int]
|
|
"""The resolution that scan images are saved at."""
|
|
|
|
image_count: int = 0
|
|
"""The number of images taken."""
|
|
|
|
duration: Optional[timedelta] = None
|
|
"""The duration of the scan.
|
|
|
|
This is automatically set when ``set_final_data()`` is run.
|
|
"""
|
|
|
|
scan_result: Optional[str] = None
|
|
"""The result of the scan.
|
|
|
|
This should be set with ``set_final_data()`` to ensure duration is set.
|
|
"""
|
|
|
|
stitching_settings: Optional[StitchingSettings]
|
|
"""The data needed to stitch a scan.
|
|
|
|
Set to None for types of scan that cannot be stitched.
|
|
"""
|
|
|
|
workflow: str
|
|
"""The class name of the workflow Thing."""
|
|
|
|
@model_validator(mode="after")
|
|
def validate_schema_version(self) -> Self:
|
|
"""Validate the schema version is as the current one."""
|
|
if self.schema_version != SCAN_DATA_SCHEMA_VERSION:
|
|
raise ValueError(
|
|
f"Unsupported schema version {self.schema_version}, "
|
|
f"expected {SCAN_DATA_SCHEMA_VERSION}"
|
|
)
|
|
return self
|
|
|
|
@field_validator("start_time", mode="before")
|
|
@classmethod
|
|
def parse_timestamp(cls, value: str | datetime) -> datetime:
|
|
"""Validate a timestamp that may be a string in Year-Month-Day_Hrs:Min:Sec format."""
|
|
if isinstance(value, str):
|
|
return datetime.strptime(value, "%Y-%m-%d_%H:%M:%S")
|
|
return value
|
|
|
|
@field_serializer("start_time")
|
|
def serialize_timestamp(self, value: datetime) -> str:
|
|
"""Serialise timestamp to Year-Month-Day_Hrs:Min:Sec format."""
|
|
return value.strftime("%Y-%m-%d_%H:%M:%S")
|
|
|
|
@field_validator("duration", mode="before")
|
|
@classmethod
|
|
def parse_timedelta(cls, value: Optional[str | timedelta]) -> Optional[timedelta]:
|
|
"""Validate a timedelta that may be a string in Hrs:Min:Sec format or "Unknown"."""
|
|
if isinstance(value, str):
|
|
if value == "Unknown":
|
|
return None
|
|
hrs, mins, secs = map(int, value.split(":"))
|
|
return timedelta(hours=hrs, minutes=mins, seconds=secs)
|
|
return value
|
|
|
|
@field_serializer("duration")
|
|
def serialize_timedelta(self, value: Optional[timedelta]) -> str:
|
|
"""Serialise timedelta to Hrs:Min:Sec (or "Unknown" if None)."""
|
|
if value is None:
|
|
return "Unknown"
|
|
# Calculate the string manually rather than use str(), as this may convert to
|
|
# days for a very very long scan, and then it is much harder to parse.
|
|
total_secs = value.total_seconds()
|
|
hrs = int(total_secs // 3600)
|
|
mins = int((total_secs % 3600) // 60)
|
|
secs = int((total_secs % 60))
|
|
return f"{hrs}:{mins:02}:{secs:02}"
|
|
|
|
@field_validator("scan_result", mode="before")
|
|
@classmethod
|
|
def parse_unknown_as_none(cls, value: Optional[str | int]) -> Optional[str | int]:
|
|
"""Validate the string "Unknown" as None."""
|
|
return None if value == "Unknown" else value
|
|
|
|
@field_serializer("scan_result")
|
|
def serialize_none_as_unknown(self, value: Optional[str | int]) -> str | int:
|
|
"""Serialise None as "Unknown" for a more human readable result."""
|
|
return "Unknown" if value is None else value
|
|
|
|
|
|
class HistoricScanData(BaseScanData):
|
|
"""A Model for the scan data that has been loaded from disk.
|
|
|
|
Any workflow specific settings are loaded as an arbitrary dictionary. Other
|
|
settings such as those which are needed for the UI or stitching are loaded and
|
|
validated by the parent class ``BaseScanData``.
|
|
"""
|
|
|
|
workflow_settings: dict
|
|
"""A dictionary of the settings for the workflow that was used workflow."""
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def coerce_legacy(cls, data: dict) -> dict:
|
|
"""Coerce any scan data from before version 2 into the version 2 format."""
|
|
# Before the current version no schema_version was set
|
|
if "schema_version" in data:
|
|
return data
|
|
|
|
if "correlation_resize" and "overlap" in data:
|
|
correlation_resize = data.pop("correlation_resize")
|
|
# Note we don't pop overlap, as it is a setting for the legacy workflow as well
|
|
# as a stitching setting.
|
|
# This is done because in future workflows the stitching overlap may be a
|
|
# directly set setting or something that is calculated from other settings.
|
|
overlap = data["overlap"]
|
|
data["stitching_settings"] = StitchingSettings(
|
|
correlation_resize=correlation_resize,
|
|
overlap=overlap,
|
|
)
|
|
else:
|
|
data["stitching_settings"] = None
|
|
|
|
# Add any legacy workflow settings that are found
|
|
legacy_keys = [
|
|
"overlap",
|
|
"max_dist",
|
|
"dx",
|
|
"dy",
|
|
"autofocus_dz",
|
|
"autofocus_on",
|
|
"skip_background",
|
|
]
|
|
workflow_settings = {}
|
|
for key in legacy_keys:
|
|
if key in data:
|
|
workflow_settings[key] = data.pop(key)
|
|
|
|
data["workflow"] = "Legacy"
|
|
data["workflow_settings"] = workflow_settings
|
|
return data
|
|
|
|
|
|
class ScanDirectoryManager:
|
|
"""A class for managing interactions with scan directories."""
|
|
|
|
_base_scan_dir: str
|
|
|
|
def __init__(self, base_scan_dir: str) -> None:
|
|
"""Initialise the scan directory manager.
|
|
|
|
:param base_scan_dir: Path of the directory that holds all scans.
|
|
"""
|
|
self._base_scan_dir = base_scan_dir
|
|
if not os.path.exists(self._base_scan_dir):
|
|
os.makedirs(self._base_scan_dir)
|
|
self._lock = threading.RLock()
|
|
|
|
@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:
|
|
"""Return 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)
|
|
|
|
@requires_lock
|
|
def get_file_path_from(
|
|
self, scan_name: str, filename: str, check_exists: bool = False
|
|
) -> Optional[str]:
|
|
"""Return the full 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 and not os.path.exists(file_path):
|
|
return None
|
|
return file_path
|
|
|
|
@requires_lock
|
|
def get_file_path_from_img_dir(
|
|
self, scan_name: str, filename: str, check_exists: bool = False
|
|
) -> Optional[str]:
|
|
"""Return the full 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 anyway
|
|
"""
|
|
file_path = os.path.join(self.img_dir_for(scan_name), filename)
|
|
if check_exists and not os.path.exists(file_path):
|
|
return None
|
|
return file_path
|
|
|
|
def get_final_stitch_path(self, scan_name: str) -> Optional[str]:
|
|
"""Return the file full path for the final stitch.
|
|
|
|
If no final stitch is found, return None
|
|
"""
|
|
stitch_fname = ScanDirectory(scan_name, self.base_dir).get_final_stitch_name()
|
|
if stitch_fname is None:
|
|
return None
|
|
return self.get_file_path_from_img_dir(scan_name, stitch_fname)
|
|
|
|
@requires_lock
|
|
def get_scan_data_path(self, scan_name: str) -> Optional[str]:
|
|
"""Return the file full path scan data JSON file.
|
|
|
|
If no scan data JSON file is found, return None
|
|
"""
|
|
scan_data_path = ScanDirectory(scan_name, self.base_dir).scan_data_path
|
|
if scan_data_path is None:
|
|
return None
|
|
if not os.path.isfile(scan_data_path):
|
|
return None
|
|
return scan_data_path
|
|
|
|
def get_scan_data(self, scan_name: str) -> Optional[HistoricScanData]:
|
|
"""Return the scan data read from a JSON file as a dict."""
|
|
return ScanDirectory(scan_name, self.base_dir).get_scan_data()
|
|
|
|
@property
|
|
@requires_lock
|
|
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()]
|
|
|
|
@requires_lock
|
|
def all_scans_info(
|
|
self, *, ongoing: Optional[str] = None, include_ongoing: bool = True
|
|
) -> list[ScanGalleryInfo]:
|
|
"""Return a lists of ScanGalleryInfo objects for each scan.
|
|
|
|
:param ongoing: The name of the ongoing scan (or None if no scan is ongoing).
|
|
:param include_ongoing: True (dfault) to include the scan info of the ongoing
|
|
scan in the return. False to exclude it.
|
|
|
|
:return: A list of ScanGalleryInfo objects for each scan.
|
|
"""
|
|
all_info: list[ScanGalleryInfo] = []
|
|
for scan_name in self.all_scans:
|
|
scan_is_ongoing = scan_name == ongoing
|
|
if scan_is_ongoing and not include_ongoing:
|
|
continue
|
|
scan_dir = ScanDirectory(scan_name, self.base_dir)
|
|
# If the scan is ongoing send flag to skip reading the json data
|
|
info = scan_dir.scan_info(skip_json=scan_is_ongoing)
|
|
all_info.append(info)
|
|
return all_info
|
|
|
|
@requires_lock
|
|
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`
|
|
"""
|
|
scan_name = make_name_safe(scan_name)
|
|
|
|
# Strip all trailing underscores from the base name
|
|
scan_name = scan_name.strip("_")
|
|
# 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_match = scan_regex.match(last_matching_scan)
|
|
|
|
if scan_match is None: # pragma: no cover
|
|
# Type narrow, we know it is a match but mypy doesn't. This code is
|
|
# Not reachable hence the no cover.
|
|
raise RuntimeError("Internal error: regex No longer matches")
|
|
|
|
scan_num = int(scan_match[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}"
|
|
|
|
@requires_lock
|
|
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)
|
|
|
|
@requires_lock
|
|
def delete_scan(self, scan_name: str) -> None:
|
|
"""Delete a scan."""
|
|
shutil.rmtree(self.path_for(scan_name))
|
|
|
|
@requires_lock
|
|
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)
|
|
|
|
@requires_lock
|
|
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."""
|
|
|
|
_name: str
|
|
_base_scan_dir: str
|
|
|
|
def __init__(self, name: str, base_scan_dir: str) -> None:
|
|
"""Initialise the scan directory.
|
|
|
|
:param name: the name of the scan (the scan directory basename).
|
|
:param base_scan_dir: Path of the directory that holds all scans.
|
|
"""
|
|
self._log_saved_at = 0.0
|
|
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 scan_data_path(self) -> Optional[str]:
|
|
"""The path to the scan data json file for this directory.
|
|
|
|
Returns None if there is no images dir to write to.
|
|
"""
|
|
if self.images_dir is not None:
|
|
return os.path.join(self.images_dir, SCAN_DATA_FILENAME)
|
|
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) -> list[str]:
|
|
"""Return a list of the files in the images dir."""
|
|
if self.images_dir is None:
|
|
return []
|
|
return os.listdir(self.images_dir)
|
|
|
|
def _extract_scan_images(self, file_list: list[str]) -> list[str]:
|
|
"""Extract files which match the naming convention for scan images.
|
|
|
|
:param file_list: The list of files to search. Normally this would be
|
|
``self.get_scan_files()``
|
|
|
|
:returns: The list of files that match the naming convention for scan images
|
|
"""
|
|
return [i for i in file_list if IMAGE_REGEX.search(i)]
|
|
|
|
def _extract_final_stitches(self, file_list: list[str]) -> list[str]:
|
|
"""Extract files which match the naming convention for final stitches.
|
|
|
|
:param file_list: The list of files to search.
|
|
|
|
:returns: The list of files that match the naming convention for final stitches
|
|
"""
|
|
return [i for i in file_list if STITCH_REGEX.search(i)]
|
|
|
|
def _extract_dzi_files(self, file_list: list[str]) -> list[str]:
|
|
"""Extract files which match the naming convention for dzi_files.
|
|
|
|
:param file_list: The list of files to search.
|
|
|
|
:returns: The list of files that match the naming convention for dzi_files
|
|
"""
|
|
return [i for i in file_list if i.endswith("dzi")]
|
|
|
|
def get_final_stitch_name(self) -> Optional[str]:
|
|
"""Return the filename for the final stitch (in the images dir).
|
|
|
|
If no final stitch is found, return None
|
|
"""
|
|
stitches = self._extract_final_stitches(self.get_scan_files())
|
|
if not stitches:
|
|
return None
|
|
return stitches[0]
|
|
|
|
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_data_dict(self) -> Optional[dict[str, Any]]:
|
|
"""Return the scan data from the json file as a dictionary.
|
|
|
|
This is safer than get_scan_data for older scans before a defined model was
|
|
used.
|
|
|
|
:return: The data as a dictionary or None if it couldn't be loaded.
|
|
"""
|
|
if self.scan_data_path is None:
|
|
return None
|
|
try:
|
|
with open(self.scan_data_path, "r", encoding="utf-8") as data_file:
|
|
return json.load(data_file)
|
|
except (json.decoder.JSONDecodeError, IOError):
|
|
return None
|
|
|
|
def get_scan_data(self) -> Optional[HistoricScanData]:
|
|
"""Return the scan data from the json file as a HistoricScanData model.
|
|
|
|
:return: The data as a HistoricScanData model or None if it couldn't be loaded or
|
|
valdiated.
|
|
"""
|
|
data_dict = self._get_scan_data_dict()
|
|
if data_dict is None:
|
|
LOGGER.warning(f"Could not load scan data for {self.name}.")
|
|
return None
|
|
try:
|
|
return HistoricScanData(**data_dict)
|
|
except ValidationError:
|
|
LOGGER.warning(f"Could not validate scan data for {self.name}.")
|
|
return None
|
|
|
|
def scan_info(self, skip_json: bool = False) -> ScanGalleryInfo:
|
|
"""Return the information to be used in the UI for the scan."""
|
|
scan_files = self.get_scan_files()
|
|
scan_images = self._extract_scan_images(scan_files)
|
|
stitches = self._extract_final_stitches(scan_files)
|
|
dzi_files = self._extract_dzi_files(scan_files)
|
|
number_of_images = len(scan_images)
|
|
stitch_available = len(stitches) > 0
|
|
dzi = None if not dzi_files else str(dzi_files[0])
|
|
|
|
scan_data = None if skip_json else self.get_scan_data()
|
|
duration = (
|
|
None
|
|
if scan_data is None or scan_data.duration is None
|
|
else scan_data.duration.total_seconds()
|
|
)
|
|
|
|
return ScanGalleryInfo(
|
|
name=self.name,
|
|
created=self.created_time,
|
|
modified=self.get_modified_time(),
|
|
duration=duration,
|
|
number_of_images=number_of_images,
|
|
stitch_available=stitch_available,
|
|
dzi=dzi,
|
|
)
|
|
|
|
def all_files(self, skip_dirs: Optional[list[str]] = None) -> list[str]:
|
|
"""Return a list of all files in the scan dir relative to the dir.
|
|
|
|
:param skip_dirs: Skip any file in a directory that is on this list. The list
|
|
should be the basename of the directory. e.g. "scan_0001_files" not
|
|
"images/scan_0001_files"
|
|
"""
|
|
if skip_dirs is None:
|
|
skip_dirs = []
|
|
files = []
|
|
for file_root, dirs, filenames in os.walk(self.dir_path, topdown=True):
|
|
# Skip any skipped directories.
|
|
# Note: we must use slice assignment to edit in place.
|
|
dirs[:] = [d for d in dirs if d not in skip_dirs]
|
|
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 save_scan_data(self, scan_data: BaseScanData) -> None:
|
|
"""Save the scan data for this scan to disk."""
|
|
if self.scan_data_path is None:
|
|
raise FileNotFoundError(
|
|
"There is no images directory to save scan data into."
|
|
)
|
|
|
|
with open(self.scan_data_path, "w", encoding="utf-8") as f:
|
|
f.write(scan_data.model_dump_json(indent=4))
|
|
|
|
def save_scan_log(self, thing: lt.Thing) -> None:
|
|
"""Save the scan log to file.
|
|
|
|
:param thing: The SmartScanThing. This is needed to exxtract the logs from the
|
|
LabThings server.
|
|
"""
|
|
self._log_saved_at = save_invocation_logs(
|
|
filename=os.path.join(self.dir_path, "log.txt"),
|
|
thing=thing,
|
|
last_saved=self._log_saved_at,
|
|
append=True,
|
|
)
|
|
|
|
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")
|
|
|
|
# Use noqa as converting this into a 1 liner is not more readable.
|
|
if os.path.isfile(zip_fname): # noqa: SIM108
|
|
# get a list of files in the existing zip
|
|
zip_files = get_files_in_zip(zip_fname)
|
|
else:
|
|
zip_files = []
|
|
|
|
# For each `filename.dzi` we need to skip the `filename_files` directory
|
|
dzi_files = self._extract_dzi_files(self.get_scan_files())
|
|
dzi_dirs = [dzi[:-4] + "_files" for dzi in dzi_files]
|
|
|
|
with zipfile.ZipFile(zip_fname, mode="a") as scan_zip:
|
|
for file in self.all_files(skip_dirs=dzi_dirs):
|
|
# Don't zip zipfiles, dzi files, or files already in the zip
|
|
if file.endswith((".zip", ".dzi")) 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: str) -> list[str]:
|
|
"""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()]
|