Merge branch 'include-sangaboard' into 'v3'
Move SangaboardThing into repo See merge request openflexure/openflexure-microscope-server!311
This commit is contained in:
Julian Stirling
commit
a0065931e4
6 changed files with 171 additions and 77 deletions
|
|
@ -25,7 +25,7 @@ from labthings_fastapi.dependencies.metadata import GetThingStates
|
|||
from labthings_fastapi.server import ThingServer
|
||||
|
||||
from . import BaseCamera, JPEGBlob, ArrayModel
|
||||
from ..stage import StageProtocol as Stage
|
||||
from ..stage import BaseStage
|
||||
|
||||
# The ratio between "motor" steps and pixels
|
||||
# higher related to a faster movement
|
||||
|
|
@ -35,7 +35,7 @@ RATIO = 0.2
|
|||
class SimulatedCamera(BaseCamera):
|
||||
"""A Thing representing an OpenCV camera"""
|
||||
|
||||
_stage: Optional[Stage] = None
|
||||
_stage: Optional[BaseStage] = None
|
||||
_server: Optional[ThingServer] = None
|
||||
|
||||
def __init__(
|
||||
|
|
|
|||
|
|
@ -1,65 +1,12 @@
|
|||
from __future__ import annotations
|
||||
from typing import Any, Protocol, TypeAlias, runtime_checkable
|
||||
from typing import TypeAlias
|
||||
from collections.abc import Sequence, Mapping
|
||||
|
||||
from labthings_fastapi.descriptors.property import PropertyDescriptor
|
||||
from labthings_fastapi.thing import Thing
|
||||
from labthings_fastapi.decorators import thing_action, thing_property
|
||||
from labthings_fastapi.dependencies.invocation import CancelHook
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from collections.abc import Sequence, Mapping
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class StageProtocol(Protocol):
|
||||
"""A protocol for the OpenFlexure translation stage"""
|
||||
|
||||
_axis_names: Sequence[str]
|
||||
|
||||
@property
|
||||
def axis_names(self) -> Sequence[str]:
|
||||
"""The names of the stage's axes, in order."""
|
||||
...
|
||||
|
||||
@property
|
||||
def position(self) -> Mapping[str, int]:
|
||||
"""Current position of the stage"""
|
||||
...
|
||||
|
||||
@property
|
||||
def moving() -> bool:
|
||||
"Whether the stage is in motion"
|
||||
...
|
||||
|
||||
@property
|
||||
def thing_state(self) -> Mapping[str, Any]:
|
||||
"""Summary metadata describing the current state of the stage"""
|
||||
...
|
||||
|
||||
def move_relative(
|
||||
self,
|
||||
cancel: CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
...
|
||||
|
||||
def move_absolute(
|
||||
self,
|
||||
cancel: CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
...
|
||||
|
||||
def set_zero_position(self):
|
||||
"""Make the current position zero in all axes
|
||||
|
||||
This action does not move the stage, but resets the position to zero.
|
||||
It is intended for use after manually or automatically recentring the
|
||||
stage.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class BaseStage(Thing):
|
||||
|
|
@ -99,16 +46,6 @@ class BaseStage(Thing):
|
|||
"""Summary metadata describing the current state of the stage"""
|
||||
return {"position": self.position}
|
||||
|
||||
|
||||
class StageStub(BaseStage):
|
||||
"""A Stub Thing for the OpenFlexure translation stage
|
||||
|
||||
As of LabThings-FastAPI 0.0.7, we can't make a thing client dependency
|
||||
based on a protocol, because the protocol is not a Thing, and its
|
||||
methods/properties are not decorated as Affordances. This stub class
|
||||
is a workaround for that limitation, and should not be used directly.
|
||||
"""
|
||||
|
||||
@thing_action
|
||||
def move_relative(
|
||||
self,
|
||||
|
|
@ -117,7 +54,9 @@ class StageStub(BaseStage):
|
|||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
raise NotImplementedError(
|
||||
"StageThings must define their own move_relative method"
|
||||
)
|
||||
|
||||
@thing_action
|
||||
def move_absolute(
|
||||
|
|
@ -127,7 +66,9 @@ class StageStub(BaseStage):
|
|||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
raise NotImplementedError(
|
||||
"StageThings must define their own move_absolute method"
|
||||
)
|
||||
|
||||
@thing_action
|
||||
def set_zero_position(self):
|
||||
|
|
@ -137,7 +78,9 @@ class StageStub(BaseStage):
|
|||
It is intended for use after manually or automatically recentring the
|
||||
stage.
|
||||
"""
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
raise NotImplementedError(
|
||||
"StageThings must define their own set_zero_position method"
|
||||
)
|
||||
|
||||
|
||||
StageDependency: TypeAlias = direct_thing_client_dependency(StageStub, "/stage/")
|
||||
StageDependency: TypeAlias = direct_thing_client_dependency(BaseStage, "/stage/")
|
||||
|
|
|
|||
|
|
@ -75,9 +75,9 @@ class DummyStage(BaseStage):
|
|||
):
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
displacement = {
|
||||
k: int(v) - self.position[k]
|
||||
for k, v in kwargs.items()
|
||||
if k in self.axis_names
|
||||
axis: int(pos) - self.position[axis]
|
||||
for axis, pos in kwargs.items()
|
||||
if axis in self.axis_names
|
||||
}
|
||||
self.move_relative(
|
||||
cancel, block_cancellation=block_cancellation, **displacement
|
||||
|
|
|
|||
151
src/openflexure_microscope_server/things/stage/sangaboard.py
Normal file
151
src/openflexure_microscope_server/things/stage/sangaboard.py
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
from __future__ import annotations
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from typing import Iterator, Literal
|
||||
from contextlib import contextmanager
|
||||
from collections.abc import Mapping
|
||||
|
||||
import sangaboard
|
||||
from labthings_fastapi.decorators import thing_action
|
||||
from labthings_fastapi.dependencies.invocation import (
|
||||
CancelHook,
|
||||
InvocationCancelledError,
|
||||
)
|
||||
|
||||
from . import BaseStage
|
||||
|
||||
|
||||
class SangaboardThing(BaseStage):
|
||||
def __init__(self, port: str = None, **kwargs):
|
||||
"""A Thing to manage a Sangaboard motor controller
|
||||
|
||||
Internally, this uses the `pysangaboard` package from PyPi. This imports
|
||||
as `sangaboard`. As `pysangaboard` does not support some features added
|
||||
to the Sangaboard firmware v1 (LED flashing, aborting moves, etc) this
|
||||
functionality is accessed by directly querying the serial interface.
|
||||
"""
|
||||
self.sangaboard_kwargs = kwargs
|
||||
self.sangaboard_kwargs["port"] = port
|
||||
|
||||
def __enter__(self):
|
||||
self._sangaboard = sangaboard.Sangaboard(**self.sangaboard_kwargs)
|
||||
self._sangaboard_lock = threading.RLock()
|
||||
with self.sangaboard() as sb:
|
||||
if sb.version_tuple[0] != 1:
|
||||
raise RuntimeError(
|
||||
"Please update your Sangaboard Firmware. v1 is required."
|
||||
)
|
||||
sb.query("blocking_moves false")
|
||||
self.update_position()
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
with self.sangaboard() as sb:
|
||||
sb.close()
|
||||
|
||||
@contextmanager
|
||||
def sangaboard(self) -> Iterator[sangaboard.Sangaboard]:
|
||||
"""Return the wrapped `sangaboard.Sangaboard` instance.
|
||||
|
||||
This is protected by a `threading.RLock`, which may change in future.
|
||||
"""
|
||||
with self._sangaboard_lock:
|
||||
yield self._sangaboard
|
||||
|
||||
def update_position(self) -> None:
|
||||
"""Read position from the stage and set the corresponding property."""
|
||||
with self.sangaboard() as sb:
|
||||
self.position = dict(zip(self.axis_names, sb.position))
|
||||
|
||||
@thing_action
|
||||
def move_relative(
|
||||
self,
|
||||
cancel: CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
) -> None:
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
displacement = [kwargs.get(axis, 0) for axis in self.axis_names]
|
||||
with self.sangaboard() as sb:
|
||||
self.moving = True
|
||||
try:
|
||||
sb.move_rel(displacement)
|
||||
if block_cancellation:
|
||||
sb.query("notify_on_stop")
|
||||
else:
|
||||
while sb.query("moving?") == "true":
|
||||
cancel.sleep(0.1)
|
||||
except 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.
|
||||
sb.query("stop")
|
||||
raise e
|
||||
finally:
|
||||
self.moving = False
|
||||
self.update_position()
|
||||
|
||||
@thing_action
|
||||
def move_absolute(
|
||||
self,
|
||||
cancel: CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
) -> None:
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
with self.sangaboard():
|
||||
self.update_position()
|
||||
displacement = {
|
||||
axis: int(pos) - self.position[axis]
|
||||
for axis, pos in kwargs.items()
|
||||
if axis in self.axis_names
|
||||
}
|
||||
self.move_relative(
|
||||
cancel, block_cancellation=block_cancellation, **displacement
|
||||
)
|
||||
|
||||
@thing_action
|
||||
def set_zero_position(self) -> None:
|
||||
"""Make the current position zero in all axes
|
||||
|
||||
This action does not move the stage, but resets the position to zero.
|
||||
It is intended for use after manually or automatically recentring the
|
||||
stage.
|
||||
"""
|
||||
with self.sangaboard() as sb:
|
||||
sb.zero_position()
|
||||
self.update_position()
|
||||
|
||||
@thing_action
|
||||
def flash_led(
|
||||
self,
|
||||
number_of_flashes: int = 10,
|
||||
dt: float = 0.5,
|
||||
led_channel: Literal["cc"] = "cc",
|
||||
) -> None:
|
||||
"""Flash the LED to identify the board
|
||||
|
||||
This is intended to be useful in situations where there are multiple
|
||||
Sangaboards in use, and it is necessary to identify which one is
|
||||
being addressed.
|
||||
"""
|
||||
led_command = f"led_{led_channel}"
|
||||
with self.sangaboard() as sb:
|
||||
return_value = sb.query(f"{led_command}?")
|
||||
if not return_value.startswith("CC LED:"):
|
||||
raise IOError("The sangaboard does not support LED control")
|
||||
|
||||
# Reading and setting LED brightness suffers from repeated reads and writes
|
||||
# decreasing the value. Rather than use the value the code warns that the value
|
||||
# cannot be used.
|
||||
intended_brightness = float(return_value[7:])
|
||||
on_brightness = 0.32
|
||||
logging.warning(
|
||||
"Brightness control is not yet implemented. Desired brightness: "
|
||||
f"{intended_brightness}. Set brightness: {on_brightness}"
|
||||
)
|
||||
for i in range(number_of_flashes):
|
||||
sb.query(f"{led_command} 0")
|
||||
time.sleep(dt)
|
||||
sb.query(f"{led_command} {on_brightness}")
|
||||
time.sleep(dt)
|
||||
Loading…
Add table
Add a link
Reference in a new issue