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

Simulation jog

Browse source
This commit is contained in:
Julian Stirling 2026-02-12 22:18:30 +00:00
parent 70dd7bd994
commit 61666a116b
3 changed files with 107 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/openflexure_microscope_server/things/stage/__init__.py
View file

@ -227,12 +227,7 @@ class BaseStage(lt.Thing):
)
def _hardware_start_move_relative(self, displacement: Sequence[int]) -> None:
"""Start a relative move.
This starts the stage moving, but does not wait for the move to complete. It
sets ``self.moving`` to ``True``: resetting it is the responsibility of the
calling code.
"""
"""Start a relative move."""
raise NotImplementedError(
"StageThings must define their own hardware_start_move_relative method"
)
@ -243,12 +238,7 @@ class BaseStage(lt.Thing):
)
def _poll_moving(self) -> bool:
"""Determine if the stage is still moving.
This also sets ``moving`` if the status has changed.
:return: whether the stage is still moving.
"""
"""Determine if the stage is still moving."""
raise NotImplementedError(
"StageThings must define their own _poll_moving method"
)

136
src/openflexure_microscope_server/things/stage/dummy.py
View file

@ -2,9 +2,11 @@
from __future__ import annotations
import threading
import time
from collections.abc import Sequence
from types import TracebackType
from typing import Any, Optional, Self
from typing import Any, Mapping, Optional, Self
import labthings_fastapi as lt
@ -32,8 +34,11 @@ class DummyStage(BaseStage):
also doing computationally heavy tasks like simulated image blurring.
"""
super().__init__(thing_server_interface, **kwargs)
self._move_thread: Optional[threading.Thread] = None
self.step_time = step_time
self.instantaneous_position = self._hardware_position
self.instantaneous_position: Mapping[str, int] = self._hardware_position
self._inst_pos_lock = threading.Lock()
self._abort_move = threading.Event()
def __enter__(self) -> Self:
"""Register the stage position when the Thing context manager is opened."""
@ -53,43 +58,111 @@ class DummyStage(BaseStage):
)
"""Used to convert coordinates between the program frame and the hardware frame."""
def _hardware_move_relative(
self,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
"""Make a relative move. Keyword arguments should be axis names."""
displacement = [kwargs.get(k, 0) for k in self.axis_names]
self.moving = True
def update_position(self) -> None:
"""Read position from the stage and set the corresponding property."""
pass
def _calc_fractional_pos(
self, displacement: Sequence[int], fraction_complete: float
) -> dict[str, int]:
"""Calculate the position a fraction through a move."""
return {
ax: self._hardware_position[ax] + int(fraction_complete * disp)
for ax, disp in zip(self.axis_names, displacement, strict=True)
}
def _apply_move(self, displacement: Sequence[int]) -> None:
try:
fraction_complete = 0.0
dt = self.step_time
max_displacement = max(abs(v) for v in displacement)
start_time = time.time()
while time.time() - start_time < dt * max_displacement:
if block_cancellation:
time.sleep(dt)
else:
lt.cancellable_sleep(dt)
if self._abort_move.is_set():
break
fraction_complete = (time.time() - start_time) / (dt * max_displacement)
self.instantaneous_position = {
ax: self._hardware_position[ax] + int(fraction_complete * disp)
for ax, disp in zip(self.axis_names, displacement, strict=True)
}
fraction_complete = 1.0
except lt.exceptions.InvocationCancelledError as e:
# If the move has been cancelled, stop it but don't handle the exception.
# We need the exception to propagate in order to stop any calling tasks,
# and to mark the invocation as "cancelled" rather than stopped.
raise e
self.instantaneous_position = self._calc_fractional_pos(
displacement, fraction_complete
)
if not self._abort_move.is_set():
fraction_complete = 1.0
finally:
self.moving = False
self._hardware_position = {
ax: self._hardware_position[ax] + int(fraction_complete * disp)
for ax, disp in zip(self.axis_names, displacement, strict=True)
}
self._hardware_position = self._calc_fractional_pos(
displacement, fraction_complete
)
self.instantaneous_position = self._hardware_position
def _hardware_start_move_relative(self, displacement: Sequence[int]) -> None:
"""Start a relative move.
This starts the stage moving, but does not wait for the move to complete. It
sets ``self.moving`` to ``True``: resetting it is the responsibility of the
calling code.
"""
with self._hardware_lock:
self.moving = True
self._abort_move.clear()
self._move_thread = threading.Thread(
target=self._apply_move, args=(displacement,)
)
self._move_thread.start()
def _hardware_stop(self) -> None:
with self._hardware_lock:
self._abort_move.set()
self.moving = False
def _poll_moving(self) -> bool:
"""Determine if the stage is still moving."""
if self._move_thread is None:
return False
moving = self._move_thread.is_alive()
if self.moving != moving:
self.moving = moving
return moving
def _hardware_move_relative(
self,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
"""Make a relative move. Keyword arguments should be axis names."""
with self._hardware_lock:
displacement = [kwargs.get(k, 0) for k in self.axis_names]
self.moving = True
# This follows a similar structure to _apply_move but need a different
# breakning mechanism. Running apply move in a thread and breaking here
# results in significantly worse autofocus
try:
fraction_complete = 0.0
dt = self.step_time
max_displacement = max(abs(v) for v in displacement)
start_time = time.time()
while time.time() - start_time < dt * max_displacement:
if block_cancellation:
time.sleep(self.step_time)
else:
lt.cancellable_sleep(self.step_time)
fraction_complete = (time.time() - start_time) / (
dt * max_displacement
)
self.instantaneous_position = self._calc_fractional_pos(
displacement, fraction_complete
)
fraction_complete = 1.0
except lt.exceptions.InvocationCancelledError as e:
# If the move has been cancelled, stop it but don't handle the
# exception. We need the exception to propagate in order to stop
# any calling tasks, and to mark the invocation as "cancelled"
# rather than stopped.
raise e
finally:
self._hardware_position = self._calc_fractional_pos(
displacement, fraction_complete
)
self.instantaneous_position = self._hardware_position
self.moving = False
def _hardware_move_absolute(
self,
block_cancellation: bool = False,
@ -113,5 +186,6 @@ class DummyStage(BaseStage):
It is intended for use after manually or automatically recentring the
stage.
"""
self._hardware_position = dict.fromkeys(self.axis_names, 0)
self.instantaneous_position = self._hardware_position
with self._hardware_lock:
self._hardware_position = dict.fromkeys(self.axis_names, 0)
self.instantaneous_position = self._hardware_position

6
src/openflexure_microscope_server/things/stage/sangaboard.py
View file

@ -137,12 +137,6 @@ class SangaboardThing(BaseStage):
self.moving = moving
return moving
def _poll_until_stopped(self) -> None:
"""Poll the stage until it reports that it has stopped moving."""
with self._hardware_lock:
while self._poll_moving():
lt.cancellable_sleep(0.1)
def _hardware_move_relative(
self,
block_cancellation: bool = False,