1088 lines
46 KiB
Python
1088 lines
46 KiB
Python
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 copy import deepcopy
|
|
from datetime import datetime
|
|
from subprocess import CompletedProcess, Popen, PIPE, SubprocessError, run
|
|
from threading import Event, Thread
|
|
import glob
|
|
import zipfile
|
|
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 BlobOutput
|
|
from labthings_sangaboard import SangaboardThing
|
|
from labthings_picamera2.thing import StreamingPiCamera2
|
|
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
|
|
|
|
StageDep = direct_thing_client_dependency(SangaboardThing, "/stage/")
|
|
CamDep = direct_thing_client_dependency(StreamingPiCamera2, "/camera/")
|
|
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", 7)
|
|
|
|
@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",
|
|
)
|
|
|
|
class JPEGBlob(BlobOutput):
|
|
media_type = "image/jpeg"
|
|
|
|
class ZipBlob(BlobOutput):
|
|
media_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(999999):
|
|
folder_path = os.path.join(self.scans_folder_path, f"{scan_name}_{j:06}")
|
|
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(f'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
|
|
ids = []
|
|
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 = 1/lum/Cr/gr
|
|
B = 1/lum/Cb/gb
|
|
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
|
|
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()
|
|
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 = rggb2rgb(raw2rggb(raw_image)) / white_norm
|
|
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_getter()
|
|
).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.")
|
|
capture_image = False
|
|
else:
|
|
capture_image = True
|
|
# 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:
|
|
autofocus.looping_autofocus(dz=self.autofocus_dz, start = 'base')
|
|
current_height = stage.position["z"]
|
|
|
|
# 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.3,
|
|
)
|
|
|
|
# if there haven't been any previous autofocuses, we have to assume this one worked
|
|
else:
|
|
result = "accept"
|
|
|
|
# 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"
|
|
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)))
|
|
|
|
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", 70000)
|
|
|
|
@max_range.setter
|
|
def max_range(self, value: int) -> None:
|
|
self.thing_settings["max_range"] = 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)}`")
|
|
output = run(cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True)
|
|
for pipe in [output.stdout, output.stderr]:
|
|
if pipe:
|
|
logger.info(pipe)
|
|
output.check_returncode()
|
|
return output
|
|
|
|
@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 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", "--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)
|
|
|
|
logger.info(current_zip)
|
|
|
|
# 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']
|
|
|
|
with zipfile.ZipFile(zip_fname, mode="a") as zip:
|
|
for file in files:
|
|
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')
|
|
elif file in current_zip:
|
|
logger.info(f'{file} is already in zip')
|
|
elif ".zip" in file:
|
|
logger.info('Not adding the .zip to itself')
|
|
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", "stitched.jpg"]:
|
|
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)
|
|
|
|
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
|
|
|