From 1493f51212fd55630a1a06f9ee96e0a771099fce Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Tue, 8 Apr 2025 10:46:22 +0100 Subject: [PATCH] Purge all windows line endings and add check to formatter. Also changes python version in pyproject.toml as it is incorrect --- pyproject.toml | 8 +- .../things/smart_scan.py | 2536 ++++++++--------- 2 files changed, 1274 insertions(+), 1270 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1d8276c1..e3ddcd89 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,7 @@ authors = [ ] description = "Back-end code for the OpenFlexure Microscope" readme = "README.md" -requires-python = ">=3.9" +requires-python = ">=3.11" classifiers = [ "Programming Language :: Python :: 3", "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", @@ -68,9 +68,13 @@ build-backend = "setuptools.build_meta" plugins = ["pydantic.mypy"] [tool.ruff] -target-version = "py39" +target-version = "py311" [tool.pytest.ini_options] addopts = [ "--import-mode=importlib", ] + +[tool.ruff.format] +# Use native line endings for all files +line-ending = "native" diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 474eb105..3c9dd7a9 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -1,1268 +1,1268 @@ -# ruff: noqa: E722 - -import shutil -import zipfile -import threading -from typing import Mapping, Optional -import cv2 -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 scipy.stats import norm -from scipy.ndimage import zoom -from scipy.interpolate import interp1d -from datetime import datetime -from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, STDOUT -from threading import Event, Thread -import glob -import json -import piexif - -from labthings_fastapi.thing import Thing -from labthings_fastapi.dependencies.metadata import GetThingStates -from labthings_fastapi.dependencies.thing import direct_thing_client_dependency -from labthings_fastapi.dependencies.invocation import ( - CancelHook, - InvocationLogger, - InvocationCancelledError, -) -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.things.autofocus import AutofocusThing -from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper -from openflexure_microscope_server.things.auto_recentre_stage import RecentringThing - - -CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/") -AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/") -RecentreStage = direct_thing_client_dependency(RecentringThing, "/auto_recentre_stage/") - - -def closest(current, focused_path): - """Finds the index of the closest x-y position in a list from the current position, - with ties split by the later element in the list (most recently taken) - - must be float64 to deal with the huge numbers involved!""" - - current_pos = np.array(current[:2], dtype="float64") - path_pos = np.asarray(focused_path, dtype="float64").T[:2].T - - dist_2 = np.sqrt( - np.sum((path_pos - current_pos) ** 2, axis=1, dtype="float64"), dtype="float64" - ) - min_dist = np.argmin(dist_2) - mask = np.where(dist_2 == dist_2[min_dist], 1, 0) - try: - closest = np.max(np.nonzero(mask)) - except: - closest = 0 - return closest - - -def unpack_autofocus(scan_data): - """Extract z, sharpness data from a move_and_measure call - - Data will start at `start_index`, i.e. `start_index` points are dropped - from the beginning of the array. - """ - jpeg_times = scan_data["jpeg_times"] - jpeg_sizes = scan_data["jpeg_sizes"] - jpeg_sizes_MB = [x / 10**3 for x in jpeg_sizes] - stage_times = scan_data["stage_times"] - stage_positions = scan_data["stage_positions"] - stage_height = [pos[2] for pos in stage_positions] - - jpeg_heights = np.interp(jpeg_times, stage_times, stage_height) - - def turningpoints(lst): - dx = np.diff(lst) - return dx[1:] * dx[:-1] < 0 - - turning = np.where(turningpoints(jpeg_heights))[0] + 1 - - return jpeg_heights[turning[0] : turning[1]], jpeg_sizes_MB[turning[0] : turning[1]] - - -def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit): - # limit is the largest ratio of change in z to change in xy that's allowed - - prev_xy = np.asarray(prev_pos, dtype="float64") - new_xy = np.asarray(new_pos, dtype="float64") - - dist = np.sqrt(np.sum(((new_xy - prev_xy) / 10**4) ** 2, dtype="float64")) - - focus_change = abs(new_z - prev_z) / 10**4 - if dist == 0: - print(f"Not moved between {prev_pos} and {new_pos}") - movement_ratio = 0 - else: - movement_ratio = np.divide(focus_change, dist, dtype="float64") - - # print('Movement ratio is {0} in z per lateral step. The limit is {1}'.format(round(movement_ratio, 4), round(limit,4))) - # print(f'This is the distance between {prev_pos}, {prev_z} and {new_pos}, {new_z}') - - if movement_ratio > limit: - return "reject" - else: - return "accept" - - -# def distance_to_site(current, next): -# current = np.array(current, dtype="float64") -# next = np.array(next, dtype="float64") -# if (next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2 < 0: -# print(f"Negative distance between {next} and {current}") -# return np.sqrt( -# (next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2, dtype="float64" -# ) - - -def steps_from_centre(current_loc, starting_loc, dx, dy): - step_size = np.array([dx, dy]) - return np.max(np.abs(np.divide(np.subtract(current_loc, starting_loc), step_size))) - - -# def set_template(microscope, pos): -# microscope.move(pos) -# background = microscope.grab_image_array() -# background_LUV = cv2.cvtColor(background, cv2.COLOR_RGB2LUV) - -# ch1 = (background_LUV.T[0]).flatten() -# ch2 = (background_LUV.T[1]).flatten() -# ch3 = (background_LUV.T[2]).flatten() - -# points = np.array([np.asarray(ch1),np.asarray(ch2),np.asarray(ch3)]).T - -# # we get the mean and standard deviation of values in each channel - -# mu, std = np.apply_along_axis(norm.fit, 0, points) -# stats_list = np.vstack([mu, std]) -# return stats_list - - -def distance_to_site(current, next): - next = np.array(next, dtype="float64") - current = np.array(current, dtype="float64") - return np.sqrt((next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2) - - -def scale_csm(csm_matrix, calibration_width, img_width): - "Account for a calibration width that may differ from image width" - scale = img_width / calibration_width # Usually >1, if we calibrated at low res - csm = np.array(csm_matrix) / scale # Decrease the CSM if pixels are smaller] - return csm - - -def generate_config( - folder_path: str, - positions: list, - names: list, - camera_to_sample_matrix, - csm_calibration_width, - img_width, - logger, -): - positions = np.array(positions) - mean_loc = np.mean(positions, axis=0) - - # TODO: positions from recent scans need to be 2x bigger - change to CSM res? - - camera_to_sample_matrix = scale_csm( - camera_to_sample_matrix, csm_calibration_width, img_width - ) - - with open(os.path.join(folder_path, "TileConfiguration.txt"), "w") as fp: - fp.write( - "# Define the number of dimensions we are working on\ndim = 2\n\n# Define the image coordinates\n" - ) - for i in range(len(names)): - loc = np.dot( - (positions[i] - mean_loc), np.linalg.inv(camera_to_sample_matrix) - ) - fp.write(f"{names[i]}; ; {loc[1], loc[0]} \n") - - -def raw2rggb(raw): - """Convert packed 10 bit raw to RGGB 8 bit""" - raw = np.asarray(raw) # ensure it's an array - rggb = np.empty((616, 820, 4), dtype=np.uint8) - raw_w = rggb.shape[1] // 2 * 5 - for plane, offset in enumerate([(1, 1), (0, 1), (1, 0), (0, 0)]): - rggb[:, ::2, plane] = raw[offset[0] :: 2, offset[1] : raw_w + offset[1] : 5] - rggb[:, 1::2, plane] = raw[ - offset[0] :: 2, offset[1] + 2 : raw_w + offset[1] + 2 : 5 - ] - return rggb - - -def rggb2rgb(rggb): - return np.stack( - [rggb[..., 0], rggb[..., 1] // 2 + rggb[..., 2] // 2, rggb[..., 3]], axis=2 - ) - - -class ChannelDistributions(BaseModel): - means: list[float] - standard_deviations: list[float] - colorspace: str = "LUV" - - -class BackgroundDetectThing(Thing): - @thing_property - def background_distributions(self) -> Optional[ChannelDistributions]: - """The statistics of the background image""" - bd = self.thing_settings.get("background_distributions", None) - if bd: - return ChannelDistributions(**bd) - else: - return None - - @background_distributions.setter - def background_distributions(self, value: Optional[ChannelDistributions]) -> None: - try: - self.thing_settings["background_distributions"] = value.model_dump() - except AttributeError: - self.thing_settings["background_distributions"] = None - - @thing_property - def tolerance(self) -> float: - """How many standard deviations to allow for the background""" - return self.thing_settings.get("tolerance", 7) - - @tolerance.setter - def tolerance(self, value: float) -> None: - self.thing_settings["tolerance"] = value - - @thing_property - def fraction(self) -> float: - """How much of the image needs to be not background to label as sample""" - return self.thing_settings.get("fraction", 25) - - @fraction.setter - def fraction(self, value: float) -> None: - self.thing_settings["fraction"] = value - - def background_mask(self, image: np.ndarray) -> np.ndarray: - """Calculate a binary image, showing whether each pixel is background - - The image should be in LUV format, the ouput will be binary with the - same shape in the first two dimensions. - """ - d = self.background_distributions - if not d: - raise RuntimeError( - "Background is not set: you need to calibrate background detection." - ) - return np.all( - np.abs(image - np.array(d.means)[np.newaxis, np.newaxis, :]) - < np.array(d.standard_deviations)[np.newaxis, np.newaxis, :] - * self.tolerance, - axis=2, - ) - - @thing_action - def background_fraction(self, cam: CamDep) -> float: - """Determine what fraction of the current image is background - - This action will acquire a new image from the preview stream, then - evaluate whether it is foreground or background, by comparing it - too the saved statistics. This is done on a per-pixel basis, and - the returned value (between 0 and 100) is the fraction of the image - that is background. - """ - current_image = cam.grab_jpeg() - current_image = np.array(Image.open(current_image.open())) - - # we're working in the LUV colourspace as it collect colours together in a human-intuitive way - current_image_LUV = cv2.cvtColor(current_image, cv2.COLOR_RGB2LUV) - mask = self.background_mask(current_image_LUV) - return np.count_nonzero(mask) / np.prod(mask.shape) * 100 - - @thing_action - def image_is_sample(self, cam: CamDep) -> bool: - """Label the current image as either background or sample""" - b_fraction = self.background_fraction(cam) - fraction_threshold = self.fraction - - return (100 - b_fraction) > fraction_threshold - - @thing_action - def set_background(self, cam: CamDep): - """Grab an image, and use its statistics to set the background - - This should be run when the microscope is looking at an empty region, - and will calculate the mean and standard deviation of the pixel values - in the LUV colourspace. These values will then be used to compare - future images to the distribution, to determine if each pixel is - foreground or background. - """ - background = cam.grab_jpeg() - background = np.array(Image.open(background.open())) - - # we're working in the LUV colourspace as it collect colours together in a human-intuitive way - background_LUV = cv2.cvtColor(background, cv2.COLOR_RGB2LUV) - - ch1 = (background_LUV.T[0]).flatten() - ch2 = (background_LUV.T[1]).flatten() - ch3 = (background_LUV.T[2]).flatten() - - points = np.array([np.asarray(ch1), np.asarray(ch2), np.asarray(ch3)]).T - - # we get the mean and standard deviation of values in each channel - mu, std = np.apply_along_axis(norm.fit, 0, points) - - self.background_distributions = ChannelDistributions( - means=mu.tolist(), - standard_deviations=std.tolist(), - colorspace="LUV", - ) - - @property - def thing_state(self) -> Mapping: - bd = self.background_distributions - return { - "background_distributions": bd.model_dump() if bd else None, - "tolerance": self.tolerance, - "fraction": self.fraction, - } - - -BackgroundDep = direct_thing_client_dependency( - BackgroundDetectThing, "/background_detect/" -) - - -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""" - du = shutil.disk_usage(path) - if du.free < min_space: - raise NotEnoughFreeSpaceError( - "There is not enough free disk space to continue." - f"(Required: {min_space}, {du})." - ) - - -class ScanInfo(BaseModel): - """ "Summary information about a scan folder""" - - name: str - created: datetime - modified: datetime - number_of_images: int - - -DOWNLOADABLE_SCAN_FILES = ("images.zip",) - -JPEGBlob = blob_type("image/jpeg") -ZipBlob = blob_type("application/zip") - - -class SmartScanThing(Thing): - def __init__(self, path_to_openflexure_stitch: str): - self._script = path_to_openflexure_stitch - self._preview_stitch_popen_lock = threading.Lock() - self._correlate_popen_lock = threading.Lock() - self._scan_lock = threading.Lock() - - @property - def scans_folder_path(self) -> str: - """This folder will hold all the scans we do.""" - # TODO: This should be determined using sensible configuration. - # If the working directory is `/var/openflexure` this will result - # in scans being saved at `/var/openflexure/scans/` - return "scans" - - _latest_scan_name = None - - @thing_property - def latest_scan_name(self) -> Optional[str]: - """The name of the last scan to be started.""" - return self._latest_scan_name - - def scan_folder_path(self, scan_name: Optional[str] = None): - """The path to the scan folder with a given name""" - if not scan_name: - if not self.latest_scan_name: - raise IOError("There is no latest scan to return") - scan_name = self.latest_scan_name - return os.path.join(self.scans_folder_path, scan_name) - - def new_scan_folder(self, scan_name: str = "scan") -> str: - """Create a new empty folder, into which we can save scan images - - The folder will be named `{scan_name}_000001/` where the number is - zero-padded to be 6 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. - """ - if not os.path.exists(self.scans_folder_path): - os.makedirs(self.scans_folder_path) - if not scan_name: - scan_name = "scan" - for j in range(9999): - folder_path = os.path.join(self.scans_folder_path, f"{scan_name}_{j:04}") - if not os.path.exists(folder_path): - os.makedirs(folder_path) - self._latest_scan_name = os.path.basename(folder_path) - return folder_path - raise FileExistsError("Could not create a new scan folder: all names in use!") - - def move_to_next_point( - self, - stage: StageDep, - logger: InvocationLogger, - path: list[list[int]], - focused_path: list[list[int]], - ) -> list[int]: - """Remove the first point from the path, and move there. - - This will move to the next XY position in `path`, taking the `z` value - either from the current z value of the stage, or from `focused_path`. - - Returns the point we have moved to. - """ - loc = [path[0][0], path[0][1]] - path.remove(path[0]) - if len(focused_path) > 1: - z_index = closest(loc, focused_path) - z = int(focused_path[z_index][2]) - else: - z = stage.position["z"] - logger.info(f"Moving to {loc}") - stage.move_absolute(x=int(loc[0]), y=int(loc[1]), z=z - self.autofocus_dz / 2) - return loc + [z] - - @thing_action - def sample_scan( - self, - cancel: CancelHook, - logger: InvocationLogger, - autofocus: AutofocusDep, - stage: StageDep, - cam: CamDep, - metadata_getter: GetThingStates, - csm: CSMDep, - background_detect: BackgroundDep, - recentre: RecentreStage, - scan_name: str = "", - ): - """Move the stage to cover an area, taking images that can be tiled together. - - The stage will move in a pattern that grows outwards from the starting point, - stopping once it is surrounded by "background" (as detected by the - background_detect Thing). - - Input: - - * `overlap` is the fraction by which images should overlap, i.e. - `0.3` means we will move by 70% of the field of view each time. - """ - # Define these variables so we can use them in the finally: block - # (after testing they are not None) - scan_folder = None - images_folder = None - starting_position = None - capture_thread = None - self._scan_lock.acquire(timeout=0.1) - try: - # Before anything else, check that we've got a background set - # It's annoying to have to wait to find out! - max_dist = self.max_range - - if self.autofocus_dz == 0: - logger.info("Running scan without autofocus") - elif self.autofocus_dz <= 200: - logger.warning( - f"Your dz range is {self.autofocus_dz} steps, which is too short to attempt to focus. Running without autofocus" - ) - - if self.skip_background: - d = background_detect.background_distributions - if not d: - raise RuntimeError( - "Background is not set: you need to calibrate background detection." - ) - else: - logger.warning( - "This scan will run in a spiral from the starting point " - f"until you cancel it, or until it has moved by {max_dist} steps " - "in every direction. Make sure you watch it run to stop it leaving " - "the area of interest, or (worse) leading the microscope's range " - "of motion." - ) - names = [] - positions = [] - - # Record the starting position so we can move back there afterwards - starting_position = stage.position - - r = cam.grab_jpeg() - arr = np.array(Image.open(r.open())) - if csm.image_resolution is None: - raise RuntimeError( - "Camera-stage mapping is not calibrated. This is required before " - "scans can be carried out." - ) - if list(arr.shape[:2]) != [int(i) for i in csm.image_resolution]: - logger.error( - f"Images are, by default, {arr.shape[:2]}, but the CSM was " - f"calibrated at {csm.image_resolution}." - ) - - # Here, we calculate the x and y step size based on the desired overlap - # TODO: Consider using CSM calibration size instead - # TODO: generalise to have 2D displacements for x and y (as the - # camera and stage may not be aligned). - CSM = csm.image_to_stage_displacement_matrix - # csm_calibration_width = csm.last_calibration["image_resolution"][1] - - overlap = self.overlap - - dx = int( - np.abs(np.dot(np.array([0, arr.shape[1] * (1 - overlap)]), CSM)[0]) - ) - dy = int( - np.abs(np.dot(np.array([arr.shape[0] * (1 - overlap), 0]), CSM)[1]) - ) - - logger.info( - f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}" - ) - - # construct a 2D scan path - path = [[stage.position["x"], stage.position["y"]]] - - focused_path = [] # This holds a list of all points where focus succeeded - true_path = [] # This holds a list of all points visited - i = 0 - start_time = time.strftime("%H_%M_%S-%d_%m_%Y") - - scan_folder = self.new_scan_folder(scan_name) - images_folder = os.path.join(scan_folder, "images") - os.mkdir(images_folder) - raw_images_folder = os.path.join(images_folder, "raw") - os.mkdir(raw_images_folder) - logger.info(f"Saving images to {images_folder}") - - data = { - "scan_name": scan_name, - "overlap": overlap, - "autofocus range": self.autofocus_dz, - "dx": dx, - "dy": dy, - "start time": start_time, - "skipping background": self.skip_background, - } - - with open( - os.path.join(images_folder, "scan_inputs.json"), "w", encoding="utf-8" - ) as f: - json.dump(data, f, ensure_ascii=False, indent=4) - - # We will capture images and process them with this function, defined once here. - # Most of the variables it needs will be "baked in" so the arguments are just the ones - # that change each iteration. - # We also pre-calculate a normalisation image based on the LST and white balance - raw_image = cam.capture_array(stream_name="raw") - # TODO: assert the image is 10-bit packed, or deal with other formats! - rgb = rggb2rgb(raw2rggb(raw_image)) - lst = dict(cam.lens_shading_tables) - lum = np.array(lst["luminance"]) - Cr = np.array(lst["Cr"]) - Cb = np.array(lst["Cb"]) - gr, gb = cam.colour_gains - G = 1 / lum - R = ( - G / Cr / gr * np.min(Cr) - ) # The extra /np.max(Cr) emulates the quirky handling of Cr in - B = G / Cb / gb * np.min(Cb) # the picamera2 pipeline - white_norm_lores = np.stack([R, G, B], axis=2) - zoom_factors = [ - i / n for i, n in zip(rgb[..., :3].shape, white_norm_lores.shape) - ] - white_norm = zoom(white_norm_lores, zoom_factors, order=1)[ - : rgb.shape[0], : rgb.shape[1], : - ] # Could use some work - colour_correction_matrix = np.array(cam.colour_correction_matrix).reshape( - (3, 3) - ) - contrast_algorithm = cam.tuning["algorithms"][9]["rpi.contrast"] - gamma = np.array(contrast_algorithm["gamma_curve"]).reshape((-1, 2)) - gamma_8bit = interp1d(gamma[:, 0] / 255, gamma[:, 1] / 255) - - def process_raw_image(img): - normed = img / white_norm - corrected = np.dot( - colour_correction_matrix, normed.reshape((-1, 3)).T - ).T.reshape(normed.shape) - corrected[corrected < 0] = 0 - corrected[corrected > 255] = 255 - return gamma_8bit(corrected) - - logger.info( - f"Generated normalisation image with shape {white_norm.shape}, " - f"max {white_norm.max(axis=(0, 1))}, min {white_norm.min(axis=(0, 1))}" - ) - norm_inputs = { - "luminance": lum, - "Cr": Cr, - "Cb": Cb, - "gain_red": gr, - "gain_blue": gb, - } - - def capture_and_save(acquired: Event, name: str) -> None: - """Capture an image and save it to disk - - This will set the event `acquired` once the image has been acquired, so - that the stage may be moved while it's saved. - """ - try: - capture_start = time.time() - metadata = metadata_getter() - raw_image = cam.capture_array(stream_name="raw") - acquired.set() - acquisition_time = time.time() - # Save the raw image - np.savez( - os.path.join(raw_images_folder, name + ".npz"), - raw_image=raw_image, - **norm_inputs, - ) - # Process it into 8 bit RGB - processed = process_raw_image(rggb2rgb(raw2rggb(raw_image))) - processed[processed > 255] = 255 - processed[processed < 0] = 0 - img = Image.fromarray(processed.astype(np.uint8), mode="RGB") - img.save( - os.path.join(images_folder, name), quality=95, subsampling=0 - ) - exif_dict = piexif.load(os.path.join(images_folder, name)) - exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps( - metadata - ).encode("utf-8") - piexif.insert( - piexif.dump(exif_dict), os.path.join(images_folder, name) - ) - save_time = time.time() - logger.info( - f"Acquired {name} in {acquisition_time - capture_start:.1f}s then {save_time - acquisition_time:.1f}s saving to disk" - ) - except Exception as e: - logger.error( - f"An error occurred while saving {name}: {e}", exc_info=e - ) - - # At the start of the loop, we simultaneously capture an image and move to the next scan point. - # We skip capturing on the first run, because we've not focused yet - and also we skip capturing if - # it looks like background. - while len(path) > 0: - loc = self.move_to_next_point( - stage, logger, path=path, focused_path=focused_path - ) - if not self.preview_stitch_running(): - self.preview_stitch_start(images_folder) - if self.stitch_automatically: - if not self.correlate_running(): - self.correlate_start(images_folder, overlap=overlap) - - ensure_free_disk_space(scan_folder) - - # Check if the image is background - if self.skip_background: - image_is_sample = background_detect.image_is_sample() - else: - image_is_sample = True - - # if more than 92% of the image is background, treat it as background and continue - if not image_is_sample: - logger.info( - f"Skipping {stage.position} as it is {round(background_detect.background_fraction(), 0)}% background." - ) - else: - # if not, it's sample. run an autofocus and use the updated height - new_pos = [ - [stage.position["x"] - dx, stage.position["y"]], - [stage.position["x"] + dx, stage.position["y"]], - [stage.position["x"], stage.position["y"] - dy], - [stage.position["x"], stage.position["y"] + dy], - ] - for pos in new_pos: - if ( - pos not in [sublist[:2] for sublist in true_path] - and pos not in path - ): - path.append(pos) - - attempts = 0 - if self.autofocus_dz > 200: - while True: - jpeg_zs, jpeg_sizes = autofocus.looping_autofocus( - dz=self.autofocus_dz, start="base" - ) - current_height = stage.position["z"] - time.sleep(0.2) - autofocus_success = autofocus.verify_focus_sharpness( - sweep_sizes=jpeg_sizes, camera=CamDep, threshold=0.92 - ) - logger.info( - f"We just tested the focus! Result was {autofocus_success}" - ) - - if autofocus_success: - # if there have been successful autofocuses in this scan, find the closest one in x-y - # test if the change in z between them exceeds a ratio (indicating a failed autofocus) - if len(focused_path) > 0: - nearest_focused_site = focused_path[ - closest(loc, focused_path) - ] - result = limit_focus_change( - nearest_focused_site[0:2], - nearest_focused_site[-1], - loc[0:2], - current_height, - 0.5, - ) - - # if there haven't been any previous autofocuses, we have to assume this one worked - else: - result = "accept" - else: - result = "reject" - - # if the autofocus worked, add the current position to the list of successful locations - if result == "accept": - loc = list(stage.position.values()) - focused_path.append(loc) - break - if attempts >= 3: - logger.warning("Could not autofocus after 3 attempts.") - break - # if the autofocus was rejected, we return to the height of the closest successful autofocus. not perfect, but better than wandering out of focus - logger.info( - "The focus has shifted further than we expect: retrying." - ) - stage.move_absolute(z=int(loc[2])) - attempts += 1 - - # Acquire the image in a thread, and continue once it's acquired (i.e. leave saving in the background) - if capture_thread: # wait for the previous capture to be saved, i.e. don't leave more than one image saving in the background - if capture_thread.is_alive(): - wait_start = time.time() - capture_thread.join() - wait_time = time.time() - wait_start - logger.info( - f"Waited {wait_time:.1f}s for the previous capture to finish saving." - ) - acquired = Event() - name = f"image_{loc[0]}_{loc[1]}.jpg" - time.sleep(0.2) - capture_thread = Thread( - target=capture_and_save, - kwargs={ - # "cam": cam, - # "logger": logger, - "acquired": acquired, - "name": name, - # "images_folder": images_folder, - # "raw_images_folder": raw_images_folder, - }, - ) - capture_thread.start() - acquired.wait() # wait until the image is acquired - # time.sleep(0.5) - positions.append(loc[:2]) - names.append(name) - - # add the current position to the list of all positions visited - true_path.append(loc) - - # if len(names) > 1: - # generate_config(images_folder, positions, names, CSM, csm_calibration_width, img_width, logger) - - temp_path = [] - - for i in path: - if distance_to_site(i, true_path[0][:2]) < max_dist: - temp_path.append(i) - else: - logger.info(f"Rejected moving to {i} as it is out of range") - path = temp_path.copy() - path = sorted( - path, - key=lambda x: ( - steps_from_centre(x, true_path[0][:2], dx, dy), - distance_to_site(loc[:2], x), - ), - ) - self.create_zip_of_scan( - logger=logger, - scan_name=scan_folder.split("scans/")[1], - download_zip=False, - ) - - except InvocationCancelledError: - logger.error("Stopping scan because it was cancelled.") - except NotEnoughFreeSpaceError as e: - logger.error( - f"Stopping scan to avoid filling up the disk: {e}", - exc_info=e, - ) - raise e - except Exception as e: - logger.error( - f"The scan stopped because of an error: {e}", - "We will attempt to stitch and archive the images acquired so far.", - exc_info=e, - ) - raise e - finally: - if capture_thread: - capture_thread.join() - try: - logger.info("Returning to starting position.") - if starting_position is not None: - stage.move_absolute(**starting_position, block_cancellation=True) - finally: - self._scan_lock.release() - self.create_zip_of_scan( - logger=logger, - scan_name=scan_folder.split("scans/")[1], - download_zip=False, - ) - logger.info("Waiting for background processes to finish...") - self.preview_stitch_wait() - self.correlate_wait() - try: - if scan_folder and self.stitch_automatically: - logger.info("Stitching final image (may take some time)...") - self.stitch_scan( - logger, os.path.basename(scan_folder), overlap=overlap - ) - except SubprocessError as e: - logger.error(f"Stitching failed: {e}", exc_info=e) - - @thing_property - def max_range(self) -> int: - """The maximum distance from the centre of the scan before we break""" - return self.thing_settings.get("max_range", 45000) - - @max_range.setter - def max_range(self, value: int) -> None: - self.thing_settings["max_range"] = value - - @thing_property - def stitch_tiff(self) -> bool: - """Whether or not to also produce a pyramidal tiff""" - return self.thing_settings.get("stitch_tiff", False) - - @stitch_tiff.setter - def stitch_tiff(self, value: bool) -> None: - self.thing_settings["stitch_tiff"] = value - - @thing_property - def skip_background(self) -> bool: - """Whether to detect and skip empty fields of view - - This uses the settings from the `background_detect` Thing. - """ - return self.thing_settings.get("skip_background", True) - - @skip_background.setter - def skip_background(self, value: bool) -> None: - self.thing_settings["skip_background"] = value - - @thing_property - def autofocus_dz(self) -> int: - """The z distance to perform an autofocus""" - return self.thing_settings.get("autofocus_dz", 1000) - - @autofocus_dz.setter - def autofocus_dz(self, value: int) -> None: - self.thing_settings["autofocus_dz"] = value - - @thing_property - def overlap(self) -> float: - """The z distance to perform an autofocus""" - return self.thing_settings.get("overlap", 0.45) - - @overlap.setter - def overlap(self, value: float) -> None: - self.thing_settings["overlap"] = value - - @thing_property - def stitch_automatically(self) -> bool: - """Should we attempt to stitch scans as we go?""" - return self.thing_settings.get("stitch_automatically", True) - - @stitch_automatically.setter - def stitch_automatically(self, value: bool) -> None: - self.thing_settings["stitch_automatically"] = value - - @thing_property - def scans(self) -> list[ScanInfo]: - """All the available scans - - Each scan has a name (which can be used to access it), along with - its modified and created times (according to the filesystem) and - the number of items in the `images` folder. Note that the number - of images reported may be confused if non-image files are present - in the `images` folder. - """ - scans: list[ScanInfo] = [] - if not os.path.isdir(self.scans_folder_path): - return scans - for f in os.listdir(self.scans_folder_path): - path = os.path.join(self.scans_folder_path, f) - if os.path.isdir(path): - images_folder = os.path.join(path, "images") - if os.path.isdir(images_folder): - number_of_images = len(os.listdir(images_folder)) - else: - number_of_images = 0 - scans.append( - ScanInfo( - name=f, - created=os.path.getctime(path), - modified=os.path.getmtime(path), - number_of_images=number_of_images, - ) - ) - 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.scans_folder_path, scan_name, file) - if not os.path.isfile(path): - raise HTTPException(404, "File not found") - return FileResponse(path) - - @fastapi_endpoint( - "delete", - "scans/{scan_name}", - responses={ - 200: {"description": "Successfully deleted scan"}, - 404: {"description": "Scan not found"}, - }, - ) - def delete_scan(self, scan_name: str) -> None: - """Delete all files from a scan. - - This endpoint allows scans to be deleted from disk. - """ - path = os.path.join(self.scans_folder_path, scan_name) - if not os.path.isdir(path): - print(f"can't find {path}") - raise HTTPException(404, "Scan not found") - shutil.rmtree(path) - - @fastapi_endpoint( - "delete", - "scans", - ) - def delete_all_scans(self) -> None: - """Delete all the scans on the microscope - - **This will irreversibly remove all smart scan data from the - microscope!** - Use with extreme caution. - """ - for scan in self.scans: - self.delete_scan(scan.name) - - def images_folder(self, scan_name: Optional[str] = None) -> str: - scan_folder = self.scan_folder_path(scan_name=scan_name) - return os.path.join(scan_folder, "images") - - @property - def latest_preview_stitch_path(self): - """The path of the latest preview stitched image""" - return os.path.join(self.images_folder(), "stitched_from_stage.jpg") - - @thing_property - def latest_preview_stitch_time(self) -> Optional[datetime]: - """The modification time of the latest preview image - - This will return `null` if there is no preview image to return. - """ - try: - fpath = self.latest_preview_stitch_path - if os.path.exists(fpath): - return os.path.getmtime(fpath) - except IOError: - return None - return None - - @fastapi_endpoint( - "get", - "latest_preview_stitch.jpg", - responses={ - 200: { - "description": "A preview-quality stitched image", - "content": {"image/jpeg": {}}, - }, - 404: {"description": "File not found"}, - }, - ) - def get_latest_preview(self) -> FileResponse: - """Retrieve the latest preview image.""" - path = self.latest_preview_stitch_path - if not os.path.isfile(path): - raise HTTPException(404, "File not found") - return FileResponse(path) - - _preview_stitch_popen = None - - def preview_stitch_start(self, images_folder: str) -> None: - """Start stitching a preview of the scan in a subprocess""" - if self.preview_stitch_running(): - raise RuntimeError("Only one subprocess is allowed at a time") - with self._preview_stitch_popen_lock: - self._preview_stitch_popen = Popen( - [self._script, "--stitching_mode", "only_stage_stitch", images_folder] - ) - - def preview_stitch_running(self) -> bool: - """Whether there is a preview stitch running in a subprocess""" - with self._preview_stitch_popen_lock: - if self._preview_stitch_popen is None: - return False - if self._preview_stitch_popen.poll() is None: - return True - return False - - def preview_stitch_wait(self): - if self.preview_stitch_running(): - with self._preview_stitch_popen_lock: - self._preview_stitch_popen.wait() - - _correlate_popen = None - - def correlate_start(self, images_folder: str, overlap: float = 0.1) -> None: - """Start stitching a preview of the scan in a subprocess""" - if self.correlate_running(): - raise RuntimeError("Only one subprocess is allowed at a time") - with self._correlate_popen_lock: - self._correlate_popen = Popen( - [ - self._script, - "--stitching_mode", - "only_correlate", - "--minimum_overlap", - f"{round(overlap * 0.9, 2)}", - images_folder, - ] - ) - - def correlate_running(self) -> bool: - """Whether there is a preview stitch running in a subprocess""" - with self._correlate_popen_lock: - if self._correlate_popen is None: - return False - if self._correlate_popen.poll() is None: - return True - return False - - def correlate_wait(self): - if self.correlate_running(): - with self._correlate_popen_lock: - self._correlate_popen.wait() - - def run_subprocess( - self, - logger: InvocationLogger, - cmd: list[str], - ) -> CompletedProcess: - """Run a subprocess and log any output""" - logger.info(f"Running command in subprocess: `{' '.join(cmd)}`") - - p = Popen(cmd, stdout=PIPE, stderr=STDOUT, bufsize=1, universal_newlines=True) - os.set_blocking(p.stdout.fileno(), False) - logger.info(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))) - while p.poll() is None: - try: - output = p.stdout.readline() - if output != "" and output is not None: - logger.info(output) - except: - pass - - for line in p.stdout: - try: - output = p.stdout.readline() - if output != "" and output is not None: - logger.info(output) - except: - pass - - logger.info("Stitching complete") - return p - - @thing_action - def stitch_scan( - self, - logger: InvocationLogger, - scan_name: Optional[str] = None, - overlap: float = 0.0, - ) -> None: - """Generate a stitched image based on stage position metadata""" - images_folder = self.images_folder(scan_name=scan_name) - - if self.stitch_tiff: - tiff_arg = "--stitch_tiff" - else: - tiff_arg = "--no-stitch_tiff" - - if overlap == 0.0: - try: - with open(os.path.join(images_folder, "scan_inputs.json")) as data_file: - data_loaded = json.load(data_file) - logger.info(data_loaded) - overlap = data_loaded["overlap"] - except: - overlap = 0.1 - self.run_subprocess( - logger, - [ - self._script, - "--stitching_mode", - "all", - f"{tiff_arg}", - "--minimum_overlap", - f"{round(overlap * 0.9, 2)}", - images_folder, - ], - ) - - @thing_action - def create_zip_of_scan( - self, - logger: InvocationLogger, - scan_name: Optional[str] = None, - download_zip=True, - ) -> ZipBlob: - """Generate a zip file that can be downloaded, with all the scan files in it.""" - images_folder = self.images_folder(scan_name=scan_name) - scan_folder = self.scan_folder_path(scan_name=scan_name) - if ( - scan_folder != os.path.dirname(images_folder) - or os.path.basename(images_folder) != "images" - ): - logger.error( - "There is a problem with filenames, the archive may be incorrect." - f"scan_folder: {scan_folder}, images_folder: {images_folder}." - ) - if not os.path.isdir(images_folder): - raise FileNotFoundError( - f"Tried to make a zip archive of {images_folder} but it does not exist." - ) - # logger.info("Creating zip archive of images (may take some time)...") - - zip_fname = f"{os.path.join(scan_folder, 'images')}.zip" - - # Create an empty zip file - we don't want to autofill it with files, - # as some of them should only be added at the end (as we can't overwrite) - # them once they change - if not os.path.isfile(zip_fname): - with zipfile.ZipFile(zip_fname, mode="w") as zip: - pass - - # 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 - folder_path = self.scan_folder_path(scan_name) - files = glob.glob(folder_path + "/**/*", recursive=True) - files = [i.split(f"{folder_path}/")[1] for i 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 - files_to_delay = [ - "TileConfiguration", - "tiling_cache", - "stitched.jp", - "stitched_from", - "stitched.om", - ] - tiff_name = "" - - with zipfile.ZipFile(zip_fname, mode="a") as zip: - for file in files: - if "stitched.jp" in file: - stitch_name = os.path.split(file)[1] - if ".ome.tiff" in file: - tiff_name = os.path.split(file)[1] - if any(banned_name in file for banned_name in files_to_delay): - # logger.info(f'we only add {file} into zip at the end of the scan') - pass - elif file in current_zip: - # logger.info(f'{file} is already in zip') - pass - elif ".zip" in file or "raw" in file: - # logger.info('Not adding the .zip to itself') - pass - else: - logger.info(f"appending {file} to zip") - zip.write(os.path.join(folder_path, file), arcname=file) - - images_folder = os.path.join(folder_path, "images") - # Promote key files to the top level of the zip only at the end of the scan (when downloading) - # and finally zip some of the final files - # TODO: if you download multiple times, you get duplicate files - is this a problem? - if download_zip: - with zipfile.ZipFile(zip_fname, mode="a") as zip: - for fname in ["stitched_from_stage.jpg", stitch_name, tiff_name]: - fpath = os.path.join(images_folder, fname) - if os.path.exists(fpath): - logger.info(f"copying {fpath} to upper level") - zip.write(fpath, arcname=fname) - for file in files: - if any(banned_name in file for banned_name in files_to_delay): - logger.info(f"we are finally adding {file} into zip") - zip.write(os.path.join(folder_path, file), arcname=file) - logger.info("about to download zip") - return ZipBlob.from_file(zip_fname) - - @thing_action - def get_files_in_zip(self, zip_path): - """List the relative paths of all files and folders in the zip folder specified""" - zip = zipfile.ZipFile(zip_path) - zip = [os.path.normpath(i) for i in zip.namelist()] - - return zip +# ruff: noqa: E722 + +import shutil +import zipfile +import threading +from typing import Mapping, Optional +import cv2 +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 scipy.stats import norm +from scipy.ndimage import zoom +from scipy.interpolate import interp1d +from datetime import datetime +from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, STDOUT +from threading import Event, Thread +import glob +import json +import piexif + +from labthings_fastapi.thing import Thing +from labthings_fastapi.dependencies.metadata import GetThingStates +from labthings_fastapi.dependencies.thing import direct_thing_client_dependency +from labthings_fastapi.dependencies.invocation import ( + CancelHook, + InvocationLogger, + InvocationCancelledError, +) +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.things.autofocus import AutofocusThing +from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper +from openflexure_microscope_server.things.auto_recentre_stage import RecentringThing + + +CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/") +AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/") +RecentreStage = direct_thing_client_dependency(RecentringThing, "/auto_recentre_stage/") + + +def closest(current, focused_path): + """Finds the index of the closest x-y position in a list from the current position, + with ties split by the later element in the list (most recently taken) + + must be float64 to deal with the huge numbers involved!""" + + current_pos = np.array(current[:2], dtype="float64") + path_pos = np.asarray(focused_path, dtype="float64").T[:2].T + + dist_2 = np.sqrt( + np.sum((path_pos - current_pos) ** 2, axis=1, dtype="float64"), dtype="float64" + ) + min_dist = np.argmin(dist_2) + mask = np.where(dist_2 == dist_2[min_dist], 1, 0) + try: + closest = np.max(np.nonzero(mask)) + except: + closest = 0 + return closest + + +def unpack_autofocus(scan_data): + """Extract z, sharpness data from a move_and_measure call + + Data will start at `start_index`, i.e. `start_index` points are dropped + from the beginning of the array. + """ + jpeg_times = scan_data["jpeg_times"] + jpeg_sizes = scan_data["jpeg_sizes"] + jpeg_sizes_MB = [x / 10**3 for x in jpeg_sizes] + stage_times = scan_data["stage_times"] + stage_positions = scan_data["stage_positions"] + stage_height = [pos[2] for pos in stage_positions] + + jpeg_heights = np.interp(jpeg_times, stage_times, stage_height) + + def turningpoints(lst): + dx = np.diff(lst) + return dx[1:] * dx[:-1] < 0 + + turning = np.where(turningpoints(jpeg_heights))[0] + 1 + + return jpeg_heights[turning[0] : turning[1]], jpeg_sizes_MB[turning[0] : turning[1]] + + +def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit): + # limit is the largest ratio of change in z to change in xy that's allowed + + prev_xy = np.asarray(prev_pos, dtype="float64") + new_xy = np.asarray(new_pos, dtype="float64") + + dist = np.sqrt(np.sum(((new_xy - prev_xy) / 10**4) ** 2, dtype="float64")) + + focus_change = abs(new_z - prev_z) / 10**4 + if dist == 0: + print(f"Not moved between {prev_pos} and {new_pos}") + movement_ratio = 0 + else: + movement_ratio = np.divide(focus_change, dist, dtype="float64") + + # print('Movement ratio is {0} in z per lateral step. The limit is {1}'.format(round(movement_ratio, 4), round(limit,4))) + # print(f'This is the distance between {prev_pos}, {prev_z} and {new_pos}, {new_z}') + + if movement_ratio > limit: + return "reject" + else: + return "accept" + + +# def distance_to_site(current, next): +# current = np.array(current, dtype="float64") +# next = np.array(next, dtype="float64") +# if (next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2 < 0: +# print(f"Negative distance between {next} and {current}") +# return np.sqrt( +# (next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2, dtype="float64" +# ) + + +def steps_from_centre(current_loc, starting_loc, dx, dy): + step_size = np.array([dx, dy]) + return np.max(np.abs(np.divide(np.subtract(current_loc, starting_loc), step_size))) + + +# def set_template(microscope, pos): +# microscope.move(pos) +# background = microscope.grab_image_array() +# background_LUV = cv2.cvtColor(background, cv2.COLOR_RGB2LUV) + +# ch1 = (background_LUV.T[0]).flatten() +# ch2 = (background_LUV.T[1]).flatten() +# ch3 = (background_LUV.T[2]).flatten() + +# points = np.array([np.asarray(ch1),np.asarray(ch2),np.asarray(ch3)]).T + +# # we get the mean and standard deviation of values in each channel + +# mu, std = np.apply_along_axis(norm.fit, 0, points) +# stats_list = np.vstack([mu, std]) +# return stats_list + + +def distance_to_site(current, next): + next = np.array(next, dtype="float64") + current = np.array(current, dtype="float64") + return np.sqrt((next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2) + + +def scale_csm(csm_matrix, calibration_width, img_width): + "Account for a calibration width that may differ from image width" + scale = img_width / calibration_width # Usually >1, if we calibrated at low res + csm = np.array(csm_matrix) / scale # Decrease the CSM if pixels are smaller] + return csm + + +def generate_config( + folder_path: str, + positions: list, + names: list, + camera_to_sample_matrix, + csm_calibration_width, + img_width, + logger, +): + positions = np.array(positions) + mean_loc = np.mean(positions, axis=0) + + # TODO: positions from recent scans need to be 2x bigger - change to CSM res? + + camera_to_sample_matrix = scale_csm( + camera_to_sample_matrix, csm_calibration_width, img_width + ) + + with open(os.path.join(folder_path, "TileConfiguration.txt"), "w") as fp: + fp.write( + "# Define the number of dimensions we are working on\ndim = 2\n\n# Define the image coordinates\n" + ) + for i in range(len(names)): + loc = np.dot( + (positions[i] - mean_loc), np.linalg.inv(camera_to_sample_matrix) + ) + fp.write(f"{names[i]}; ; {loc[1], loc[0]} \n") + + +def raw2rggb(raw): + """Convert packed 10 bit raw to RGGB 8 bit""" + raw = np.asarray(raw) # ensure it's an array + rggb = np.empty((616, 820, 4), dtype=np.uint8) + raw_w = rggb.shape[1] // 2 * 5 + for plane, offset in enumerate([(1, 1), (0, 1), (1, 0), (0, 0)]): + rggb[:, ::2, plane] = raw[offset[0] :: 2, offset[1] : raw_w + offset[1] : 5] + rggb[:, 1::2, plane] = raw[ + offset[0] :: 2, offset[1] + 2 : raw_w + offset[1] + 2 : 5 + ] + return rggb + + +def rggb2rgb(rggb): + return np.stack( + [rggb[..., 0], rggb[..., 1] // 2 + rggb[..., 2] // 2, rggb[..., 3]], axis=2 + ) + + +class ChannelDistributions(BaseModel): + means: list[float] + standard_deviations: list[float] + colorspace: str = "LUV" + + +class BackgroundDetectThing(Thing): + @thing_property + def background_distributions(self) -> Optional[ChannelDistributions]: + """The statistics of the background image""" + bd = self.thing_settings.get("background_distributions", None) + if bd: + return ChannelDistributions(**bd) + else: + return None + + @background_distributions.setter + def background_distributions(self, value: Optional[ChannelDistributions]) -> None: + try: + self.thing_settings["background_distributions"] = value.model_dump() + except AttributeError: + self.thing_settings["background_distributions"] = None + + @thing_property + def tolerance(self) -> float: + """How many standard deviations to allow for the background""" + return self.thing_settings.get("tolerance", 7) + + @tolerance.setter + def tolerance(self, value: float) -> None: + self.thing_settings["tolerance"] = value + + @thing_property + def fraction(self) -> float: + """How much of the image needs to be not background to label as sample""" + return self.thing_settings.get("fraction", 25) + + @fraction.setter + def fraction(self, value: float) -> None: + self.thing_settings["fraction"] = value + + def background_mask(self, image: np.ndarray) -> np.ndarray: + """Calculate a binary image, showing whether each pixel is background + + The image should be in LUV format, the ouput will be binary with the + same shape in the first two dimensions. + """ + d = self.background_distributions + if not d: + raise RuntimeError( + "Background is not set: you need to calibrate background detection." + ) + return np.all( + np.abs(image - np.array(d.means)[np.newaxis, np.newaxis, :]) + < np.array(d.standard_deviations)[np.newaxis, np.newaxis, :] + * self.tolerance, + axis=2, + ) + + @thing_action + def background_fraction(self, cam: CamDep) -> float: + """Determine what fraction of the current image is background + + This action will acquire a new image from the preview stream, then + evaluate whether it is foreground or background, by comparing it + too the saved statistics. This is done on a per-pixel basis, and + the returned value (between 0 and 100) is the fraction of the image + that is background. + """ + current_image = cam.grab_jpeg() + current_image = np.array(Image.open(current_image.open())) + + # we're working in the LUV colourspace as it collect colours together in a human-intuitive way + current_image_LUV = cv2.cvtColor(current_image, cv2.COLOR_RGB2LUV) + mask = self.background_mask(current_image_LUV) + return np.count_nonzero(mask) / np.prod(mask.shape) * 100 + + @thing_action + def image_is_sample(self, cam: CamDep) -> bool: + """Label the current image as either background or sample""" + b_fraction = self.background_fraction(cam) + fraction_threshold = self.fraction + + return (100 - b_fraction) > fraction_threshold + + @thing_action + def set_background(self, cam: CamDep): + """Grab an image, and use its statistics to set the background + + This should be run when the microscope is looking at an empty region, + and will calculate the mean and standard deviation of the pixel values + in the LUV colourspace. These values will then be used to compare + future images to the distribution, to determine if each pixel is + foreground or background. + """ + background = cam.grab_jpeg() + background = np.array(Image.open(background.open())) + + # we're working in the LUV colourspace as it collect colours together in a human-intuitive way + background_LUV = cv2.cvtColor(background, cv2.COLOR_RGB2LUV) + + ch1 = (background_LUV.T[0]).flatten() + ch2 = (background_LUV.T[1]).flatten() + ch3 = (background_LUV.T[2]).flatten() + + points = np.array([np.asarray(ch1), np.asarray(ch2), np.asarray(ch3)]).T + + # we get the mean and standard deviation of values in each channel + mu, std = np.apply_along_axis(norm.fit, 0, points) + + self.background_distributions = ChannelDistributions( + means=mu.tolist(), + standard_deviations=std.tolist(), + colorspace="LUV", + ) + + @property + def thing_state(self) -> Mapping: + bd = self.background_distributions + return { + "background_distributions": bd.model_dump() if bd else None, + "tolerance": self.tolerance, + "fraction": self.fraction, + } + + +BackgroundDep = direct_thing_client_dependency( + BackgroundDetectThing, "/background_detect/" +) + + +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""" + du = shutil.disk_usage(path) + if du.free < min_space: + raise NotEnoughFreeSpaceError( + "There is not enough free disk space to continue." + f"(Required: {min_space}, {du})." + ) + + +class ScanInfo(BaseModel): + """ "Summary information about a scan folder""" + + name: str + created: datetime + modified: datetime + number_of_images: int + + +DOWNLOADABLE_SCAN_FILES = ("images.zip",) + +JPEGBlob = blob_type("image/jpeg") +ZipBlob = blob_type("application/zip") + + +class SmartScanThing(Thing): + def __init__(self, path_to_openflexure_stitch: str): + self._script = path_to_openflexure_stitch + self._preview_stitch_popen_lock = threading.Lock() + self._correlate_popen_lock = threading.Lock() + self._scan_lock = threading.Lock() + + @property + def scans_folder_path(self) -> str: + """This folder will hold all the scans we do.""" + # TODO: This should be determined using sensible configuration. + # If the working directory is `/var/openflexure` this will result + # in scans being saved at `/var/openflexure/scans/` + return "scans" + + _latest_scan_name = None + + @thing_property + def latest_scan_name(self) -> Optional[str]: + """The name of the last scan to be started.""" + return self._latest_scan_name + + def scan_folder_path(self, scan_name: Optional[str] = None): + """The path to the scan folder with a given name""" + if not scan_name: + if not self.latest_scan_name: + raise IOError("There is no latest scan to return") + scan_name = self.latest_scan_name + return os.path.join(self.scans_folder_path, scan_name) + + def new_scan_folder(self, scan_name: str = "scan") -> str: + """Create a new empty folder, into which we can save scan images + + The folder will be named `{scan_name}_000001/` where the number is + zero-padded to be 6 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. + """ + if not os.path.exists(self.scans_folder_path): + os.makedirs(self.scans_folder_path) + if not scan_name: + scan_name = "scan" + for j in range(9999): + folder_path = os.path.join(self.scans_folder_path, f"{scan_name}_{j:04}") + if not os.path.exists(folder_path): + os.makedirs(folder_path) + self._latest_scan_name = os.path.basename(folder_path) + return folder_path + raise FileExistsError("Could not create a new scan folder: all names in use!") + + def move_to_next_point( + self, + stage: StageDep, + logger: InvocationLogger, + path: list[list[int]], + focused_path: list[list[int]], + ) -> list[int]: + """Remove the first point from the path, and move there. + + This will move to the next XY position in `path`, taking the `z` value + either from the current z value of the stage, or from `focused_path`. + + Returns the point we have moved to. + """ + loc = [path[0][0], path[0][1]] + path.remove(path[0]) + if len(focused_path) > 1: + z_index = closest(loc, focused_path) + z = int(focused_path[z_index][2]) + else: + z = stage.position["z"] + logger.info(f"Moving to {loc}") + stage.move_absolute(x=int(loc[0]), y=int(loc[1]), z=z - self.autofocus_dz / 2) + return loc + [z] + + @thing_action + def sample_scan( + self, + cancel: CancelHook, + logger: InvocationLogger, + autofocus: AutofocusDep, + stage: StageDep, + cam: CamDep, + metadata_getter: GetThingStates, + csm: CSMDep, + background_detect: BackgroundDep, + recentre: RecentreStage, + scan_name: str = "", + ): + """Move the stage to cover an area, taking images that can be tiled together. + + The stage will move in a pattern that grows outwards from the starting point, + stopping once it is surrounded by "background" (as detected by the + background_detect Thing). + + Input: + + * `overlap` is the fraction by which images should overlap, i.e. + `0.3` means we will move by 70% of the field of view each time. + """ + # Define these variables so we can use them in the finally: block + # (after testing they are not None) + scan_folder = None + images_folder = None + starting_position = None + capture_thread = None + self._scan_lock.acquire(timeout=0.1) + try: + # Before anything else, check that we've got a background set + # It's annoying to have to wait to find out! + max_dist = self.max_range + + if self.autofocus_dz == 0: + logger.info("Running scan without autofocus") + elif self.autofocus_dz <= 200: + logger.warning( + f"Your dz range is {self.autofocus_dz} steps, which is too short to attempt to focus. Running without autofocus" + ) + + if self.skip_background: + d = background_detect.background_distributions + if not d: + raise RuntimeError( + "Background is not set: you need to calibrate background detection." + ) + else: + logger.warning( + "This scan will run in a spiral from the starting point " + f"until you cancel it, or until it has moved by {max_dist} steps " + "in every direction. Make sure you watch it run to stop it leaving " + "the area of interest, or (worse) leading the microscope's range " + "of motion." + ) + names = [] + positions = [] + + # Record the starting position so we can move back there afterwards + starting_position = stage.position + + r = cam.grab_jpeg() + arr = np.array(Image.open(r.open())) + if csm.image_resolution is None: + raise RuntimeError( + "Camera-stage mapping is not calibrated. This is required before " + "scans can be carried out." + ) + if list(arr.shape[:2]) != [int(i) for i in csm.image_resolution]: + logger.error( + f"Images are, by default, {arr.shape[:2]}, but the CSM was " + f"calibrated at {csm.image_resolution}." + ) + + # Here, we calculate the x and y step size based on the desired overlap + # TODO: Consider using CSM calibration size instead + # TODO: generalise to have 2D displacements for x and y (as the + # camera and stage may not be aligned). + CSM = csm.image_to_stage_displacement_matrix + # csm_calibration_width = csm.last_calibration["image_resolution"][1] + + overlap = self.overlap + + dx = int( + np.abs(np.dot(np.array([0, arr.shape[1] * (1 - overlap)]), CSM)[0]) + ) + dy = int( + np.abs(np.dot(np.array([arr.shape[0] * (1 - overlap), 0]), CSM)[1]) + ) + + logger.info( + f"Based on an overlap of {overlap}, we will make steps of {dx}, {dy}" + ) + + # construct a 2D scan path + path = [[stage.position["x"], stage.position["y"]]] + + focused_path = [] # This holds a list of all points where focus succeeded + true_path = [] # This holds a list of all points visited + i = 0 + start_time = time.strftime("%H_%M_%S-%d_%m_%Y") + + scan_folder = self.new_scan_folder(scan_name) + images_folder = os.path.join(scan_folder, "images") + os.mkdir(images_folder) + raw_images_folder = os.path.join(images_folder, "raw") + os.mkdir(raw_images_folder) + logger.info(f"Saving images to {images_folder}") + + data = { + "scan_name": scan_name, + "overlap": overlap, + "autofocus range": self.autofocus_dz, + "dx": dx, + "dy": dy, + "start time": start_time, + "skipping background": self.skip_background, + } + + with open( + os.path.join(images_folder, "scan_inputs.json"), "w", encoding="utf-8" + ) as f: + json.dump(data, f, ensure_ascii=False, indent=4) + + # We will capture images and process them with this function, defined once here. + # Most of the variables it needs will be "baked in" so the arguments are just the ones + # that change each iteration. + # We also pre-calculate a normalisation image based on the LST and white balance + raw_image = cam.capture_array(stream_name="raw") + # TODO: assert the image is 10-bit packed, or deal with other formats! + rgb = rggb2rgb(raw2rggb(raw_image)) + lst = dict(cam.lens_shading_tables) + lum = np.array(lst["luminance"]) + Cr = np.array(lst["Cr"]) + Cb = np.array(lst["Cb"]) + gr, gb = cam.colour_gains + G = 1 / lum + R = ( + G / Cr / gr * np.min(Cr) + ) # The extra /np.max(Cr) emulates the quirky handling of Cr in + B = G / Cb / gb * np.min(Cb) # the picamera2 pipeline + white_norm_lores = np.stack([R, G, B], axis=2) + zoom_factors = [ + i / n for i, n in zip(rgb[..., :3].shape, white_norm_lores.shape) + ] + white_norm = zoom(white_norm_lores, zoom_factors, order=1)[ + : rgb.shape[0], : rgb.shape[1], : + ] # Could use some work + colour_correction_matrix = np.array(cam.colour_correction_matrix).reshape( + (3, 3) + ) + contrast_algorithm = cam.tuning["algorithms"][9]["rpi.contrast"] + gamma = np.array(contrast_algorithm["gamma_curve"]).reshape((-1, 2)) + gamma_8bit = interp1d(gamma[:, 0] / 255, gamma[:, 1] / 255) + + def process_raw_image(img): + normed = img / white_norm + corrected = np.dot( + colour_correction_matrix, normed.reshape((-1, 3)).T + ).T.reshape(normed.shape) + corrected[corrected < 0] = 0 + corrected[corrected > 255] = 255 + return gamma_8bit(corrected) + + logger.info( + f"Generated normalisation image with shape {white_norm.shape}, " + f"max {white_norm.max(axis=(0, 1))}, min {white_norm.min(axis=(0, 1))}" + ) + norm_inputs = { + "luminance": lum, + "Cr": Cr, + "Cb": Cb, + "gain_red": gr, + "gain_blue": gb, + } + + def capture_and_save(acquired: Event, name: str) -> None: + """Capture an image and save it to disk + + This will set the event `acquired` once the image has been acquired, so + that the stage may be moved while it's saved. + """ + try: + capture_start = time.time() + metadata = metadata_getter() + raw_image = cam.capture_array(stream_name="raw") + acquired.set() + acquisition_time = time.time() + # Save the raw image + np.savez( + os.path.join(raw_images_folder, name + ".npz"), + raw_image=raw_image, + **norm_inputs, + ) + # Process it into 8 bit RGB + processed = process_raw_image(rggb2rgb(raw2rggb(raw_image))) + processed[processed > 255] = 255 + processed[processed < 0] = 0 + img = Image.fromarray(processed.astype(np.uint8), mode="RGB") + img.save( + os.path.join(images_folder, name), quality=95, subsampling=0 + ) + exif_dict = piexif.load(os.path.join(images_folder, name)) + exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps( + metadata + ).encode("utf-8") + piexif.insert( + piexif.dump(exif_dict), os.path.join(images_folder, name) + ) + save_time = time.time() + logger.info( + f"Acquired {name} in {acquisition_time - capture_start:.1f}s then {save_time - acquisition_time:.1f}s saving to disk" + ) + except Exception as e: + logger.error( + f"An error occurred while saving {name}: {e}", exc_info=e + ) + + # At the start of the loop, we simultaneously capture an image and move to the next scan point. + # We skip capturing on the first run, because we've not focused yet - and also we skip capturing if + # it looks like background. + while len(path) > 0: + loc = self.move_to_next_point( + stage, logger, path=path, focused_path=focused_path + ) + if not self.preview_stitch_running(): + self.preview_stitch_start(images_folder) + if self.stitch_automatically: + if not self.correlate_running(): + self.correlate_start(images_folder, overlap=overlap) + + ensure_free_disk_space(scan_folder) + + # Check if the image is background + if self.skip_background: + image_is_sample = background_detect.image_is_sample() + else: + image_is_sample = True + + # if more than 92% of the image is background, treat it as background and continue + if not image_is_sample: + logger.info( + f"Skipping {stage.position} as it is {round(background_detect.background_fraction(), 0)}% background." + ) + else: + # if not, it's sample. run an autofocus and use the updated height + new_pos = [ + [stage.position["x"] - dx, stage.position["y"]], + [stage.position["x"] + dx, stage.position["y"]], + [stage.position["x"], stage.position["y"] - dy], + [stage.position["x"], stage.position["y"] + dy], + ] + for pos in new_pos: + if ( + pos not in [sublist[:2] for sublist in true_path] + and pos not in path + ): + path.append(pos) + + attempts = 0 + if self.autofocus_dz > 200: + while True: + jpeg_zs, jpeg_sizes = autofocus.looping_autofocus( + dz=self.autofocus_dz, start="base" + ) + current_height = stage.position["z"] + time.sleep(0.2) + autofocus_success = autofocus.verify_focus_sharpness( + sweep_sizes=jpeg_sizes, camera=CamDep, threshold=0.92 + ) + logger.info( + f"We just tested the focus! Result was {autofocus_success}" + ) + + if autofocus_success: + # if there have been successful autofocuses in this scan, find the closest one in x-y + # test if the change in z between them exceeds a ratio (indicating a failed autofocus) + if len(focused_path) > 0: + nearest_focused_site = focused_path[ + closest(loc, focused_path) + ] + result = limit_focus_change( + nearest_focused_site[0:2], + nearest_focused_site[-1], + loc[0:2], + current_height, + 0.5, + ) + + # if there haven't been any previous autofocuses, we have to assume this one worked + else: + result = "accept" + else: + result = "reject" + + # if the autofocus worked, add the current position to the list of successful locations + if result == "accept": + loc = list(stage.position.values()) + focused_path.append(loc) + break + if attempts >= 3: + logger.warning("Could not autofocus after 3 attempts.") + break + # if the autofocus was rejected, we return to the height of the closest successful autofocus. not perfect, but better than wandering out of focus + logger.info( + "The focus has shifted further than we expect: retrying." + ) + stage.move_absolute(z=int(loc[2])) + attempts += 1 + + # Acquire the image in a thread, and continue once it's acquired (i.e. leave saving in the background) + if capture_thread: # wait for the previous capture to be saved, i.e. don't leave more than one image saving in the background + if capture_thread.is_alive(): + wait_start = time.time() + capture_thread.join() + wait_time = time.time() - wait_start + logger.info( + f"Waited {wait_time:.1f}s for the previous capture to finish saving." + ) + acquired = Event() + name = f"image_{loc[0]}_{loc[1]}.jpg" + time.sleep(0.2) + capture_thread = Thread( + target=capture_and_save, + kwargs={ + # "cam": cam, + # "logger": logger, + "acquired": acquired, + "name": name, + # "images_folder": images_folder, + # "raw_images_folder": raw_images_folder, + }, + ) + capture_thread.start() + acquired.wait() # wait until the image is acquired + # time.sleep(0.5) + positions.append(loc[:2]) + names.append(name) + + # add the current position to the list of all positions visited + true_path.append(loc) + + # if len(names) > 1: + # generate_config(images_folder, positions, names, CSM, csm_calibration_width, img_width, logger) + + temp_path = [] + + for i in path: + if distance_to_site(i, true_path[0][:2]) < max_dist: + temp_path.append(i) + else: + logger.info(f"Rejected moving to {i} as it is out of range") + path = temp_path.copy() + path = sorted( + path, + key=lambda x: ( + steps_from_centre(x, true_path[0][:2], dx, dy), + distance_to_site(loc[:2], x), + ), + ) + self.create_zip_of_scan( + logger=logger, + scan_name=scan_folder.split("scans/")[1], + download_zip=False, + ) + + except InvocationCancelledError: + logger.error("Stopping scan because it was cancelled.") + except NotEnoughFreeSpaceError as e: + logger.error( + f"Stopping scan to avoid filling up the disk: {e}", + exc_info=e, + ) + raise e + except Exception as e: + logger.error( + f"The scan stopped because of an error: {e}", + "We will attempt to stitch and archive the images acquired so far.", + exc_info=e, + ) + raise e + finally: + if capture_thread: + capture_thread.join() + try: + logger.info("Returning to starting position.") + if starting_position is not None: + stage.move_absolute(**starting_position, block_cancellation=True) + finally: + self._scan_lock.release() + self.create_zip_of_scan( + logger=logger, + scan_name=scan_folder.split("scans/")[1], + download_zip=False, + ) + logger.info("Waiting for background processes to finish...") + self.preview_stitch_wait() + self.correlate_wait() + try: + if scan_folder and self.stitch_automatically: + logger.info("Stitching final image (may take some time)...") + self.stitch_scan( + logger, os.path.basename(scan_folder), overlap=overlap + ) + except SubprocessError as e: + logger.error(f"Stitching failed: {e}", exc_info=e) + + @thing_property + def max_range(self) -> int: + """The maximum distance from the centre of the scan before we break""" + return self.thing_settings.get("max_range", 45000) + + @max_range.setter + def max_range(self, value: int) -> None: + self.thing_settings["max_range"] = value + + @thing_property + def stitch_tiff(self) -> bool: + """Whether or not to also produce a pyramidal tiff""" + return self.thing_settings.get("stitch_tiff", False) + + @stitch_tiff.setter + def stitch_tiff(self, value: bool) -> None: + self.thing_settings["stitch_tiff"] = value + + @thing_property + def skip_background(self) -> bool: + """Whether to detect and skip empty fields of view + + This uses the settings from the `background_detect` Thing. + """ + return self.thing_settings.get("skip_background", True) + + @skip_background.setter + def skip_background(self, value: bool) -> None: + self.thing_settings["skip_background"] = value + + @thing_property + def autofocus_dz(self) -> int: + """The z distance to perform an autofocus""" + return self.thing_settings.get("autofocus_dz", 1000) + + @autofocus_dz.setter + def autofocus_dz(self, value: int) -> None: + self.thing_settings["autofocus_dz"] = value + + @thing_property + def overlap(self) -> float: + """The z distance to perform an autofocus""" + return self.thing_settings.get("overlap", 0.45) + + @overlap.setter + def overlap(self, value: float) -> None: + self.thing_settings["overlap"] = value + + @thing_property + def stitch_automatically(self) -> bool: + """Should we attempt to stitch scans as we go?""" + return self.thing_settings.get("stitch_automatically", True) + + @stitch_automatically.setter + def stitch_automatically(self, value: bool) -> None: + self.thing_settings["stitch_automatically"] = value + + @thing_property + def scans(self) -> list[ScanInfo]: + """All the available scans + + Each scan has a name (which can be used to access it), along with + its modified and created times (according to the filesystem) and + the number of items in the `images` folder. Note that the number + of images reported may be confused if non-image files are present + in the `images` folder. + """ + scans: list[ScanInfo] = [] + if not os.path.isdir(self.scans_folder_path): + return scans + for f in os.listdir(self.scans_folder_path): + path = os.path.join(self.scans_folder_path, f) + if os.path.isdir(path): + images_folder = os.path.join(path, "images") + if os.path.isdir(images_folder): + number_of_images = len(os.listdir(images_folder)) + else: + number_of_images = 0 + scans.append( + ScanInfo( + name=f, + created=os.path.getctime(path), + modified=os.path.getmtime(path), + number_of_images=number_of_images, + ) + ) + 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.scans_folder_path, scan_name, file) + if not os.path.isfile(path): + raise HTTPException(404, "File not found") + return FileResponse(path) + + @fastapi_endpoint( + "delete", + "scans/{scan_name}", + responses={ + 200: {"description": "Successfully deleted scan"}, + 404: {"description": "Scan not found"}, + }, + ) + def delete_scan(self, scan_name: str) -> None: + """Delete all files from a scan. + + This endpoint allows scans to be deleted from disk. + """ + path = os.path.join(self.scans_folder_path, scan_name) + if not os.path.isdir(path): + print(f"can't find {path}") + raise HTTPException(404, "Scan not found") + shutil.rmtree(path) + + @fastapi_endpoint( + "delete", + "scans", + ) + def delete_all_scans(self) -> None: + """Delete all the scans on the microscope + + **This will irreversibly remove all smart scan data from the + microscope!** + Use with extreme caution. + """ + for scan in self.scans: + self.delete_scan(scan.name) + + def images_folder(self, scan_name: Optional[str] = None) -> str: + scan_folder = self.scan_folder_path(scan_name=scan_name) + return os.path.join(scan_folder, "images") + + @property + def latest_preview_stitch_path(self): + """The path of the latest preview stitched image""" + return os.path.join(self.images_folder(), "stitched_from_stage.jpg") + + @thing_property + def latest_preview_stitch_time(self) -> Optional[datetime]: + """The modification time of the latest preview image + + This will return `null` if there is no preview image to return. + """ + try: + fpath = self.latest_preview_stitch_path + if os.path.exists(fpath): + return os.path.getmtime(fpath) + except IOError: + return None + return None + + @fastapi_endpoint( + "get", + "latest_preview_stitch.jpg", + responses={ + 200: { + "description": "A preview-quality stitched image", + "content": {"image/jpeg": {}}, + }, + 404: {"description": "File not found"}, + }, + ) + def get_latest_preview(self) -> FileResponse: + """Retrieve the latest preview image.""" + path = self.latest_preview_stitch_path + if not os.path.isfile(path): + raise HTTPException(404, "File not found") + return FileResponse(path) + + _preview_stitch_popen = None + + def preview_stitch_start(self, images_folder: str) -> None: + """Start stitching a preview of the scan in a subprocess""" + if self.preview_stitch_running(): + raise RuntimeError("Only one subprocess is allowed at a time") + with self._preview_stitch_popen_lock: + self._preview_stitch_popen = Popen( + [self._script, "--stitching_mode", "only_stage_stitch", images_folder] + ) + + def preview_stitch_running(self) -> bool: + """Whether there is a preview stitch running in a subprocess""" + with self._preview_stitch_popen_lock: + if self._preview_stitch_popen is None: + return False + if self._preview_stitch_popen.poll() is None: + return True + return False + + def preview_stitch_wait(self): + if self.preview_stitch_running(): + with self._preview_stitch_popen_lock: + self._preview_stitch_popen.wait() + + _correlate_popen = None + + def correlate_start(self, images_folder: str, overlap: float = 0.1) -> None: + """Start stitching a preview of the scan in a subprocess""" + if self.correlate_running(): + raise RuntimeError("Only one subprocess is allowed at a time") + with self._correlate_popen_lock: + self._correlate_popen = Popen( + [ + self._script, + "--stitching_mode", + "only_correlate", + "--minimum_overlap", + f"{round(overlap * 0.9, 2)}", + images_folder, + ] + ) + + def correlate_running(self) -> bool: + """Whether there is a preview stitch running in a subprocess""" + with self._correlate_popen_lock: + if self._correlate_popen is None: + return False + if self._correlate_popen.poll() is None: + return True + return False + + def correlate_wait(self): + if self.correlate_running(): + with self._correlate_popen_lock: + self._correlate_popen.wait() + + def run_subprocess( + self, + logger: InvocationLogger, + cmd: list[str], + ) -> CompletedProcess: + """Run a subprocess and log any output""" + logger.info(f"Running command in subprocess: `{' '.join(cmd)}`") + + p = Popen(cmd, stdout=PIPE, stderr=STDOUT, bufsize=1, universal_newlines=True) + os.set_blocking(p.stdout.fileno(), False) + logger.info(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))) + while p.poll() is None: + try: + output = p.stdout.readline() + if output != "" and output is not None: + logger.info(output) + except: + pass + + for line in p.stdout: + try: + output = p.stdout.readline() + if output != "" and output is not None: + logger.info(output) + except: + pass + + logger.info("Stitching complete") + return p + + @thing_action + def stitch_scan( + self, + logger: InvocationLogger, + scan_name: Optional[str] = None, + overlap: float = 0.0, + ) -> None: + """Generate a stitched image based on stage position metadata""" + images_folder = self.images_folder(scan_name=scan_name) + + if self.stitch_tiff: + tiff_arg = "--stitch_tiff" + else: + tiff_arg = "--no-stitch_tiff" + + if overlap == 0.0: + try: + with open(os.path.join(images_folder, "scan_inputs.json")) as data_file: + data_loaded = json.load(data_file) + logger.info(data_loaded) + overlap = data_loaded["overlap"] + except: + overlap = 0.1 + self.run_subprocess( + logger, + [ + self._script, + "--stitching_mode", + "all", + f"{tiff_arg}", + "--minimum_overlap", + f"{round(overlap * 0.9, 2)}", + images_folder, + ], + ) + + @thing_action + def create_zip_of_scan( + self, + logger: InvocationLogger, + scan_name: Optional[str] = None, + download_zip=True, + ) -> ZipBlob: + """Generate a zip file that can be downloaded, with all the scan files in it.""" + images_folder = self.images_folder(scan_name=scan_name) + scan_folder = self.scan_folder_path(scan_name=scan_name) + if ( + scan_folder != os.path.dirname(images_folder) + or os.path.basename(images_folder) != "images" + ): + logger.error( + "There is a problem with filenames, the archive may be incorrect." + f"scan_folder: {scan_folder}, images_folder: {images_folder}." + ) + if not os.path.isdir(images_folder): + raise FileNotFoundError( + f"Tried to make a zip archive of {images_folder} but it does not exist." + ) + # logger.info("Creating zip archive of images (may take some time)...") + + zip_fname = f"{os.path.join(scan_folder, 'images')}.zip" + + # Create an empty zip file - we don't want to autofill it with files, + # as some of them should only be added at the end (as we can't overwrite) + # them once they change + if not os.path.isfile(zip_fname): + with zipfile.ZipFile(zip_fname, mode="w") as zip: + pass + + # 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 + folder_path = self.scan_folder_path(scan_name) + files = glob.glob(folder_path + "/**/*", recursive=True) + files = [i.split(f"{folder_path}/")[1] for i 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 + files_to_delay = [ + "TileConfiguration", + "tiling_cache", + "stitched.jp", + "stitched_from", + "stitched.om", + ] + tiff_name = "" + + with zipfile.ZipFile(zip_fname, mode="a") as zip: + for file in files: + if "stitched.jp" in file: + stitch_name = os.path.split(file)[1] + if ".ome.tiff" in file: + tiff_name = os.path.split(file)[1] + if any(banned_name in file for banned_name in files_to_delay): + # logger.info(f'we only add {file} into zip at the end of the scan') + pass + elif file in current_zip: + # logger.info(f'{file} is already in zip') + pass + elif ".zip" in file or "raw" in file: + # logger.info('Not adding the .zip to itself') + pass + else: + logger.info(f"appending {file} to zip") + zip.write(os.path.join(folder_path, file), arcname=file) + + images_folder = os.path.join(folder_path, "images") + # Promote key files to the top level of the zip only at the end of the scan (when downloading) + # and finally zip some of the final files + # TODO: if you download multiple times, you get duplicate files - is this a problem? + if download_zip: + with zipfile.ZipFile(zip_fname, mode="a") as zip: + for fname in ["stitched_from_stage.jpg", stitch_name, tiff_name]: + fpath = os.path.join(images_folder, fname) + if os.path.exists(fpath): + logger.info(f"copying {fpath} to upper level") + zip.write(fpath, arcname=fname) + for file in files: + if any(banned_name in file for banned_name in files_to_delay): + logger.info(f"we are finally adding {file} into zip") + zip.write(os.path.join(folder_path, file), arcname=file) + logger.info("about to download zip") + return ZipBlob.from_file(zip_fname) + + @thing_action + def get_files_in_zip(self, zip_path): + """List the relative paths of all files and folders in the zip folder specified""" + zip = zipfile.ZipFile(zip_path) + zip = [os.path.normpath(i) for i in zip.namelist()] + + return zip