Completes all necessary movements along a single axis
This commit is contained in:
Chish36
Julian Stirling
parent
3d3ca9ebcf
commit
eb01aaf017
1 changed files with 188 additions and 0 deletions
188
src/openflexure_microscope_server/things/stage_measure.py
Normal file
188
src/openflexure_microscope_server/things/stage_measure.py
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
import numpy as np
|
||||
import logging
|
||||
import cv2
|
||||
import json
|
||||
from PIL import Image
|
||||
import time
|
||||
from typing import Annotated, Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Sequence, Tuple
|
||||
from scipy.optimize import curve_fit
|
||||
from camera_stage_mapping import camera_stage_tracker
|
||||
from camera_stage_mapping import fft_image_tracking
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from labthings_fastapi.thing import Thing
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from labthings_fastapi.dependencies.invocation import CancelHook, InvocationLogger, InvocationCancelledError
|
||||
from labthings_fastapi.decorators import thing_action, thing_property
|
||||
from .stage import StageDependency as StageDep
|
||||
from labthings_sangaboard import SangaboardThing
|
||||
from labthings_picamera2.thing import StreamingPiCamera2
|
||||
from labthings_fastapi.types.numpy import NDArray, denumpify, DenumpifyingDict
|
||||
from openflexure_microscope_server.things.autofocus import AutofocusThing
|
||||
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
|
||||
|
||||
StageDep = direct_thing_client_dependency(SangaboardThing, "/stage/")
|
||||
CamDep = direct_thing_client_dependency(StreamingPiCamera2, "/camera/")
|
||||
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
|
||||
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
|
||||
|
||||
def dict_generate(dict_steps, stream_resolution, dir, factor = 1):
|
||||
'''
|
||||
Creates a single dictionary of step sizes
|
||||
'''
|
||||
dict = {
|
||||
'x':(dict_steps/100) * stream_resolution[0] * factor * dir,
|
||||
'y':(dict_steps/100) * stream_resolution[1] * factor * dir
|
||||
}
|
||||
return dict
|
||||
|
||||
def steps_generate(small_step, z_perc, big_step, dir, stream_resolution):
|
||||
'''
|
||||
Creates all required dictionaries of all necessary step sizes.
|
||||
'''
|
||||
step_sizes_big = dict_generate(big_step, stream_resolution, dir)
|
||||
step_sizes_small = dict_generate(small_step, stream_resolution, dir)
|
||||
minimum_offset_small = dict_generate(small_step, stream_resolution, dir, factor = 0.8)
|
||||
z_steps = dict_generate(z_perc, stream_resolution, dir)
|
||||
minimum_offset_z = dict_generate(z_perc, stream_resolution, dir, factor = 0.8)
|
||||
|
||||
return step_sizes_small, minimum_offset_small, z_steps, minimum_offset_z, step_sizes_big
|
||||
|
||||
class RangeofMotionThing(Thing):
|
||||
def rom_axis(
|
||||
self,
|
||||
autofocus: AutofocusDep,
|
||||
stage: StageDep,
|
||||
cam: CamDep,
|
||||
csm: CSMDep,
|
||||
cancel: CancelHook,
|
||||
logger: InvocationLogger,
|
||||
axis: str,
|
||||
direction: int
|
||||
):
|
||||
"""
|
||||
Measure the range of motion in a single axis and direction.
|
||||
"""
|
||||
|
||||
try:
|
||||
# Generate required dictionaries for step sizes and minimum offsets
|
||||
stream_resolution = cam.stream_resolution
|
||||
|
||||
step_sizes_small, minimum_offset_small, z_steps, minimum_offset_z, step_sizes_big = steps_generate(20, 50, 200, direction, stream_resolution=stream_resolution)
|
||||
|
||||
delta = {
|
||||
'x':0,
|
||||
'y':0
|
||||
}
|
||||
|
||||
logger.info(f"Using the follwing steps: {step_sizes_small, minimum_offset_small, z_steps, minimum_offset_z, step_sizes_big}")
|
||||
|
||||
focus_data = autofocus.looping_autofocus(dz = 1000)
|
||||
|
||||
starting_position = list(stage.position.values())
|
||||
|
||||
logger.info("Moving the stage in 4 medium sized steps.")
|
||||
|
||||
# Medium sized steps
|
||||
for loop in range(4):
|
||||
image1 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
if axis == 'x':
|
||||
csm.move_in_image_coordinates(x = z_steps['x'], y = 0)
|
||||
else:
|
||||
csm.move_in_image_coordinates(x = 0, y = z_steps['y'])
|
||||
focus_data = autofocus.looping_autofocus(dz = 800)
|
||||
image2 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
offset = [x * 1 for x in fft_image_tracking.displacement_between_images(image_0 = image1, image_1 = image2, sigma=10, fractional_threshold=0.1, pad=True)] # Units is pixels
|
||||
delta['x'] = int(offset[1])
|
||||
delta['y'] = int(offset[0])
|
||||
logger.info(f"Offset measured as {delta[axis]}")
|
||||
|
||||
# 1 big step followed by 3 small steps
|
||||
while np.abs(delta[axis]) > minimum_offset_small[axis]:
|
||||
|
||||
if axis == 'x':
|
||||
csm.move_in_image_coordinates(x = step_sizes_big['x'], y = 0)
|
||||
else:
|
||||
csm.move_in_image_coordinates(x = 0, y = step_sizes_big['y'])
|
||||
|
||||
for loop in range(3):
|
||||
image1 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
if axis == 'x':
|
||||
csm.move_in_image_coordinates(x = step_sizes_small['x'], y = 0)
|
||||
else:
|
||||
csm.move_in_image_coordinates(x = 0, y = step_sizes_small['y'])
|
||||
focus_data = autofocus.looping_autofocus(dz = 800)
|
||||
image2 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
offset = [x * 1 for x in fft_image_tracking.displacement_between_images(image_0 = image1, image_1 = image2, sigma=10, fractional_threshold=0.1, pad=True)] # Units is pixels
|
||||
delta['x'] = int(offset[1])
|
||||
delta['y'] = int(offset[0])
|
||||
logger.info(f"Offset measured as {delta[axis]}")
|
||||
|
||||
if np.abs(delta[axis]) < minimum_offset_small[axis]: # this means the edge has been found
|
||||
logger.info(f"Edge has been found.")
|
||||
break
|
||||
|
||||
# Motion detection
|
||||
|
||||
logger.info(f"Running motion detection")
|
||||
|
||||
displacements = np.array([1,2,4,8,16,32,64,128,256,512,1024]) # Array of increasing step sizes
|
||||
motion_minimum = 20 # minimum nuber of pixels for motion to be detected
|
||||
|
||||
this_motion_step = {
|
||||
'x':np.zeros(np.shape(displacements)[0]),
|
||||
'y':np.zeros(np.shape(displacements)[0]),
|
||||
'z':0
|
||||
}
|
||||
|
||||
this_motion_step[axis] = displacements * direction * -1
|
||||
|
||||
for loop in range(np.shape(displacements)[0]):
|
||||
logger.info(f"Testing with step size {this_motion_step[axis][loop]}")
|
||||
image1 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
stage.move_relative(x = this_motion_step['x'][loop], y = this_motion_step['y'][loop], z = this_motion_step['z'])
|
||||
image2 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1)
|
||||
offset = [x * 1 for x in fft_image_tracking.displacement_between_images(
|
||||
image_0 = image1, image_1 = image2, sigma=10, fractional_threshold=0.1, pad=True)] # Units is pixels
|
||||
delta['x'] = int(offset[1])
|
||||
delta['y'] = int(offset[0])
|
||||
logger.info(f"Offset measured as {np.abs(delta[axis])}")
|
||||
if np.abs(delta[axis]) > motion_minimum:
|
||||
logger.info("Motion detected.")
|
||||
break
|
||||
|
||||
stage.move_absolute(x = starting_position[0], y = starting_position[1], z = starting_position[2], block_cancellation=True)
|
||||
|
||||
except:
|
||||
logger.error("Stopping measurement because it was cancelled by the user")
|
||||
stage.move_absolute(x = starting_position[0], y = starting_position[1], z = starting_position[2], block_cancellation=True)
|
||||
raise Exception
|
||||
return focus_data
|
||||
|
||||
@thing_action
|
||||
def rom_main(
|
||||
self,
|
||||
autofocus: AutofocusDep,
|
||||
stage: StageDep,
|
||||
cam: CamDep,
|
||||
csm: CSMDep,
|
||||
cancel: CancelHook,
|
||||
logger: InvocationLogger
|
||||
):
|
||||
"""
|
||||
Measures the range of motion of the stage across the x and y axes.
|
||||
"""
|
||||
logger.info("Using the stage to measure the Range of Motion. Please ensure you are using a big enough sample.")
|
||||
x_pos_results = self.rom_axis(
|
||||
autofocus,
|
||||
stage,
|
||||
cam,
|
||||
csm,
|
||||
cancel,
|
||||
logger,
|
||||
axis = 'x',
|
||||
direction = 1
|
||||
)
|
||||
return x_pos_results
|
||||
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue