Remove dependencies from stages

This commit is contained in:
Julian Stirling 2025-12-16 13:45:51 +00:00
parent 4b845fcf18
commit 426178ba31
3 changed files with 11 additions and 38 deletions

View file

@ -133,24 +133,15 @@ class BaseStage(lt.Thing):
self.axis_inverted = direction
@lt.action
def move_relative(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
def move_relative(self, block_cancellation: bool = False, **kwargs: int) -> None:
"""Make a relative move. Keyword arguments should be axis names."""
self._hardware_move_relative(
cancel=cancel,
block_cancellation=block_cancellation,
**self._apply_axis_direction(kwargs),
)
def _hardware_move_relative(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
self, block_cancellation: bool = False, **kwargs: int
) -> Never:
"""Make a relative move in the coordinate system used by the physical hardware.
@ -161,22 +152,15 @@ class BaseStage(lt.Thing):
)
@lt.action
def move_absolute(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
def move_absolute(self, block_cancellation: bool = False, **kwargs: int) -> None:
"""Make an absolute move. Keyword arguments should be axis names."""
self._hardware_move_absolute(
cancel=cancel,
block_cancellation=block_cancellation,
**self._apply_axis_direction(kwargs),
)
def _hardware_move_absolute(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> Never:
@ -212,20 +196,16 @@ class BaseStage(lt.Thing):
return (position_dict["x"], position_dict["y"], position_dict["z"])
@lt.action
def move_to_xyz_position(
self, cancel: lt.deps.CancelHook, xyz_pos: tuple[int, int, int]
) -> None:
def move_to_xyz_position(self, xyz_pos: tuple[int, int, int]) -> None:
"""Move to the location specified by an (x, y, z) tuple.
:param cancel: A cancel hook for cancelling the move. This dependency should be
injected automatically by LabThings-FastAPI
:param xyz_pos: The (x, y, z) position to move to.
:raises KeyError: if this stage does not have axes named "x", "y", and "z".
This method provides the interface expected by the camera_stage_mapping.
"""
self.move_absolute(cancel=cancel, x=xyz_pos[0], y=xyz_pos[1], z=xyz_pos[2])
self.move_absolute(x=xyz_pos[0], y=xyz_pos[1], z=xyz_pos[2])
StageDependency = lt.deps.direct_thing_client_dependency(BaseStage, "stage")

View file

@ -55,7 +55,6 @@ class DummyStage(BaseStage):
def _hardware_move_relative(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
@ -71,7 +70,7 @@ class DummyStage(BaseStage):
if block_cancellation:
time.sleep(dt)
else:
cancel.sleep(dt)
lt.cancellable_sleep(dt)
fraction_complete = (time.time() - start_time) / (dt * max_displacement)
self.instantaneous_position = {
ax: self._hardware_position[ax] + int(fraction_complete * disp)
@ -93,7 +92,6 @@ class DummyStage(BaseStage):
def _hardware_move_absolute(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
@ -104,7 +102,7 @@ class DummyStage(BaseStage):
if axis in self.axis_names
}
self._hardware_move_relative(
cancel, block_cancellation=block_cancellation, **displacement
block_cancellation=block_cancellation, **displacement
)
@lt.action

View file

@ -2,7 +2,6 @@
from __future__ import annotations
import logging
import threading
import time
from collections.abc import Mapping
@ -18,8 +17,6 @@ import sangaboard
from . import BaseStage
LOGGER = logging.getLogger(__name__)
REQUIRED_VERSION = semver.Version.parse("1.0.0")
RECOMMENDED_VERSION = semver.Version.parse("1.0.4")
@ -117,14 +114,13 @@ class SangaboardThing(BaseStage):
# Warn if version is below recommended
if version < RECOMMENDED_VERSION:
LOGGER.warning(
self.logger.warning(
f"Sangaboard firmware version {version} is below the recommended "
f"{RECOMMENDED_VERSION}."
)
def _hardware_move_relative(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
@ -138,7 +134,7 @@ class SangaboardThing(BaseStage):
sb.query("notify_on_stop")
else:
while sb.query("moving?") == "true":
cancel.sleep(0.1)
lt.cancellable_sleep(0.1)
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,
@ -151,7 +147,6 @@ class SangaboardThing(BaseStage):
def _hardware_move_absolute(
self,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
) -> None:
@ -164,7 +159,7 @@ class SangaboardThing(BaseStage):
if axis in self.axis_names
}
self._hardware_move_relative(
cancel, block_cancellation=block_cancellation, **displacement
block_cancellation=block_cancellation, **displacement
)
@lt.action
@ -203,7 +198,7 @@ class SangaboardThing(BaseStage):
# cannot be used.
intended_brightness = float(return_value[7:])
on_brightness = 0.32
LOGGER.warning(
self.logger.warning(
"Brightness control is not yet implemented. Desired brightness: "
f"{intended_brightness}. Set brightness: {on_brightness}"
)