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

Merge branch 'update-for-labthings0.0.12' into 'v3'

Browse source 
Update to be compatible with labthings-fastapi 0.0.12/0.0.13

See merge request openflexure/openflexure-microscope-server!450
This commit is contained in:
Julian Stirling 2025-12-16 14:18:27 +00:00
commit 7038ecdf0b
36 changed files with 457 additions and 519 deletions
Download patch file Download diff file Expand all files Collapse all files

50
hardware-specific-tests/picamera2/cam_test_utils/__init__.py
View file

@ -8,13 +8,36 @@ from fastapi.testclient import TestClient
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
@contextmanager
def camera_test_client_and_server(settings_folder: Optional[str] = None):
"""Yield a camera ThingClient and the associated camera server.
This is a context manager, not a pytest fixture, as it needs to be created
multiple times in some tests.
:param cam: The camera Thing to be used. If not supplied a new one will be created.
:param settings_folder: The settings folder for the camera, if none is supplied, new
temporary directory will be used as the settings folder.
"""
# Create a temp dir, if the setting folder is set it isn't really needed
# but doesn't add much overhead.
with tempfile.TemporaryDirectory() as tmpdir:
if settings_folder is None:
settings_folder = tmpdir
thing_conf = {
"camera": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2",
}
server = lt.ThingServer(things=thing_conf, settings_folder=settings_folder)
with TestClient(server.app) as test_client:
client = lt.ThingClient.from_url("/camera/", client=test_client)
yield client, server
del server
@contextmanager
def camera_test_client(
cam: Optional[StreamingPiCamera2] = None, settings_folder: Optional[str] = None
):
def camera_test_client(settings_folder: Optional[str] = None):
"""Yield a camera ThingClient on a camera server.
This is a context manager, not a pytest fixture, as it needs to be created
@ -24,18 +47,7 @@ def camera_test_client(
:param settings_folder: The settings folder for the camera, if none is supplied, new
temporary directory will be used as the settings folder.
"""
if cam is None:
cam = StreamingPiCamera2()
# Create a temp dir, if the setting folder is set it isn't really needed
# but doesn't add much overhead.
with tempfile.TemporaryDirectory() as tmpdir:
if settings_folder is None:
settings_folder = tmpdir
server = lt.ThingServer(settings_folder=settings_folder)
server.add_thing(cam, "/camera/")
with TestClient(server.app) as test_client:
client = lt.ThingClient.from_url("/camera/", client=test_client)
yield client
del server
del cam
with camera_test_client_and_server(
settings_folder=settings_folder
) as client_and_server:
yield client_and_server[0]

28
hardware-specific-tests/picamera2/conftest.py
View file

@ -4,30 +4,32 @@ import pytest
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
from .cam_test_utils import camera_test_client
from .cam_test_utils import camera_test_client_and_server
@pytest.fixture
def picamera_thing() -> StreamingPiCamera2:
"""Return a StreamingPiCamera2 Thing.
def picamera_client_and_server() -> lt.ThingClient:
"""Initialise a test picamera_client and server for the StreamingPiCamera2 Thing.
This is the Thing that the fixture picamera_client uses. It can be used to probe
the Thing directly to check actions had the expected response.
This fixture:
* Sets up a ThingServer,
* Registers a StreamingPiCamera2 instance at the "camera" endpoint
* Yields a ThingClient and the server for interacting with it during tests.
* The picamera thing can be found at server.things["camera"]
"""
return StreamingPiCamera2()
with camera_test_client_and_server() as client_and_server:
yield client_and_server
@pytest.fixture
def picamera_client(picamera_thing) -> lt.ThingClient:
def picamera_client(picamera_client_and_server) -> lt.ThingClient:
"""Initialise a test picamera_client for the StreamingPiCamera2 Thing.
This fixture:
* Sets up a ThingServer,
* Registers a StreamingPiCamera2 instance at the "/camera/" endpoint
* Provides a ThingClient for interacting with it during tests.
* Registers a StreamingPiCamera2 instance at the "camera" endpoint
* return a ThingClient for interacting with it during tests.
"""
with camera_test_client(cam=picamera_thing) as picamera_client:
yield picamera_client
return picamera_client_and_server[0]

4
hardware-specific-tests/picamera2/test_calibration.py
View file

@ -6,8 +6,10 @@ from copy import deepcopy
from .cam_test_utils import camera_test_client
def test_calibration(picamera_thing, picamera_client):
def test_calibration(picamera_client_and_server):
"""Check that full auto calibrate completes and set the expected values."""
picamera_client, server = picamera_client_and_server
picamera_thing = server.things["camera"]
# Check the calibration_required property used by the calibration wizard
assert picamera_thing.calibration_required
# Save copy of default tuning file for end of test

14
ofm_config_full.json
View file

@ -1,22 +1,22 @@
{
"things": {
"/camera/": {
"camera": {
"class": "openflexure_microscope_server.things.camera.picamera:StreamingPiCamera2",
"kwargs": {
"camera_board": "picamera_v2"
}
},
"/stage/": "openflexure_microscope_server.things.stage.sangaboard:SangaboardThing",
"/autofocus/": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"/camera_stage_mapping/": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
"/system/": "openflexure_microscope_server.things.system:OpenFlexureSystem",
"/smart_scan/": {
"stage": "openflexure_microscope_server.things.stage.sangaboard:SangaboardThing",
"autofocus": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"camera_stage_mapping": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
"system": "openflexure_microscope_server.things.system:OpenFlexureSystem",
"smart_scan": {
"class": "openflexure_microscope_server.things.smart_scan:SmartScanThing",
"kwargs": {
"scans_folder": "/var/openflexure/scans/"
}
},
"/stage_measure/":"openflexure_microscope_server.things.stage_measure:RangeofMotionThing"
"stage_measure":"openflexure_microscope_server.things.stage_measure:RangeofMotionThing"
},
"settings_folder": "/var/openflexure/settings/",
"log_folder": "/var/openflexure/logs/"

4
ofm_config_manual.json
View file

@ -1,10 +1,10 @@
{
"things": {
"/camera/": {
"camera": {
"class": "openflexure_microscope_server.things.camera.opencv:OpenCVCamera",
"kwargs": {"camera_index": 0}
},
"/system/": "openflexure_microscope_server.things.system:OpenFlexureSystem"
"system": "openflexure_microscope_server.things.system:OpenFlexureSystem"
},
"settings_folder": "./openflexure/settings/",
"log_folder": "./openflexure/logs/"

12
ofm_config_simulation.json
View file

@ -1,11 +1,11 @@
{
"things": {
"/camera/": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"/stage/": "openflexure_microscope_server.things.stage.dummy:DummyStage",
"/autofocus/": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"/camera_stage_mapping/": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
"/system/": "openflexure_microscope_server.things.system:OpenFlexureSystem",
"/smart_scan/": {
"camera": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"stage": "openflexure_microscope_server.things.stage.dummy:DummyStage",
"autofocus": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"camera_stage_mapping": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
"system": "openflexure_microscope_server.things.system:OpenFlexureSystem",
"smart_scan": {
"class": "openflexure_microscope_server.things.smart_scan:SmartScanThing",
"kwargs": {
"scans_folder": "./openflexure/scans/"

BIN
picamera_coverage.zip
View file

Binary file not shown.

7
pyproject.toml
View file

@ -18,12 +18,11 @@ classifiers = [
"Operating System :: OS Independent",
]
dependencies = [
"labthings-fastapi == 0.0.11",
"fastapi <= 0.123.6", # This can be removed once #216 on labthings-fastapi is fixed
"labthings-fastapi==0.0.13",
"sangaboard",
"camera-stage-mapping ~= 0.1.10",
"opencv-python ~= 4.11.0",
"openflexure-stitching[libvips]==0.2.2",
"openflexure-stitching[libvips]==0.2.3",
"pillow ~= 10.4",
"anyio ~= 4.0",
"numpy ~= 2.2",
@ -161,7 +160,7 @@ ignore = [
[tool.ruff.lint.pydocstyle]
# This lets the D401 checker understand that decorated thing properties and thing
# settings act like properties so should be documented as such.
property-decorators = ["labthings_fastapi.thing_property", "labthings_fastapi.thing_setting"]
property-decorators = ["labthings_fastapi.property", "labthings_fastapi.setting"]
[tool.ruff.lint.pylint]
max-args = 7

18
src/openflexure_microscope_server/server/__init__.py
View file

@ -12,6 +12,7 @@ import uvicorn
from uvicorn.main import Server
import labthings_fastapi as lt
from labthings_fastapi.server import fallback
from openflexure_microscope_server.utilities import load_patched_config
@ -62,11 +63,11 @@ def customise_server(
def _get_scans_dir(config: dict) -> Optional[str]:
"""Read the config and return the scans directory.
Return is None if there is no /smart_scan/ thing loaded.
Return is None if there is no smart_scan thing loaded.
"""
if "/smart_scan/" in config["things"]:
if "smart_scan" in config["things"]:
try:
return config["things"]["/smart_scan/"]["kwargs"]["scans_folder"]
return config["things"]["smart_scan"]["kwargs"]["scans_folder"]
except KeyError as e:
msg = "Configuration error: smart scan should have scans_folder kwarg set"
raise RuntimeError(msg) from e
@ -87,15 +88,15 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
server = None
try:
config = _full_config_from_args(args)
log_folder = config.get("log_folder", "./openflexure/logs")
log_folder = config.pop("log_folder", "./openflexure/logs")
scans_folder = _get_scans_dir(config)
server = lt.cli.server_from_config(config)
server = lt.ThingServer.from_config(config)
customise_server(server, log_folder, scans_folder)
def shutdown_call() -> None:
try:
# Kill any mjpeg streams so that StreamingResponses close.
server.things["/camera/"].kill_mjpeg_streams()
server.things["camera"].kill_mjpeg_streams()
except BaseException as e:
# Catch anything and log as it is essential that this
# function cannot raise an unhandled exception or Uvicorn
@ -119,9 +120,8 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
# Allow printing to the terminal for fallback errors so they are not
# presented in the fallback logs.
print(f"Error: {e}") # noqa: T201
fallback_server = "labthings_fastapi.server.fallback:app"
print(f"Starting fallback server {fallback_server}.") # noqa: T201
app = lt.cli.object_reference_to_object(fallback_server)
print("Starting fallback server.") # noqa: T201
app = fallback.app
app.labthings_config = config
app.labthings_server = server
app.labthings_error = e

2
src/openflexure_microscope_server/server/legacy_api.py
View file

@ -38,7 +38,7 @@ def add_v2_endpoints(thing_server: lt.ThingServer) -> None:
@app.head("/api/v2/streams/snapshot")
async def thumbnail() -> JPEGResponse:
"""Return a low-resolution snapshot, for compatibility with OF connect."""
blob = await thing_server.things["/camera/"].lores_mjpeg_stream.grab_frame()
blob = await thing_server.things["camera"].lores_mjpeg_stream.grab_frame()
return JPEGResponse(blob)
@app.get("/api/v2/instrument/settings/name")

22
src/openflexure_microscope_server/things/autofocus.py
View file

@ -350,7 +350,7 @@ class AutofocusThing(lt.Thing):
field of view to assess focus (autofocus and testing the success of a z-stack)
"""
@lt.thing_action
@lt.action
def fast_autofocus(
self,
sharpness_monitor: SharpnessMonitorDep,
@ -381,7 +381,7 @@ class AutofocusThing(lt.Thing):
# Return all focus data
return sharpness_monitor.data_dict()
@lt.thing_action
@lt.action
def z_move_and_measure_sharpness(
self,
sharpness_monitor: SharpnessMonitorDep,
@ -407,7 +407,7 @@ class AutofocusThing(lt.Thing):
sharpness_monitor.focus_rel(current_dz)
return sharpness_monitor.data_dict()
@lt.thing_action
@lt.action
def looping_autofocus(
self,
stage: Stage,
@ -455,19 +455,13 @@ class AutofocusThing(lt.Thing):
"Looping autofocus couldn't converge on a focus location."
)
stack_images_to_save = lt.ThingSetting(
initial_value=1,
model=int,
)
stack_images_to_save: int = lt.setting(default=1)
"""The number of images to save in a stack.
Defaults to 1 unless you need to see either side of focus
"""
stack_min_images_to_test = lt.ThingSetting(
initial_value=9,
model=int,
)
stack_min_images_to_test: int = lt.setting(default=9)
"""The minimum number of images to capture in a stack.
This many images are captures and tested for focus, if the focus is not central
@ -477,7 +471,7 @@ class AutofocusThing(lt.Thing):
Defaults to 9 which balances reliability and speed.
"""
stack_dz = lt.ThingSetting(initial_value=50, model=int)
stack_dz: int = lt.setting(default=50)
"""Distance in steps between images in a z-stack.
Suggested values:
@ -487,7 +481,7 @@ class AutofocusThing(lt.Thing):
* 200 for 20x
"""
@lt.thing_action
@lt.action
def create_stack_params(
self,
images_dir: str,
@ -560,7 +554,7 @@ class AutofocusThing(lt.Thing):
save_resolution=save_resolution,
)
@lt.thing_action
@lt.action
def run_smart_stack(
self,
cam: CameraClient,

75
src/openflexure_microscope_server/things/camera/__init__.py
View file

@ -175,7 +175,7 @@ class BaseCamera(lt.Thing):
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
def __init__(self) -> None:
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise the base camera, this creates the background detectors.
This must be run by all child camera classes.
@ -183,7 +183,7 @@ class BaseCamera(lt.Thing):
To add a new background detector to the server it must be added to the
dictionary in this function. Configuration will be added at a later date.
"""
super().__init__()
super().__init__(thing_server_interface)
self.background_detectors = {
"Colour Channels (LUV)": ColourChannelDetectLUV(),
"Channel Deviations (LUV)": ChannelDeviationLUV(),
@ -203,7 +203,7 @@ class BaseCamera(lt.Thing):
"""Close hardware connection when the Thing context manager is closed."""
raise NotImplementedError("CameraThings must define their own __exit__ method")
@lt.thing_property
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating.
@ -212,7 +212,7 @@ class BaseCamera(lt.Thing):
"""
return False
@lt.thing_action
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int], buffer_count: int
) -> None:
@ -239,21 +239,21 @@ class BaseCamera(lt.Thing):
self.mjpeg_stream._streaming = False
self.lores_mjpeg_stream._streaming = False
@lt.thing_property
@lt.property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active."""
raise NotImplementedError(
"CameraThings must define their own stream_active method"
)
@lt.thing_action
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
raise NotImplementedError(
"CameraThings must define their own discard_frames method"
)
@lt.thing_action
@lt.action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
@ -264,10 +264,10 @@ class BaseCamera(lt.Thing):
"CameraThings must define their own capture_array method"
)
downsampled_array_factor = lt.ThingProperty(int, 2)
downsampled_array_factor: int = lt.property(default=2)
"""The downsampling factor when calling capture_downsampled_array."""
@lt.thing_action
@lt.action
def capture_downsampled_array(self) -> ArrayModel:
"""Acquire one image from the camera, downsample, and return as an array.
@ -279,7 +279,7 @@ class BaseCamera(lt.Thing):
img = self.capture_array()
return downsample(self.downsampled_array_factor, img)
@lt.thing_action
@lt.action
def capture_jpeg(
self,
metadata_getter: lt.deps.GetThingStates,
@ -316,10 +316,9 @@ class BaseCamera(lt.Thing):
return JPEGBlob.from_temporary_directory(directory, fname)
@lt.thing_action
@lt.action
def grab_jpeg(
self,
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as blob of JPEG data.
@ -337,13 +336,12 @@ class BaseCamera(lt.Thing):
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
frame = portal.call(stream.grab_frame)
frame = self._thing_server_interface.call_async_task(stream.grab_frame)
return JPEGBlob.from_bytes(frame)
@lt.thing_action
@lt.action
def grab_as_array(
self,
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> ArrayModel:
"""Acquire one image from the preview stream and return as an array.
@ -361,23 +359,22 @@ class BaseCamera(lt.Thing):
tries = 0
while tries < 3:
try:
frame = portal.call(stream.grab_frame)
frame = self._thing_server_interface.call_async_task(stream.grab_frame)
return np.asarray(Image.open(io.BytesIO(frame)))
except OSError:
tries += 1
raise OSError("Could not open frames from MJPEG stream.")
@lt.thing_action
@lt.action
def grab_jpeg_size(
self,
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size."""
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
return portal.call(stream.next_frame_size)
return self._thing_server_interface.call_async_task(stream.next_frame_size)
def capture_image(
self,
@ -389,7 +386,7 @@ class BaseCamera(lt.Thing):
"CameraThings must define their own capture_image method"
)
@lt.thing_action
@lt.action
def capture_and_save(
self,
jpeg_path: str,
@ -420,7 +417,7 @@ class BaseCamera(lt.Thing):
save_resolution,
)
@lt.thing_action
@lt.action
def capture_to_memory(
self,
logger: lt.deps.InvocationLogger,
@ -444,7 +441,7 @@ class BaseCamera(lt.Thing):
image, metadata = self._robust_image_capture(metadata_getter, logger)
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
@lt.thing_action
@lt.action
def save_from_memory(
self,
jpeg_path: str,
@ -473,7 +470,7 @@ class BaseCamera(lt.Thing):
save_resolution=save_resolution,
)
@lt.thing_action
@lt.action
def clear_buffers(self) -> None:
"""Clear all images in memory."""
self._memory_buffer.clear()
@ -600,10 +597,10 @@ class BaseCamera(lt.Thing):
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e
settling_time = lt.ThingSetting(float, 0.2)
settling_time: float = lt.setting(default=0.2)
"""The settling time when calling the ``settle()`` method."""
@lt.thing_action
@lt.action
def settle(self) -> None:
"""Sleep for the settling time, ready to provide a fresh frame.
@ -616,24 +613,24 @@ class BaseCamera(lt.Thing):
time.sleep(self.settling_time)
self.discard_frames()
@lt.thing_property
@lt.property
def primary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions for both calibration wizard and settings panel."""
return []
@lt.thing_property
@lt.property
def secondary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions that appear only in settings panel."""
return []
@lt.thing_property
@lt.property
def manual_camera_settings(self) -> list[PropertyControl]:
"""The camera settings to expose as property controls in the settings panel."""
return []
# Note that the default detector name is set at init. This is over written if
# setting is loaded from disk.
@lt.thing_setting
@lt.setting
def detector_name(self) -> str:
"""The name of the active background selector."""
return self._detector_name
@ -650,12 +647,12 @@ class BaseCamera(lt.Thing):
"""The active background detector instance."""
return self.background_detectors[self.detector_name]
@lt.thing_property
@lt.property
def background_detector_status(self) -> BackgroundDetectorStatus:
"""The status of the active detector for the UI."""
return self.active_detector.status
@lt.thing_action
@lt.action
def update_detector_settings(self, data: dict[str, Any]) -> None:
"""Update the settings of the current detector.
@ -671,7 +668,7 @@ class BaseCamera(lt.Thing):
# Manually save settings as the setter is not called.
self.save_settings()
@lt.thing_setting
@lt.setting
def background_detector_data(self) -> dict:
"""The data for each background detector, used to save to disk."""
data = {}
@ -704,14 +701,14 @@ class BaseCamera(lt.Thing):
f"No background detector named {name}, settings will be discarded."
)
@lt.thing_action
def image_is_sample(self, portal: lt.deps.BlockingPortal) -> tuple[bool, str]:
@lt.action
def image_is_sample(self) -> tuple[bool, str]:
"""Label the current image as either background or sample."""
current_image = self.grab_as_array(portal, stream_name="lores")
current_image = self.grab_as_array(stream_name="lores")
return self.active_detector.image_is_sample(current_image)
@lt.thing_action
def set_background(self, portal: lt.deps.BlockingPortal) -> None:
@lt.action
def set_background(self) -> None:
"""Grab an image, and use its statistics to set the background.
This should be run when the microscope is looking at an empty region,
@ -720,7 +717,7 @@ class BaseCamera(lt.Thing):
future images to the distribution, to determine if each pixel is
foreground or background.
"""
background = self.grab_as_array(portal, stream_name="lores")
background = self.grab_as_array(stream_name="lores")
self.active_detector.set_background(background)
# Manually save settings as the setter is not called.
self.save_settings()
@ -731,7 +728,7 @@ class BaseCamera(lt.Thing):
return {}
CameraDependency = lt.deps.direct_thing_client_dependency(BaseCamera, "/camera/")
CameraDependency = lt.deps.direct_thing_client_dependency(BaseCamera, "camera")
RawCameraDependency = lt.deps.raw_thing_dependency(BaseCamera)

17
src/openflexure_microscope_server/things/camera/opencv.py
View file

@ -27,12 +27,14 @@ LOGGER = logging.getLogger(__name__)
class OpenCVCamera(BaseCamera):
"""A Thing that provides and interface to an OpenCV Camera."""
def __init__(self, camera_index: int = 0) -> None:
def __init__(
self, thing_server_interface: lt.ThingServerInterface, camera_index: int = 0
) -> None:
"""Iniatilise the thing storing the index of the camera to use.
:param camera_index: The index of the camera to use for the microscope.
"""
super().__init__()
super().__init__(thing_server_interface)
self.camera_index = camera_index
self._capture_thread: Optional[Thread] = None
self._capture_enabled = False
@ -60,7 +62,7 @@ class OpenCVCamera(BaseCamera):
self._capture_thread.join()
self.cap.release()
@lt.thing_property
@lt.property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active."""
if self._capture_enabled and self._capture_thread:
@ -68,25 +70,24 @@ class OpenCVCamera(BaseCamera):
return False
def _capture_frames(self) -> None:
portal = lt.get_blocking_portal(self)
while self._capture_enabled:
ret, frame = self.cap.read()
if not ret:
LOGGER.error(f"Failed to capture frame from camera {self.camera_index}")
break
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
self.mjpeg_stream.add_frame(jpeg, portal)
self.mjpeg_stream.add_frame(jpeg)
jpeg_lores = cv2.imencode(".jpg", cv2.resize(frame, (320, 240)))[
1
].tobytes()
self.lores_mjpeg_stream.add_frame(jpeg_lores, portal)
self.lores_mjpeg_stream.add_frame(jpeg_lores)
@lt.thing_action
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
self.capture_array()
@lt.thing_action
@lt.action
def capture_array(
self,
stream_name: Literal["main", "full"] = "full",

123
src/openflexure_microscope_server/things/camera/picamera.py
View file

@ -35,7 +35,7 @@ from PIL import Image
from pydantic import BaseModel, BeforeValidator
import labthings_fastapi as lt
from labthings_fastapi.exceptions import NotConnectedToServerError
from labthings_fastapi.exceptions import ServerNotRunningError
from openflexure_microscope_server.background_detect import ChannelBlankError
from openflexure_microscope_server.ui import (
@ -68,19 +68,13 @@ class MissingCalibrationError(RuntimeError):
class PicameraStreamOutput(Output):
"""An Output class that sends frames to a stream."""
def __init__(
self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal
) -> None:
def __init__(self, stream: lt.outputs.MJPEGStream) -> None:
"""Create an output that puts frames in an MJPEGStream.
We need to pass the stream object, and also the blocking portal, because
new frame notifications happen in the anyio event loop and frames are
sent from a thread. The blocking portal enables thread-to-async
communication.
:param stream: The labthings MJPEGStream to send frames to.
"""
Output.__init__(self)
self.stream = stream
self.portal = portal
def outputframe(
self,
@ -91,7 +85,7 @@ class PicameraStreamOutput(Output):
_audio: bool = False,
) -> None:
"""Add a frame to the stream's ringbuffer."""
self.stream.add_frame(frame, self.portal)
self.stream.add_frame(frame)
class SensorMode(BaseModel):
@ -128,7 +122,12 @@ class StreamingPiCamera2(BaseCamera):
generalisation.
"""
def __init__(self, camera_num: int = 0, camera_board: str = "picamera_v2") -> None:
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
camera_num: int = 0,
camera_board: str = "picamera_v2",
) -> None:
"""Initialise the camera with the given camera number.
This makes no connection to the camera (except to get the default tuning file).
@ -138,7 +137,7 @@ class StreamingPiCamera2(BaseCamera):
:param camera_board: The camera board used. Supported options are "picamera_v2"
and "picamera_hq".
"""
super().__init__()
super().__init__(thing_server_interface)
self._setting_save_in_progress = False
self._camera_num = camera_num
self._camera_board = camera_board
@ -160,40 +159,29 @@ class StreamingPiCamera2(BaseCamera):
# connected to the server if tuning is saved to disk.
try:
self.tuning = copy.deepcopy(self.default_tuning)
except NotConnectedToServerError:
except ServerNotRunningError:
# This will throw an error after setting as we are not connected to
# a server. But we know this, so we ignore the error.
pass
# Also set the colour gains based on the tuning. Set to _colour_gains to not
# trigger a NotConnectedToServerError
# trigger a ServerNotRunningError
self._colour_gains = tf_utils.get_colour_gains_from_lst(self.tuning)
stream_resolution = lt.ThingProperty(
tuple[int, int],
initial_value=(820, 616),
)
stream_resolution: tuple[int, int] = lt.property(default=(820, 616))
"""Resolution to use for the MJPEG stream."""
mjpeg_bitrate = lt.ThingProperty(
Optional[int],
initial_value=100000000,
)
mjpeg_bitrate: Optional[int] = lt.property(default=100000000)
"""Bitrate for MJPEG stream (None for default)."""
stream_active = lt.ThingProperty(
bool,
initial_value=False,
observable=True,
readonly=True,
)
stream_active: bool = lt.property(default=False, readonly=True)
"""Whether the MJPEG stream is active."""
def save_settings(self) -> None:
"""Override save_settings to ensure that camera properties don't recurse.
This method is run by any Thing when a ThingSetting is saved. However, the
method reads the thing_setting. As reading the thing setting talks to the
This method is run by any Thing when a setting is saved. However, the
method reads the setting. As reading the setting talks to the
camera and calls save_settings if the value is not as expected, this could
cause recursion. Also this means that saving one setting causes all others
to be read each time.
@ -204,7 +192,7 @@ class StreamingPiCamera2(BaseCamera):
finally:
self._setting_save_in_progress = False
@lt.thing_property
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating."""
# Check if the lens shading table is calibrated.
@ -214,7 +202,7 @@ class StreamingPiCamera2(BaseCamera):
_analogue_gain: float = 1.0
@lt.thing_setting
@lt.setting
def analogue_gain(self) -> float:
"""The Analogue gain applied by the camera sensor."""
if not self._setting_save_in_progress and self.streaming:
@ -234,7 +222,7 @@ class StreamingPiCamera2(BaseCamera):
_colour_gains: tuple[float, float] = (1.0, 1.0)
@lt.thing_setting
@lt.setting
def colour_gains(self) -> tuple[float, float]:
"""The red and blue colour gains, must be between 0.0 and 32.0."""
if not self._setting_save_in_progress and self.streaming:
@ -254,7 +242,7 @@ class StreamingPiCamera2(BaseCamera):
_exposure_time: int = 500
@lt.thing_setting
@lt.setting
def exposure_time(self) -> int:
"""The camera exposure time in microseconds.
@ -296,7 +284,7 @@ class StreamingPiCamera2(BaseCamera):
_sensor_modes = None
@lt.thing_property
@lt.property
def sensor_modes(self) -> list[SensorMode]:
"""All the available modes the current sensor supports."""
if not self._sensor_modes:
@ -306,7 +294,7 @@ class StreamingPiCamera2(BaseCamera):
_sensor_mode: Optional[dict] = None
@lt.thing_property
@lt.property
def sensor_mode(self) -> Optional[SensorModeSelector]:
"""The intended sensor mode of the camera."""
if self._sensor_mode is None:
@ -329,13 +317,13 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera(pause_stream=True):
pass
@lt.thing_property
@lt.property
def sensor_resolution(self) -> Optional[tuple[int, int]]:
"""The native resolution of the camera's sensor."""
with self._streaming_picamera() as cam:
return cam.sensor_resolution
tuning = lt.ThingSetting(Optional[dict], None, readonly=True)
tuning: Optional[dict] = lt.setting(default=None, readonly=True)
"""The Raspberry PiCamera Tuning File JSON."""
def _initialise_picamera(self, check_sensor_model: bool = False) -> None:
@ -437,7 +425,7 @@ class StreamingPiCamera2(BaseCamera):
cam.close()
del self._picamera
@lt.thing_action
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 6
) -> None:
@ -486,18 +474,12 @@ class StreamingPiCamera2(BaseCamera):
stream_name = "lores" if main_resolution[0] > 1280 else "main"
picam.start_recording(
MJPEGEncoder(self.mjpeg_bitrate),
PicameraStreamOutput(
self.mjpeg_stream,
lt.get_blocking_portal(self),
),
PicameraStreamOutput(self.mjpeg_stream),
name=stream_name,
)
picam.start_encoder(
MJPEGEncoder(100000000),
PicameraStreamOutput(
self.lores_mjpeg_stream,
lt.get_blocking_portal(self),
),
PicameraStreamOutput(self.lores_mjpeg_stream),
name="lores",
)
except Exception as e:
@ -509,7 +491,7 @@ class StreamingPiCamera2(BaseCamera):
"Started MJPEG stream at %s on port %s", self.stream_resolution, 1
)
@lt.thing_action
@lt.action
def stop_streaming(self, stop_web_stream: bool = True) -> None:
"""Stop the MJPEG stream."""
with self._streaming_picamera() as picam:
@ -521,15 +503,14 @@ class StreamingPiCamera2(BaseCamera):
else:
self.stream_active = False
if stop_web_stream:
portal = lt.get_blocking_portal(self)
self.mjpeg_stream.stop(portal)
self.lores_mjpeg_stream.stop(portal)
self.mjpeg_stream.stop()
self.lores_mjpeg_stream.stop()
LOGGER.info("Stopped MJPEG stream.")
# Adding a sleep to prevent camera getting confused by rapid commands
time.sleep(self._sensor_info.short_pause)
@lt.thing_action
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh."""
with self._streaming_picamera() as cam:
@ -587,7 +568,7 @@ class StreamingPiCamera2(BaseCamera):
else:
raise ValueError(f'Unknown stream name "{stream_name}"')
@lt.thing_action
@lt.action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
@ -620,7 +601,7 @@ class StreamingPiCamera2(BaseCamera):
# as array
return np.array(self.capture_image(stream_name, wait))
@lt.thing_property
@lt.property
def camera_configuration(self) -> Mapping:
"""The "configuration" dictionary of the picamera2 object.
@ -635,13 +616,13 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera() as cam:
return cam.camera_configuration()
@lt.thing_property
@lt.property
def capture_metadata(self) -> dict:
"""Return the metadata from the camera."""
with self._streaming_picamera() as cam:
return cam.capture_metadata()
@lt.thing_action
@lt.action
def auto_expose_from_minimum(
self,
target_white_level: Optional[int] = None,
@ -671,7 +652,7 @@ class StreamingPiCamera2(BaseCamera):
percentile=percentile,
)
@lt.thing_action
@lt.action
def calibrate_lens_shading(self) -> None:
"""Take an image and use it for flat-field correction.
@ -699,7 +680,7 @@ class StreamingPiCamera2(BaseCamera):
self.colour_gains = tf_utils.get_colour_gains_from_lst(self.tuning)
@lt.thing_property
@lt.property
def colour_correction_matrix(
self,
) -> tuple[float, float, float, float, float, float, float, float, float]:
@ -724,7 +705,7 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera(pause_stream=True):
self._initialise_picamera()
@lt.thing_action
@lt.action
def reset_ccm(self) -> None:
"""Overwrite the colour correction matrix in camera tuning with default values."""
self.tuning = tf_utils.copy_algo_from_other_tuning(
@ -733,7 +714,7 @@ class StreamingPiCamera2(BaseCamera):
copy_from=self.default_tuning,
)
@lt.thing_action
@lt.action
def set_static_green_equalisation(self, offset: int = 65535) -> None:
"""Set the green equalisation to a static value.
@ -748,7 +729,7 @@ class StreamingPiCamera2(BaseCamera):
self.tuning = tf_utils.set_static_geq(self.tuning, offset)
self._initialise_picamera()
@lt.thing_action
@lt.action
def set_ce_enable_to_off(self) -> None:
"""Set the contrast enhancement to disabled.
@ -759,8 +740,8 @@ class StreamingPiCamera2(BaseCamera):
self.tuning = tf_utils.set_ce_to_disabled(self.tuning)
self._initialise_picamera()
@lt.thing_action
def full_auto_calibrate(self, portal: lt.deps.BlockingPortal) -> None:
@lt.action
def full_auto_calibrate(self) -> None:
"""Perform a full auto-calibration.
This function will call the other calibration actions in sequence:
@ -779,7 +760,7 @@ class StreamingPiCamera2(BaseCamera):
for _i in range(3):
try:
time.sleep(self._sensor_info.long_pause)
self.set_background(portal)
self.set_background()
# Return if background is set
return
except ChannelBlankError:
@ -787,7 +768,7 @@ class StreamingPiCamera2(BaseCamera):
pass
raise RuntimeError("Couldn't set background")
@lt.thing_property
@lt.property
def primary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions for both calibration wizard and settings panel."""
return [
@ -805,7 +786,7 @@ class StreamingPiCamera2(BaseCamera):
),
]
@lt.thing_property
@lt.property
def secondary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions that appear only in settings panel."""
return [
@ -847,7 +828,7 @@ class StreamingPiCamera2(BaseCamera):
),
]
@lt.thing_property
@lt.property
def manual_camera_settings(self) -> list[PropertyControl]:
"""The camera settings to expose as property controls in the settings panel."""
return [
@ -874,7 +855,7 @@ class StreamingPiCamera2(BaseCamera):
),
]
@lt.thing_property
@lt.property
def lens_shading_tables(self) -> Optional[tf_utils.LensShading]:
"""The current lens shading (i.e. flat-field correction).
@ -901,7 +882,7 @@ class StreamingPiCamera2(BaseCamera):
)
self._initialise_picamera()
@lt.thing_action
@lt.action
def flat_lens_shading(self) -> None:
"""Disable flat-field correction.
@ -916,7 +897,7 @@ class StreamingPiCamera2(BaseCamera):
self.tuning = tf_utils.flatten_lst(self.tuning)
self._initialise_picamera()
@lt.thing_action
@lt.action
def flat_lens_shading_chrominance(self) -> None:
"""Disable flat-field correction for colour only.
@ -928,7 +909,7 @@ class StreamingPiCamera2(BaseCamera):
self.tuning = tf_utils.flatten_lst(self.tuning, keep_luminance=True)
self._initialise_picamera()
@lt.thing_action
@lt.action
def reset_lens_shading(self) -> None:
"""Revert to default lens shading settings.

69
src/openflexure_microscope_server/things/camera/simulation.py
View file

@ -13,7 +13,7 @@ import logging
import time
from threading import Thread
from types import TracebackType
from typing import Literal, Mapping, Optional
from typing import Literal, Optional
import numpy as np
from PIL import Image, ImageFilter
@ -27,7 +27,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from ..stage import BaseStage
from ..stage.dummy import DummyStage
from . import ArrayModel, BaseCamera
LOGGER = logging.getLogger(__name__)
@ -46,12 +46,13 @@ RNG = np.random.default_rng()
class SimulatedCamera(BaseCamera):
"""A Thing that simulates a camera for testing."""
_stage: Optional[BaseStage] = None
_server: Optional[lt.ThingServer] = None
_stage: DummyStage = lt.thing_slot()
_show_sample: bool = True
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
shape: tuple[int, int, int] = (616, 820, 3),
glyph_shape: tuple[int, int, int] = (121, 121, 3),
canvas_shape: tuple[int, int, int] = (3000, 4000, 3),
@ -71,7 +72,7 @@ class SimulatedCamera(BaseCamera):
:param frame_interval: Nominally the time between frames on the MJPEG stream,
however the rate may be slower due to calculation time for focus.
"""
super().__init__()
super().__init__(thing_server_interface)
self.shape = shape
self.glyph_shape = glyph_shape
self.canvas_shape = canvas_shape
@ -86,7 +87,7 @@ class SimulatedCamera(BaseCamera):
self.generate_blobs()
self.generate_canvas()
@lt.thing_property
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating."""
return not self.background_detector_status.ready
@ -228,31 +229,10 @@ class SimulatedCamera(BaseCamera):
image[image > 255] = 255
return Image.fromarray(image.astype("uint8"))
def attach_to_server(
self, server: lt.ThingServer, path: str, setting_storage_path: str
) -> None:
"""Wrap the attach_to_server method so the server instance can be stored.
Direct access to the server instance is needed to get the stage position while
maintaining the same public API as a real camera that doesn't need this access.
"""
self._server = server
super().attach_to_server(server, path, setting_storage_path)
def get_stage_position(self) -> Mapping[str, int]:
"""Return the stage position.
The simulation camera has access to the stage position so it can generate a
different image as the stage moves.
"""
if not self._stage and self._server:
self._stage = self._server.things["/stage/"]
return self._stage.instantaneous_position
def generate_frame(self) -> Image:
"""Generate a frame with blobs based on the stage coordinates."""
try:
pos = self.get_stage_position()
pos = self._stage.instantaneous_position
except Exception as e:
LOGGER.debug(f"Failed to get stage position: {e}")
pos = {"x": 0, "y": 0, "z": 0}
@ -274,7 +254,7 @@ class SimulatedCamera(BaseCamera):
self._capture_enabled = False
self._capture_thread.join()
@lt.thing_action
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 1
) -> None:
@ -300,40 +280,39 @@ class SimulatedCamera(BaseCamera):
self._capture_thread = Thread(target=self._capture_frames)
self._capture_thread.start()
@lt.thing_property
@lt.property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active."""
if self._capture_enabled and self._capture_thread:
return self._capture_thread.is_alive()
return False
noise_level = lt.ThingProperty(float, 2.0)
noise_level: float = lt.property(default=2.0)
def _capture_frames(self) -> None:
portal = lt.get_blocking_portal(self)
last_frame_t = time.time()
while self._capture_enabled:
wait_time = last_frame_t - time.time() - self.frame_interval
wait_time = self.frame_interval - (time.time() - last_frame_t)
if wait_time > 0:
time.sleep(wait_time)
last_frame_t = time.time()
try:
frame = self.generate_frame()
self.mjpeg_stream.add_frame(_frame2bytes(frame), portal)
self.mjpeg_stream.add_frame(_frame2bytes(frame))
ds_frame = frame.resize((320, 240), resample=Image.NEAREST)
self.lores_mjpeg_stream.add_frame(_frame2bytes(ds_frame), portal)
self.lores_mjpeg_stream.add_frame(_frame2bytes(ds_frame))
except Exception as e:
LOGGER.exception(f"Failed to capture frame: {e}, retrying...")
@lt.thing_action
@lt.action
def discard_frames(self) -> None:
"""Discard frames so that the next frame captured is fresh.
There is nothing to do as this is a simulation!
"""
@lt.thing_action
@lt.action
def capture_array(
self,
stream_name: Literal["main", "full"] = "full",
@ -370,8 +349,8 @@ class SimulatedCamera(BaseCamera):
LOGGER.warning(f"Simulation camera camera doesn't respect {stream_name=}")
return self.generate_frame()
@lt.thing_action
def full_auto_calibrate(self, portal: lt.deps.BlockingPortal) -> None:
@lt.action
def full_auto_calibrate(self) -> None:
"""Perform a full auto-calibration.
For the simulation microscope the process is:
@ -382,25 +361,25 @@ class SimulatedCamera(BaseCamera):
"""
self.remove_sample()
time.sleep(0.2)
self.set_background(portal)
self.set_background()
time.sleep(0.2)
self.load_sample()
@lt.thing_action
@lt.action
def remove_sample(self) -> None:
"""Show the simulated background with no sample."""
if not self._show_sample:
raise RuntimeError("Sample is already removed.")
self._show_sample = False
@lt.thing_action
@lt.action
def load_sample(self) -> None:
"""Show the simulated sample."""
if self._show_sample:
raise RuntimeError("Sample is already in place.")
self._show_sample = True
@lt.thing_property
@lt.property
def primary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions for both calibration wizard and settings panel."""
return [
@ -409,7 +388,7 @@ class SimulatedCamera(BaseCamera):
),
]
@lt.thing_property
@lt.property
def secondary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions that appear only in settings panel."""
return [
@ -417,7 +396,7 @@ class SimulatedCamera(BaseCamera):
action_button_for(self.remove_sample, submit_label="Remove Sample"),
]
@lt.thing_property
@lt.property
def manual_camera_settings(self) -> list[PropertyControl]:
"""The camera settings to expose as property controls in the settings panel."""
return [property_control_for(self, "noise_level", label="Noise Level")]

22
src/openflexure_microscope_server/things/camera_stage_mapping.py
View file

@ -118,7 +118,7 @@ class CameraStageMapper(lt.Thing):
override the ``get_xyz_position()`` and ``move_to_xyz_position()`` methods.
"""
@lt.thing_action
@lt.action
def calibrate_1d(
self,
camera: CameraClient,
@ -154,7 +154,7 @@ class CameraStageMapper(lt.Thing):
result["image_resolution"] = camera.capture_downsampled_array().shape[:2]
return result
@lt.thing_action
@lt.action
def calibrate_xy(
self, camera: CameraClient, stage: Stage, logger: lt.deps.InvocationLogger
) -> DenumpifyingDict:
@ -200,7 +200,7 @@ class CameraStageMapper(lt.Thing):
return data
@lt.thing_property
@lt.property
def image_to_stage_displacement_matrix(
self,
) -> Optional[List[List[float]]]: # 2x2 integer array
@ -230,19 +230,17 @@ class CameraStageMapper(lt.Thing):
]
return np.array(displacement_matrix).tolist()
last_calibration = lt.ThingSetting(
initial_value=None, model=Optional[dict], readonly=True
)
last_calibration: Optional[dict] = lt.setting(default=None, readonly=True)
"""The most recent CSM calibration."""
@lt.thing_property
@lt.property
def image_resolution(self) -> Optional[Tuple[float, float]]:
"""The image size used to calibrate the image_to_stage_displacement_matrix."""
if self.last_calibration is None:
return None
return self.last_calibration["image_resolution"]
@lt.thing_property
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera stage mapper needs calibrating."""
return self.image_to_stage_displacement_matrix is None
@ -254,7 +252,7 @@ class CameraStageMapper(lt.Thing):
# added by CSMUncalibratedError
raise CSMUncalibratedError() # noqa: RSE102
@lt.thing_action
@lt.action
def move_in_image_coordinates(
self,
stage: Stage,
@ -275,7 +273,7 @@ class CameraStageMapper(lt.Thing):
self.assert_calibrated()
stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y))
@lt.thing_action
@lt.action
def convert_image_to_stage_coordinates(
self, x: float, y: float, **_kwargs: float
) -> Mapping[str, int]:
@ -283,7 +281,7 @@ class CameraStageMapper(lt.Thing):
self.assert_calibrated()
return csm_img_to_stage(self.image_to_stage_displacement_matrix, x=x, y=y)
@lt.thing_action
@lt.action
def convert_stage_to_image_coordinates(
self, x: int, y: int, **_kwargs: int
) -> Mapping[str, float]:
@ -291,7 +289,7 @@ class CameraStageMapper(lt.Thing):
self.assert_calibrated()
return csm_stage_to_img(self.image_to_stage_displacement_matrix, x=x, y=y)
@lt.thing_property
@lt.property
def thing_state(self) -> Mapping[str, Any]:
"""Summary metadata describing the current state of the Thing."""
return {

74
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -42,9 +42,9 @@ T = TypeVar("T")
P = ParamSpec("P")
CSMDep = lt.deps.direct_thing_client_dependency(
CameraStageMapper, "/camera_stage_mapping/"
CameraStageMapper, "camera_stage_mapping"
)
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "autofocus")
class ScanListInfo(BaseModel):
@ -103,13 +103,16 @@ class SmartScanThing(lt.Thing):
past scans.
"""
def __init__(self, scans_folder: str) -> None:
def __init__(
self, thing_server_interface: lt.ThingServerInterface, scans_folder: str
) -> None:
"""Initialise a SmartScanThing saving to and loading from the input directory.
:param scans_folder: This is the path to the directory where all scans will be
saved. Any scans already in this directory will be accessible through the
HTTP interface.
"""
super().__init__(thing_server_interface)
self._scan_dir_manager = scan_directories.ScanDirectoryManager(scans_folder)
self._scan_lock = threading.Lock()
@ -117,7 +120,7 @@ class SmartScanThing(lt.Thing):
self._latest_scan_name: Optional[str] = None
# Scan logger is the invocation logger labthings-fastapi creates
# when the `sample_scan` lt.thing_action is called. It is saved as
# when the `sample_scan` lt.action is called. It is saved as
# private class variable along with many others here.
# Access to these variables requires a scan to be running,
# any method that calls these should be decorated with
@ -133,7 +136,7 @@ class SmartScanThing(lt.Thing):
self._scan_data: Optional[scan_directories.ScanData] = None
self._preview_stitcher: Optional[stitching.PreviewStitcher] = None
@lt.thing_action
@lt.action
def sample_scan(
self,
cancel: lt.deps.CancelHook,
@ -231,7 +234,7 @@ class SmartScanThing(lt.Thing):
"of motion."
)
@lt.thing_property
@lt.property
def latest_scan_name(self) -> Optional[str]:
"""The name of the last scan to be started."""
return self._latest_scan_name
@ -523,7 +526,7 @@ class SmartScanThing(lt.Thing):
overlap=self._scan_data.overlap,
)
@lt.fastapi_endpoint(
@lt.endpoint(
"get",
"scans/stitched_thumbnail.jpg",
responses={
@ -546,48 +549,27 @@ class SmartScanThing(lt.Thing):
raise HTTPException(404, "File not found")
return FileResponse(preview_path)
save_resolution = lt.ThingSetting(
initial_value=(1640, 1232),
model=tuple[int, int],
)
save_resolution: tuple[int, int] = lt.setting(default=(1640, 1232))
"""A tuple of the image resolution to capture."""
max_range = lt.ThingSetting(
initial_value=45000,
model=int,
)
max_range: int = lt.setting(default=45000)
"""The maximum distance in steps from the centre of the scan."""
stitch_tiff = lt.ThingSetting(
initial_value=False,
model=bool,
)
stitch_tiff: bool = lt.setting(default=False)
"""Whether or not to also produce a pyramidal tiff at the end of a scan."""
skip_background = lt.ThingSetting(
initial_value=True,
model=bool,
)
skip_background: bool = lt.setting(default=True)
"""Whether to detect and skip empty fields of view.
This uses the settings from the ``BackgroundDetectThing``."""
autofocus_dz = lt.ThingSetting(
initial_value=1000,
model=int,
)
autofocus_dz: int = lt.setting(default=1000)
"""The z distance to perform an autofocus in steps."""
overlap = lt.ThingSetting(
initial_value=0.45,
model=float,
)
overlap: float = lt.setting(default=0.45)
"""The fraction (0-1) that adjacent images should overlap in x or y."""
stitch_automatically = lt.ThingSetting(
initial_value=True,
model=bool,
)
stitch_automatically: bool = lt.setting(default=True)
"""Whether to run a final stitch at the end of a successful scan."""
def _get_all_scan_info(self) -> list[scan_directories.ScanInfo]:
@ -600,7 +582,7 @@ class SmartScanThing(lt.Thing):
ongoing_name = None if self._ongoing_scan is None else self._ongoing_scan.name
return self._scan_dir_manager.all_scans_info(ongoing=ongoing_name)
@lt.thing_property
@lt.property
def scans(self) -> ScanListInfo:
"""All the available scans.
@ -615,7 +597,7 @@ class SmartScanThing(lt.Thing):
ongoing=None if self._ongoing_scan is None else self._ongoing_scan.name,
)
@lt.fastapi_endpoint(
@lt.endpoint(
"get",
"get_stitch/{scan_name}",
responses={
@ -639,7 +621,7 @@ class SmartScanThing(lt.Thing):
raise HTTPException(404, "File not found")
return FileResponse(stitch_path)
@lt.fastapi_endpoint(
@lt.endpoint(
"delete",
"scans/{scan_name}",
responses={
@ -661,7 +643,7 @@ class SmartScanThing(lt.Thing):
if not deleted_scan_success:
raise HTTPException(400, "Couldn't delete scan, check log for details")
@lt.fastapi_endpoint(
@lt.endpoint(
"delete",
"scans",
)
@ -675,7 +657,7 @@ class SmartScanThing(lt.Thing):
for scan_name in self._scan_dir_manager.all_scans:
self._delete_scan(scan_name, logger)
@lt.thing_action
@lt.action
def purge_empty_scans(self, logger: lt.deps.InvocationLogger) -> None:
"""Delete all scan folders containing no images at the top level."""
# JSON is ignored as it's created before any images are captured
@ -712,7 +694,7 @@ class SmartScanThing(lt.Thing):
scan_name=self.latest_scan_name, filename="preview.jpg", check_exists=True
)
@lt.thing_property
@lt.property
def latest_preview_stitch_time(self) -> Optional[float]:
"""The modification time of the latest preview image, to allow live updating.
@ -728,7 +710,7 @@ class SmartScanThing(lt.Thing):
return None
return os.path.getmtime(self.latest_preview_stitch_path)
@lt.fastapi_endpoint(
@lt.endpoint(
"get",
"latest_preview_stitch.jpg",
responses={
@ -746,7 +728,7 @@ class SmartScanThing(lt.Thing):
raise HTTPException(404, "File not found")
return FileResponse(preview_path)
@lt.thing_action
@lt.action
def stitch_scan(
self,
logger: lt.deps.InvocationLogger,
@ -757,7 +739,7 @@ class SmartScanThing(lt.Thing):
) -> None:
"""Generate a stitched image based on stage position metadata.
Note that as this is a lt.thing_action it needs the logger passed as
Note that as this is a lt.action it needs the logger passed as
a variable if called from another thing action
"""
scan_data_dict = self._scan_dir_manager.get_scan_data_dict(scan_name)
@ -780,7 +762,7 @@ class SmartScanThing(lt.Thing):
except SubprocessError as e:
self._scan_logger.error(f"Stitching failed: {e}", exc_info=e)
@lt.thing_action
@lt.action
def download_zip(
self,
scan_name: str,
@ -792,7 +774,7 @@ class SmartScanThing(lt.Thing):
zip_fname = self._scan_dir_manager.zip_scan(scan_name, final_version=True)
return ZipBlob.from_file(zip_fname)
@lt.thing_action
@lt.action
def stitch_all_scans(
self, logger: lt.deps.InvocationLogger, cancel: lt.deps.CancelHook
) -> None:

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

@ -46,7 +46,7 @@ class BaseStage(lt.Thing):
_axis_names = ("x", "y", "z")
def __init__(self) -> None:
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise the stage.
:raises RedefinedBaseMovementError: if ``move_relative`` and/or
@ -54,6 +54,7 @@ class BaseStage(lt.Thing):
``_hardware_move_relative`` and/or ``_hardware_move_absolute`` instead so
that all code in the child class uses the hardware reference frame.
"""
super().__init__(thing_server_interface)
self._hardware_position = dict.fromkeys(self._axis_names, 0)
# This must be the last thing the function does in case it is caught in a try.
@ -68,28 +69,21 @@ class BaseStage(lt.Thing):
"_hardware_move_relative and/or _hardware_move_absolute instead."
)
@lt.thing_property
@lt.property
def axis_names(self) -> Sequence[str]:
"""The names of the stage's axes, in order."""
return self._axis_names
@lt.thing_property
@lt.property
def position(self) -> Mapping[str, int]:
"""Current position of the stage."""
return self._apply_axis_direction(self._hardware_position)
moving = lt.ThingProperty(
bool,
False,
readonly=True,
observable=True,
)
moving: bool = lt.property(default=False, readonly=True)
"""Whether the stage is in motion."""
axis_inverted = lt.ThingSetting(
initial_value={"x": False, "y": False, "z": False},
model=Mapping[str, bool],
readonly=True,
axis_inverted: Mapping[str, bool] = lt.setting(
default={"x": False, "y": False, "z": False}, readonly=True
)
"""Used to convert coordinates between the program frame and the hardware frame."""
@ -124,7 +118,7 @@ class BaseStage(lt.Thing):
"""Summary metadata describing the current state of the stage."""
return {"position": self.position}
@lt.thing_action
@lt.action
def invert_axis_direction(self, axis: Literal["x", "y", "z"]) -> None:
"""Invert the direction setting of the given axis.
@ -138,7 +132,7 @@ class BaseStage(lt.Thing):
raise KeyError(f"The axis {axis} is not defined.") from e
self.axis_inverted = direction
@lt.thing_action
@lt.action
def move_relative(
self,
cancel: lt.deps.CancelHook,
@ -166,7 +160,7 @@ class BaseStage(lt.Thing):
"StageThings must define their own _hardware_move_relative method"
)
@lt.thing_action
@lt.action
def move_absolute(
self,
cancel: lt.deps.CancelHook,
@ -194,7 +188,7 @@ class BaseStage(lt.Thing):
"StageThings must define their own move_absolute method"
)
@lt.thing_action
@lt.action
def set_zero_position(self) -> None:
"""Make the current position zero in all axes.
@ -206,7 +200,7 @@ class BaseStage(lt.Thing):
"StageThings must define their own set_zero_position method"
)
@lt.thing_action
@lt.action
def get_xyz_position(self) -> tuple[int, int, int]:
"""Return a tuple containing (x, y, z) position.
@ -217,7 +211,7 @@ class BaseStage(lt.Thing):
position_dict = self.position
return (position_dict["x"], position_dict["y"], position_dict["z"])
@lt.thing_action
@lt.action
def move_to_xyz_position(
self, cancel: lt.deps.CancelHook, xyz_pos: tuple[int, int, int]
) -> None:
@ -234,4 +228,4 @@ class BaseStage(lt.Thing):
self.move_absolute(cancel=cancel, x=xyz_pos[0], y=xyz_pos[1], z=xyz_pos[2])
StageDependency = lt.deps.direct_thing_client_dependency(BaseStage, "/stage/")
StageDependency = lt.deps.direct_thing_client_dependency(BaseStage, "stage")

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

@ -19,7 +19,12 @@ class DummyStage(BaseStage):
hardware attached.
"""
def __init__(self, step_time: float = 0.001, **kwargs: Any) -> None:
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
step_time: float = 0.001,
**kwargs: Any,
) -> None:
"""Initialise the Dummy stage, setting the step_time to adjust the speed.
:param step_time: The time in seconds per "motor" step. The default of 0.001
@ -27,7 +32,7 @@ class DummyStage(BaseStage):
so the speed can be increased. Increasing it too far is problematic if
also doing computationally heavy tasks like simulated image blurring.
"""
super().__init__(**kwargs)
super().__init__(thing_server_interface, **kwargs)
self.step_time = step_time
self.instantaneous_position = self._hardware_position
@ -43,10 +48,8 @@ class DummyStage(BaseStage):
) -> None:
"""Nothing to do when the Thing context manager is closed."""
axis_inverted = lt.ThingSetting(
initial_value={"x": True, "y": False, "z": False},
model=Mapping[str, bool],
readonly=True,
axis_inverted: Mapping[str, bool] = lt.setting(
default={"x": True, "y": False, "z": False}, readonly=True
)
"""Used to convert coordinates between the program frame and the hardware frame."""
@ -104,7 +107,7 @@ class DummyStage(BaseStage):
cancel, block_cancellation=block_cancellation, **displacement
)
@lt.thing_action
@lt.action
def set_zero_position(self) -> None:
"""Make the current position zero in all axes.

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

@ -33,7 +33,12 @@ class SangaboardThing(BaseStage):
functionality is accessed by directly querying the serial interface.
"""
def __init__(self, port: str = None, **kwargs: Any) -> None:
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
port: str = None,
**kwargs: Any,
) -> None:
"""Initialise SangaboardThing.
Initialise the "Thing", but do not initialise an underlying
@ -49,7 +54,7 @@ class SangaboardThing(BaseStage):
self.sangaboard_kwargs = copy(kwargs)
self.sangaboard_kwargs["port"] = port
self._sangaboard_lock = threading.RLock()
super().__init__(**kwargs)
super().__init__(thing_server_interface, **kwargs)
def __enter__(self) -> None:
"""Connect to the sangaboard when the Thing context manager is opened."""
@ -78,10 +83,8 @@ class SangaboardThing(BaseStage):
with self._sangaboard_lock:
yield self._sangaboard
axis_inverted = lt.ThingSetting(
initial_value={"x": True, "y": False, "z": True},
model=Mapping[str, bool],
readonly=True,
axis_inverted: Mapping[str, bool] = lt.setting(
default={"x": True, "y": False, "z": True}, readonly=True
)
"""Used to convert coordinates between the program frame and the hardware frame."""
@ -164,7 +167,7 @@ class SangaboardThing(BaseStage):
cancel, block_cancellation=block_cancellation, **displacement
)
@lt.thing_action
@lt.action
def set_zero_position(self) -> None:
"""Make the current position zero in all axes.
@ -176,7 +179,7 @@ class SangaboardThing(BaseStage):
sb.zero_position()
self.update_position()
@lt.thing_action
@lt.action
def flash_led(
self,
number_of_flashes: int = 10,

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

@ -30,9 +30,9 @@ from .camera_stage_mapping import CameraStageMapper
from .stage import StageDependency as StageDep
CSMDep = lt.deps.direct_thing_client_dependency(
CameraStageMapper, "/camera_stage_mapping/"
CameraStageMapper, "camera_stage_mapping"
)
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "autofocus")
## Size of movement in percentage of field of view
SMALL_STEP = 20
@ -136,18 +136,16 @@ class ParasiticMotionError(Exception):
class RangeofMotionThing(lt.Thing):
"""A class used to measure the range of motion of the stage in X and Y."""
calibrated_range = lt.ThingSetting(
initial_value=None, model=Optional[list[int, int]], readonly=True
)
calibrated_range: Optional[list[int, int]] = lt.setting(default=None, readonly=True)
def __init__(self) -> None:
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise and create the lock."""
super().__init__()
super().__init__(thing_server_interface)
self._lock = Lock()
self._stream_resolution: Optional[tuple[int, int]] = None
self._rom_data = RomDataTracker()
@lt.thing_action
@lt.action
def perform_rom_test(
self,
autofocus: AutofocusDep,
@ -214,7 +212,7 @@ class RangeofMotionThing(lt.Thing):
"Step Range": step_range,
}
@lt.thing_action
@lt.action
def perform_recentre(
self,
autofocus: AutofocusDep,

14
src/openflexure_microscope_server/things/system.py
View file

@ -43,7 +43,7 @@ class OpenFlexureSystem(lt.Thing):
_microscope_id: Optional[str] = None
@lt.thing_setting
@lt.setting
def microscope_id(self) -> UUID:
"""A unique identifier for this microscope."""
if self._microscope_id is None:
@ -54,14 +54,14 @@ class OpenFlexureSystem(lt.Thing):
def microscope_id(self, uuid: UUID) -> None:
self._microscope_id = uuid
@lt.thing_property
@lt.property
def hostname(self) -> str:
"""The hostname of the microscope, as reported by its operating system."""
return socket.gethostname()
_version_data: Optional[VersionData] = None
@lt.thing_property
@lt.property
def version_data(self) -> VersionData:
"""The version string and version source for the server.
@ -76,12 +76,12 @@ class OpenFlexureSystem(lt.Thing):
self._version_data = robust_version_strings()
return self._version_data
@lt.thing_property
@lt.property
def is_raspberrypi(self) -> bool:
"""Return True if running on a Raspberry Pi."""
return os.path.exists("/usr/bin/raspi-config")
@lt.thing_action
@lt.action
def shutdown(self) -> CommandOutput:
"""Attempt to shutdown the device."""
if not self.is_raspberrypi:
@ -103,7 +103,7 @@ class OpenFlexureSystem(lt.Thing):
out, err = p.communicate()
return CommandOutput(output=out, error=err)
@lt.thing_action
@lt.action
def reboot(self) -> CommandOutput:
"""Attempt to reboot the device."""
if not self.is_raspberrypi:
@ -120,7 +120,7 @@ class OpenFlexureSystem(lt.Thing):
out, err = p.communicate()
return CommandOutput(output=out, error=err)
@lt.thing_action
@lt.action
def get_things_state(self, metadata_getter: lt.deps.GetThingStates) -> Mapping:
"""Metadata summarising the current state of all Things in the server."""
return metadata_getter()

6
src/openflexure_microscope_server/ui.py
View file

@ -54,9 +54,8 @@ def action_button_for(action: Callable[..., Any], **kwargs: Any) -> ActionButton
:param kwargs: Any attribute of `ActionButton` except for ``thing`` or ``action``.
"""
thing = action.args[0]
thing_path = thing.path.strip("/")
action_name = action.func.__name__
return ActionButton(thing=thing_path, action=action_name, **kwargs)
return ActionButton(thing=thing.name, action=action_name, **kwargs)
class PropertyControl(BaseModel):
@ -94,7 +93,6 @@ def property_control_for(
:param kwargs: Any attribute of `PropertyControl` except for ``thing`` or
``property_name``. If label is not set here it will be the property name.
"""
thing_path = thing.path.strip("/")
if "label" not in kwargs:
kwargs["label"] = property_name
return PropertyControl(thing=thing_path, property_name=property_name, **kwargs)
return PropertyControl(thing=thing.name, property_name=property_name, **kwargs)

4
tests/test_autofocus.py
View file

@ -6,6 +6,8 @@ This doesn't check the behaviour of the JPEG shaprness monitor.
import numpy as np
import pytest
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.autofocus import (
AutofocusThing,
NoFocusFoundError,
@ -77,7 +79,7 @@ def test_looping_autofocus(start_z, max_loc, centre, attempts_expected, passes,
sharpness_monitor.move_data.side_effect = return_sharpness
autofocus_thing = AutofocusThing()
autofocus_thing = create_thing_without_server(AutofocusThing)
if passes:
autofocus_thing.looping_autofocus(
stage=stage,

40
tests/test_camera.py
View file

@ -1,7 +1,6 @@
"""Use the Simulation camera to test base camera functionality."""
import tempfile
from contextlib import contextmanager
import numpy as np
import pytest
@ -10,29 +9,30 @@ from PIL import Image
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
@contextmanager
def camera_server(camera: SimulatedCamera) -> lt.ThingClient:
@pytest.fixture
def camera_server() -> lt.ThingClient:
"""Add the camera to a ThingServer and start a TestClient application.
The test client application is needed for the camera to have a blocking portal.
The test client will be needed for the camera to run async frame generation code.
"""
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(camera, "/camera/")
with TestClient(server.app):
yield server
thing_conf = {
"camera": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"stage": "openflexure_microscope_server.things.stage.dummy:DummyStage",
}
server = lt.ThingServer(things=thing_conf, settings_folder=tmpdir)
yield server
def test_handle_broken_frame():
def test_handle_broken_frame(camera_server):
"""Monkey patch the the mjpeg steam so 1 in 5 frames are broken, then test operation.
This simulates the very occasional broken frames that can occur when grabbing
directly from the MJPEG stream.
"""
camera = SimulatedCamera()
camera = camera_server.things["camera"]
# Money patch the mjpeg_stream grab_frame to break 1 in 5 frames.
frame_number = 0
@ -50,9 +50,8 @@ def test_handle_broken_frame():
return frame
camera.mjpeg_stream.grab_frame = flaky_grabber
with camera_server(camera):
portal = lt.get_blocking_portal(camera)
with TestClient(camera_server.app):
# Check that this does cause broken frames.
# The noqa is because we don't know exactly when the error is thrown so we
# can't have a single simple statement in the pytest raises.
@ -60,22 +59,21 @@ def test_handle_broken_frame():
OSError, match="broken data stream when reading image file"
):
for _i in range(15):
jpeg = camera.grab_jpeg(portal)
jpeg = camera.grab_jpeg()
np.asarray(Image.open(jpeg.open()))
# Check that grab_as_array handles the broken frames and completes without
# the same error.
for _i in range(15):
array = camera.grab_as_array(portal)
array = camera.grab_as_array()
assert isinstance(array, np.ndarray)
def test_simulation_cam_calibration():
def test_simulation_cam_calibration(camera_server):
"""Test that the simulated camera can be calibrated and reports calibration correctly."""
camera = SimulatedCamera()
with camera_server(camera):
portal = lt.get_blocking_portal(camera)
camera = camera_server.things["camera"]
with TestClient(camera_server.app):
assert camera.calibration_required
camera.full_auto_calibrate(portal)
camera.full_auto_calibrate()
assert not camera.calibration_required
assert camera.background_detector_status.ready

14
tests/test_cameras.py
View file

@ -6,6 +6,8 @@ on camera functionality using the simulation camera are in "test_camera".
import pytest
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.camera import BaseCamera
from openflexure_microscope_server.things.camera.opencv import OpenCVCamera
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
@ -30,7 +32,7 @@ def mock_picam_thing(mocker):
from openflexure_microscope_server.things.camera.picamera import StreamingPiCamera2
return StreamingPiCamera2()
return create_thing_without_server(StreamingPiCamera2)
def _get_clean_camera_description(camera_thing):
@ -49,8 +51,6 @@ def _get_clean_camera_description(camera_thing):
* "manual_properties" - Properties explicitly exposed to the UI as manual
camera settings.
"""
camera_thing.path = "/mock/"
td = camera_thing.thing_description()
actions = set(td.actions.keys())
properties = set(td.properties.keys())
@ -82,12 +82,14 @@ def test_thing_description_equivalence(mock_picam_thing):
to this test, prompting discussion of whether the action belongs in the subclass
or the base camera class.
"""
base_td = BaseCamera().thing_description()
base_td = create_thing_without_server(BaseCamera).thing_description()
base_actions = set(base_td.actions.keys())
base_props = set(base_td.properties.keys())
sim_description = _get_clean_camera_description(SimulatedCamera())
opencv_description = _get_clean_camera_description(OpenCVCamera())
sim_camera = create_thing_without_server(SimulatedCamera)
sim_description = _get_clean_camera_description(sim_camera)
opencv_camera = create_thing_without_server(OpenCVCamera)
opencv_description = _get_clean_camera_description(opencv_camera)
picamera_description = _get_clean_camera_description(mock_picam_thing)
# Note. These are the actions and properties that are not exposed to the UI via

40
tests/test_dummy_server.py
View file

@ -20,27 +20,31 @@ from PIL import Image
import labthings_fastapi as lt
from openflexure_microscope_server.things.autofocus import AutofocusThing
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
from openflexure_microscope_server.things.stage.dummy import DummyStage
@pytest.fixture
def thing_server():
"""Yield a server with a very basic configuration."""
temp_folder = tempfile.TemporaryDirectory()
server = lt.ThingServer(settings_folder=temp_folder.name)
server.add_thing(
SimulatedCamera(
shape=(240, 320, 3), canvas_shape=(1000, 1500, 3), frame_interval=0.01
),
"/camera/",
)
server.add_thing(DummyStage(step_time=0.000001), "/stage/")
server.add_thing(AutofocusThing(), "/autofocus/")
server.add_thing(CameraStageMapper(), "/camera_stage_mapping/")
assert os.path.exists(os.path.join(temp_folder.name, "camera/"))
thing_conf = {
"camera": {
"class": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"kwargs": {
"shape": (240, 320, 3),
"canvas_shape": (1000, 1500, 3),
"frame_interval": 0.01,
},
},
"stage": {
"class": "openflexure_microscope_server.things.stage.dummy:DummyStage",
"kwargs": {"step_time": 0.000001},
},
"autofocus": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"camera_stage_mapping": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
}
server = lt.ThingServer(things=thing_conf, settings_folder=temp_folder.name)
assert os.path.exists(os.path.join(temp_folder.name, "camera"))
# Note: yield is important. If return is used the temp folder gets deleted
# before the test runs. Silence PT022 as ruff doesn't think yield is needed.
yield server # noqa: PT022
@ -60,7 +64,7 @@ def slower_client(thing_server):
The step time for the stage is 100 microseconds rather than
1 microsecond.
"""
thing_server.things["/stage/"].step_time = 0.0001
thing_server.things["stage"].step_time = 0.0001
with TestClient(thing_server.app) as client:
yield client
@ -87,7 +91,7 @@ def test_capture_jpeg_metadata(client):
exif_dict = piexif.load(image.info["exif"])
encoded_metadata = exif_dict["Exif"][piexif.ExifIFD.UserComment]
metadata = json.loads(encoded_metadata)
assert "position" in metadata["/stage/"]
assert "position" in metadata["stage"]
def test_stage(client):

4
tests/test_legacy_api.py
View file

@ -12,8 +12,8 @@ def test_v2_endpoints(mocker):
"""Check that the expected v2 endpoints are added."""
mock_server = mocker.Mock()
# Mock the camera thing to mocke the lores_mjpeg stream get_frame()
mock_server.things = {"/camera/": mocker.Mock()}
mock_server.things["/camera/"].lores_mjpeg_stream.grab_frame = mocker.AsyncMock(
mock_server.things = {"camera": mocker.Mock()}
mock_server.things["camera"].lores_mjpeg_stream.grab_frame = mocker.AsyncMock(
return_value="Mock Frame"
)
mock_app = mock_server.app

4
tests/test_sangaboard.py
View file

@ -4,6 +4,8 @@ import logging
import pytest
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.stage.sangaboard import (
RECOMMENDED_VERSION,
REQUIRED_VERSION,
@ -14,7 +16,7 @@ from openflexure_microscope_server.things.stage.sangaboard import (
@pytest.fixture
def mock_sanga_thing(mocker):
"""Return a Sangaboard thing with a MagicMock for self._sangaboard."""
sanga_thing = SangaboardThing()
sanga_thing = create_thing_without_server(SangaboardThing)
sanga_thing._sangaboard = mocker.MagicMock()
return sanga_thing

6
tests/test_server_cli.py
View file

@ -28,10 +28,10 @@ def test_successful_start(mocker, caplog):
# Create a mock for the camera so we can check the MJPEG streams are closed on
# shutdown.
mock_camera = mocker.Mock()
mock_server.things = {"/camera/": mock_camera}
mock_server.things = {"camera": mock_camera}
# Mock the LabThings function that returns the server so we have a mock server
mocker.patch(
"openflexure_microscope_server.server.lt.cli.server_from_config",
"openflexure_microscope_server.server.lt.ThingServer.from_config",
return_value=mock_server,
)
# Also mock customisation or it will try to access the hard drive and make files.
@ -78,7 +78,7 @@ def test_failed_customise(mocker):
mock_server = mocker.Mock()
# Mock the LabThings function that returns the server so we have a mock server
mocker.patch(
"openflexure_microscope_server.server.lt.cli.server_from_config",
"openflexure_microscope_server.server.lt.ThingServer.from_config",
return_value=mock_server,
)
# Also mock customisation or it will try to access the hard drive and make files.

6
tests/test_server_config.py
View file

@ -102,7 +102,7 @@ def test_get_scans_dir_no_smart_scan():
with open(FULL_CONFIG, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
# Delete smart scan
del config_dict["things"]["/smart_scan/"]
del config_dict["things"]["smart_scan"]
# No SmartScanThing, should return None
assert ofm_server._get_scans_dir(config_dict) is None
@ -116,14 +116,14 @@ def test_get_scans_dir_bad_smart_scan_config():
# Copy the dictionary
broken_config = deepcopy(config_dict)
# Delete all the smart scan kwargs
del broken_config["things"]["/smart_scan/"]["kwargs"]
del broken_config["things"]["smart_scan"]["kwargs"]
# Creates an Error
with pytest.raises(RuntimeError):
ofm_server._get_scans_dir(broken_config)
# Same thing should happen if just the scans_folder key is deleted
broken_config = deepcopy(config_dict)
del broken_config["things"]["/smart_scan/"]["kwargs"]
del broken_config["things"]["smart_scan"]["kwargs"]
# Creates an Error
with pytest.raises(RuntimeError):
ofm_server._get_scans_dir(broken_config)

7
tests/test_smart_scan.py
View file

@ -24,6 +24,7 @@ import pytest
from fastapi import HTTPException
from labthings_fastapi.exceptions import InvocationCancelledError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import (
NotEnoughFreeSpaceError,
@ -56,7 +57,7 @@ def _clear_scan_dir() -> None:
@pytest.fixture
def smart_scan_thing():
"""Return a smart scan thing as a fixture."""
return SmartScanThing(SCAN_DIR)
return create_thing_without_server(SmartScanThing, scans_folder=SCAN_DIR)
def test_initial_properties(smart_scan_thing):
@ -198,7 +199,9 @@ def _run_only_outer_scan(adjust_initial_state: Optional[Callable] = None):
assert self._csm is csm_mock
# mock smart scan thing
mock_ss_thing = MockedSmartScanThing(SCAN_DIR)
mock_ss_thing = create_thing_without_server(
MockedSmartScanThing, scans_folder=SCAN_DIR
)
if adjust_initial_state is not None:
adjust_initial_state(mock_ss_thing)

13
tests/test_stack.py
View file

@ -1,17 +1,15 @@
"""Tests for the smart and fast stacking."""
import logging
import tempfile
from random import randint
from typing import Optional
import numpy as np
import pytest
from fastapi.testclient import TestClient
from hypothesis import given
from hypothesis import strategies as st
import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.scan_directories import IMAGE_REGEX
from openflexure_microscope_server.things.autofocus import (
@ -266,13 +264,8 @@ def test_retrieval_of_captures(start):
@pytest.fixture
def autofocus_thing():
"""Yield an autofocus thing connected to a server."""
autofocus_thing = AutofocusThing()
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(autofocus_thing, "/autofocus/")
with TestClient(server.app):
yield autofocus_thing
"""Return an autofocus thing connected to a server."""
return create_thing_without_server(AutofocusThing)
def test_create_stack(autofocus_thing, caplog):

141
tests/test_stage.py
View file

@ -10,7 +10,7 @@ from hypothesis import HealthCheck, given, settings
from hypothesis import strategies as st
import labthings_fastapi as lt
from labthings_fastapi.exceptions import NotConnectedToServerError
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things.stage import (
BaseStage,
@ -33,25 +33,21 @@ path3d = st.lists(point3d, min_size=5, max_size=10)
@pytest.fixture
def dummy_stage():
"""Return a dummy stage with a very low step time."""
return DummyStage(step_time=0.000001)
return create_thing_without_server(DummyStage, step_time=0.000001)
@pytest.fixture
def thing_server(dummy_stage):
def stage_server():
"""Yield a server with a very basic configuration."""
thing_conf = {
"camera": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"stage": "openflexure_microscope_server.things.stage.dummy:DummyStage",
}
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(dummy_stage, "/stage/")
server = lt.ThingServer(things=thing_conf, settings_folder=tmpdir)
yield server
@pytest.fixture
def stage_client(thing_server):
"""Yield a labthings ThingClient for the stage."""
with TestClient(thing_server.app) as test_client:
yield lt.ThingClient.from_url("/stage/", client=test_client)
def test_override_base_movement():
"""Child classes of stage should implement functions in the hardware reference frame.
@ -63,7 +59,7 @@ def test_override_base_movement():
"""
class BadStage1(BaseStage):
@lt.thing_action
@lt.action
def move_relative(
self,
cancel: lt.deps.CancelHook,
@ -73,10 +69,10 @@ def test_override_base_movement():
pass
with pytest.raises(RedefinedBaseMovementError):
BadStage1()
create_thing_without_server(BadStage1)
class BadStage2(BaseStage):
@lt.thing_action
@lt.action
def move_absolute(
self,
cancel: lt.deps.CancelHook,
@ -86,21 +82,13 @@ def test_override_base_movement():
pass
with pytest.raises(RedefinedBaseMovementError):
BadStage2()
def _set_axis_direction(dummy_stage, direction):
"""Set the axis direction directly even if not connected to server."""
try:
dummy_stage.axis_inverted = direction
except NotConnectedToServerError:
pass
create_thing_without_server(BadStage2)
def test_apply_axis_direction_all_pos(dummy_stage):
"""Test the apply axis direction function behaves as expected when axis +v3."""
# Directly create a stage not through a ThingServer to access private methods
_set_axis_direction(dummy_stage, {"x": False, "y": False, "z": False})
dummy_stage.axis_inverted = {"x": False, "y": False, "z": False}
# A list of positions to try
positions = [
@ -120,7 +108,7 @@ def test_apply_axis_direction_all_pos(dummy_stage):
def test_apply_axis_direction_mixed(dummy_stage):
"""Test the apply axis direction function behaves as expected when axis dirs are mixed."""
# Make x and z negative
_set_axis_direction(dummy_stage, {"x": True, "y": False, "z": True})
dummy_stage.axis_inverted = {"x": True, "y": False, "z": True}
# A list of (input position, output position) to try
position_pairs = [
@ -149,48 +137,61 @@ def test_apply_axis_errors(dummy_stage):
dummy_stage._apply_axis_direction({"x": -1, "y": 2, "up": -3})
def test_default_values(stage_client, dummy_stage):
def test_default_values(dummy_stage):
"""Check the default values for the dummy stage."""
# axes are x, y, z (note that going through the thing, client the tuple is
# converted to a list.
assert stage_client.axis_names == ["x", "y", "z"]
assert dummy_stage.axis_names == ("x", "y", "z")
# position starts at 0, 0, 0
assert stage_client.position == {"x": 0, "y": 0, "z": 0}
assert dummy_stage.position == {"x": 0, "y": 0, "z": 0}
# axis direction starts is -1, 1, 1 for the dummy stage
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# And check the thing state
assert dummy_stage.thing_state == {"position": {"x": 0, "y": 0, "z": 0}}
def test_direction_inversion(stage_client, dummy_stage):
def test_direction_inversion(dummy_stage):
"""Check axes invert as expected when called."""
# Can't set an arbitrary value via a client as read only:
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.axis_inverted = {"x": 2, "y": 1, "z": 1}
# Read only should set a 405 error code
assert excinfo.value.response.status_code == 405
# And not modify th initial value
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="x")
assert stage_client.axis_inverted == {"x": False, "y": False, "z": False}
stage_client.invert_axis_direction(axis="x")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="y")
assert stage_client.axis_inverted == {"x": True, "y": True, "z": False}
stage_client.invert_axis_direction(axis="y")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
stage_client.invert_axis_direction(axis="z")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": True}
stage_client.invert_axis_direction(axis="z")
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Check initial value
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# Start inverting
dummy_stage.invert_axis_direction(axis="x")
assert dummy_stage.axis_inverted == {"x": False, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="x")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="y")
assert dummy_stage.axis_inverted == {"x": True, "y": True, "z": False}
dummy_stage.invert_axis_direction(axis="y")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
dummy_stage.invert_axis_direction(axis="z")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": True}
dummy_stage.invert_axis_direction(axis="z")
assert dummy_stage.axis_inverted == {"x": True, "y": False, "z": False}
# Should error if axis doesn't exist, this is a KeyError in the server
with pytest.raises(KeyError):
dummy_stage.invert_axis_direction(axis="theta")
# But a 422 over HTTP
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.invert_axis_direction(axis="theta")
assert excinfo.value.response.status_code == 422
def test_direction_errors_local_and_http(stage_server):
"""Check for expected errors both locally and over http."""
dummy_stage = stage_server.things["stage"]
with TestClient(stage_server.app) as test_client:
stage_client = lt.ThingClient.from_url("/stage/", client=test_client)
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Can't set an arbitrary value via a client as read only:
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.axis_inverted = {"x": 2, "y": 1, "z": 1}
# Read only should set a 405 error code ...
assert excinfo.value.response.status_code == 405
# ... and should not modify the initial value
assert stage_client.axis_inverted == {"x": True, "y": False, "z": False}
# Should error if axis doesn't exist, this is a KeyError in the server
with pytest.raises(KeyError):
dummy_stage.invert_axis_direction(axis="theta")
# But a 422 over HTTP
with pytest.raises(HTTPStatusError) as excinfo:
stage_client.invert_axis_direction(axis="theta")
assert excinfo.value.response.status_code == 422
def _test_move_relative(dummy_stage, axis_inverted, path):
@ -200,7 +201,7 @@ def _test_move_relative(dummy_stage, axis_inverted, path):
:param path: The 3d path to move over, generated by hypothesis.
"""
cancel = MockCancel()
_set_axis_direction(dummy_stage, axis_inverted)
dummy_stage.axis_inverted = axis_inverted
# Explicitly do axes calculation here to check logic in main code.
x_dir = -1 if axis_inverted["x"] else 1
y_dir = -1 if axis_inverted["y"] else 1
@ -225,22 +226,17 @@ def _test_move_relative(dummy_stage, axis_inverted, path):
suppress_health_check=[HealthCheck.function_scoped_fixture],
deadline=10000,
)
def test_move_relative(stage_client, dummy_stage, path):
def test_move_relative(dummy_stage, path):
"""Loop over different inversion options and check that the stage moves as expected.
3 paths are tried for each case of axis inversion. This checks both that the
reported position changes as is input in path, and that the hardware position is
inverted when appropriate.
NOTE: it is essential that `stage_client` is imported even if any time it is used
dummy stage could be used. This is because the fixture that creates stage client
handles adding the dummy_stage to a server. It needs to have been added to a
server for it not to throw errors about not being connected to a server.
"""
# Note that the fixture is not reset. This is fine, because the stage_should work
# no matter the starting position.
axis_names = stage_client.axis_names
axis_names = dummy_stage.axis_names
# Create every combination of True/False for x,y,z.
inversion_combinations = [
dict(zip(axis_names, inverted, strict=True))
@ -262,7 +258,7 @@ def _test_move_absolute(dummy_stage, axis_inverted, path):
:param path: The 3d path to move over, generated by hypothesis.
"""
cancel = MockCancel()
_set_axis_direction(dummy_stage, axis_inverted)
dummy_stage.axis_inverted = axis_inverted
# Explicitly do axes calculation here to check logic in main code.
x_dir = -1 if axis_inverted["x"] else 1
y_dir = -1 if axis_inverted["y"] else 1
@ -286,22 +282,17 @@ def _test_move_absolute(dummy_stage, axis_inverted, path):
suppress_health_check=[HealthCheck.function_scoped_fixture],
deadline=10000,
)
def test_move_absolute(stage_client, dummy_stage, path):
def test_move_absolute(dummy_stage, path):
"""Loop over different inversion options and check that the stage moves as expected.
3 paths are tried for each case of axis inversion. This checks both that the
reported position changes as is input in path, and that the hardware position is
inverted when appropriate.
NOTE: it is essential that `stage_client` is imported even if any time it is used
dummy stage could be used. This is because the fixture that creates stage client
handles adding the dummy_stage to a server. It needs to have been added to a
server for it not to throw errors about not being connected to a server.
"""
# Note that the fixture is not reset. This is fine, because the stage_should work
# no matter the starting position.
axis_names = stage_client.axis_names
axis_names = dummy_stage.axis_names
# Create every combination of True/False for x,y,z.
inversion_combinations = [
dict(zip(axis_names, inverted, strict=True))
@ -328,7 +319,7 @@ def test_thing_description_equivalence(dummy_stage, mocker):
# Flash LED isn't a standard stage action, most stages do not control illumination.
extra_sanga_actions = ["flash_led"]
base_td = BaseStage().thing_description()
base_td = create_thing_without_server(BaseStage).thing_description()
base_actions = set(base_td.actions.keys())
base_properties = set(base_td.properties.keys())
@ -336,7 +327,7 @@ def test_thing_description_equivalence(dummy_stage, mocker):
dummy_actions = set(dummy_td.actions.keys())
dummy_properties = set(dummy_td.properties.keys())
sanga_td = SangaboardThing().thing_description()
sanga_td = create_thing_without_server(SangaboardThing).thing_description()
# Remove known extra actions
sanga_actions = list(sanga_td.actions.keys())
for action in extra_sanga_actions:

16
tests/test_stage_measure.py
View file

@ -2,14 +2,12 @@
import dataclasses
import logging
import tempfile
from copy import copy
import numpy as np
import pytest
from fastapi.testclient import TestClient
import labthings_fastapi as lt
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things import stage_measure
from openflexure_microscope_server.things.camera_stage_mapping import (
@ -145,22 +143,18 @@ def test_parasitic_detect(par_fraction, too_high):
def test_error_if_no_stream_res_set_when_requesting_img_coords():
"""Check a RuntimeError thrown when requesting image coordinates if resolution unset."""
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing)
with pytest.raises(RuntimeError, match="Stream resolution must be set"):
stage_measure.RangeofMotionThing()._img_percentage_to_img_coords(20, "x")
rom_thing._img_percentage_to_img_coords(20, "x")
@pytest.fixture
def rom_thing(example_rom_data) -> stage_measure.RangeofMotionThing:
"""Yield a RangeofMotionThing already populated with some example rom_data."""
rom_thing = stage_measure.RangeofMotionThing()
rom_thing = create_thing_without_server(stage_measure.RangeofMotionThing)
rom_thing._stream_resolution = [800, 600]
rom_thing._rom_data = example_rom_data
with tempfile.TemporaryDirectory() as tmpdir:
server = lt.ThingServer(settings_folder=tmpdir)
server.add_thing(rom_thing, "/rom_thing/")
with TestClient(server.app):
yield rom_thing
return rom_thing
@pytest.fixture

34
tests/test_system_thing.py
View file

@ -4,10 +4,20 @@ import os
from signal import SIGTERM
from uuid import UUID
import pytest
from labthings_fastapi.testing import create_thing_without_server
from openflexure_microscope_server.things import system
from openflexure_microscope_server.utilities import VersionData, robust_version_strings
@pytest.fixture
def system_thing():
"""Return a OpenFlexureSystem with a mocked server interface."""
return create_thing_without_server(system.OpenFlexureSystem)
def _is_raspberrypi() -> bool:
"""Check if the machine running the test is a Raspberry Pi.
@ -22,17 +32,16 @@ def _is_raspberrypi() -> bool:
return "Raspberry Pi" in model_name
def test_is_raspberry():
def test_is_raspberry(system_thing):
"""Check the thing property reports whether this is a Raspberry Pi correctly."""
assert system.OpenFlexureSystem().is_raspberrypi == _is_raspberrypi()
assert system_thing.is_raspberrypi == _is_raspberrypi()
def test_version_data():
def test_version_data(system_thing):
"""Check VersionData is returned.
The content of robust_version_strings() is tested elsewhere.
"""
system_thing = system.OpenFlexureSystem()
data = system_thing.version_data
assert isinstance(data, VersionData)
assert data == robust_version_strings()
@ -43,27 +52,24 @@ def test_version_data():
assert data == fake_data
def test_hostname(mocker):
def test_hostname(system_thing, mocker):
"""Check the hostname matches what socket.gethostname() returns."""
mock_gethostname = mocker.patch("socket.gethostname", return_value="foobar")
system_thing = system.OpenFlexureSystem()
assert system_thing.hostname == "foobar"
mock_gethostname.assert_called_once_with()
def test_microscope_id():
def test_microscope_id(system_thing):
"""Check the microscope UUID is a valid UUID and doesn't change when read again."""
system_thing = system.OpenFlexureSystem()
microscope_id = system_thing.microscope_id
assert isinstance(microscope_id, UUID)
assert microscope_id == system_thing.microscope_id
def test_thing_state(mocker):
def test_thing_state(system_thing, mocker):
"""Check the thing state contains version data, hostname, and a UUID string."""
mocker.patch("socket.gethostname", return_value="foobar")
version_data = robust_version_strings()
system_thing = system.OpenFlexureSystem()
state_dict = system_thing.thing_state
assert state_dict["hostname"] == "foobar"
# Check the UUID in the dictionary is a string
@ -93,7 +99,7 @@ def test_pi_shutdown(mocker):
# Mock the shutdown command as we don't want to shutdown when running tests.
mocker.patch.object(system, "SHUTDOWN_CMD", new=["echo", "shutdown"])
# Call shutdown on a MockPiSystem
system_thing = MockPiSystem()
system_thing = create_thing_without_server(MockPiSystem)
result = system_thing.shutdown()
# Check the result of the echo mock command was returned
@ -108,7 +114,7 @@ def test_pi_reboot(mocker):
# Mock the reboot command as we don't want to reboot when running tests.
mocker.patch.object(system, "REBOOT_CMD", new=["echo", "restart"])
# Call reboot on a MockPiSystem
system_thing = MockPiSystem()
system_thing = create_thing_without_server(MockPiSystem)
result = system_thing.reboot()
# Check the result of the echo mock command was returned
@ -125,7 +131,7 @@ def test_non_pi_shutdown(mocker):
mock_kill = mocker.patch("os.kill")
# Get this subprocess
pid = os.getpid()
system_thing = MockNonPiSystem()
system_thing = create_thing_without_server(MockNonPiSystem)
result = system_thing.shutdown()
# Check it tried to kill this process with SIGTERM
@ -137,7 +143,7 @@ def test_non_pi_shutdown(mocker):
def test_non_pi_reboot():
"""Check that a server not on a pi refuses to restart."""
system_thing = MockNonPiSystem()
system_thing = create_thing_without_server(MockNonPiSystem)
result = system_thing.reboot()
# Check output is an appropriate error message