Remove old plugins
This commit is contained in:
Joel Collins
parent
bd7cea0fdf
commit
94aac61925
22 changed files with 0 additions and 1438 deletions
|
|
@ -1,2 +0,0 @@
|
|||
__all__ = ["AutofocusPlugin"]
|
||||
from .plugin import AutofocusPlugin
|
||||
|
|
@ -1,64 +0,0 @@
|
|||
import numpy as np
|
||||
import logging
|
||||
|
||||
from openflexure_microscope.devel import (
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
taskify,
|
||||
abort,
|
||||
)
|
||||
|
||||
|
||||
class MeasureSharpnessAPI(MicroscopeViewPlugin):
|
||||
def post(self):
|
||||
payload = JsonResponse(request)
|
||||
return jsonify({"sharpness": self.plugin.measure_sharpness()})
|
||||
|
||||
|
||||
class AutofocusAPI(MicroscopeViewPlugin):
|
||||
"""
|
||||
Run a standard autofocus
|
||||
"""
|
||||
|
||||
def post(self):
|
||||
payload = JsonResponse(request)
|
||||
|
||||
# Figure out the range of z values to use
|
||||
dz = payload.param("dz", default=np.linspace(-300, 300, 7), convert=np.array)
|
||||
|
||||
if self.microscope.has_real_stage():
|
||||
logging.info("Running autofocus...")
|
||||
task = taskify(self.plugin.autofocus)(dz)
|
||||
|
||||
# return a handle on the autofocus task
|
||||
return jsonify(task.state), 201
|
||||
|
||||
else:
|
||||
abort(503, "No stage connected. Unable to autofocus.")
|
||||
|
||||
|
||||
class FastAutofocusAPI(MicroscopeViewPlugin):
|
||||
"""
|
||||
Run a fast autofocus
|
||||
"""
|
||||
|
||||
def post(self):
|
||||
payload = JsonResponse(request)
|
||||
|
||||
# Figure out the parameters to use
|
||||
dz = payload.param("dz", default=2000, convert=int)
|
||||
backlash = payload.param("backlash", default=0, convert=int)
|
||||
if backlash < 0:
|
||||
backlash = 0
|
||||
|
||||
if self.microscope.has_real_stage():
|
||||
logging.info("Running autofocus...")
|
||||
task = taskify(self.plugin.fast_autofocus)(dz, backlash=backlash)
|
||||
|
||||
# return a handle on the autofocus task
|
||||
return jsonify(task.state), 201
|
||||
|
||||
else:
|
||||
abort(503, "No stage connected. Unable to autofocus.")
|
||||
|
|
@ -1,129 +0,0 @@
|
|||
import time
|
||||
import numpy as np
|
||||
import threading
|
||||
import logging
|
||||
from scipy import ndimage
|
||||
|
||||
|
||||
class JPEGSharpnessMonitor:
|
||||
def __init__(self, microscope, timeout=60):
|
||||
self.microscope = microscope
|
||||
self.camera = microscope.camera
|
||||
self.stage = microscope.stage
|
||||
self.jpeg_sizes = []
|
||||
self.jpeg_times = []
|
||||
self.stage_positions = []
|
||||
self.stage_times = []
|
||||
self.stop_event = threading.Event()
|
||||
self.timeout = timeout
|
||||
self.keep_alive()
|
||||
self.background_thread = None
|
||||
|
||||
def is_alive(self):
|
||||
if self.background_thread is None:
|
||||
return False
|
||||
else:
|
||||
return self.background_thread.is_alive()
|
||||
|
||||
def should_stop(self):
|
||||
import time
|
||||
|
||||
return time.time() - self.kept_alive > self.timeout
|
||||
|
||||
def keep_alive(self):
|
||||
import time
|
||||
|
||||
self.kept_alive = time.time()
|
||||
|
||||
def start(self):
|
||||
"Start monitoring sharpness by looking at JPEG size"
|
||||
self.background_thread = threading.Thread(target=self._measure_jpegs)
|
||||
self.background_thread.start()
|
||||
return self
|
||||
|
||||
def stop(self):
|
||||
"Stop the background thread"
|
||||
self.stop_event.set()
|
||||
self.background_thread.join()
|
||||
|
||||
def _measure_jpegs(self):
|
||||
"Function that runs in a background thread to record sharpness"
|
||||
logging.info("Starting sharpness measurement in background thread")
|
||||
self.keep_alive()
|
||||
while not self.stop_event.is_set() and not self.should_stop():
|
||||
self.jpeg_sizes.append(self.jpeg_size())
|
||||
self.jpeg_times.append(time.time())
|
||||
if self.stop_event.is_set():
|
||||
logging.info("Cleanly stopped sharpness measurement in background thread")
|
||||
if self.should_stop():
|
||||
logging.info("Sharpness measurement timed out and has stopped")
|
||||
|
||||
def jpeg_size(self):
|
||||
"""Return the size of a frame from the MJPEG stream"""
|
||||
return len(self.camera.get_frame())
|
||||
|
||||
def focus_rel(self, dz, backlash=False, **kwargs):
|
||||
self.keep_alive()
|
||||
self.stage_times.append(time.time())
|
||||
self.stage_positions.append(self.stage.position)
|
||||
self.stage.move_rel([0, 0, dz], backlash=backlash, **kwargs)
|
||||
self.stage_times.append(time.time())
|
||||
self.stage_positions.append(self.stage.position)
|
||||
i = len(self.stage_positions) - 2
|
||||
return i, self.stage_positions[-1][2]
|
||||
|
||||
def move_data(self, istart, istop=None):
|
||||
"Extract sharpness as a function of (interpolated) z"
|
||||
global np, logging
|
||||
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]
|
||||
start = np.argmax(jpeg_times > stage_times[0])
|
||||
stop = np.argmax(jpeg_times > stage_times[1])
|
||||
if stop < 1:
|
||||
stop = len(jpeg_times)
|
||||
logging.debug("changing stop to {}".format(stop))
|
||||
jpeg_times = jpeg_times[start:stop]
|
||||
jpeg_zs = np.interp(jpeg_times, stage_times, stage_zs)
|
||||
return jpeg_times, jpeg_zs, jpeg_sizes[start:stop]
|
||||
|
||||
def sharpest_z_on_move(self, index):
|
||||
"""Return the z position of the sharpest image on a given move"""
|
||||
jt, jz, js = self.move_data(index)
|
||||
return jz[np.argmax(js)]
|
||||
|
||||
def data_dict(self):
|
||||
"""Return the gathered data as a single convenient dictionary"""
|
||||
data = {}
|
||||
for k in ["jpeg_times", "jpeg_sizes", "stage_times", "stage_positions"]:
|
||||
data[k] = getattr(self, k)
|
||||
return data
|
||||
|
||||
|
||||
def decimate_to(shape, image):
|
||||
"""Decimate an image to reduce its size if it's too big."""
|
||||
decimation = np.max(
|
||||
np.ceil(np.array(image.shape, dtype=np.float)[: len(shape)] / np.array(shape))
|
||||
)
|
||||
return image[:: int(decimation), :: int(decimation), ...]
|
||||
|
||||
|
||||
def sharpness_sum_lap2(rgb_image):
|
||||
"""Return an image sharpness metric: sum(laplacian(image)**")"""
|
||||
# image_bw=np.mean(decimate_to((1000,1000), rgb_image),2)
|
||||
image_bw = np.mean(rgb_image, 2)
|
||||
image_lap = ndimage.filters.laplace(image_bw)
|
||||
return np.mean(image_lap.astype(np.float) ** 4)
|
||||
|
||||
|
||||
def sharpness_edge(image):
|
||||
"""Return a sharpness metric optimised for vertical lines"""
|
||||
gray = np.mean(image.astype(float), 2)
|
||||
n = 20
|
||||
edge = np.array([[-1] * n + [1] * n])
|
||||
return np.sum(
|
||||
[np.sum(ndimage.filters.convolve(gray, W) ** 2) for W in [edge, edge.T]]
|
||||
)
|
||||
|
|
@ -1,168 +0,0 @@
|
|||
import time
|
||||
import logging
|
||||
import numpy as np
|
||||
from contextlib import contextmanager
|
||||
|
||||
from openflexure_microscope.utilities import set_properties
|
||||
|
||||
from .focus_utils import sharpness_sum_lap2, JPEGSharpnessMonitor
|
||||
from .api import MeasureSharpnessAPI, AutofocusAPI, FastAutofocusAPI
|
||||
|
||||
from openflexure_microscope.devel import MicroscopePlugin
|
||||
|
||||
|
||||
class AutofocusPlugin(MicroscopePlugin):
|
||||
"""
|
||||
Basic autofocus plugin
|
||||
"""
|
||||
|
||||
api_views = {
|
||||
"/measure_sharpness": MeasureSharpnessAPI,
|
||||
"/autofocus": AutofocusAPI,
|
||||
"/fast_autofocus": FastAutofocusAPI,
|
||||
}
|
||||
|
||||
### SLOW AUTOFOCUS
|
||||
|
||||
def autofocus(self, dz, settle=0.5, metric_fn=sharpness_sum_lap2):
|
||||
"""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
|
||||
evaulated. We then move back to the position where the sharpness was
|
||||
highest. No interpolation is performed.
|
||||
dz is assumed to be in ascending order (starting at -ve values)
|
||||
"""
|
||||
camera = self.microscope.camera
|
||||
stage = self.microscope.stage
|
||||
|
||||
with set_properties(stage, backlash=256), stage.lock, camera.lock:
|
||||
sharpnesses = []
|
||||
positions = []
|
||||
camera.annotate_text = ""
|
||||
|
||||
for _ in stage.scan_z(dz, return_to_start=False):
|
||||
positions.append(stage.position[2])
|
||||
time.sleep(settle)
|
||||
sharpnesses.append(self.measure_sharpness(metric_fn))
|
||||
|
||||
newposition = positions[np.argmax(sharpnesses)]
|
||||
stage.move_rel([0, 0, newposition - stage.position[2]])
|
||||
|
||||
return positions, sharpnesses
|
||||
|
||||
def measure_sharpness(self, metric_fn=sharpness_sum_lap2):
|
||||
"""Measure the sharpness of the camera's current view."""
|
||||
return metric_fn(self.microscope.camera.array(use_video_port=True))
|
||||
|
||||
### FAST AUTOFOCUS
|
||||
|
||||
@contextmanager
|
||||
def monitor_sharpness(self):
|
||||
m = JPEGSharpnessMonitor(self.microscope)
|
||||
m.start()
|
||||
try:
|
||||
yield m
|
||||
finally:
|
||||
m.stop()
|
||||
|
||||
def move_and_find_focus(self, dz):
|
||||
"""Make a relative Z move and return the peak sharpness position"""
|
||||
with self.monitor_sharpness() as m:
|
||||
m.focus_rel(dz)
|
||||
return m.sharpest_z_on_move(0)
|
||||
|
||||
def fast_autofocus(self, dz=2000, backlash=None):
|
||||
"""Perform a down-up-down-up autofocus"""
|
||||
with self.monitor_sharpness() as m:
|
||||
i, z = m.focus_rel(-dz / 2)
|
||||
i, z = m.focus_rel(dz)
|
||||
fz = m.sharpest_z_on_move(i)
|
||||
if backlash is None:
|
||||
i, z = m.focus_rel(
|
||||
-dz
|
||||
) # move all the way to the start so it's consistent
|
||||
else:
|
||||
i, z = m.focus_rel(fz - z - backlash)
|
||||
m.focus_rel(fz - z)
|
||||
return m.data_dict()
|
||||
|
||||
def fast_up_down_up_autofocus(
|
||||
self, dz=2000, target_z=0, initial_move_up=True, mini_backlash=150
|
||||
):
|
||||
"""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.
|
||||
The sequence of moves it performs is:
|
||||
|
||||
1. Move to the top of the range `dz/2` (can be disabled)
|
||||
|
||||
2. Move down by `dz` while monitoring JPEG size to find the focus.
|
||||
|
||||
3. Move back up to the `target_z` position, relative to the sharpest image.
|
||||
|
||||
4. Measure the sharpness, and compare against the curve recorded in (2) to \\
|
||||
estimate how much further we need to go. Make this move, to reach our \\
|
||||
target position.
|
||||
|
||||
Moving back to the target position in two steps allows us to correct for
|
||||
backlash, by using the sharpness-vs-z curve as a rough encoder for Z.
|
||||
|
||||
Parameters:
|
||||
dz: number of steps over which to scan (optional, default 2000)
|
||||
|
||||
target_z: we aim to finish at this position, relative to focus. This may
|
||||
be useful if, for example, you want to acquire a stack of images in Z.
|
||||
It is optional, and the default value of 0 will finish at the focus.
|
||||
|
||||
initial_move_up: (optional, default True) set this to `False` to move down
|
||||
from the starting position. Mostly useful if you're able to combine
|
||||
the initial move with something else, e.g. moving to the next scan point.
|
||||
|
||||
mini_backlash: (optional, default 50) is a small extra move made in step
|
||||
3 to help counteract backlash. It should be small enough that you
|
||||
would always expect there to be greater backlash than this. Too small
|
||||
might slightly hurt accuracy, but is unlikely to be a big issue. Too big
|
||||
may cause you to overshoot, which is a problem.
|
||||
"""
|
||||
with self.monitor_sharpness() as m, self.microscope.camera.lock:
|
||||
# Ensure the MJPEG stream has started
|
||||
self.microscope.camera.start_stream_recording()
|
||||
|
||||
df = dz # TODO: refactor so I actually use dz in the code below!
|
||||
if initial_move_up:
|
||||
m.focus_rel(df / 2)
|
||||
# move down
|
||||
i, z = m.focus_rel(-df)
|
||||
# now inspect where the sharpest point is, and estimate the sharpness
|
||||
# (JPEG size) that we should find at the start of the Z stack
|
||||
jt, jz, js = m.move_data(i)
|
||||
best_z = jz[np.argmax(js)]
|
||||
target_s = np.interp(
|
||||
[best_z + target_z], jz[::-1], js[::-1]
|
||||
) # NB jz is decreasing
|
||||
|
||||
# now move to the start of the z stack
|
||||
i, z = m.focus_rel(
|
||||
best_z + target_z - z + mini_backlash
|
||||
) # takes us to the start of the stack
|
||||
|
||||
# We've deliberately undershot - figure out how much further we should move based on the curve
|
||||
current_js = m.jpeg_size()
|
||||
imax = 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
|
||||
) # use the curve we recorded to estimate our position
|
||||
# TODO: fancy interpolation stuff
|
||||
|
||||
# So, the Z position corresponding to our current sharpness value is zs[inow]
|
||||
# That means we should move forwards, by best_z - zs[inow]
|
||||
correction_move = best_z + target_z - jz[inow]
|
||||
logging.debug(
|
||||
"Fast autofocus scan: correcting backlash by moving {} steps".format(
|
||||
correction_move
|
||||
)
|
||||
)
|
||||
m.focus_rel(correction_move)
|
||||
return m.data_dict()
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
__all__ = ["Plugin"]
|
||||
from .plugin import AutocalibrationPlugin
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
from openflexure_microscope.devel import (
|
||||
MicroscopePlugin,
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
taskify,
|
||||
)
|
||||
|
||||
import logging
|
||||
|
||||
from .recalibrate_utils import recalibrate_camera, auto_expose_and_freeze_settings
|
||||
|
||||
|
||||
class RecalibrateAPIView(MicroscopeViewPlugin):
|
||||
def post(self):
|
||||
logging.info("Starting microscope recalibration...")
|
||||
task = taskify(self.plugin.recalibrate)()
|
||||
|
||||
# Return a handle on the autofocus task
|
||||
return jsonify(task.state), 201
|
||||
|
||||
|
||||
class AutocalibrationPlugin(MicroscopePlugin):
|
||||
"""
|
||||
Auto-calibration plugin
|
||||
"""
|
||||
|
||||
api_views = {"/recalibrate": RecalibrateAPIView}
|
||||
|
||||
def recalibrate(self):
|
||||
"""Reset the camera's settings.
|
||||
|
||||
This generates new gains, exposure time, and lens shading
|
||||
table such that the background is as uniform as possible
|
||||
with a gray level of 230. It takes a little while to run.
|
||||
"""
|
||||
scamera = self.microscope.camera
|
||||
with scamera.lock:
|
||||
assert not scamera.status[
|
||||
"record_active"
|
||||
], "Can't recalibrate while recording!"
|
||||
streaming = scamera.status["stream_active"]
|
||||
if streaming:
|
||||
logging.info("Stopping stream before recalibration")
|
||||
scamera.stop_stream_recording(resolution=(640, 480))
|
||||
old_resolution = scamera.camera.resolution
|
||||
try:
|
||||
scamera.camera.resolution = (640, 480)
|
||||
auto_expose_and_freeze_settings(scamera.camera)
|
||||
recalibrate_camera(scamera.camera)
|
||||
finally:
|
||||
scamera.camera.resolution = old_resolution
|
||||
self.microscope.save_settings()
|
||||
if streaming:
|
||||
logging.info("Restarting stream after recalibration")
|
||||
scamera.start_stream_recording()
|
||||
|
|
@ -1,191 +0,0 @@
|
|||
import numpy as np
|
||||
import time
|
||||
|
||||
from picamera import PiCamera
|
||||
from picamera.array import PiRGBArray, PiBayerArray
|
||||
|
||||
|
||||
def rgb_image(camera, resize=None, **kwargs):
|
||||
"""Capture an image and return an RGB numpy array"""
|
||||
with PiRGBArray(camera, size=resize) as output:
|
||||
camera.capture(output, format="rgb", resize=resize, **kwargs)
|
||||
return output.array
|
||||
|
||||
|
||||
def flat_lens_shading_table(camera):
|
||||
"""Return a flat (i.e. unity gain) lens shading table.
|
||||
|
||||
This is mostly useful because it makes it easy to get the size
|
||||
of the array correct. NB if you are not using the forked picamera
|
||||
library (with lens shading table support) it will raise an error.
|
||||
"""
|
||||
if not hasattr(PiCamera, "lens_shading_table"):
|
||||
raise ImportError(
|
||||
"This program requires the forked picamera library with lens shading support"
|
||||
)
|
||||
return np.zeros(camera._lens_shading_table_shape(), dtype=np.uint8) + 32
|
||||
|
||||
|
||||
def adjust_exposure_to_setpoint(camera, setpoint):
|
||||
"""Adjust the camera's exposure time until the maximum pixel value is <setpoint>."""
|
||||
print("Adjusting shutter speed to hit setpoint {}".format(setpoint), end="")
|
||||
for i in range(3):
|
||||
print(".", end="")
|
||||
camera.shutter_speed = int(
|
||||
camera.shutter_speed * setpoint / np.max(rgb_image(camera))
|
||||
)
|
||||
time.sleep(1)
|
||||
print("done")
|
||||
|
||||
|
||||
def auto_expose_and_freeze_settings(camera):
|
||||
"""Freeze the settings after auto-exposing to white illumination"""
|
||||
print("Allowing the camera to auto-expose")
|
||||
camera.awb_mode = "auto"
|
||||
camera.exposure_mode = "auto"
|
||||
camera.iso = (
|
||||
0
|
||||
) # This is important, if it's on a fixed ISO, gain might not set properly.
|
||||
for i in range(6):
|
||||
print(".", end="")
|
||||
time.sleep(0.5)
|
||||
print("done")
|
||||
|
||||
print("Freezing the camera settings...")
|
||||
camera.shutter_speed = camera.exposure_speed
|
||||
print("Shutter speed = {}".format(camera.shutter_speed))
|
||||
camera.exposure_mode = "off"
|
||||
print("Auto exposure disabled")
|
||||
g = camera.awb_gains
|
||||
camera.awb_mode = "off"
|
||||
camera.awb_gains = g
|
||||
print("Auto white balance disabled, gains are {}".format(g))
|
||||
print(
|
||||
"Analogue gain: {}, Digital gain: {}".format(
|
||||
camera.analog_gain, camera.digital_gain
|
||||
)
|
||||
)
|
||||
adjust_exposure_to_setpoint(camera, 215)
|
||||
|
||||
|
||||
def channels_from_bayer_array(bayer_array):
|
||||
"""Given the 'array' from a PiBayerArray, return the 4 channels."""
|
||||
bayer_pattern = [(i // 2, i % 2) for i in range(4)]
|
||||
channels = np.zeros(
|
||||
(4, bayer_array.shape[0] // 2, bayer_array.shape[1] // 2),
|
||||
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(
|
||||
bayer_array[offset[0] :: 2, offset[1] :: 2, :], axis=2
|
||||
)
|
||||
|
||||
return channels
|
||||
|
||||
|
||||
def lst_from_channels(channels):
|
||||
"""Given the 4 Bayer colour channels from a white image, generate a LST."""
|
||||
full_resolution = np.array(channels.shape[1:]) * 2 # channels have been binned
|
||||
# lst_resolution = list(np.ceil(full_resolution / 64.0).astype(int))
|
||||
lst_resolution = [(r // 64) + 1 for r in full_resolution]
|
||||
# NB the size of the LST is 1/64th of the image, but rounded UP.
|
||||
print("Generating a lens shading table at {}x{}".format(*lst_resolution))
|
||||
lens_shading = np.zeros([channels.shape[0]] + lst_resolution, dtype=np.float)
|
||||
for i in range(lens_shading.shape[0]):
|
||||
image_channel = channels[i, :, :]
|
||||
iw, ih = image_channel.shape
|
||||
ls_channel = lens_shading[i, :, :]
|
||||
lw, lh = ls_channel.shape
|
||||
# The lens shading table is rounded **up** in size to 1/64th of the size of
|
||||
# the image. Rather than handle edge images separately, I'm just going to
|
||||
# pad the image by copying edge pixels, so that it is exactly 32 times the
|
||||
# size of the lens shading table (NB 32 not 64 because each channel is only
|
||||
# half the size of the full image - remember the Bayer pattern... This
|
||||
# should give results very close to 6by9's solution, albeit considerably
|
||||
# less computationally efficient!
|
||||
padded_image_channel = np.pad(
|
||||
image_channel, [(0, lw * 32 - iw), (0, lh * 32 - ih)], mode="edge"
|
||||
) # Pad image to the right and bottom
|
||||
print(
|
||||
"Channel shape: {}x{}, shading table shape: {}x{}, after padding {}".format(
|
||||
iw, ih, lw * 32, lh * 32, padded_image_channel.shape
|
||||
)
|
||||
)
|
||||
# Next, fill the shading table (except edge pixels). Please excuse the
|
||||
# for loop - I know it's not fast but this code needn't be!
|
||||
box = 3 # We average together a square of this side length for each pixel.
|
||||
# NB this isn't quite what 6by9's program does - it averages 3 pixels
|
||||
# horizontally, but not vertically.
|
||||
for dx in np.arange(box) - box // 2:
|
||||
for dy in np.arange(box) - box // 2:
|
||||
ls_channel[:, :] += (
|
||||
padded_image_channel[16 + dx :: 32, 16 + dy :: 32] - 64
|
||||
)
|
||||
ls_channel /= box ** 2
|
||||
# The original C code written by 6by9 normalises to the central 64 pixels in each channel.
|
||||
# ls_channel /= np.mean(image_channel[iw//2-4:iw//2+4, ih//2-4:ih//2+4])
|
||||
# I have had better results just normalising to the maximum:
|
||||
ls_channel /= np.max(ls_channel)
|
||||
# NB the central pixel should now be *approximately* 1.0 (may not be exactly
|
||||
# due to different averaging widths between the normalisation & shading table)
|
||||
# For most sensible lenses I'd expect that 1.0 is the maximum value.
|
||||
# NB ls_channel should be a "view" of the whole lens shading array, so we don't
|
||||
# need to update the big array here.
|
||||
|
||||
# What we actually want to calculate is the gains needed to compensate for the
|
||||
# lens shading - that's 1/lens_shading_table_float as we currently have it.
|
||||
gains = 32.0 / lens_shading # 32 is unity gain
|
||||
gains[gains > 255] = 255 # clip at 255, maximum gain is 255/32
|
||||
gains[gains < 32] = 32 # clip at 32, minimum gain is 1 (is this necessary?)
|
||||
lens_shading_table = gains.astype(np.uint8)
|
||||
return lens_shading_table[::-1, :, :].copy()
|
||||
|
||||
|
||||
def recalibrate_camera(camera):
|
||||
"""Reset the lens shading table and exposure settings.
|
||||
|
||||
This method first resets to a flat lens shading table, then auto-exposes,
|
||||
then generates a new lens shading table to make the current view uniform.
|
||||
It should be run when the camera is looking at a uniform white scene.
|
||||
|
||||
NB the only parameter ``camera`` is a ``PiCamera`` instance and **not** a
|
||||
``StreamingCamera``.
|
||||
"""
|
||||
camera.lens_shading_table = flat_lens_shading_table(camera)
|
||||
_ = rgb_image(camera) # for some reason the camera won't work unless I do this!
|
||||
|
||||
with PiBayerArray(camera) as a:
|
||||
camera.capture(a, format="jpeg", bayer=True)
|
||||
raw_image = a.array.copy()
|
||||
|
||||
# Now we need to calculate a lens shading table that would make this flat.
|
||||
# raw_image is a 3D array, with full resolution and 3 colour channels. No
|
||||
# de-mosaicing has been done, so 2/3 of the values are zero (3/4 for R and B
|
||||
# channels, 1/2 for green because there's twice as many green pixels).
|
||||
channels = channels_from_bayer_array(raw_image)
|
||||
lens_shading_table = lst_from_channels(channels)
|
||||
|
||||
camera.lens_shading_table = lens_shading_table
|
||||
_ = rgb_image(camera)
|
||||
|
||||
# Fix the AWB gains so the image is neutral
|
||||
channel_means = np.mean(np.mean(rgb_image(camera), axis=0, dtype=np.float), axis=0)
|
||||
old_gains = camera.awb_gains
|
||||
camera.awb_gains = (
|
||||
channel_means[1] / channel_means[0] * old_gains[0],
|
||||
channel_means[1] / channel_means[2] * old_gains[1],
|
||||
)
|
||||
time.sleep(1)
|
||||
# Ensure the background is bright but not saturated
|
||||
adjust_exposure_to_setpoint(camera, 230)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
with PiCamera() as camera:
|
||||
camera.start_preview()
|
||||
time.sleep(3)
|
||||
print("Recalibrating...")
|
||||
recalibrate_camera(camera)
|
||||
print("Done.")
|
||||
time.sleep(2)
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
__all__ = ["ScanPlugin"]
|
||||
from .plugin import ScanPlugin
|
||||
|
|
@ -1,63 +0,0 @@
|
|||
from openflexure_microscope.devel import (
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
abort,
|
||||
taskify,
|
||||
)
|
||||
|
||||
import logging
|
||||
|
||||
|
||||
class TileScanAPI(MicroscopeViewPlugin):
|
||||
def post(self):
|
||||
payload = JsonResponse(request)
|
||||
|
||||
# Get params
|
||||
filename = payload.param("filename")
|
||||
temporary = payload.param("temporary", default=False, convert=bool)
|
||||
|
||||
step_size = payload.param("step_size", default=[2000, 1500, 100], convert=list)
|
||||
step_size = [int(i) for i in step_size]
|
||||
|
||||
grid = payload.param("grid", default=[3, 3, 5], convert=list)
|
||||
grid = [int(i) for i in grid]
|
||||
|
||||
style = payload.param("style", default="raster", convert=str)
|
||||
autofocus_dz = payload.param("autofocus_dz", default=50, convert=int)
|
||||
fast_autofocus = payload.param("fast_autofocus", default=False, convert=bool)
|
||||
|
||||
use_video_port = payload.param("use_video_port", default=True, convert=bool)
|
||||
resize = payload.param("size", default=None)
|
||||
if resize:
|
||||
if ("width" in resize) and ("height" in resize):
|
||||
resize = (
|
||||
int(resize["width"]),
|
||||
int(resize["height"]),
|
||||
) # Convert dict to tuple
|
||||
else:
|
||||
abort(404)
|
||||
|
||||
bayer = payload.param("bayer", default=False, convert=bool)
|
||||
metadata = payload.param("metadata", default={}, convert=dict)
|
||||
tags = payload.param("tags", default=[], convert=list)
|
||||
|
||||
logging.info("Running tile scan...")
|
||||
task = taskify(self.plugin.tile)(
|
||||
basename=filename,
|
||||
temporary=temporary,
|
||||
step_size=step_size,
|
||||
grid=grid,
|
||||
style=style,
|
||||
autofocus_dz=autofocus_dz,
|
||||
use_video_port=use_video_port,
|
||||
resize=resize,
|
||||
bayer=bayer,
|
||||
fast_autofocus=fast_autofocus,
|
||||
metadata=metadata,
|
||||
tags=tags,
|
||||
)
|
||||
|
||||
# return a handle on the scan task
|
||||
return jsonify(task.state), 201
|
||||
|
|
@ -1,328 +0,0 @@
|
|||
import time
|
||||
from typing import Tuple
|
||||
from functools import reduce
|
||||
import uuid
|
||||
import itertools
|
||||
import logging
|
||||
|
||||
from openflexure_microscope.camera.base import generate_basename
|
||||
|
||||
from openflexure_microscope.devel import (
|
||||
MicroscopePlugin,
|
||||
update_task_progress,
|
||||
update_task_data,
|
||||
)
|
||||
|
||||
from .api import TileScanAPI
|
||||
|
||||
|
||||
def construct_grid(initial, step_sizes, n_steps, style="raster"):
|
||||
"""
|
||||
Given an initial position, step sizes, and number of steps,
|
||||
construct a 2-dimensional list of scan x-y positions.
|
||||
"""
|
||||
arr = []
|
||||
|
||||
for i in range(n_steps[0]): # x axis
|
||||
arr.append([])
|
||||
for j in range(n_steps[1]): # y axis
|
||||
# Create a coordinate array
|
||||
coord = [initial[ax] + [i, j][ax] * step_sizes[ax] for ax in range(2)]
|
||||
# Append coordinate array to position grid
|
||||
arr[i].append(tuple(coord))
|
||||
|
||||
# Style modifiers
|
||||
if style == "snake":
|
||||
for i, line in enumerate(arr):
|
||||
if i % 2 != 0:
|
||||
line.reverse()
|
||||
|
||||
return arr
|
||||
|
||||
|
||||
def flatten_grid(grid):
|
||||
"""
|
||||
Convert a 3D list of scan positions into a flat list
|
||||
of sequential positions.
|
||||
"""
|
||||
|
||||
grid = list(itertools.chain(*grid))
|
||||
return grid
|
||||
|
||||
|
||||
class ScanPlugin(MicroscopePlugin):
|
||||
"""
|
||||
Stack and tile plugin
|
||||
"""
|
||||
|
||||
api_views = {"/tile": TileScanAPI}
|
||||
|
||||
def __init__(self):
|
||||
MicroscopePlugin.__init__(self)
|
||||
|
||||
self.images_to_be_captured: int = 1
|
||||
update_task_data({"images_to_be_captured": self.images_to_be_captured})
|
||||
|
||||
@property
|
||||
def progress(self):
|
||||
progress = (self.images_captured_so_far / self.images_to_be_captured) * 100
|
||||
logging.info(progress)
|
||||
return progress
|
||||
|
||||
def capture(
|
||||
self,
|
||||
basename,
|
||||
scan_id,
|
||||
temporary: bool = False,
|
||||
use_video_port: bool = False,
|
||||
resize: Tuple[int, int] = None,
|
||||
bayer: bool = False,
|
||||
metadata: dict = {},
|
||||
tags: list = [],
|
||||
):
|
||||
|
||||
# Construct a tile filename
|
||||
filename = "{}_{}_{}_{}".format(basename, *self.microscope.stage.position)
|
||||
folder = "SCAN_{}".format(basename)
|
||||
|
||||
# Create output object
|
||||
output = self.microscope.camera.new_image(
|
||||
temporary=temporary, filename=filename, folder=folder
|
||||
)
|
||||
|
||||
# Capture
|
||||
self.microscope.camera.capture(
|
||||
output.file, use_video_port=use_video_port, resize=resize, bayer=bayer
|
||||
)
|
||||
|
||||
# Affix metadata
|
||||
if "scan" not in tags:
|
||||
tags.append("scan")
|
||||
|
||||
# Inject system metadata
|
||||
output.put_metadata(self.microscope.metadata, system=True)
|
||||
|
||||
# Insert custom metadata
|
||||
output.put_metadata(metadata)
|
||||
|
||||
# Insert custom tags
|
||||
output.put_tags(tags)
|
||||
|
||||
def tile(
|
||||
self,
|
||||
basename: str = None,
|
||||
temporary: bool = False,
|
||||
step_size: int = [2000, 1500, 100],
|
||||
grid: list = [3, 3, 5],
|
||||
style="raster",
|
||||
autofocus_dz: int = 50,
|
||||
use_video_port: bool = False,
|
||||
resize: Tuple[int, int] = None,
|
||||
bayer: bool = False,
|
||||
fast_autofocus=False,
|
||||
metadata: dict = {},
|
||||
tags: list = [],
|
||||
):
|
||||
|
||||
# Keep task progress
|
||||
# TODO: Make this line not nasty
|
||||
self.images_to_be_captured = reduce((lambda x, y: x * y), grid)
|
||||
self.images_captured_so_far = 0
|
||||
|
||||
# Generate a basename if none given
|
||||
if not basename:
|
||||
basename = generate_basename()
|
||||
|
||||
# Generate a stack ID
|
||||
scan_id = uuid.uuid4()
|
||||
|
||||
# Store initial position
|
||||
initial_position = self.microscope.stage.position
|
||||
|
||||
# Add scan metadata
|
||||
if "time" not in metadata:
|
||||
metadata["time"] = generate_basename()
|
||||
|
||||
metadata.update(
|
||||
{
|
||||
"scan_id": scan_id,
|
||||
"basename": basename,
|
||||
"scan_parameters": {
|
||||
"step_size": step_size,
|
||||
"grid": grid,
|
||||
"style": style,
|
||||
"autofocus_dz": autofocus_dz,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
# Check if autofocus is enabled
|
||||
if (
|
||||
autofocus_dz
|
||||
and hasattr(self.microscope.plugins, "default_autofocus")
|
||||
and self.microscope.has_real_stage()
|
||||
and self.microscope.has_real_camera()
|
||||
):
|
||||
autofocus_enabled = True
|
||||
else:
|
||||
autofocus_enabled = False
|
||||
|
||||
if fast_autofocus and not hasattr(
|
||||
self.microscope.plugins.default_autofocus, "monitor_sharpness"
|
||||
):
|
||||
logging.warning(
|
||||
"Can't use fast autofocus in the scan - the default plugin doesn't support monitor_sharpness; maybe it is too old?"
|
||||
)
|
||||
fast_autofocus = False
|
||||
z_stack_dz = (
|
||||
grid[2] * step_size[2] if grid[2] > 1 else 0
|
||||
) # shorthand for Z stack range
|
||||
|
||||
# Construct an x-y grid (worry about z later)
|
||||
x_y_grid = construct_grid(
|
||||
initial_position, step_size[:2], grid[:2], style=style
|
||||
)
|
||||
|
||||
# Keep the initial Z position the same as our current position
|
||||
next_z = initial_position[2]
|
||||
if fast_autofocus: # If fast autofocus is enabled, make
|
||||
next_z += autofocus_dz / 2 # sure we start from the top of the range
|
||||
initial_z = next_z # Save this value for use in raster scans
|
||||
|
||||
# Now step through each point in the x-y coordinate array
|
||||
for line in x_y_grid:
|
||||
# If rastering, rather than snake (or eventually spiral)
|
||||
# Return focus to initial position
|
||||
if style == "raster":
|
||||
next_z = initial_z
|
||||
logging.debug("Returning to initial z position")
|
||||
self.microscope.stage.move_abs(
|
||||
[line[0][0], line[0][1], next_z]
|
||||
) # RWB: I think this line is redundant
|
||||
|
||||
for x_y in line:
|
||||
# Move to new grid position without changing z
|
||||
logging.debug("Moving to step {}".format([x_y[0], x_y[1], next_z]))
|
||||
self.microscope.stage.move_abs([x_y[0], x_y[1], next_z])
|
||||
# Refocus
|
||||
if autofocus_enabled:
|
||||
if fast_autofocus:
|
||||
self.microscope.plugins.default_autofocus.fast_up_down_up_autofocus(
|
||||
dz=autofocus_dz,
|
||||
target_z=-z_stack_dz / 2.0, # Finish below the focus
|
||||
initial_move_up=False, # We're already at the top of the scan
|
||||
)
|
||||
# TODO: save the focus data for future reference? Use it for diagnostics?
|
||||
else:
|
||||
logging.debug("Running autofocus")
|
||||
self.microscope.plugins.default_autofocus.autofocus(
|
||||
range(-3 * autofocus_dz, 4 * autofocus_dz, autofocus_dz)
|
||||
)
|
||||
logging.debug("Finished autofocus")
|
||||
time.sleep(1) # TODO: Remove
|
||||
|
||||
# If we're not doing a z-stack, just capture
|
||||
if grid[2] <= 1:
|
||||
self.capture(
|
||||
basename,
|
||||
scan_id,
|
||||
temporary=temporary,
|
||||
use_video_port=use_video_port,
|
||||
resize=resize,
|
||||
bayer=bayer,
|
||||
metadata=metadata,
|
||||
tags=tags,
|
||||
)
|
||||
# Update task progress
|
||||
self.images_captured_so_far += 1
|
||||
update_task_progress(self.progress)
|
||||
else:
|
||||
logging.debug("Entering z-stack")
|
||||
self.stack(
|
||||
basename=basename,
|
||||
temporary=temporary,
|
||||
scan_id=scan_id,
|
||||
step_size=step_size[2],
|
||||
steps=grid[2],
|
||||
center=not fast_autofocus, # fast_autofocus does this for us!
|
||||
return_to_start=not fast_autofocus,
|
||||
use_video_port=use_video_port,
|
||||
resize=resize,
|
||||
bayer=bayer,
|
||||
metadata=metadata,
|
||||
tags=tags,
|
||||
)
|
||||
# Make sure we use our current best estimate of focus (i.e. the current position) next point
|
||||
next_z = self.microscope.stage.position[2]
|
||||
if fast_autofocus:
|
||||
next_z += (
|
||||
autofocus_dz / 2
|
||||
) # Fast autofocus requires us to start at the top of the range
|
||||
if grid[2] > 1:
|
||||
next_z -= int(
|
||||
grid[2] / 2.0 * step_size[2]
|
||||
) # Z stacking means we're higher up to start with
|
||||
|
||||
logging.debug("Returning to {}".format(initial_position))
|
||||
self.microscope.stage.move_abs(initial_position)
|
||||
|
||||
def stack(
|
||||
self,
|
||||
basename: str = None,
|
||||
temporary: bool = False,
|
||||
scan_id: str = None,
|
||||
step_size: int = 100,
|
||||
steps: int = 5,
|
||||
center: bool = True,
|
||||
return_to_start: bool = True,
|
||||
use_video_port: bool = False,
|
||||
resize: Tuple[int, int] = None,
|
||||
bayer: bool = False,
|
||||
metadata: dict = {},
|
||||
tags: list = [],
|
||||
):
|
||||
|
||||
# Generate a basename if none given
|
||||
if not basename:
|
||||
basename = generate_basename()
|
||||
|
||||
# Generate a stack ID
|
||||
if not scan_id:
|
||||
scan_id = uuid.uuid4()
|
||||
|
||||
# Add scan metadata
|
||||
if not "time" in metadata:
|
||||
metadata["time"] = generate_basename()
|
||||
|
||||
# Store initial position
|
||||
initial_position = self.microscope.stage.position
|
||||
|
||||
with self.microscope.lock:
|
||||
# Move to center scan
|
||||
if center:
|
||||
logging.debug("Moving to starting position")
|
||||
self.microscope.stage.move_rel([0, 0, int((-step_size * steps) / 2)])
|
||||
|
||||
for i in range(steps):
|
||||
time.sleep(0.1)
|
||||
logging.debug("Capturing...")
|
||||
self.capture(
|
||||
basename,
|
||||
scan_id,
|
||||
temporary=temporary,
|
||||
use_video_port=use_video_port,
|
||||
resize=resize,
|
||||
bayer=bayer,
|
||||
metadata=metadata,
|
||||
tags=tags,
|
||||
)
|
||||
# Update task progress
|
||||
self.images_captured_so_far += 1
|
||||
update_task_progress(self.progress)
|
||||
|
||||
if i != steps - 1:
|
||||
logging.debug("Moving z by {}".format(step_size))
|
||||
self.microscope.stage.move_rel([0, 0, step_size])
|
||||
if return_to_start:
|
||||
logging.debug("Returning to {}".format(initial_position))
|
||||
self.microscope.stage.move_abs(initial_position)
|
||||
|
|
@ -1 +0,0 @@
|
|||
from .plugin import ZipBuilderPlugin
|
||||
|
|
@ -1,138 +0,0 @@
|
|||
from openflexure_microscope.devel import (
|
||||
MicroscopePlugin,
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
taskify,
|
||||
update_task_progress,
|
||||
)
|
||||
|
||||
from flask import send_file, abort
|
||||
|
||||
import uuid
|
||||
import os
|
||||
import zipfile
|
||||
import tempfile
|
||||
import logging
|
||||
|
||||
|
||||
class ZipBuilderAPIView(MicroscopeViewPlugin):
|
||||
def post(self):
|
||||
|
||||
ids = list(JsonResponse(request).json)
|
||||
|
||||
task = taskify(self.plugin.build_zip_from_capture_ids)(ids)
|
||||
|
||||
# Return a handle on the autofocus task
|
||||
return jsonify(task.state), 201
|
||||
|
||||
|
||||
class ZipListAPIView(MicroscopeViewPlugin):
|
||||
def get(self):
|
||||
return jsonify(self.plugin.session_zips)
|
||||
|
||||
|
||||
class ZipGetterAPIView(MicroscopeViewPlugin):
|
||||
def get(self, session_id):
|
||||
if not session_id in self.plugin.session_zips:
|
||||
return abort(404) # 404 Not Found
|
||||
|
||||
logging.info(f"Session ID: {session_id}")
|
||||
|
||||
return send_file(
|
||||
self.plugin.zip_from_id(session_id).name,
|
||||
mimetype="application/zip",
|
||||
as_attachment=True,
|
||||
attachment_filename=f"{session_id}.zip",
|
||||
)
|
||||
|
||||
def delete(self, session_id):
|
||||
if not session_id in self.plugin.session_zips:
|
||||
return abort(404) # 404 Not Found
|
||||
|
||||
logging.info(f"Session ID: {session_id}")
|
||||
|
||||
fp = self.plugin.zip_from_id(session_id)
|
||||
logging.debug(fp.name)
|
||||
fp.close()
|
||||
os.unlink(fp.name)
|
||||
|
||||
assert not os.path.exists(fp.name)
|
||||
|
||||
del self.plugin.session_zips[session_id]
|
||||
|
||||
return jsonify({"return": session_id})
|
||||
|
||||
|
||||
class ZipBuilderPlugin(MicroscopePlugin):
|
||||
"""
|
||||
ZIP-builder plugin
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
self.session_zips = {}
|
||||
|
||||
self.add_view("/get/<string:session_id>", ZipGetterAPIView)
|
||||
self.add_view("/get", ZipListAPIView)
|
||||
|
||||
self.add_view("/build", ZipBuilderAPIView)
|
||||
|
||||
def build_zip_from_capture_ids(self, capture_id_list):
|
||||
logging.debug(capture_id_list)
|
||||
|
||||
# Get array of captures from IDs
|
||||
capture_list = [
|
||||
self.microscope.camera.image_from_id(capture_id)
|
||||
for capture_id in capture_id_list
|
||||
]
|
||||
# Remove Nones from list (missing/invalid captures)
|
||||
capture_list = [capture for capture in capture_list if capture]
|
||||
|
||||
# Get size (in bytes) of each capture
|
||||
capture_sizes = [
|
||||
os.path.getsize(capture_obj.file) for capture_obj in capture_list
|
||||
]
|
||||
# Calculate size of input data in megabytes
|
||||
data_size_megabytes = sum(capture_sizes) * 1e-6
|
||||
|
||||
# If more than 1GB
|
||||
if data_size_megabytes > 1000:
|
||||
# Throw exception
|
||||
raise Exception(
|
||||
"Zip data cannot exceed 1GB. Please transfer data manually."
|
||||
)
|
||||
|
||||
# Number of files to add (used for task progress)
|
||||
n_files = len(capture_id_list)
|
||||
|
||||
# Create temporary file
|
||||
fp = tempfile.NamedTemporaryFile(delete=False)
|
||||
|
||||
# Open temp file as a ZIP file
|
||||
with zipfile.ZipFile(fp, "w") as zipObj:
|
||||
for index, capture_obj in enumerate(capture_list):
|
||||
# Add to ZIP file if it exists
|
||||
file_path = capture_obj.file
|
||||
rel_path = os.path.relpath(
|
||||
file_path, self.microscope.camera.paths["default"]
|
||||
)
|
||||
zipObj.write(file_path, arcname=rel_path)
|
||||
# Update task progress
|
||||
update_task_progress(int((index / n_files) * 100))
|
||||
|
||||
session_id = uuid.uuid4()
|
||||
# self.session_zips[session_id] = fp
|
||||
self.session_zips[session_id] = {
|
||||
"id": session_id,
|
||||
"fp": fp,
|
||||
"data_size": data_size_megabytes,
|
||||
"zip_size": os.path.getsize(fp.name) * 1e-6,
|
||||
}
|
||||
|
||||
return self.session_zips[session_id]
|
||||
|
||||
def zip_from_id(self, session_id):
|
||||
return self.session_zips[session_id]["fp"]
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
from .plugin import DynamicExamplePlugin
|
||||
from . import api
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
from openflexure_microscope.devel import (
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
taskify,
|
||||
)
|
||||
|
||||
import logging
|
||||
|
||||
|
||||
class DoAPI(MicroscopeViewPlugin):
|
||||
"""
|
||||
A dynamic example API plugin
|
||||
"""
|
||||
|
||||
def get(self):
|
||||
values = {"val_int": self.plugin.val_int}
|
||||
return jsonify(values)
|
||||
|
||||
def post(self):
|
||||
self.plugin.val_int += 1
|
||||
|
||||
return jsonify({"response": "completed"})
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
import random
|
||||
import time
|
||||
import os
|
||||
import json
|
||||
from openflexure_microscope.devel import MicroscopePlugin, update_task_progress
|
||||
|
||||
from .api import DoAPI
|
||||
|
||||
|
||||
class DynamicExamplePlugin(MicroscopePlugin):
|
||||
"""
|
||||
An example plugin using a comprehensive form
|
||||
"""
|
||||
|
||||
api_views = {"/do": DoAPI}
|
||||
|
||||
def __init__(self):
|
||||
MicroscopePlugin.__init__(self)
|
||||
|
||||
self.val_int = 0
|
||||
self.val_str = "Hello"
|
||||
|
||||
self.set_gui(self.dynamic_form)
|
||||
|
||||
def dynamic_form(self):
|
||||
return {
|
||||
"id": "test-plugin",
|
||||
"icon": "pets",
|
||||
"forms": [
|
||||
{
|
||||
"name": "Simple request",
|
||||
"isCollapsible": False,
|
||||
"isTask": False,
|
||||
"selfUpdate": True,
|
||||
"route": "/do",
|
||||
"submitLabel": "Do things",
|
||||
"schema": [
|
||||
{
|
||||
"fieldType": "numberInput",
|
||||
"name": "val_int",
|
||||
"label": "Number value",
|
||||
"minValue": 0,
|
||||
},
|
||||
{
|
||||
"fieldType": "htmlBlock",
|
||||
"name": "html_block",
|
||||
"content": f"<i>Value is: </i><br>{self.val_int}.",
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
from .plugin import ExamplePlugin
|
||||
from . import api
|
||||
|
|
@ -1,60 +0,0 @@
|
|||
from openflexure_microscope.devel import (
|
||||
MicroscopeViewPlugin,
|
||||
JsonResponse,
|
||||
request,
|
||||
jsonify,
|
||||
taskify,
|
||||
)
|
||||
|
||||
import logging
|
||||
|
||||
|
||||
class DoAPI(MicroscopeViewPlugin):
|
||||
"""
|
||||
A simple example API plugin
|
||||
"""
|
||||
|
||||
def get(self):
|
||||
return jsonify(self.plugin.get_values_dict())
|
||||
|
||||
def post(self):
|
||||
# Get payload JSON
|
||||
payload = JsonResponse(request)
|
||||
|
||||
# Extract a values from the JSON payload.
|
||||
val_int = payload.param("val_int", default=None, convert=int)
|
||||
val_str = payload.param("val_str", default=None, convert=str)
|
||||
val_radio = payload.param("val_radio", default=None)
|
||||
val_check = payload.param("val_check", default=None)
|
||||
val_select = payload.param("val_select", default=None)
|
||||
val_disposable = payload.param("val_disposable", default=None)
|
||||
|
||||
self.plugin.set_values(
|
||||
val_int, val_str, val_radio, val_check, val_select, val_disposable
|
||||
)
|
||||
|
||||
print(self.plugin.get_values_dict())
|
||||
|
||||
return jsonify({"response": "completed"})
|
||||
|
||||
|
||||
class TaskAPI(MicroscopeViewPlugin):
|
||||
"""
|
||||
A task example API plugin
|
||||
"""
|
||||
|
||||
def get(self):
|
||||
return jsonify({"run_time": self.plugin.run_time})
|
||||
|
||||
def post(self):
|
||||
# Get payload JSON
|
||||
payload = JsonResponse(request)
|
||||
|
||||
# Extract a values from the JSON payload.
|
||||
val_int = payload.param("run_time", default=5, convert=int)
|
||||
|
||||
logging.info("Running task...")
|
||||
task = taskify(self.plugin.generate_random_numbers_for_a_while)(val_int)
|
||||
|
||||
# return a handle on the autofocus task
|
||||
return jsonify(task.state), 201
|
||||
|
|
@ -1,73 +0,0 @@
|
|||
{
|
||||
"id": "test-plugin",
|
||||
"icon": "pets",
|
||||
"forms": [{
|
||||
"name": "Simple request",
|
||||
"isCollapsible": false,
|
||||
"isTask": false,
|
||||
"selfUpdate": true,
|
||||
"route": "/do",
|
||||
"submitLabel": "Do things",
|
||||
"schema": [
|
||||
[{
|
||||
"fieldType": "numberInput",
|
||||
"name": "val_int",
|
||||
"label": "Number value",
|
||||
"minValue": 0
|
||||
},
|
||||
{
|
||||
"fieldType": "textInput",
|
||||
"placeholder": "Some string",
|
||||
"label": "String value",
|
||||
"name": "val_str"
|
||||
}
|
||||
],
|
||||
[{
|
||||
"fieldType": "radioList",
|
||||
"name": "val_radio",
|
||||
"label": "Radio value",
|
||||
"options": ["First", "Second", "Third"]
|
||||
},
|
||||
{
|
||||
"fieldType": "checkList",
|
||||
"name": "val_check",
|
||||
"label": "Checklist values",
|
||||
"options": ["Foo", "Bar", "Baz"]
|
||||
}
|
||||
],
|
||||
{
|
||||
"fieldType": "htmlBlock",
|
||||
"name": "html_block",
|
||||
"content": "<i>This is a block of HTML in a plugin!</i><br>I can do paragraph breaks and stuff."
|
||||
},
|
||||
{
|
||||
"fieldType": "selectList",
|
||||
"name": "val_select",
|
||||
"multi": false,
|
||||
"label": "Some selection",
|
||||
"options": ["Most", "Average", "Least"]
|
||||
},
|
||||
|
||||
{
|
||||
"fieldType": "textInput",
|
||||
"label": "Non-persistent string",
|
||||
"default": "A default value",
|
||||
"name": "val_disposable"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "Task form",
|
||||
"isTask": true,
|
||||
"selfUpdate": true,
|
||||
"route": "/task",
|
||||
"submitLabel": "Start task",
|
||||
"schema": [{
|
||||
"fieldType": "numberInput",
|
||||
"name": "run_time",
|
||||
"label": "Run time (seconds)",
|
||||
"minValue": 1
|
||||
}]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -1,80 +0,0 @@
|
|||
import random
|
||||
import time
|
||||
import os
|
||||
import json
|
||||
from openflexure_microscope.devel import MicroscopePlugin, update_task_progress
|
||||
|
||||
from .api import DoAPI, TaskAPI
|
||||
|
||||
HERE = os.path.dirname(os.path.realpath(__file__))
|
||||
FORM_PATH = os.path.join(HERE, "forms.json")
|
||||
|
||||
|
||||
class ExamplePlugin(MicroscopePlugin):
|
||||
"""
|
||||
An example plugin using a comprehensive form
|
||||
"""
|
||||
|
||||
global FORM_PATH
|
||||
|
||||
with open(FORM_PATH, "r") as sc:
|
||||
api_form = json.load(sc)
|
||||
|
||||
api_views = {"/do": DoAPI, "/task": TaskAPI}
|
||||
|
||||
def __init__(self):
|
||||
MicroscopePlugin.__init__(self)
|
||||
|
||||
self.val_int = 10
|
||||
self.val_str = "Hello"
|
||||
self.val_radio = "First"
|
||||
self.val_check = ["Foo", "Bar"]
|
||||
self.val_select = "Most"
|
||||
self.val_unused = "I'm an unused string, here to confuse the form parsing"
|
||||
|
||||
self.run_time = 5
|
||||
|
||||
def set_values(
|
||||
self, val_int, val_str, val_radio, val_check, val_select, val_disposable
|
||||
):
|
||||
"""
|
||||
Demonstrate a plugin with form
|
||||
"""
|
||||
if val_int:
|
||||
self.val_int = int(val_int)
|
||||
if val_str:
|
||||
self.val_str = str(val_str)
|
||||
if val_radio:
|
||||
self.val_radio = val_radio
|
||||
if val_check is not None:
|
||||
print(val_check)
|
||||
self.val_check = val_check if (type(val_check) is list) else [val_check]
|
||||
if val_select:
|
||||
self.val_select = val_select
|
||||
|
||||
if val_disposable:
|
||||
print("DISPOSABLE VALUE: {}".format(val_disposable))
|
||||
|
||||
def get_values_dict(self):
|
||||
return {
|
||||
"val_int": self.val_int,
|
||||
"val_str": self.val_str,
|
||||
"val_radio": self.val_radio,
|
||||
"val_check": self.val_check,
|
||||
"val_select": self.val_select,
|
||||
"val_unused": self.val_unused,
|
||||
}
|
||||
|
||||
def generate_random_numbers_for_a_while(self, run_time: int):
|
||||
self.run_time = run_time
|
||||
vals = []
|
||||
|
||||
for t in range(run_time):
|
||||
vals.append(random.random())
|
||||
time.sleep(1)
|
||||
|
||||
# Update task progress (if running as a task)
|
||||
percent_complete = int(((t + 1) / run_time) * 100)
|
||||
update_task_progress(percent_complete)
|
||||
|
||||
return vals
|
||||
Loading…
Add table
Add a link
Reference in a new issue