rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

2.9 dev numpy types

Browse source
This commit is contained in:
Joel Collins 2020-12-04 13:35:11 +00:00
parent f2a2d880f2
commit 63b633ba44
10 changed files with 156 additions and 90 deletions
Download patch file Download diff file Expand all files Collapse all files

98
openflexure_microscope/api/default_extensions/autofocus.py
View file

@ -1,7 +1,7 @@
import logging
import time
from contextlib import contextmanager
from typing import Callable, List, Optional, Tuple
from typing import Callable, Dict, List, Optional, Tuple
import numpy as np
from labthings import current_action, fields, find_component
@ -39,7 +39,7 @@ class JPEGSharpnessMonitor:
self.camera.stream.stop_tracking()
self.camera.stream.reset_tracking()
def focus_rel(self, dz: int, backlash: bool = False, **kwargs):
def focus_rel(self, dz: int, backlash: bool = False, **kwargs) -> Tuple[int, int]:
# Store the start time and position
self.camera.stream.start_tracking()
self.stage_times.append(time.time())
@ -62,22 +62,28 @@ class JPEGSharpnessMonitor:
self.camera.stream.reset_tracking()
# Index of the data for this movement
data_index = len(self.stage_positions) - 2
data_index: int = len(self.stage_positions) - 2
# Final z position after move
final_z_position = self.stage_positions[-1][2]
final_z_position: int = self.stage_positions[-1][2]
return data_index, final_z_position
def move_data(self, istart: int, istop: Optional[int] = None):
def move_data(
self, istart: int, istop: Optional[int] = None
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Extract sharpness as a function of (interpolated) z"""
if istop is None:
istop = istart + 2
jpeg_times = np.array(self.jpeg_times)
jpeg_sizes = np.array(self.jpeg_sizes)
stage_times = np.array(self.stage_times)[istart:istop]
stage_zs = np.array(self.stage_positions)[istart:istop, 2]
jpeg_times: np.ndarray = np.array(self.jpeg_times) # np.ndarray[float]
jpeg_sizes: np.ndarray = np.array(self.jpeg_sizes) # np.ndarray[int]
stage_times: np.ndarray = np.array(self.stage_times)[
istart:istop
] # np.ndarray[float]
stage_zs: np.ndarray = np.array(self.stage_positions)[
istart:istop, 2
] # np.ndarray[int]
try:
start = np.argmax(jpeg_times > stage_times[0])
stop = np.argmax(jpeg_times > stage_times[1])
start: int = int(np.argmax(jpeg_times > stage_times[0]))
stop: int = int(np.argmax(jpeg_times > stage_times[1]))
except ValueError as e:
if np.sum(jpeg_times > stage_times[0]) == 0:
raise ValueError(
@ -89,10 +95,12 @@ class JPEGSharpnessMonitor:
stop = len(jpeg_times)
logging.debug("changing stop to %s", (stop))
jpeg_times = jpeg_times[start:stop]
jpeg_zs = np.interp(jpeg_times, stage_times, stage_zs)
jpeg_zs: np.ndarray = np.interp(
jpeg_times, stage_times, stage_zs
) # np.ndarray[float]
return jpeg_times, jpeg_zs, jpeg_sizes[start:stop]
def sharpest_z_on_move(self, index: int):
def sharpest_z_on_move(self, index: int) -> int:
"""Return the z position of the sharpest image on a given move"""
_, jz, js = self.move_data(index)
if len(js) == 0:
@ -101,7 +109,7 @@ class JPEGSharpnessMonitor:
)
return jz[np.argmax(js)]
def data_dict(self):
def data_dict(self) -> Dict[str, np.ndarray]:
"""Return the gathered data as a single convenient dictionary"""
data = {}
for k in ["jpeg_times", "jpeg_sizes", "stage_times", "stage_positions"]:
@ -111,7 +119,7 @@ class JPEGSharpnessMonitor:
@contextmanager
def monitor_sharpness(microscope: Microscope):
m = JPEGSharpnessMonitor(microscope)
m: JPEGSharpnessMonitor = JPEGSharpnessMonitor(microscope)
m.start()
try:
yield m
@ -119,18 +127,18 @@ def monitor_sharpness(microscope: Microscope):
m.stop()
def sharpness_sum_lap2(rgb_image: np.ndarray):
def sharpness_sum_lap2(rgb_image: np.ndarray) -> np.float:
"""Return an image sharpness metric: sum(laplacian(image)**")"""
image_bw = np.mean(rgb_image, 2)
image_lap = ndimage.filters.laplace(image_bw)
image_bw: np.float = np.mean(rgb_image, 2)
image_lap: np.float = ndimage.filters.laplace(image_bw)
return np.mean(image_lap.astype(np.float) ** 4)
def sharpness_edge(image: np.ndarray):
def sharpness_edge(image: np.ndarray) -> np.float:
"""Return a sharpness metric optimised for vertical lines"""
gray = np.mean(image.astype(float), 2)
n = 20
edge = np.array([[-1] * n + [1] * n])
gray: np.float = np.mean(image.astype(float), 2)
n: int = 20
edge: np.ndarray = np.array([[-1] * n + [1] * n])
return np.sum(
[np.sum(ndimage.filters.convolve(gray, W) ** 2) for W in [edge, edge.T]]
)
@ -161,7 +169,7 @@ class AutofocusExtension(BaseExtension):
def measure_sharpness(
self, microscope: Microscope, metric_fn: Callable = sharpness_sum_lap2
):
) -> np.float:
"""Measure the sharpness of the camera's current view."""
if hasattr(microscope.camera, "array") and callable(
@ -177,7 +185,7 @@ class AutofocusExtension(BaseExtension):
dz: List[int],
settle: float = 0.5,
metric_fn: Callable = sharpness_sum_lap2,
):
) -> Tuple[List[int], List[np.float]]:
"""Perform a simple autofocus routine.
The stage is moved to z positions (relative to current position) in dz,
and at each position an image is captured and the sharpness function
@ -185,12 +193,12 @@ class AutofocusExtension(BaseExtension):
highest. No interpolation is performed.
dz is assumed to be in ascending order (starting at -ve values)
"""
camera = microscope.camera
stage = microscope.stage
camera: BaseCamera = microscope.camera
stage: BaseStage = microscope.stage
with set_properties(stage, backlash=256), stage.lock, camera.lock:
sharpnesses = []
positions = []
sharpnesses: List[np.float] = []
positions: List[int] = []
# Some cameras may not have annotate_text. Reset if it does
if getattr(camera, "annotate_text"):
@ -198,29 +206,33 @@ class AutofocusExtension(BaseExtension):
for _ in stage.scan_z(dz, return_to_start=False):
if current_action() and current_action().stopped:
return
return [], []
positions.append(stage.position[2])
time.sleep(settle)
sharpnesses.append(self.measure_sharpness(microscope, metric_fn))
newposition = positions[np.argmax(sharpnesses)]
newposition: int = positions[int(np.argmax(sharpnesses))]
stage.move_rel((0, 0, newposition - stage.position[2]))
return positions, sharpnesses
def move_and_find_focus(self, microscope: Microscope, dz: int):
def move_and_find_focus(self, microscope: Microscope, dz: int) -> int:
"""Make a relative Z move and return the peak sharpness position"""
with monitor_sharpness(microscope) as m:
m.focus_rel(dz)
return m.sharpest_z_on_move(0)
def move_and_measure(self, microscope: Microscope, dz: int):
def move_and_measure(
self, microscope: Microscope, dz: int
) -> Dict[str, np.ndarray]:
"""Make a relative Z move and return the sharpness data"""
with monitor_sharpness(microscope) as m:
m.focus_rel(dz)
return m.data_dict()
def fast_autofocus(self, microscope: Microscope, dz: int = 2000):
def fast_autofocus(
self, microscope: Microscope, dz: int = 2000
) -> Dict[str, np.ndarray]:
"""Perform a down-up-down-up autofocus"""
with microscope.camera.lock, microscope.stage.lock:
with monitor_sharpness(microscope) as m:
@ -231,7 +243,7 @@ class AutofocusExtension(BaseExtension):
# z: Final z position after move
i, z = m.focus_rel(dz)
# Get the z position with highest sharpness from the previous move (index i)
fz = m.sharpest_z_on_move(i)
fz: int = m.sharpest_z_on_move(i)
# Move all the way to the start so it's consistent
# Store final absolute z position from this return move
i, z = m.focus_rel(-dz)
@ -248,7 +260,7 @@ class AutofocusExtension(BaseExtension):
target_z: int = 0,
initial_move_up: bool = True,
mini_backlash: int = 25,
):
) -> Dict[str, np.ndarray]:
"""Autofocus by measuring on the way down, and moving back up with feedback.
This autofocus method is very efficient, as it only passes the peak once.
@ -294,11 +306,15 @@ class AutofocusExtension(BaseExtension):
m.focus_rel(dz / 2)
# move down
logging.debug("Move down")
i: int
z: int
i, z = m.focus_rel(-dz)
# now inspect where the sharpest point is, and estimate the sharpness
# (JPEG size) that we should find at the start of the Z stack
jz: np.ndarray # np.ndarray[float]
js: np.ndarray # np.ndarray[float]
_, jz, js = m.move_data(i)
best_z = jz[np.argmax(js)]
best_z: int = jz[np.argmax(js)]
# now move to the start of the z stack
logging.debug("Move to the start of the z stack")
@ -309,17 +325,19 @@ class AutofocusExtension(BaseExtension):
# We've deliberately undershot - figure out how much further we should move based on the curve
logging.debug("Calculate remining movement")
current_js = m.camera.stream.last.size
imax = np.argmax(js) # we want to crop out just the bit below the peak
imax: int = int(
np.argmax(js)
) # we want to crop out just the bit below the peak
js = js[imax:] # NB z is in DECREASING order
jz = jz[imax:]
inow = np.argmax(
js < current_js
inow: int = int(
np.argmax(js < current_js)
) # use the curve we recorded to estimate our position
# So, the Z position corresponding to our current sharpness value is zs[inow]
# That means we should move forwards, by best_z - zs[inow]
logging.debug("Correction move")
correction_move = best_z + target_z - jz[inow]
correction_move: int = best_z + target_z - jz[inow]
logging.debug(
"Fast autofocus scan: correcting backlash by moving %s steps",
(correction_move),

8
openflexure_microscope/api/default_extensions/picamera_autocalibrate/recalibrate_utils.py
View file

@ -78,10 +78,12 @@ def auto_expose_and_freeze_settings(camera: PiCamera):
def channels_from_bayer_array(bayer_array: np.ndarray) -> np.ndarray:
"""Given the 'array' from a PiBayerArray, return the 4 channels."""
bayer_pattern: List[Tuple[int, int]] = [(0, 0), (0, 1), (1, 0), (1, 1)]
channels: np.ndarray = np.zeros(
(4, bayer_array.shape[0] // 2, bayer_array.shape[1] // 2),
dtype=bayer_array.dtype,
channels_shape: Tuple[int, ...] = (
4,
bayer_array.shape[0] // 2,
bayer_array.shape[1] // 2,
)
channels: np.ndarray = np.zeros(channels_shape, dtype=bayer_array.dtype)
for i, offset in enumerate(bayer_pattern):
# We simplify life by dealing with only one channel at a time.
channels[i, :, :] = np.sum(

4
openflexure_microscope/api/dev_extensions/__init__.py
View file

@ -1,3 +1,3 @@
from .tools import devtools_extension_v2
from .tools import DevToolsExtension
LABTHINGS_EXTENSIONS = [devtools_extension_v2]
LABTHINGS_EXTENSIONS = [DevToolsExtension]

21
openflexure_microscope/api/dev_extensions/tools.py
View file

@ -6,6 +6,17 @@ from labthings.extensions import BaseExtension
from labthings.views import ActionView
class DevToolsExtension(BaseExtension):
def __init__(self) -> None:
super().__init__(
"org.openflexure.dev.tools",
version="0.1.0",
description="Actions to cause various traumatic events in the microscope, used for testing.",
)
self.add_view(RaiseException, "/raise")
self.add_view(SleepFor, "/sleep")
class RaiseException(ActionView):
def post(self):
raise Exception("The developer raised an exception")
@ -23,13 +34,3 @@ class SleepFor(ActionView):
end = time.time()
logging.info("Waking up!")
return {"TimeAsleep": (end - start)}
devtools_extension_v2 = BaseExtension(
"org.openflexure.dev.tools",
version="0.1.0",
description="Actions to cause various traumatic events in the microscope, used for testing.",
)
devtools_extension_v2.add_view(RaiseException, "/raise")
devtools_extension_v2.add_view(SleepFor, "/sleep")

19
openflexure_microscope/camera/base.py
View file

@ -3,14 +3,16 @@ import io
import logging
import time
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from types import TracebackType
from typing import BinaryIO, List, Optional, Tuple, Type, Union
from typing import BinaryIO, List, NamedTuple, Optional, Tuple, Type, Union
from labthings import ClientEvent, StrictLock
# Class to store a frames metadata
TrackerFrame = namedtuple("TrackerFrame", ["size", "time"])
class TrackerFrame(NamedTuple):
size: int
time: float
class FrameStream(io.BytesIO):
@ -172,16 +174,17 @@ class BaseCamera(metaclass=ABCMeta):
def start_worker(self, **_) -> bool:
"""Start the background camera thread if it isn't running yet."""
logging.debug(
logging.warning(
"`start_worker` method has been deprecated and is no longer required. Please avoid calling this method."
)
return True
def get_frame(self):
"""Return the current camera frame."""
logging.debug(
"`camera.get_frame` method has been deprecated. Please use `camera.stream.getframe()."
)
"""
Return the current camera frame.
Just an alias of self.stream.getframe()
"""
return self.stream.getframe()
def __enter__(self):

2
openflexure_microscope/stage/sanga.py
View file

@ -101,7 +101,7 @@ class SangaStage(BaseStage):
if "backlash" in config:
# Construct backlash array
backlash = axes_to_array(config["backlash"], ["x", "y", "z"], [0, 0, 0])
self.backlash = backlash
self.backlash = np.array(backlash)
if "settle_time" in config:
self.settle_time = config.get("settle_time")

43
openflexure_microscope/utilities.py
View file

@ -1,18 +1,25 @@
import base64
import copy
import logging
import sys
import time
from contextlib import contextmanager
from typing import Dict, List, Optional
from typing import Dict, List, Optional, Tuple, Type, Union
import numpy as np
# TypedDict was added to typing in 3.8. Use typing_extensions for <3.8
if sys.version_info >= (3, 8):
from typing import TypedDict # pylint: disable=no-name-in-module
else:
from typing_extensions import TypedDict
class Timer(object):
def __init__(self, name):
self.name = name
self.start = None
self.end = None
def __init__(self, name: str):
self.name: str = name
self.start: Optional[float] = None
self.end: Optional[float] = None
def __enter__(self):
self.start = time.time()
@ -22,19 +29,27 @@ class Timer(object):
logging.debug("%s time: %s", self.name, self.end - self.start)
def deserialise_array_b64(b64_string: str, dtype: str, shape: List[int]):
flat_arr = np.frombuffer(base64.b64decode(b64_string), dtype)
JSONArrayType = TypedDict(
"JSONArrayType",
{"@type": str, "base64": str, "dtype": str, "shape": Tuple[int, ...]},
)
def deserialise_array_b64(
b64_string: str, dtype: Union[Type[np.dtype], str], shape: Tuple[int, ...]
):
flat_arr: np.ndarray = np.frombuffer(base64.b64decode(b64_string), dtype)
return flat_arr.reshape(shape)
def serialise_array_b64(npy_arr: np.ndarray):
b64_string = base64.b64encode(npy_arr).decode("ascii")
dtype = str(npy_arr.dtype)
shape = npy_arr.shape
def serialise_array_b64(npy_arr: np.ndarray) -> Tuple[str, str, Tuple[int, ...]]:
b64_string: str = base64.b64encode(npy_arr.tobytes()).decode("ascii")
dtype: str = str(npy_arr.dtype)
shape: Tuple[int, ...] = npy_arr.shape
return b64_string, dtype, shape
def ndarray_to_json(arr: np.ndarray):
def ndarray_to_json(arr: np.ndarray) -> JSONArrayType:
if isinstance(arr, memoryview):
# We can transparently convert memoryview objects to arrays
# This comes in very handy for the lens shading table.
@ -43,7 +58,7 @@ def ndarray_to_json(arr: np.ndarray):
return {"@type": "ndarray", "dtype": dtype, "shape": shape, "base64": b64_string}
def json_to_ndarray(json_dict: dict):
def json_to_ndarray(json_dict: JSONArrayType):
if not json_dict.get("@type") != "ndarray":
logging.warning("No valid @type attribute found. Conversion may fail.")
for required_param in ("dtype", "shape", "base64"):
@ -52,7 +67,7 @@ def json_to_ndarray(json_dict: dict):
b64_string: Optional[str] = json_dict.get("base64")
dtype: Optional[str] = json_dict.get("dtype")
shape: Optional[List[int]] = json_dict.get("shape")
shape: Optional[Tuple[int, ...]] = json_dict.get("shape")
if b64_string and dtype and shape:
return deserialise_array_b64(b64_string, dtype, shape)