Code format

This commit is contained in:
Joel Collins 2020-04-28 13:20:38 +01:00
parent ed8057ce04
commit 9646058c37
14 changed files with 381 additions and 201 deletions

View file

@ -1,5 +1,6 @@
#!/usr/bin/env python #!/usr/bin/env python
from gevent import monkey from gevent import monkey
monkey.patch_all() monkey.patch_all()
import time import time

View file

@ -2,6 +2,7 @@ import logging
import traceback import traceback
from contextlib import contextmanager from contextlib import contextmanager
@contextmanager @contextmanager
def handle_extension_error(extension_name): def handle_extension_error(extension_name):
"""'gracefully' log an error if an extension fails to load.""" """'gracefully' log an error if an extension fails to load."""
@ -12,6 +13,7 @@ def handle_extension_error(extension_name):
f"Exception loading builtin extension picamera_autocalibrate: \n{traceback.format_exc()}" f"Exception loading builtin extension picamera_autocalibrate: \n{traceback.format_exc()}"
) )
with handle_extension_error("autofocus"): with handle_extension_error("autofocus"):
from .autofocus import autofocus_extension_v2 from .autofocus import autofocus_extension_v2
with handle_extension_error("scan"): with handle_extension_error("scan"):

View file

@ -51,7 +51,9 @@ class JPEGSharpnessMonitor:
def start(self): def start(self):
"Start monitoring sharpness by looking at JPEG size" "Start monitoring sharpness by looking at JPEG size"
if not self.camera.stream_active: if not self.camera.stream_active:
logging.warn("Autofocus sharpness monitor was started but the camera isn't streaming. Attempting to start the stream...") logging.warn(
"Autofocus sharpness monitor was started but the camera isn't streaming. Attempting to start the stream..."
)
self.camera.start_stream_recording() self.camera.start_stream_recording()
self.background_thread = Thread(target=self._measure_jpegs) self.background_thread = Thread(target=self._measure_jpegs)
self.background_thread.start() self.background_thread.start()
@ -106,7 +108,9 @@ class JPEGSharpnessMonitor:
stop = np.argmax(jpeg_times > stage_times[1]) stop = np.argmax(jpeg_times > stage_times[1])
except ValueError as e: except ValueError as e:
if np.sum(jpeg_times > stage_times[0]) == 0: if np.sum(jpeg_times > stage_times[0]) == 0:
raise ValueError("No images were captured during the move of the stage. Perhaps the camera is not streaming images?") raise ValueError(
"No images were captured during the move of the stage. Perhaps the camera is not streaming images?"
)
else: else:
raise e raise e
if stop < 1: if stop < 1:
@ -120,7 +124,9 @@ class JPEGSharpnessMonitor:
"""Return the z position of the sharpest image on a given move""" """Return the z position of the sharpest image on a given move"""
jt, jz, js = self.move_data(index) jt, jz, js = self.move_data(index)
if len(js) == 0: if len(js) == 0:
raise ValueError("No images were captured during the move of the stage. Perhaps the camera is not streaming images?") raise ValueError(
"No images were captured during the move of the stage. Perhaps the camera is not streaming images?"
)
return jz[np.argmax(js)] return jz[np.argmax(js)]
def data_dict(self): def data_dict(self):
@ -212,7 +218,9 @@ def move_and_find_focus(microscope, dz):
def fast_autofocus(microscope, dz=2000, backlash=None): def fast_autofocus(microscope, dz=2000, backlash=None):
"""Perform a down-up-down-up autofocus""" """Perform a down-up-down-up autofocus"""
with monitor_sharpness(microscope) as m, microscope.camera.lock, microscope.stage.lock: with monitor_sharpness(
microscope
) as m, microscope.camera.lock, microscope.stage.lock:
i, z = m.focus_rel(-dz / 2) i, z = m.focus_rel(-dz / 2)
i, z = m.focus_rel(dz) i, z = m.focus_rel(dz)
fz = m.sharpest_z_on_move(i) fz = m.sharpest_z_on_move(i)
@ -262,7 +270,9 @@ def fast_up_down_up_autofocus(
might slightly hurt accuracy, but is unlikely to be a big issue. Too big might slightly hurt accuracy, but is unlikely to be a big issue. Too big
may cause you to overshoot, which is a problem. may cause you to overshoot, which is a problem.
""" """
with monitor_sharpness(microscope) as m, microscope.camera.lock, microscope.stage.lock: with monitor_sharpness(
microscope
) as m, microscope.camera.lock, microscope.stage.lock:
# Ensure the MJPEG stream has started # Ensure the MJPEG stream has started
microscope.camera.start_stream_recording() microscope.camera.start_stream_recording()
@ -372,7 +382,9 @@ class FastAutofocusAPI(View):
if microscope.has_real_stage(): if microscope.has_real_stage():
logging.debug("Running autofocus...") logging.debug("Running autofocus...")
task = taskify(fast_up_down_up_autofocus)(microscope, dz=dz, mini_backlash=backlash) task = taskify(fast_up_down_up_autofocus)(
microscope, dz=dz, mini_backlash=backlash
)
# return a handle on the autofocus task # return a handle on the autofocus task
return task return task
@ -382,12 +394,15 @@ class FastAutofocusAPI(View):
autofocus_extension_v2 = BaseExtension( autofocus_extension_v2 = BaseExtension(
"org.openflexure.autofocus", version="2.0.0", "org.openflexure.autofocus",
description="Actions to move the microscope in Z and pick the point with the sharpest image." version="2.0.0",
description="Actions to move the microscope in Z and pick the point with the sharpest image.",
) )
autofocus_extension_v2.add_method(fast_autofocus, "fast_autofocus") autofocus_extension_v2.add_method(fast_autofocus, "fast_autofocus")
autofocus_extension_v2.add_method(fast_up_down_up_autofocus, "fast_up_down_up_autofocus") autofocus_extension_v2.add_method(
fast_up_down_up_autofocus, "fast_up_down_up_autofocus"
)
autofocus_extension_v2.add_method(autofocus, "autofocus") autofocus_extension_v2.add_method(autofocus, "autofocus")
autofocus_extension_v2.add_view(MeasureSharpnessAPI, "/measure_sharpness") autofocus_extension_v2.add_view(MeasureSharpnessAPI, "/measure_sharpness")

View file

@ -7,13 +7,14 @@ Created on Tue May 26 08:08:14 2015
import numpy as np import numpy as np
class AttributeDict(dict): class AttributeDict(dict):
"""This class extends a dictionary to have a "create" method for """This class extends a dictionary to have a "create" method for
compatibility with h5py attrs objects.""" compatibility with h5py attrs objects."""
def create(self, name, data): def create(self, name, data):
self[name] = data self[name] = data
def modify(self, name, data): def modify(self, name, data):
self[name] = data self[name] = data
@ -22,7 +23,8 @@ class AttributeDict(dict):
for k in list(self.keys()): for k in list(self.keys()):
if isinstance(self[k], np.ndarray): if isinstance(self[k], np.ndarray):
self[k] = np.copy(self[k]) self[k] = np.copy(self[k])
def ensure_attribute_dict(obj, copy=False): def ensure_attribute_dict(obj, copy=False):
"""Given a mapping that may or not be an AttributeDict, return an """Given a mapping that may or not be an AttributeDict, return an
AttributeDict object that either is, or copies the data of, the input.""" AttributeDict object that either is, or copies the data of, the input."""
@ -33,15 +35,17 @@ def ensure_attribute_dict(obj, copy=False):
if copy: if copy:
out.copy_arrays() out.copy_arrays()
return out return out
def ensure_attrs(obj): def ensure_attrs(obj):
"""Return an ArrayWithAttrs version of an array-like object, may be the """Return an ArrayWithAttrs version of an array-like object, may be the
original object if it already has attrs.""" original object if it already has attrs."""
if hasattr(obj, 'attrs'): if hasattr(obj, "attrs"):
return obj #if it has attrs, do nothing return obj # if it has attrs, do nothing
else: else:
return ArrayWithAttrs(obj) #otherwise, wrap it return ArrayWithAttrs(obj) # otherwise, wrap it
class ArrayWithAttrs(np.ndarray): class ArrayWithAttrs(np.ndarray):
"""A numpy ndarray, with an AttributeDict accessible as array.attrs. """A numpy ndarray, with an AttributeDict accessible as array.attrs.
@ -50,7 +54,7 @@ class ArrayWithAttrs(np.ndarray):
a lot to the ``InfoArray`` example in `numpy` documentation on subclassing a lot to the ``InfoArray`` example in `numpy` documentation on subclassing
`numpy.ndarray`. `numpy.ndarray`.
""" """
def __new__(cls, input_array, attrs={}): def __new__(cls, input_array, attrs={}):
"""Make a new ndarray, based on an existing one, with an attrs dict. """Make a new ndarray, based on an existing one, with an attrs dict.
@ -64,26 +68,29 @@ class ArrayWithAttrs(np.ndarray):
obj.attrs = ensure_attribute_dict(attrs) obj.attrs = ensure_attribute_dict(attrs)
# return the new object # return the new object
return obj return obj
def __array_finalize__(self, obj): def __array_finalize__(self, obj):
# this is called by numpy when the object is created (__new__ may or # this is called by numpy when the object is created (__new__ may or
# may not get called) # may not get called)
if obj is None: return # if obj is None, __new__ was called - do nothing if obj is None:
return # if obj is None, __new__ was called - do nothing
# if we didn't create the object with __new__, we must add the attrs # if we didn't create the object with __new__, we must add the attrs
# dictionary. We copy this from the source object if possible (while # dictionary. We copy this from the source object if possible (while
# ensuring it's the right type) or create a new, empty one if not. # ensuring it's the right type) or create a new, empty one if not.
# NB we don't use ensure_attribute_dict because we want to make sure the # NB we don't use ensure_attribute_dict because we want to make sure the
# dict object is *copied* not merely referenced. # dict object is *copied* not merely referenced.
self.attrs = ensure_attribute_dict(getattr(obj, 'attrs', {}), copy=True) self.attrs = ensure_attribute_dict(getattr(obj, "attrs", {}), copy=True)
def attribute_bundler(attrs): def attribute_bundler(attrs):
"""Return a function that bundles the supplied attributes with an array.""" """Return a function that bundles the supplied attributes with an array."""
def bundle_attrs(array): def bundle_attrs(array):
return ArrayWithAttrs(array, attrs=attrs) return ArrayWithAttrs(array, attrs=attrs)
class DummyHDF5Group(dict): class DummyHDF5Group(dict):
def __init__(self,dictionary, attrs ={}, name="DummyHDF5Group"): def __init__(self, dictionary, attrs={}, name="DummyHDF5Group"):
super(DummyHDF5Group, self).__init__() super(DummyHDF5Group, self).__init__()
self.attrs = attrs self.attrs = attrs
for key in dictionary: for key in dictionary:
@ -92,4 +99,4 @@ class DummyHDF5Group(dict):
self.basename = name self.basename = name
file = None file = None
parent = None parent = None

View file

@ -12,9 +12,11 @@ from numpy.linalg import norm
from .camera_stage_tracker import Tracker, move_until_motion_detected from .camera_stage_tracker import Tracker, move_until_motion_detected
import logging import logging
def displacements(positions): def displacements(positions):
"""Calculate the absolute distance of each point from the first point.""" """Calculate the absolute distance of each point from the first point."""
return norm(positions - positions[0,:][np.newaxis,:], axis=1) return norm(positions - positions[0, :][np.newaxis, :], axis=1)
def direction_from_points(points): def direction_from_points(points):
"""Given an Nx2 array of points, figure out the principal component. """Given an Nx2 array of points, figure out the principal component.
@ -25,7 +27,8 @@ def direction_from_points(points):
points = points.astype(np.float) points = points.astype(np.float)
points -= np.mean(points, axis=0)[np.newaxis, :] points -= np.mean(points, axis=0)[np.newaxis, :]
eigenvalues, eigenvectors = np.linalg.eig(np.cov(points.T)) eigenvalues, eigenvectors = np.linalg.eig(np.cov(points.T))
return eigenvectors[:,np.argmax(eigenvalues)] return eigenvectors[:, np.argmax(eigenvalues)]
def apply_backlash(x, backlash=0, start_unwound=True): def apply_backlash(x, backlash=0, start_unwound=True):
"""Apply a basic model of backlash to a set of coordinates. """Apply a basic model of backlash to a set of coordinates.
@ -42,14 +45,15 @@ def apply_backlash(x, backlash=0, start_unwound=True):
y[0] = x[0] + initial_direction * backlash y[0] = x[0] + initial_direction * backlash
else: else:
y[0] = x[0] y[0] = x[0]
for i in range(1,len(x)): for i in range(1, len(x)):
d = x[i] - y[i-1] d = x[i] - y[i - 1]
if np.abs(d) >= backlash: if np.abs(d) >= backlash:
y[i] = x[i] - np.sign(d) * backlash y[i] = x[i] - np.sign(d) * backlash
else: else:
y[i] = y[i-1] y[i] = y[i - 1]
return y return y
def fit_backlash(moves): def fit_backlash(moves):
"""Given a set of linear moves forwards and back, estimate backlash. """Given a set of linear moves forwards and back, estimate backlash.
@ -99,7 +103,7 @@ def fit_backlash(moves):
residuals = yfit - (xfit_blsh * m + c) residuals = yfit - (xfit_blsh * m + c)
return m, c, np.std(residuals, ddof=3) return m, c, np.std(residuals, ddof=3)
max_backlash = (np.max(xfit) - np.min(xfit))/3 max_backlash = (np.max(xfit) - np.min(xfit)) / 3
backlash_values = [] backlash_values = []
residual_values = [] residual_values = []
backlash = 0 backlash = 0
@ -107,18 +111,18 @@ def fit_backlash(moves):
m, c, residual = fit_motion(xfit, yfit, backlash) m, c, residual = fit_motion(xfit, yfit, backlash)
residual_values.append(residual) residual_values.append(residual)
backlash_values.append(backlash) backlash_values.append(backlash)
backlash += max(1, backlash/3) backlash += max(1, backlash / 3)
backlash = backlash_values[np.argmin(residual_values)] backlash = backlash_values[np.argmin(residual_values)]
m, c, residual = fit_motion(xfit, yfit, backlash) m, c, residual = fit_motion(xfit, yfit, backlash)
fractional_error = residual/norm(np.diff(yfit)) fractional_error = residual / norm(np.diff(yfit))
if fractional_error > 0.1: if fractional_error > 0.1:
raise ValueError("The fit didn't look successful") raise ValueError("The fit didn't look successful")
return { return {
"backlash": backlash, "backlash": backlash,
"pixels_per_step": m, "pixels_per_step": m,
"fractional_error": fractional_error, "fractional_error": fractional_error,
"stage_direction": stage_direction, "stage_direction": stage_direction,
"image_direction": image_direction, "image_direction": image_direction,
@ -126,36 +130,42 @@ def fit_backlash(moves):
} }
def calibrate_backlash_1d(tracker, move, direction=np.array([1,0,0])): def calibrate_backlash_1d(tracker, move, direction=np.array([1, 0, 0])):
"""Figure out reasonable step sizes for calibration, and estimate the backlash.""" """Figure out reasonable step sizes for calibration, and estimate the backlash."""
try: # Ensure that the tracker has a template set try: # Ensure that the tracker has a template set
_ = tracker.template _ = tracker.template
except: except:
tracker.acquire_template() tracker.acquire_template()
assert tracker.stage_positions.shape[0] == 1 assert tracker.stage_positions.shape[0] == 1
original_stage_pos = tracker.stage_positions[-1,:] original_stage_pos = tracker.stage_positions[-1, :]
direction = direction / np.sum(direction**2)**0.5 # ensure "direction" is normalised direction = (
direction / np.sum(direction ** 2) ** 0.5
) # ensure "direction" is normalised
logging.info("Moving the stage until we see motion...") logging.info("Moving the stage until we see motion...")
# Move the stage until we can see a significant amount of motion # Move the stage until we can see a significant amount of motion
i, m = move_until_motion_detected( i, m = move_until_motion_detected(
tracker, move, direction, threshold=tracker.max_safe_displacement * 0.2) tracker, move, direction, threshold=tracker.max_safe_displacement * 0.2
)
logging.info("Moving the stage to the edge of the field of view...") logging.info("Moving the stage to the edge of the field of view...")
i, m = move_until_motion_detected( i, m = move_until_motion_detected(
tracker, move, direction, tracker,
move,
direction,
threshold=tracker.max_safe_displacement * 0.7, threshold=tracker.max_safe_displacement * 0.7,
multipliers=m/2.0 * np.arange(20), multipliers=m / 2.0 * np.arange(20),
detect_cumulative_motion=True) detect_cumulative_motion=True,
)
exponential_moves = tracker.history exponential_moves = tracker.history
# Include this final step, and make a rough estimate of the scaling from stage to image # Include this final step, and make a rough estimate of the scaling from stage to image
stage_pos, image_pos = tracker.history stage_pos, image_pos = tracker.history
stage_step = stage_pos[-1, :] - stage_pos[-1 - i, :] stage_step = stage_pos[-1, :] - stage_pos[-1 - i, :]
image_step = image_pos[-1, :] - image_pos[-1 - i, :] image_step = image_pos[-1, :] - image_pos[-1 - i, :]
steps_per_pixel = norm(stage_step)/norm(image_step) steps_per_pixel = norm(stage_step) / norm(image_step)
# Calculate a step that moves roughly 0.2 times the max. displacement (i.e. 0.1 times the FoV) # Calculate a step that moves roughly 0.2 times the max. displacement (i.e. 0.1 times the FoV)
sensible_step = direction * tracker.max_safe_displacement * 0.2 * steps_per_pixel sensible_step = direction * tracker.max_safe_displacement * 0.2 * steps_per_pixel
tracker.reset_history() tracker.reset_history()
@ -167,27 +177,35 @@ def calibrate_backlash_1d(tracker, move, direction=np.array([1,0,0])):
starting_stage_pos, starting_camera_pos = tracker.append_point() starting_stage_pos, starting_camera_pos = tracker.append_point()
for i in range(15): for i in range(15):
move(starting_stage_pos - sensible_step * (i + 1)) move(starting_stage_pos - sensible_step * (i + 1))
#print(".", end="") # print(".", end="")
stage_pos, image_pos = tracker.append_point() stage_pos, image_pos = tracker.append_point()
if (i > 3 and tracker.moving_away_from_centre if (
and norm(image_pos) > 0.65 * tracker.max_safe_displacement): i > 3
break # Stop once we have moved far enough and tracker.moving_away_from_centre
and norm(image_pos) > 0.65 * tracker.max_safe_displacement
):
break # Stop once we have moved far enough
logging.info("Moving the stage forwards to measure backlash (2/2)") logging.info("Moving the stage forwards to measure backlash (2/2)")
# Move forwards again, in 10 steps # Move forwards again, in 10 steps
starting_stage_pos, starting_camera_pos = tracker.append_point() starting_stage_pos, starting_camera_pos = tracker.append_point()
for i in range(15): for i in range(15):
move(starting_stage_pos + sensible_step * (i + 1)) move(starting_stage_pos + sensible_step * (i + 1))
#print(".", end="") # print(".", end="")
stage_pos, image_pos = tracker.append_point() stage_pos, image_pos = tracker.append_point()
if (i > 3 and tracker.moving_away_from_centre if (
and norm(image_pos) > 0.65 * tracker.max_safe_displacement): i > 3
break # Stop once we have moved far enough and tracker.moving_away_from_centre
and norm(image_pos) > 0.65 * tracker.max_safe_displacement
):
break # Stop once we have moved far enough
linear_moves = tracker.history linear_moves = tracker.history
try: try:
res = fit_backlash(linear_moves) res = fit_backlash(linear_moves)
backlash_correction = sensible_step / norm(sensible_step) * res["backlash"] * 1.5 backlash_correction = (
sensible_step / norm(sensible_step) * res["backlash"] * 1.5
)
# Finally, move back to the starting position, doing backlash-corrected moves. # Finally, move back to the starting position, doing backlash-corrected moves.
logging.info("Moving back to the start, correcting for backlash...") logging.info("Moving back to the start, correcting for backlash...")
@ -200,32 +218,39 @@ def calibrate_backlash_1d(tracker, move, direction=np.array([1,0,0])):
backlash_corrected_moves = tracker.history backlash_corrected_moves = tracker.history
move(original_stage_pos - backlash_correction) move(original_stage_pos - backlash_correction)
except ValueError: except ValueError:
return {"exponential_moves": exponential_moves, "linear_moves": linear_moves,} return {"exponential_moves": exponential_moves, "linear_moves": linear_moves}
finally: finally:
# Reset position # Reset position
move(original_stage_pos) move(original_stage_pos)
logging.info(f"Estimated backlash {res['backlash']:.0f} steps") logging.info(f"Estimated backlash {res['backlash']:.0f} steps")
logging.info(f"Stage-to-image ratio {np.abs(res['pixels_per_step']):.3f} pixels/step") logging.info(
logging.info(f"Residuals were about {res['fractional_error']:.2f} times the step size") f"Stage-to-image ratio {np.abs(res['pixels_per_step']):.3f} pixels/step"
)
logging.info(
f"Residuals were about {res['fractional_error']:.2f} times the step size"
)
res.update({ res.update(
"exponential_moves": exponential_moves, {
"linear_moves": linear_moves, "exponential_moves": exponential_moves,
"backlash_corrected_moves": backlash_corrected_moves "linear_moves": linear_moves,
}) "backlash_corrected_moves": backlash_corrected_moves,
}
)
return res return res
def plot_1d_backlash_calibration(results): def plot_1d_backlash_calibration(results):
"""Plot the results of a calibration run""" """Plot the results of a calibration run"""
from matplotlib import pyplot as plt from matplotlib import pyplot as plt
f, ax = plt.subplots(1,2)
f, ax = plt.subplots(1, 2)
for k in ["exponential", "linear", "backlash_corrected"]: for k in ["exponential", "linear", "backlash_corrected"]:
moves = results[k+"_moves"] moves = results[k + "_moves"]
if moves is not None: if moves is not None:
ax[0].plot(moves[1][:,0], moves[1][:,1], 'o-') ax[0].plot(moves[1][:, 0], moves[1][:, 1], "o-")
ax[0].set_aspect(1, adjustable="datalim") ax[0].set_aspect(1, adjustable="datalim")
image_direction = results["image_direction"] image_direction = results["image_direction"]
@ -237,19 +262,20 @@ def plot_1d_backlash_calibration(results):
image_1d = np.sum(image_pos * image_direction[np.newaxis, :], axis=1) image_1d = np.sum(image_pos * image_direction[np.newaxis, :], axis=1)
return stage_1d, image_1d return stage_1d, image_1d
ax[1].plot(*convert_moves(results["exponential_moves"]), 'o-') ax[1].plot(*convert_moves(results["exponential_moves"]), "o-")
stage_pos, image_pos = convert_moves(results["linear_moves"]) stage_pos, image_pos = convert_moves(results["linear_moves"])
model = apply_backlash(stage_pos, results["backlash"]) model = apply_backlash(stage_pos, results["backlash"])
model *= results["pixels_per_step"] model *= results["pixels_per_step"]
model += np.mean(image_pos) - np.mean(model) model += np.mean(image_pos) - np.mean(model)
ax[1].plot(stage_pos, model, '-') ax[1].plot(stage_pos, model, "-")
ax[1].plot(stage_pos, image_pos, 'o') ax[1].plot(stage_pos, image_pos, "o")
if results["backlash_corrected_moves"] is not None: if results["backlash_corrected_moves"] is not None:
ax[1].plot(*convert_moves(results["backlash_corrected_moves"]), '+') ax[1].plot(*convert_moves(results["backlash_corrected_moves"]), "+")
return f, ax return f, ax
def image_to_stage_displacement_from_1d(calibrations): def image_to_stage_displacement_from_1d(calibrations):
"""Combine X and Y calibrations """Combine X and Y calibrations
@ -271,9 +297,11 @@ def image_to_stage_displacement_from_1d(calibrations):
c_blash = np.abs(cal["backlash"] * cal["stage_direction"]) c_blash = np.abs(cal["backlash"] * cal["stage_direction"])
backlash[backlash < c_blash] = c_blash[backlash < c_blash] backlash[backlash < c_blash] = c_blash[backlash < c_blash]
A, res, rank, s = np.linalg.lstsq(image_vectors, stage_vectors) # we solve image*A = stage A, res, rank, s = np.linalg.lstsq(
image_vectors, stage_vectors
) # we solve image*A = stage
return { return {
"image_to_stage_displacement": A, "image_to_stage_displacement": A,
"backlash_vector": backlash, "backlash_vector": backlash,
"backlash": np.max(backlash), "backlash": np.max(backlash),
} }

View file

@ -12,10 +12,11 @@ from numpy.linalg import norm
import cv2 import cv2
from scipy import ndimage from scipy import ndimage
def central_half(image): def central_half(image):
"""Return the central 50% (in X and Y) of an image""" """Return the central 50% (in X and Y) of an image"""
w, h = image.shape[:2] w, h = image.shape[:2]
return image[int(w/4):int(3*w/4),int(h/4):int(3*h/4), ...] return image[int(w / 4) : int(3 * w / 4), int(h / 4) : int(3 * h / 4), ...]
def datum_pixel(image): def datum_pixel(image):
@ -23,7 +24,8 @@ def datum_pixel(image):
try: try:
return np.array(image.datum_pixel) return np.array(image.datum_pixel)
except: except:
return (np.array(image.shape[:2]) - 1) / 2. return (np.array(image.shape[:2]) - 1) / 2.0
def locate_feature_in_image(image, feature, margin=0, restrict=False): def locate_feature_in_image(image, feature, margin=0, restrict=False):
"""Find the given feature (small image) and return the position of its datum (or centre) in the image's pixels. """Find the given feature (small image) and return the position of its datum (or centre) in the image's pixels.
@ -47,31 +49,51 @@ def locate_feature_in_image(image, feature, margin=0, restrict=False):
image to yield the position in the sample of the feature you're looking for. image to yield the position in the sample of the feature you're looking for.
""" """
# The line below is superfluous if we keep the datum-aware code below it. # The line below is superfluous if we keep the datum-aware code below it.
assert image.shape[0] > feature.shape[0] and image.shape[1] > feature.shape[1], "Image must be larger than feature!" assert (
image.shape[0] > feature.shape[0] and image.shape[1] > feature.shape[1]
), "Image must be larger than feature!"
# Check that there's enough space around the feature image # Check that there's enough space around the feature image
lower_margin = datum_pixel(image) - datum_pixel(feature) lower_margin = datum_pixel(image) - datum_pixel(feature)
upper_margin = (image.shape[:2] - datum_pixel(image)) - (feature.shape[:2] - datum_pixel(feature)) upper_margin = (image.shape[:2] - datum_pixel(image)) - (
assert np.all(np.array([lower_margin, upper_margin]) >= margin), "The feature image is too large." feature.shape[:2] - datum_pixel(feature)
#TODO: sensible auto-crop of the template if it's too large? )
image_shift = np.array((0,0)) assert np.all(
np.array([lower_margin, upper_margin]) >= margin
), "The feature image is too large."
# TODO: sensible auto-crop of the template if it's too large?
image_shift = np.array((0, 0))
if restrict: if restrict:
# if requested, crop the larger image so that our search area is (2*margin + 1) square. # if requested, crop the larger image so that our search area is (2*margin + 1) square.
image_shift = np.array(lower_margin - margin,dtype = int) image_shift = np.array(lower_margin - margin, dtype=int)
image = image[image_shift[0]:image_shift[0] + feature.shape[0] + 2 * margin + 1, image = image[
image_shift[1]:image_shift[1] + feature.shape[1] + 2 * margin + 1, ...] image_shift[0] : image_shift[0] + feature.shape[0] + 2 * margin + 1,
image_shift[1] : image_shift[1] + feature.shape[1] + 2 * margin + 1,
...,
]
corr = cv2.matchTemplate(image, feature, corr = cv2.matchTemplate(
cv2.TM_SQDIFF_NORMED) # correlate them: NB the match position is the MINIMUM image, feature, cv2.TM_SQDIFF_NORMED
corr = -corr # invert the image so we can find a peak ) # correlate them: NB the match position is the MINIMUM
corr += (corr.max() - corr.min()) * 0.1 - corr.max() # background-subtract 90% of maximum corr = -corr # invert the image so we can find a peak
corr += (
corr.max() - corr.min()
) * 0.1 - corr.max() # background-subtract 90% of maximum
corr = cv2.threshold(corr, 0, 0, cv2.THRESH_TOZERO)[ corr = cv2.threshold(corr, 0, 0, cv2.THRESH_TOZERO)[
1] # zero out any negative pixels - but there should always be > 0 nonzero pixels 1
assert np.sum(corr) > 0, "Error: the correlation image doesn't have any nonzero pixels." ] # zero out any negative pixels - but there should always be > 0 nonzero pixels
peak = ndimage.measurements.center_of_mass(corr) # take the centroid (NB this is of grayscale values, not binary) assert (
pos = np.array(peak) + image_shift + datum_pixel(feature) # return the position of the feature's datum point. np.sum(corr) > 0
), "Error: the correlation image doesn't have any nonzero pixels."
peak = ndimage.measurements.center_of_mass(
corr
) # take the centroid (NB this is of grayscale values, not binary)
pos = (
np.array(peak) + image_shift + datum_pixel(feature)
) # return the position of the feature's datum point.
return pos return pos
class Tracker():
class Tracker:
def __init__(self, grab_image, get_position, settle=None): def __init__(self, grab_image, get_position, settle=None):
"""A class to manage moving the stage and following motion in the image """A class to manage moving the stage and following motion in the image
@ -98,11 +120,11 @@ class Tracker():
self.margin = np.array([0, 0]) self.margin = np.array([0, 0])
self._template_position = np.array([0.0, 0.0]) self._template_position = np.array([0.0, 0.0])
self.image_shape = None self.image_shape = None
def get_position(self): def get_position(self):
"""Get the position of the stage""" """Get the position of the stage"""
return np.array(self._get_position()) return np.array(self._get_position())
def settle(self): def settle(self):
"""Wait a short time and discard an image so the stage is no longer wobbling.""" """Wait a short time and discard an image so the stage is no longer wobbling."""
if self._settle is not None: if self._settle is not None:
@ -110,7 +132,7 @@ class Tracker():
else: else:
time.sleep(0.3) time.sleep(0.3)
self._grab_image() self._grab_image()
@property @property
def template(self): def template(self):
"""The template image (should be a numpy array)""" """The template image (should be a numpy array)"""
@ -118,12 +140,14 @@ class Tracker():
raise ValueError("Attempt to use the tracker before setting the template") raise ValueError("Attempt to use the tracker before setting the template")
else: else:
return self._template return self._template
@template.setter @template.setter
def template(self, new_value): def template(self, new_value):
self._template = new_value self._template = new_value
def acquire_template(self, settle=True, reset_history=True, relative_positions=True): def acquire_template(
self, settle=True, reset_history=True, relative_positions=True
):
"""Take a new image, and use it as the template. NB this records the initial point. """Take a new image, and use it as the template. NB this records the initial point.
We will wait for the stage to settle, then acquire a new image to use as the template. We will wait for the stage to settle, then acquire a new image to use as the template.
@ -151,26 +175,32 @@ class Tracker():
self.margin = np.array(image.shape)[:2] - np.array(self.template.shape)[:2] self.margin = np.array(image.shape)[:2] - np.array(self.template.shape)[:2]
if reset_history: if reset_history:
self.reset_history() self.reset_history()
self._template_position = np.array([0., 0.]) self._template_position = np.array([0.0, 0.0])
if relative_positions: if relative_positions:
self._template_position = self.track_image(image) # Position should be zero initially self._template_position = self.track_image(
image
) # Position should be zero initially
self.append_point(settle=False) self.append_point(settle=False)
@property @property
def max_displacement(self): def max_displacement(self):
"""The highest position values that can be tracked""" """The highest position values that can be tracked"""
return self.margin // 2 # TODO: be cleverer about non-trivial values of template_position return (
self.margin // 2
) # TODO: be cleverer about non-trivial values of template_position
@property @property
def min_displacement(self): def min_displacement(self):
"""The lowest position values that can be tracked""" """The lowest position values that can be tracked"""
return -self.max_displacement # TODO: be cleverer about non-trivial template_position values return (
-self.max_displacement
) # TODO: be cleverer about non-trivial template_position values
@property @property
def max_safe_displacement(self): def max_safe_displacement(self):
"""The biggest displacement we can safely attempt to track without knowing direction.""" """The biggest displacement we can safely attempt to track without knowing direction."""
return np.min(np.concatenate([self.max_displacement, -self.min_displacement])) return np.min(np.concatenate([self.max_displacement, -self.min_displacement]))
def track_image(self, image): def track_image(self, image):
"""Find the position of the image relative to the template """Find the position of the image relative to the template
@ -182,8 +212,8 @@ class Tracker():
a minus sign in front of `locate_feature_in_image` in the source a minus sign in front of `locate_feature_in_image` in the source
code. code.
""" """
return - locate_feature_in_image(image, self.template) - self._template_position return -locate_feature_in_image(image, self.template) - self._template_position
def append_point(self, settle=True, image=None): def append_point(self, settle=True, image=None):
"""Find the current position using both stage and image, and append it""" """Find the current position using both stage and image, and append it"""
if settle: if settle:
@ -195,22 +225,22 @@ class Tracker():
self._image_positions.append(image_pos) self._image_positions.append(image_pos)
self._stage_positions.append(stage_pos) self._stage_positions.append(stage_pos)
return stage_pos, image_pos return stage_pos, image_pos
@property @property
def stage_positions(self): def stage_positions(self):
"""An array of positions we have moved the stage to""" """An array of positions we have moved the stage to"""
return np.array(self._stage_positions) return np.array(self._stage_positions)
@property @property
def image_positions(self): def image_positions(self):
"""An array of positions we have moved the stage to""" """An array of positions we have moved the stage to"""
return np.array(self._image_positions) return np.array(self._image_positions)
@property @property
def history(self): def history(self):
"""Return arrays of stage, image positions""" """Return arrays of stage, image positions"""
return self.stage_positions, self.image_positions return self.stage_positions, self.image_positions
def reset_history(self, leave_first_point=False): def reset_history(self, leave_first_point=False):
"""Reset the positions and displacements recorded""" """Reset the positions and displacements recorded"""
if leave_first_point: if leave_first_point:
@ -232,10 +262,17 @@ class Tracker():
if len(self.image_positions) < 2: if len(self.image_positions) < 2:
return None return None
else: else:
return norm(self.image_positions[-1,:]) > norm(self.image_positions[-2]) return norm(self.image_positions[-1, :]) > norm(self.image_positions[-2])
def move_until_motion_detected(tracker, move, displacement, threshold=10, multipliers=2**np.arange(16), detect_cumulative_motion=False): def move_until_motion_detected(
tracker,
move,
displacement,
threshold=10,
multipliers=2 ** np.arange(16),
detect_cumulative_motion=False,
):
"""Move the stage until we can detect motion in the camera. """Move the stage until we can detect motion in the camera.
We move the stage in the direction given by ``displacement`` until the We move the stage in the direction given by ``displacement`` until the
@ -259,14 +296,21 @@ def move_until_motion_detected(tracker, move, displacement, threshold=10, multip
`displacement * m`. `displacement * m`.
""" """
displacement = np.array(displacement) displacement = np.array(displacement)
starting_image_position = tracker.image_positions[0 if detect_cumulative_motion else -1, :] starting_image_position = tracker.image_positions[
0 if detect_cumulative_motion else -1, :
]
starting_stage_position = tracker.stage_positions[-1, :] starting_stage_position = tracker.stage_positions[-1, :]
for i, m in enumerate(multipliers): for i, m in enumerate(multipliers):
move(starting_stage_position + displacement * m) move(starting_stage_position + displacement * m)
tracker.append_point() tracker.append_point()
if norm(tracker.image_positions[-1, :] - starting_image_position) >= threshold: if norm(tracker.image_positions[-1, :] - starting_image_position) >= threshold:
return i + 1, m return i + 1, m
raise Exception("Moved the stage by {} but saw no motion.".format(multipliers[-1] * displacement)) raise Exception(
"Moved the stage by {} but saw no motion.".format(
multipliers[-1] * displacement
)
)
def concatenate_tracker_histories(histories): def concatenate_tracker_histories(histories):
"""Combine a number of separate tracker history entries into one """Combine a number of separate tracker history entries into one
@ -286,4 +330,3 @@ def concatenate_tracker_histories(histories):
""" """
components = zip(*histories) components = zip(*histories)
return tuple(np.concatenate(c, axis=1) for c in components) return tuple(np.concatenate(c, axis=1) for c in components)

View file

@ -6,7 +6,12 @@ This file contains the HTTP API for camera/stage calibration.
from labthings.server.view import View from labthings.server.view import View
from labthings.server.find import find_component from labthings.server.find import find_component
from labthings.server.extensions import BaseExtension from labthings.server.extensions import BaseExtension
from labthings.server.decorators import marshal_task, ThingAction, use_args, ThingProperty from labthings.server.decorators import (
marshal_task,
ThingAction,
use_args,
ThingProperty,
)
from labthings.server import fields from labthings.server import fields
from labthings.core.tasks import taskify from labthings.core.tasks import taskify
@ -23,7 +28,10 @@ import io
import os import os
import json import json
from .camera_stage_calibration_1d import calibrate_backlash_1d, image_to_stage_displacement_from_1d from .camera_stage_calibration_1d import (
calibrate_backlash_1d,
image_to_stage_displacement_from_1d,
)
from .camera_stage_tracker import Tracker from .camera_stage_tracker import Tracker
from openflexure_microscope.utilities import axes_to_array from openflexure_microscope.utilities import axes_to_array
@ -33,15 +41,15 @@ from openflexure_microscope.config import JSONEncoder
CSM_DATAFILE_NAME = "csm_calibration.json" CSM_DATAFILE_NAME = "csm_calibration.json"
CSM_DATAFILE_PATH = data_file_path(CSM_DATAFILE_NAME) CSM_DATAFILE_PATH = data_file_path(CSM_DATAFILE_NAME)
class CSMExtension(BaseExtension): class CSMExtension(BaseExtension):
""" """
Use the camera as an encoder, so we can relate camera and stage coordinates Use the camera as an encoder, so we can relate camera and stage coordinates
""" """
def __init__(self): def __init__(self):
BaseExtension.__init__( BaseExtension.__init__(
self, self, "org.openflexure.camera_stage_mapping", version="0.0.1"
"org.openflexure.camera_stage_mapping",
version="0.0.1",
) )
_microscope = None _microscope = None
@ -52,10 +60,10 @@ class CSMExtension(BaseExtension):
if self._microscope is None: if self._microscope is None:
self._microscope = find_component("org.openflexure.microscope") self._microscope = find_component("org.openflexure.microscope")
return self._microscope return self._microscope
def update_settings(self, settings): def update_settings(self, settings):
"""Update the stored extension settings dictionary""" """Update the stored extension settings dictionary"""
keys = ["extensions",self.name] keys = ["extensions", self.name]
dictionary = create_from_path(keys) dictionary = create_from_path(keys)
set_by_path(dictionary, keys, settings) set_by_path(dictionary, keys, settings)
logging.info(f"Updating settings with {dictionary}") logging.info(f"Updating settings with {dictionary}")
@ -64,20 +72,20 @@ class CSMExtension(BaseExtension):
def get_settings(self): def get_settings(self):
"""Retrieve the settings for this extension""" """Retrieve the settings for this extension"""
keys = ["extensions",self.name] keys = ["extensions", self.name]
return get_by_path(self.microscope.read_settings(), keys) return get_by_path(self.microscope.read_settings(), keys)
def camera_stage_functions(self): def camera_stage_functions(self):
"""Return functions that allow us to interface with the microscope""" """Return functions that allow us to interface with the microscope"""
self.microscope.camera.start_worker() # ensure the worker thread is running, so there is an MJPEG stream self.microscope.camera.start_worker() # ensure the worker thread is running, so there is an MJPEG stream
def grab_image(): def grab_image():
jpeg = self.microscope.camera.get_frame() jpeg = self.microscope.camera.get_frame()
return np.array(PIL.Image.open(io.BytesIO(jpeg))) return np.array(PIL.Image.open(io.BytesIO(jpeg)))
def get_position(): def get_position():
return self.microscope.stage.position return self.microscope.stage.position
move = self.microscope.stage.move_abs move = self.microscope.stage.move_abs
return grab_image, get_position, move return grab_image, get_position, move
@ -96,10 +104,10 @@ class CSMExtension(BaseExtension):
def calibrate_xy(self): def calibrate_xy(self):
"""Move the microscope's stage in X and Y, to calibrate its relationship to the camera""" """Move the microscope's stage in X and Y, to calibrate its relationship to the camera"""
logging.info("Calibrating X axis:") logging.info("Calibrating X axis:")
cal_x = self.calibrate_1d(np.array([1,0,0])) cal_x = self.calibrate_1d(np.array([1, 0, 0]))
logging.info("Calibrating Y axis:") logging.info("Calibrating Y axis:")
cal_y = self.calibrate_1d(np.array([0,1,0])) cal_y = self.calibrate_1d(np.array([0, 1, 0]))
# Combine X and Y calibrations to make a 2D calibration # Combine X and Y calibrations to make a 2D calibration
cal_xy = image_to_stage_displacement_from_1d([cal_x, cal_y]) cal_xy = image_to_stage_displacement_from_1d([cal_x, cal_y])
self.update_settings(cal_xy) self.update_settings(cal_xy)
@ -110,7 +118,7 @@ class CSMExtension(BaseExtension):
"linear_calibration_y": cal_y, "linear_calibration_y": cal_y,
} }
with open(CSM_DATAFILE_PATH, 'w') as f: with open(CSM_DATAFILE_PATH, "w") as f:
json.dump(data, f, cls=JSONEncoder) json.dump(data, f, cls=JSONEncoder)
return data return data
@ -130,27 +138,28 @@ class CSMExtension(BaseExtension):
self.microscope.stage.move_rel([relative_move[0], relative_move[1], 0]) self.microscope.stage.move_rel([relative_move[0], relative_move[1], 0])
csm_extension = CSMExtension() csm_extension = CSMExtension()
@ThingAction @ThingAction
class Calibrate1DView(View): class Calibrate1DView(View):
@use_args({ @use_args(
"direction": fields.List(fields.Float(), required=True, example=[1,0,0]) {"direction": fields.List(fields.Float(), required=True, example=[1, 0, 0])}
}) )
@marshal_task @marshal_task
def post(self, args): def post(self, args):
"""Calibrate one axis of the microscope stage against the camera.""" """Calibrate one axis of the microscope stage against the camera."""
direction = np.array(args.get("direction")) direction = np.array(args.get("direction"))
task = taskify(csm_extension.calibrate_1d)(direction) task = taskify(csm_extension.calibrate_1d)(direction)
return task return task
csm_extension.add_view(Calibrate1DView, "/calibrate_1d") csm_extension.add_view(Calibrate1DView, "/calibrate_1d")
@ThingAction @ThingAction
class CalibrateXYView(View): class CalibrateXYView(View):
@marshal_task @marshal_task
@ -160,24 +169,39 @@ class CalibrateXYView(View):
return task return task
csm_extension.add_view(CalibrateXYView, "/calibrate_xy") csm_extension.add_view(CalibrateXYView, "/calibrate_xy")
@ThingAction @ThingAction
class MoveInImageCoordinatesView(View): class MoveInImageCoordinatesView(View):
@use_args({ @use_args(
"x": fields.Float(description="The number of pixels to move in X", required=True, example=100), {
"y": fields.Float(description="The number of pixels to move in Y", required=True, example=100), "x": fields.Float(
}) description="The number of pixels to move in X",
required=True,
example=100,
),
"y": fields.Float(
description="The number of pixels to move in Y",
required=True,
example=100,
),
}
)
def post(self, args): def post(self, args):
logging.debug("moving in pixels") logging.debug("moving in pixels")
"""Move the microscope stage, such that we move by a given number of pixels on the camera""" """Move the microscope stage, such that we move by a given number of pixels on the camera"""
csm_extension.move_in_image_coordinates(np.array([args.get("x"), args.get("y")])) csm_extension.move_in_image_coordinates(
np.array([args.get("x"), args.get("y")])
)
return csm_extension.microscope.state["stage"]["position"] return csm_extension.microscope.state["stage"]["position"]
csm_extension.add_view(MoveInImageCoordinatesView, "/move_in_image_coordinates") csm_extension.add_view(MoveInImageCoordinatesView, "/move_in_image_coordinates")
@ThingProperty @ThingProperty
class GetCalibrationFile(View): class GetCalibrationFile(View):
def get(self): def get(self):
@ -186,9 +210,10 @@ class GetCalibrationFile(View):
datafile_path = CSM_DATAFILE_PATH datafile_path = CSM_DATAFILE_PATH
if os.path.isfile(datafile_path): if os.path.isfile(datafile_path):
with open(datafile_path, 'rb') as f: with open(datafile_path, "rb") as f:
return json.load(f) return json.load(f)
else: else:
return {} return {}
csm_extension.add_view(GetCalibrationFile, "/get_calibration")
csm_extension.add_view(GetCalibrationFile, "/get_calibration")

View file

@ -38,9 +38,11 @@ from past.utils import old_div
import numpy as np import numpy as np
from array_with_attrs import ArrayWithAttrs, ensure_attrs from array_with_attrs import ArrayWithAttrs, ensure_attrs
import cv2 import cv2
#import cv2.cv
# import cv2.cv
from scipy import ndimage from scipy import ndimage
class ImageWithLocation(ArrayWithAttrs): class ImageWithLocation(ArrayWithAttrs):
"""An image, as a numpy array, with attributes to provide location information """An image, as a numpy array, with attributes to provide location information
@ -49,9 +51,10 @@ class ImageWithLocation(ArrayWithAttrs):
that we use to store the crucial mapping from pixels in the image to position in the that we use to store the crucial mapping from pixels in the image to position in the
sample. sample.
""" """
# def __array_finalize__(self, obj):
# """Ensure that the object is a properly set-up ImageWithLocation""" # def __array_finalize__(self, obj):
# ArrayWithAttrs.__array_finalize__(self, obj) # Ensure we have self.attrs # """Ensure that the object is a properly set-up ImageWithLocation"""
# ArrayWithAttrs.__array_finalize__(self, obj) # Ensure we have self.attrs
def __getitem__(self, item): def __getitem__(self, item):
"""Update the metadata when we extract a slice""" """Update the metadata when we extract a slice"""
try: try:
@ -61,18 +64,24 @@ class ImageWithLocation(ArrayWithAttrs):
assert isinstance(item[0], slice), "First index was not a slice" assert isinstance(item[0], slice), "First index was not a slice"
assert isinstance(item[1], slice), "Second index was not a slice" assert isinstance(item[1], slice), "Second index was not a slice"
start = np.array([item[i].start for i in range(2)]) start = np.array([item[i].start for i in range(2)])
start = np.where(start == np.array(None), 0, start) # missing start points are equivalent to zero start = np.where(
start == np.array(None), 0, start
) # missing start points are equivalent to zero
step = np.array([item[i].step for i in range(2)]) step = np.array([item[i].step for i in range(2)])
step = np.where(step == np.array(None), 1, step) # missing step is equivalent to step==1 step = np.where(
step == np.array(None), 1, step
) # missing step is equivalent to step==1
except: except:
# If the above doesn't work, assume we're not dealing with a 2D slice and give up. # If the above doesn't work, assume we're not dealing with a 2D slice and give up.
return super(ImageWithLocation, self).__getitem__(item) # pass it on up return super(ImageWithLocation, self).__getitem__(item) # pass it on up
out = super(ImageWithLocation, self).__getitem__(item) # retrieve the slice out = super(ImageWithLocation, self).__getitem__(item) # retrieve the slice
out.datum_pixel -= start # adjust the datum pixel so it refers to the same part of the image out.datum_pixel -= (
start
) # adjust the datum pixel so it refers to the same part of the image
# Next, we adjust the constant part of the pixel-sample matrix so pixels stay in the same place # Next, we adjust the constant part of the pixel-sample matrix so pixels stay in the same place
location_shift = np.dot(ensure_3d(start), self.pixel_to_sample_matrix[:3,:3]) location_shift = np.dot(ensure_3d(start), self.pixel_to_sample_matrix[:3, :3])
out.pixel_to_sample_matrix[3,:3] += location_shift out.pixel_to_sample_matrix[3, :3] += location_shift
if not np.all(step == 1): if not np.all(step == 1):
# if we're downsampling, remember to scale datum_pixel accordingly # if we're downsampling, remember to scale datum_pixel accordingly
out.datum_pixel = old_div(out.datum_pixel, step) out.datum_pixel = old_div(out.datum_pixel, step)
@ -105,18 +114,22 @@ class ImageWithLocation(ArrayWithAttrs):
A 2- or 3- element position, to match the size of location passed in. A 2- or 3- element position, to match the size of location passed in.
""" """
l = ensure_2d(location) l = ensure_2d(location)
l = l[:2]-self.pixel_to_sample_matrix[3,:2] l = l[:2] - self.pixel_to_sample_matrix[3, :2]
p = np.dot(l, np.linalg.inv(self.pixel_to_sample_matrix[:2,:2])) p = np.dot(l, np.linalg.inv(self.pixel_to_sample_matrix[:2, :2]))
if check_bounds: if check_bounds:
assert np.all(0 <= p[0:2]), "The location was not within the image" assert np.all(0 <= p[0:2]), "The location was not within the image"
assert np.all(p[0:2] <= self.shape[0:2]), "The location was not within the image" assert np.all(
assert np.abs(p[2]) < z_tolerance, "The location was too far away from the plane of the image" p[0:2] <= self.shape[0:2]
), "The location was not within the image"
assert (
np.abs(p[2]) < z_tolerance
), "The location was too far away from the plane of the image"
if len(location) == 2: if len(location) == 2:
return p[:2] return p[:2]
else: else:
return p[:3] return p[:3]
def feature_at(self, centre_position, size=(100,100), set_datum_to_centre=True): def feature_at(self, centre_position, size=(100, 100), set_datum_to_centre=True):
"""Return a thumbnail cropped out of this image, centred on a particular pixel position. """Return a thumbnail cropped out of this image, centred on a particular pixel position.
This is simply a convenience method that saves typing over the usual slice syntax. Below are two equivalent This is simply a convenience method that saves typing over the usual slice syntax. Below are two equivalent
@ -139,14 +152,25 @@ class ImageWithLocation(ArrayWithAttrs):
float(size[0]) float(size[0])
float(size[1]) float(size[1])
except: except:
raise IndexError("Error: arguments of feature_at were invalid: {}, {}".format(centre_position, size)) raise IndexError(
"Error: arguments of feature_at were invalid: {}, {}".format(
centre_position, size
)
)
pos = centre_position pos = centre_position
# For now, rely on numpy to complain if the feature is outside the image. May do bound-checking at some point. # For now, rely on numpy to complain if the feature is outside the image. May do bound-checking at some point.
# If so, we might need to think carefully about the datum pixel of the resulting image. # If so, we might need to think carefully about the datum pixel of the resulting image.
thumb = self[pos[0] - old_div(size[0],2):pos[0] + old_div(size[0],2), pos[1] - old_div(size[1],2):pos[1] + old_div(size[1],2), ...] thumb = self[
pos[0] - old_div(size[0], 2) : pos[0] + old_div(size[0], 2),
pos[1] - old_div(size[1], 2) : pos[1] + old_div(size[1], 2),
...,
]
if set_datum_to_centre: if set_datum_to_centre:
thumb.datum_pixel = (old_div(size[0],2), old_div(size[1],2)) # Make the datum point of the new image its centre. thumb.datum_pixel = (
old_div(size[0], 2),
old_div(size[1], 2),
) # Make the datum point of the new image its centre.
return thumb return thumb
def downsample(self, n): def downsample(self, n):
@ -156,7 +180,9 @@ class ImageWithLocation(ArrayWithAttrs):
to noise. Currently it just decimates (i.e. throws away rows and columns). to noise. Currently it just decimates (i.e. throws away rows and columns).
""" """
assert n > 0, "The downsampling factor must be an integer greater than 0" assert n > 0, "The downsampling factor must be an integer greater than 0"
return self[::int(n), ::int(n), ...] # The slicing code handles updating metadata return self[
:: int(n), :: int(n), ...
] # The slicing code handles updating metadata
@property @property
def datum_pixel(self): def datum_pixel(self):
@ -165,14 +191,16 @@ class ImageWithLocation(ArrayWithAttrs):
Usually the datum pixel is the central pixel, and if the metadata required is not present, Usually the datum pixel is the central pixel, and if the metadata required is not present,
we will silently assume that this is the case. we will silently assume that this is the case.
""" """
datum = self.attrs.get('datum_pixel', old_div((np.array(self.shape[:2]) - 1),2)) datum = self.attrs.get(
"datum_pixel", old_div((np.array(self.shape[:2]) - 1), 2)
)
assert len(datum) == 2, "The datum pixel didn't have length 2!" assert len(datum) == 2, "The datum pixel didn't have length 2!"
return datum return datum
@datum_pixel.setter @datum_pixel.setter
def datum_pixel(self, datum): def datum_pixel(self, datum):
assert len(datum) == 2, "The datum pixel didn't have length 2!" assert len(datum) == 2, "The datum pixel didn't have length 2!"
self.attrs['datum_pixel'] = datum self.attrs["datum_pixel"] = datum
@property @property
def datum_location(self): def datum_location(self):
@ -186,34 +214,36 @@ class ImageWithLocation(ArrayWithAttrs):
np.dot(p, M) yields a location for the given pixel, where p is [x,y,0,1] and M is this matrix. The location np.dot(p, M) yields a location for the given pixel, where p is [x,y,0,1] and M is this matrix. The location
given will be 4 elements long, and will have 1 as the final element. given will be 4 elements long, and will have 1 as the final element.
""" """
M = self.attrs['pixel_to_sample_matrix'] M = self.attrs["pixel_to_sample_matrix"]
assert M.shape == (4, 4), "The pixel-to-sample matrix is the wrong shape!" assert M.shape == (4, 4), "The pixel-to-sample matrix is the wrong shape!"
assert M.dtype.kind == "f", "The pixel-to-sample matrix is not floating point!" assert M.dtype.kind == "f", "The pixel-to-sample matrix is not floating point!"
return M return M
@pixel_to_sample_matrix.setter @pixel_to_sample_matrix.setter
def pixel_to_sample_matrix(self, M): def pixel_to_sample_matrix(self, M):
M = np.asanyarray(M) #ensure it's an ndarray subclass M = np.asanyarray(M) # ensure it's an ndarray subclass
assert M.shape == (4, 4), "The pixel-to-sample matrix must be 4x4!" assert M.shape == (4, 4), "The pixel-to-sample matrix must be 4x4!"
assert M.dtype.kind == "f", "The pixel-to-sample matrix must be floating point!" assert M.dtype.kind == "f", "The pixel-to-sample matrix must be floating point!"
self.attrs['pixel_to_sample_matrix'] = M self.attrs["pixel_to_sample_matrix"] = M
# TODO: split the data type out of this module and put it somewhere sensible
#TODO: split the data type out of this module and put it somewhere sensible
def add_location_metadata(image, pixel_to_sample_matrix, datum_pixel=None): def add_location_metadata(image, pixel_to_sample_matrix, datum_pixel=None):
"""Wrap an image if needed, and set its pixel to sample matrix.""" """Wrap an image if needed, and set its pixel to sample matrix."""
awa = ensure_attrs(image) # if needed, convert the image to an ArrayWithAttrs awa = ensure_attrs(image) # if needed, convert the image to an ArrayWithAttrs
awa.attrs['pixel_to_sample_matrix'] = pixel_to_sample_matrix awa.attrs["pixel_to_sample_matrix"] = pixel_to_sample_matrix
if datum_pixel is not None: if datum_pixel is not None:
awa.attrs['datum_pixel'] = datum_pixel awa.attrs["datum_pixel"] = datum_pixel
return awa return awa
def datum_pixel(image): def datum_pixel(image):
"""Get the datum pixel of an image - if no property is present, assume the central pixel.""" """Get the datum pixel of an image - if no property is present, assume the central pixel."""
try: try:
return np.array(image.datum_pixel) return np.array(image.datum_pixel)
except: except:
return (np.array(image.shape[:2]) - 1) / 2. return (np.array(image.shape[:2]) - 1) / 2.0
def ensure_3d(vector): def ensure_3d(vector):
@ -223,7 +253,9 @@ def ensure_3d(vector):
elif len(vector) == 2: elif len(vector) == 2:
return np.array([vector[0], vector[1], 0]) return np.array([vector[0], vector[1], 0])
else: else:
raise ValueError("Tried to ensure a vector was 3D, but it had neither 2 nor 3 elements!") raise ValueError(
"Tried to ensure a vector was 3D, but it had neither 2 nor 3 elements!"
)
def ensure_2d(vector): def ensure_2d(vector):
@ -233,5 +265,6 @@ def ensure_2d(vector):
elif len(vector) == 3: elif len(vector) == 3:
return np.array(vector[:2]) return np.array(vector[:2])
else: else:
raise ValueError("Tried to ensure a vector was 2D, but it had neither 2 nor 3 elements!") raise ValueError(
"Tried to ensure a vector was 2D, but it had neither 2 nor 3 elements!"
)

View file

@ -13,7 +13,12 @@ import logging
# Type hinting # Type hinting
from typing import Tuple from typing import Tuple
from .recalibrate_utils import recalibrate_camera, auto_expose_and_freeze_settings, flat_lens_shading_table from .recalibrate_utils import (
recalibrate_camera,
auto_expose_and_freeze_settings,
flat_lens_shading_table,
)
@contextmanager @contextmanager
def pause_stream(scamera, resolution: Tuple[int, int] = None): def pause_stream(scamera, resolution: Tuple[int, int] = None):
@ -23,7 +28,9 @@ def pause_stream(scamera, resolution: Tuple[int, int] = None):
block has finished. block has finished.
""" """
with scamera.lock: with scamera.lock:
assert not scamera.record_active, "We can't pause the camera's video stream while a recording is in progress." assert (
not scamera.record_active
), "We can't pause the camera's video stream while a recording is in progress."
streaming = scamera.stream_active streaming = scamera.stream_active
old_resolution = scamera.camera.resolution old_resolution = scamera.camera.resolution
if streaming: if streaming:
@ -37,6 +44,7 @@ def pause_stream(scamera, resolution: Tuple[int, int] = None):
logging.info("Restarting stream in pause_stream context manager") logging.info("Restarting stream in pause_stream context manager")
scamera.start_stream_recording() scamera.start_stream_recording()
def recalibrate(microscope): def recalibrate(microscope):
"""Reset the camera's settings. """Reset the camera's settings.
@ -45,7 +53,9 @@ def recalibrate(microscope):
with a gray level of 230. It takes a little while to run. with a gray level of 230. It takes a little while to run.
""" """
with pause_stream(microscope.camera) as scamera: with pause_stream(microscope.camera) as scamera:
auto_expose_and_freeze_settings(scamera.camera) # scamera.camera is the PiCamera object auto_expose_and_freeze_settings(
scamera.camera
) # scamera.camera is the PiCamera object
recalibrate_camera(scamera.camera) recalibrate_camera(scamera.camera)
microscope.save_settings() microscope.save_settings()
@ -63,13 +73,17 @@ class RecalibrateView(View):
return taskify(recalibrate)(microscope) return taskify(recalibrate)(microscope)
@ThingAction @ThingAction
class FlattenLSTView(View): class FlattenLSTView(View):
def post(self): def post(self):
microscope = find_component("org.openflexure.microscope") microscope = find_component("org.openflexure.microscope")
if not microscope: if not microscope:
abort(503, "No microscope connected. Unable to flatten the lens shading table.") abort(
503,
"No microscope connected. Unable to flatten the lens shading table.",
)
try: try:
with pause_stream(microscope.camera) as scamera: with pause_stream(microscope.camera) as scamera:
@ -78,7 +92,11 @@ class FlattenLSTView(View):
microscope.save_settings() microscope.save_settings()
except: except:
logging.exception("Error flattening the lens shading table.") logging.exception("Error flattening the lens shading table.")
abort(503, "Couldn't flatten the lens shading table - do you have the forked PiCamera library installed?") abort(
503,
"Couldn't flatten the lens shading table - do you have the forked PiCamera library installed?",
)
@ThingAction @ThingAction
class DeleteLSTView(View): class DeleteLSTView(View):
@ -86,7 +104,10 @@ class DeleteLSTView(View):
microscope = find_component("org.openflexure.microscope") microscope = find_component("org.openflexure.microscope")
if not microscope: if not microscope:
abort(503, "No microscope connected. Unable to flatten the lens shading table.") abort(
503,
"No microscope connected. Unable to flatten the lens shading table.",
)
try: try:
with pause_stream(microscope.camera) as scamera: with pause_stream(microscope.camera) as scamera:
@ -94,11 +115,16 @@ class DeleteLSTView(View):
microscope.save_settings() microscope.save_settings()
except: except:
logging.exception("Error deleting the lens shading table.") logging.exception("Error deleting the lens shading table.")
abort(503, "Couldn't flatten the lens shading table - do you have the forked PiCamera library installed?") abort(
503,
"Couldn't flatten the lens shading table - do you have the forked PiCamera library installed?",
)
lst_extension_v2 = BaseExtension( lst_extension_v2 = BaseExtension(
"org.openflexure.calibration.picamera", version="2.0.0-beta.1", description="Routines to perform flat-field correction on the camera." "org.openflexure.calibration.picamera",
version="2.0.0-beta.1",
description="Routines to perform flat-field correction on the camera.",
) )
lst_extension_v2.add_method( lst_extension_v2.add_method(

View file

@ -199,7 +199,7 @@ def tile(
# Run slow autofocus. Client should provide dz ~ 50 # Run slow autofocus. Client should provide dz ~ 50
autofocus_extension.autofocus( autofocus_extension.autofocus(
microscope, microscope,
range(-3 * autofocus_dz, 4 * autofocus_dz, autofocus_dz) range(-3 * autofocus_dz, 4 * autofocus_dz, autofocus_dz),
) )
logging.debug("Finished autofocus") logging.debug("Finished autofocus")
time.sleep(1) time.sleep(1)

View file

@ -58,7 +58,7 @@ def enabled_root_actions():
return {k: v for k, v in _actions.items() if v["conditions"]} return {k: v for k, v in _actions.items() if v["conditions"]}
#@Tag("actions") # @Tag("actions")
class ActionsView(View): class ActionsView(View):
def get(self): def get(self):
""" """

View file

@ -203,4 +203,3 @@ with open(DEFAULT_CONFIGURATION_FILE_PATH, "r") as default_configuration:
user_configuration = OpenflexureSettingsFile( user_configuration = OpenflexureSettingsFile(
path=CONFIGURATION_FILE_PATH, defaults=DEFAULT_CONFIGURATION path=CONFIGURATION_FILE_PATH, defaults=DEFAULT_CONFIGURATION
) )

View file

@ -175,7 +175,12 @@ class Microscope:
don't get removed from the settings file. don't get removed from the settings file.
""" """
settings_current = {"id": self.id, "name": self.name, "fov": self.fov, "extensions": self.extension_settings} settings_current = {
"id": self.id,
"name": self.name,
"fov": self.fov,
"extensions": self.extension_settings,
}
# If attached to a camera # If attached to a camera
if self.camera: if self.camera:

View file

@ -28,12 +28,7 @@ def ndarray_to_json(arr: np.ndarray):
# This comes in very handy for the lens shading table. # This comes in very handy for the lens shading table.
arr = np.array(arr) arr = np.array(arr)
b64_string, dtype, shape = serialise_array_b64(arr) b64_string, dtype, shape = serialise_array_b64(arr)
return { return {"@type": "ndarray", "dtype": dtype, "shape": shape, "base64": b64_string}
"@type": "ndarray",
"dtype": dtype,
"shape": shape,
"base64": b64_string
}
def json_to_ndarray(json_dict: dict): def json_to_ndarray(json_dict: dict):
@ -42,9 +37,10 @@ def json_to_ndarray(json_dict: dict):
for required_param in ("dtype", "shape", "base64"): for required_param in ("dtype", "shape", "base64"):
if not json_dict.get(required_param): if not json_dict.get(required_param):
raise KeyError(f"Missing required key {required_param}") raise KeyError(f"Missing required key {required_param}")
return deserialise_array_b64(json_dict.get("base64"), json_dict.get("dtype"), json_dict.get("shape"))
return deserialise_array_b64(
json_dict.get("base64"), json_dict.get("dtype"), json_dict.get("shape")
)
@contextmanager @contextmanager