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

Ruff formatting stuff

Browse source
This commit is contained in:
Chish36 2025-07-30 15:44:02 +01:00 • committed by Julian Stirling
parent 6579a65f84
commit d6c47bd43e
2 changed files with 142 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

249
src/openflexure_microscope_server/things/stage_measure.py
View file

@ -35,11 +35,9 @@ CSMDep = lt.deps.direct_thing_client_dependency(
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/") AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
def generate_move_dicts(fov_perc: int, def generate_move_dicts(
stream_resolution: list[int], fov_perc: int, stream_resolution: list[int], direction: int, factor: float = 1
direction: int, ) -> dict[str, float]:
factor: float = 1
) -> dict[str, float]:
"""Create a single dictionary of x and y moves for moving in image coordinates. """Create a single dictionary of x and y moves for moving in image coordinates.
:param fov_perc: The percentage of field of view the stage should move by. :param fov_perc: The percentage of field of view the stage should move by.
@ -54,12 +52,9 @@ def generate_move_dicts(fov_perc: int,
} }
def predict_z(positions: list, def predict_z(
axis: str, positions: list, axis: str, relative_move: float, stage: StageDep, csm: CSMDep
relative_move: float, ) -> float:
stage: StageDep,
csm: CSMDep
) -> float:
"""Predict the next z position for a move using previous positions. """Predict the next z position for a move using previous positions.
:params positions: The list of positions used for predicting z. :params positions: The list of positions used for predicting z.
@ -70,26 +65,30 @@ def predict_z(positions: list,
:return: A number of pixels the stage needs to move in z. :return: A number of pixels the stage needs to move in z.
""" """
pixel_step = { pixel_step = {
'x':1/csm.image_to_stage_displacement_matrix[0][1], "x": 1 / csm.image_to_stage_displacement_matrix[0][1],
'y':1/csm.image_to_stage_displacement_matrix[1][0] "y": 1 / csm.image_to_stage_displacement_matrix[1][0],
} }
lateral_positions = [i[axis] for i in positions] lateral_positions = [i[axis] for i in positions]
z_positions = [i['z'] for i in positions] z_positions = [i["z"] for i in positions]
parameters, _ = curve_fit(quadratic, lateral_positions, z_positions) parameters, _ = curve_fit(quadratic, lateral_positions, z_positions)
z_dest = quadratic(stage.position[axis] + (relative_move/pixel_step[axis]), *parameters) z_dest = quadratic(
stage.position[axis] + (relative_move / pixel_step[axis]), *parameters
)
return z_dest - stage.position["z"] return z_dest - stage.position["z"]
def move_and_measure( def move_and_measure(
step_size: dict[str, float], step_size: dict[str, float],
axis: str, axis: str,
data, data,
image1, image1,
autofocus_proc: bool, autofocus_proc: bool,
csm: CSMDep, csm: CSMDep,
autofocus: AutofocusDep, autofocus: AutofocusDep,
cam:CamDep) -> tuple: cam: CamDep,
) -> tuple:
"""Move the stage and measure the offset between the two positions. """Move the stage and measure the offset between the two positions.
:params step_size: A dictionary with keys 'x' and 'y' with pixel distances. :params step_size: A dictionary with keys 'x' and 'y' with pixel distances.
@ -100,26 +99,29 @@ def move_and_measure(
:return: All required data for the next move. This includes the updated delta value and offset. :return: All required data for the next move. This includes the updated delta value and offset.
Also returns what wrong_axis is i.e. if the direction is 'x', wrong_axis = 'y'. Also returns what wrong_axis is i.e. if the direction is 'x', wrong_axis = 'y'.
""" """
if axis == 'x': if axis == "x":
csm.move_in_image_coordinates(x = step_size['x'], y = 0) csm.move_in_image_coordinates(x=step_size["x"], y=0)
wrong_axis = 'y' wrong_axis = "y"
else: else:
csm.move_in_image_coordinates(x = 0, y = step_size['y']) csm.move_in_image_coordinates(x=0, y=step_size["y"])
wrong_axis = 'x' wrong_axis = "x"
if autofocus_proc: if autofocus_proc:
autofocus.looping_autofocus(dz = 800) autofocus.looping_autofocus(dz=800)
image2 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), dsize=(0,0), fx= 1, fy= 1) image2 = cv2.resize(
offset = [x * 1 for x in fft_image_tracking.displacement_between_images( np.array(Image.open(cam.grab_jpeg().open())), dsize=(0, 0), fx=1, fy=1
image_0 = image1, )
image_1 = image2, offset = [
sigma=10, x * 1
fractional_threshold=0.1, for x in fft_image_tracking.displacement_between_images(
pad=True)] # Units is pixels image_0=image1, image_1=image2, sigma=10, fractional_threshold=0.1, pad=True
data.delta['x'] = int(offset[1]) )
data.delta['y'] = int(offset[0]) ] # Units is pixels
data.delta["x"] = int(offset[1])
data.delta["y"] = int(offset[0])
return offset, wrong_axis return offset, wrong_axis
def acquire_z_predict_points( def acquire_z_predict_points(
stream_resolution: list[int], stream_resolution: list[int],
direction: int, direction: int,
@ -162,7 +164,10 @@ def acquire_z_predict_points(
data.measure(stage.position, offset) data.measure(stage.position, offset)
assert(np.abs(data.delta[wrong_axis]) < np.abs(wrong_axis_max_medium[wrong_axis])) assert np.abs(data.delta[wrong_axis]) < np.abs(
wrong_axis_max_medium[wrong_axis]
)
def check_stage_operation( def check_stage_operation(
small_step: int, small_step: int,
@ -191,10 +196,8 @@ def check_stage_operation(
""" """
failure_count = 0 failure_count = 0
wrong_axis_max_small = generate_move_dicts( wrong_axis_max_small = generate_move_dicts(
small_step, small_step, stream_resolution, direction, factor=0.1
stream_resolution, )
direction,
factor=0.1)
for _loop in range(3): for _loop in range(3):
image1 = cv2.resize( image1 = cv2.resize(
@ -250,6 +253,7 @@ def check_stage_operation(
logger.info("Edge has been found.") logger.info("Edge has been found.")
break break
def motion_detection( def motion_detection(
axis: str, axis: str,
direction: int, direction: int,
@ -265,39 +269,46 @@ def motion_detection(
previous to motion detection being used. previous to motion detection being used.
:return: The stage coordinates where motion was detected. :return: The stage coordinates where motion was detected.
""" """
displacements = [1,2,4,8,16,32,64,128,256,512] # Array of increasing step sizes displacements = [
1,
2,
4,
8,
16,
32,
64,
128,
256,
512,
] # Array of increasing step sizes
motion_minimum = 20 # minimum number of pixels for motion to be detected motion_minimum = 20 # minimum number of pixels for motion to be detected
this_motion_step = { this_motion_step = {"x": 0, "y": 0}
'x': 0,
'y': 0
}
delta = { delta = {"x": 0, "y": 0}
'x': 0,
'y': 0
}
for loop in range(np.shape(displacements)[0]): for loop in range(np.shape(displacements)[0]):
this_motion_step[axis] = displacements[loop] * direction * -1 this_motion_step[axis] = displacements[loop] * direction * -1
logger.info(f"Testing with step size {this_motion_step[axis]}") logger.info(f"Testing with step size {this_motion_step[axis]}")
image1 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), image1 = cv2.resize(
dsize=(0,0), np.array(Image.open(cam.grab_jpeg().open())), dsize=(0, 0), fx=1, fy=1
fx= 1, )
fy= 1) csm.move_in_image_coordinates(x=this_motion_step["x"], y=this_motion_step["y"])
csm.move_in_image_coordinates(x = this_motion_step['x'], y = this_motion_step['y']) image2 = cv2.resize(
image2 = cv2.resize(np.array(Image.open(cam.grab_jpeg().open())), np.array(Image.open(cam.grab_jpeg().open())), dsize=(0, 0), fx=1, fy=1
dsize=(0,0), )
fx= 1, offset = [
fy= 1) x * 1
offset = [x * 1 for x in fft_image_tracking.displacement_between_images( for x in fft_image_tracking.displacement_between_images(
image_0 = image1, image_0=image1,
image_1 = image2, image_1=image2,
sigma=10, sigma=10,
fractional_threshold=0.1, fractional_threshold=0.1,
pad=True)] pad=True,
delta['x'] = int(offset[1]) )
delta['y'] = int(offset[0]) ]
delta["x"] = int(offset[1])
delta["y"] = int(offset[0])
logger.info(f"Offset measured as {np.abs(delta[axis])}") logger.info(f"Offset measured as {np.abs(delta[axis])}")
if np.abs(delta[axis]) > motion_minimum: if np.abs(delta[axis]) > motion_minimum:
logger.info("Motion detected.") logger.info("Motion detected.")
@ -305,10 +316,16 @@ def motion_detection(
return stage.position return stage.position
class RomDataTracker():
class RomDataTracker:
"""Class for tracking range of motion data.""" """Class for tracking range of motion data."""
def __init__(self, stage_coords: list[dict[str, int]] = [], cor_lat_steps: list[list[float]] = [], delta: dict = {'x':0, 'y':0}): def __init__(
self,
stage_coords: list[dict[str, int]] = [],
cor_lat_steps: list[list[float]] = [],
delta: dict = {"x": 0, "y": 0},
):
"""Define useful data tracked throughout test.""" """Define useful data tracked throughout test."""
self.stage_coords = stage_coords self.stage_coords = stage_coords
self.cor_lat_steps = cor_lat_steps self.cor_lat_steps = cor_lat_steps
@ -319,6 +336,7 @@ class RomDataTracker():
self.stage_coords.append(current_pos) self.stage_coords.append(current_pos)
self.cor_lat_steps.append(cor) self.cor_lat_steps.append(cor)
class RangeofMotionThing(lt.Thing): class RangeofMotionThing(lt.Thing):
"""A class used to measure the range of motion of the stage in X and Y.""" """A class used to measure the range of motion of the stage in X and Y."""
@ -339,7 +357,7 @@ class RangeofMotionThing(lt.Thing):
:return: Results dictionary containing stage positions, :return: Results dictionary containing stage positions,
correlations and the final position. correlations and the final position.
""" """
autofocus.looping_autofocus(dz = 1000) autofocus.looping_autofocus(dz=1000)
starting_position = list(stage.position.values()) starting_position = list(stage.position.values())
@ -352,7 +370,7 @@ class RangeofMotionThing(lt.Thing):
logger.info(f"Beginning the {axis}-axis in the {dir_word} direction") logger.info(f"Beginning the {axis}-axis in the {dir_word} direction")
# Generate required dictionaries for step sizes and minimum offsets # Generate required dictionaries for step sizes and minimum offsets
stream_resolution = [820,616] stream_resolution = [820, 616]
big_step = 200 big_step = 200
small_step = 20 small_step = 20
step_sizes_big = generate_move_dicts(big_step, stream_resolution, direction) step_sizes_big = generate_move_dicts(big_step, stream_resolution, direction)
@ -380,23 +398,24 @@ class RangeofMotionThing(lt.Thing):
while np.abs(rom_data.delta[axis]) > np.abs(minimum_offset_small[axis]): while np.abs(rom_data.delta[axis]) > np.abs(minimum_offset_small[axis]):
z_diff = predict_z( z_diff = predict_z(
positions = rom_data.stage_coords, positions=rom_data.stage_coords,
axis = axis, axis=axis,
relative_move = step_sizes_big[axis], relative_move=step_sizes_big[axis],
stage = stage, stage=stage,
csm = csm) csm=csm,
)
logger.info("Z calibration complete.") logger.info("Z calibration complete.")
stage.move_relative(z = z_diff) stage.move_relative(z=z_diff)
logger.info(f"Moved in z by {z_diff}") logger.info(f"Moved in z by {z_diff}")
# Big step # Big step
if axis == 'x': if axis == "x":
csm.move_in_image_coordinates(x = step_sizes_big['x'], y = 0) csm.move_in_image_coordinates(x=step_sizes_big["x"], y=0)
else: else:
csm.move_in_image_coordinates(x = 0, y = step_sizes_big['y']) csm.move_in_image_coordinates(x=0, y=step_sizes_big["y"])
autofocus.looping_autofocus(dz = 800) autofocus.looping_autofocus(dz=800)
rom_data.stage_coords.append(stage.position) rom_data.stage_coords.append(stage.position)
check_stage_operation( check_stage_operation(
@ -404,7 +423,7 @@ class RangeofMotionThing(lt.Thing):
stream_resolution=stream_resolution, stream_resolution=stream_resolution,
direction=direction, direction=direction,
axis=axis, axis=axis,
data = rom_data, data=rom_data,
minimum_offset_small=minimum_offset_small, minimum_offset_small=minimum_offset_small,
csm=csm, csm=csm,
cam=cam, cam=cam,
@ -415,22 +434,32 @@ class RangeofMotionThing(lt.Thing):
# Motion detection # Motion detection
logger.info("Running motion detection") logger.info("Running motion detection")
final_pos = motion_detection(axis = axis, direction = direction, csm = csm, stage = stage, cam = cam, logger = logger) final_pos = motion_detection(
rom_data.stage_coords[np.shape(np.array(rom_data.stage_coords))[0] - 1] = final_pos axis=axis,
direction=direction,
csm=csm,
stage=stage,
cam=cam,
logger=logger,
)
rom_data.stage_coords[np.shape(np.array(rom_data.stage_coords))[0] - 1] = (
final_pos
)
axis_results = { axis_results = {
"correlation_lateral_steps": rom_data.cor_lat_steps, "correlation_lateral_steps": rom_data.cor_lat_steps,
"stage_positions": rom_data.stage_coords, "stage_positions": rom_data.stage_coords,
"final_position": final_pos "final_position": final_pos,
} }
except AssertionError: except AssertionError:
logger.info("Parasitic motion detected.") logger.info("Parasitic motion detected.")
finally: finally:
stage.move_absolute( stage.move_absolute(
x = starting_position[0], x=starting_position[0],
y = starting_position[1], y=starting_position[1],
z = starting_position[2], z=starting_position[2],
block_cancellation=True) block_cancellation=True,
)
return axis_results return axis_results
@ -447,30 +476,36 @@ class RangeofMotionThing(lt.Thing):
:return: Results dictionary separated into keys of each axis and direction. :return: Results dictionary separated into keys of each axis and direction.
""" """
logger.info("Using the stage to measure the Range of Motion.\ logger.info(
Please ensure you are using a big enough sample.") "Using the stage to measure the Range of Motion.\
Please ensure you are using a big enough sample."
)
start_time = time.time() start_time = time.time()
rom_results = {} rom_results = {}
for axis_dir in [['x', 1],['x', -1],['y', 1],['y', -1]]: for axis_dir in [["x", 1], ["x", -1], ["y", 1], ["y", -1]]:
axis_dir_results = self.rom_axis( axis_dir_results = self.rom_axis(
autofocus, autofocus,
stage, stage,
cam, cam,
csm, csm,
logger, logger,
axis = axis_dir[0], axis=axis_dir[0],
direction = axis_dir[1] direction=axis_dir[1],
) )
rom_results[f"{axis_dir}"] = axis_dir_results rom_results[f"{axis_dir}"] = axis_dir_results
end_time = time.time() end_time = time.time()
total_time = (end_time - start_time)/60 total_time = (end_time - start_time) / 60
x_range = abs(rom_results["['x', 1]"]["final_position"]["x"] - x_range = abs(
rom_results["['x', -1]"]["final_position"]["x"]) rom_results["['x', 1]"]["final_position"]["x"]
y_range = abs(rom_results["['y', 1]"]["final_position"]["y"] - - rom_results["['x', -1]"]["final_position"]["x"]
rom_results["['y', -1]"]["final_position"]["y"]) )
y_range = abs(
rom_results["['y', 1]"]["final_position"]["y"]
- rom_results["['y', -1]"]["final_position"]["y"]
)
step_range = [x_range, y_range] step_range = [x_range, y_range]
logger.info(f"Range of motion is {x_range} X {y_range}") logger.info(f"Range of motion is {x_range} X {y_range}")
@ -480,11 +515,9 @@ class RangeofMotionThing(lt.Thing):
self.last_calibration = DenumpifyingDict(rom_results).model_dump() self.last_calibration = DenumpifyingDict(rom_results).model_dump()
with open("/var/openflexure/ROM_Test_Results.json", 'w') as file_object: with open("/var/openflexure/ROM_Test_Results.json", "w") as file_object:
json.dump(rom_results, file_object, indent = 3) json.dump(rom_results, file_object, indent=3)
return rom_results return rom_results
last_calibration = lt.ThingSetting( last_calibration = lt.ThingSetting(initial_value=None, model=dict, readonly=True)
initial_value=None, model=dict, readonly=True
)

1
src/openflexure_microscope_server/utilities.py
View file

@ -328,6 +328,7 @@ def _get_version_from_toml(toml_path: str) -> str:
LOGGER.error("Problem opening pyproject.toml") LOGGER.error("Problem opening pyproject.toml")
return "Undefined" return "Undefined"
def quadratic(x, a, b, c): def quadratic(x, a, b, c):
"""Quadratic function. Used for predicting z. """Quadratic function. Used for predicting z.