# ruff: noqa: E722 import shutil import zipfile import threading from typing import Optional from fastapi import HTTPException from fastapi.responses import FileResponse import numpy as np import os import time from PIL import Image from pydantic import BaseModel from datetime import datetime from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, STDOUT from threading import Event 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.utilities import ErrorCapturingThread from openflexure_microscope_server.things.autofocus import AutofocusThing from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper from openflexure_microscope_server.things.background_detect import BackgroundDetectThing CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/") AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/") BackgroundDep = direct_thing_client_dependency( BackgroundDetectThing, "/background_detect/" ) 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") if movement_ratio > limit: return "reject" else: return "accept" 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 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? # TODO: fully test this with whether it works in Fiji as expected 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") 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, 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_sucessful = True scan_folder = None images_folder = None starting_position = None capture_thread = None self._scan_lock.acquire(timeout=0.1) try: 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" ) # Before anything else, check that we've got a background set # It's annoying to have to wait to find out! 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 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) # 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 wait_start = time.time() thread_was_alive = capture_thread.is_alive() # If the capture thread has thrown an exception it will be raised when join is called, # this will cause the scan to end. If we want to retry captures at a later date # this is where we will need to catch the IOError or CaptureError from the thread. capture_thread.join() time.sleep(0.2) if thread_was_alive: 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" jpeg_path = os.path.join(images_folder, name) time.sleep(0.2) # Use ErrorCapturingThread intead of Thread. This will raise errors in the calling # thread only when join() is called, allowing us to handle this appropriately. capture_thread = ErrorCapturingThread( target=self.capture_and_save, kwargs={ "acquired": acquired, "jpeg_path": jpeg_path, "cam": cam, "logger": logger, "metadata_getter": metadata_getter, }, ) capture_thread.start() acquired.wait() # wait until the image is acquired positions.append(loc[:2]) names.append(name) # add the current position to the list of all positions visited true_path.append(loc) 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: scan_sucessful = False logger.error("Stopping scan because it was cancelled.") except NotEnoughFreeSpaceError as e: scan_sucessful = False logger.error( f"Stopping scan to avoid filling up the disk: {e}", exc_info=e, ) raise e except Exception as e: scan_sucessful = False logger.error( f"The scan stopped because of an error: {e} " "Attempting to stitch and archive the images acquired so far.", exc_info=e, ) raise e finally: if capture_thread: # If the capture thread had an error we capture it here try: capture_thread.join() except Exception as e: # If the thread has already ended due to an exception we will # ignore any excppetions, but if it appeared to be succesfull # we will log an error. if scan_sucessful: logger.error( "The appears to have been successful however the final capture raised" f"the following error: {e}." "Attempting to stitch and archive images.", exc_info=e, ) 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) def capture_and_save( self, acquired: Event, jpeg_path: str, cam: CamDep, metadata_getter: GetThingStates, logger: InvocationLogger, ) -> 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. """ capture_start = time.time() image, metadata = self.capture_image(cam, metadata_getter, logger) acquired.set() acquisition_time = time.time() self.save_capture(jpeg_path, image, metadata, logger) save_time = time.time() acquisition_duration = round(acquisition_time - capture_start, 1) saving_duration = round(save_time - acquisition_time, 1) logger.debug( f"Acquired {jpeg_path} in {acquisition_duration}s then {saving_duration}s saving to disk" ) def capture_image( self, cam: CamDep, metadata_getter: GetThingStates, logger: InvocationLogger, ) -> tuple[np.ndarray, dict]: """Capture an image in memory and return it with metadata This will set the event `acquired` once the image has been acquired, so that the stage may be moved while it's saved. CaptureError raised if the capture fails for any reason returns tuple with numpy array of image data, and dict of metadata """ try: metadata = metadata_getter() image = cam.capture_array()[..., :3] except Exception as e: raise CaptureError( "An error occurred while capturing: {}".format(e), exc_info=e ) return image, metadata def save_capture( self, jpeg_path: str, image: np.ndarray, metadata: dict, logger: InvocationLogger, ) -> None: """Saving the captured image and metadata to disk logger warning (via InvocationLogger) is raised if metadata is failed to be added IOError is raised if the file cannot be saved nothing is returned on success""" try: Image.fromarray(image.astype("uint8"), "RGB").save( jpeg_path, quality=95, subsampling=0 ) try: exif_dict = piexif.load(jpeg_path) exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps( metadata ).encode("utf-8") piexif.insert(piexif.dump(exif_dict), jpeg_path) except: logger.warning(f"Failed to add metadata to {jpeg_path}") except Exception as e: raise IOError( f"An error occurred while saving {jpeg_path}: {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." ) 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): pass elif file in current_zip: pass elif ".zip" in file or "raw" in file: 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 class CaptureError(RuntimeError): """An error trying to capture from Picamera"""