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

Merge branch 'type-fixes' into 'v3'

Browse source 
Type fixes

See merge request openflexure/openflexure-microscope-server!455
This commit is contained in:
Julian Stirling 2026-01-12 16:10:25 +00:00
commit 1e36811796
32 changed files with 521 additions and 483 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -9,6 +9,9 @@ __pycache__/
# Mypy cache
.mypy_cache*
.dmypy.json
mypy/
mypy-out.txt
# C extensions
*.so

9
.gitlab-ci.yml
View file

@ -95,12 +95,15 @@ mypy:
extends: .python
script:
- mkdir -p mypy
# "|| true" is used for preventing job failure when mypy find errors
- mypy src --cobertura-xml-report mypy --no-error-summary > mypy-out.txt || true
- mypy src --cobertura-xml-report mypy --no-error-summary > mypy-out.txt
after_script:
# Installed here rather than with [dev] as the after_script runs in a new shell
- pip install mypy-gitlab-code-quality
- mypy-gitlab-code-quality < mypy-out.txt > codequality.json
allow_failure: true
- cat mypy-out.txt
artifacts:
when: always
reports:
codequality: codequality.json
coverage_report:

BIN
picamera_coverage.zip
View file

Binary file not shown.

3
pyproject.toml
View file

@ -18,7 +18,7 @@ classifiers = [
"Operating System :: OS Independent",
]
dependencies = [
"labthings-fastapi==0.0.13",
"labthings-fastapi==0.0.14",
"sangaboard",
"camera-stage-mapping ~= 0.1.10",
"opencv-python ~= 4.11.0",
@ -37,7 +37,6 @@ dependencies = [
dev = [
"ruff~=0.13.0",
"mypy[reports]",
"mypy-gitlab-code-quality",
# "pytest-gitlab-code-quality", # pytest version constraint clashes with labthings
"pytest",
"pytest-cov",

87
src/openflexure_microscope_server/background_detect.py
View file

@ -5,15 +5,17 @@ for analysis. Information from these images is used to detect whether an image f
current camera field of view contains sample.
"""
from typing import Any, Optional
from typing import Any, Generic, Optional, TypeVar
import cv2
import numpy as np
from pydantic import BaseModel, ConfigDict, Field
from pydantic.errors import PydanticUserError
from labthings_fastapi.thing_description import type_to_dataschema
SettingsType = TypeVar("SettingsType", bound=BaseModel)
BackgroundType = TypeVar("BackgroundType", bound=BaseModel)
class MissingBackgroundDataError(RuntimeError):
"""An error raised if checking for sample without background data set."""
@ -56,22 +58,24 @@ class BackgroundDetectorStatus(BaseModel):
"""
class BackgroundDetectAlgorithm:
class BackgroundDetectAlgorithm(Generic[SettingsType, BackgroundType]):
"""The base class for defining background detect algorithms."""
background_data_model: BaseModel = BaseModel
background_data_model: type[BackgroundType]
"""The data model of the background data. This must be set by child classes"""
settings_data_model: BaseModel = BaseModel
settings_data_model: type[SettingsType]
"""The data model of algorithm settings. This must be set by child classes"""
def __init__(self) -> None:
"""Initialise the algorithm settings."""
try:
self._settings: BaseModel = self.settings_data_model()
except PydanticUserError as e:
if not hasattr(self, "background_data_model") or not hasattr(
self, "settings_data_model"
):
raise NotImplementedError(
"BackgroundDetectAlgorithms must set their own settings data model."
) from e
"All BackgroundDetectAlgorithm subclesses must set their own settings "
"and background data models."
)
self._settings: SettingsType = self.settings_data_model()
@property
def status(self) -> BackgroundDetectorStatus:
@ -86,10 +90,10 @@ class BackgroundDetectAlgorithm:
# Requires a getter and a setter to support being a BaseModel but being
# saved to file as a dict
_background_data: Optional[BaseModel] = None
_background_data: Optional[BackgroundType] = None
@property
def background_data(self) -> Optional[BaseModel]:
def background_data(self) -> Optional[BackgroundType]:
"""The statistics of the background image."""
bd = self._background_data
if bd is None:
@ -97,36 +101,31 @@ class BackgroundDetectAlgorithm:
return bd
@background_data.setter
def background_data(self, value: Optional[BaseModel | dict]) -> None:
def background_data(self, value: Optional[BackgroundType | dict]) -> None:
"""Set the statistics for the background image.
This should be None, of no data is available. It can be set from either
a dictionary or a base model of the type specified in
``self.background_data_model``.
"""
try:
if value is None:
self._background_data = None
elif isinstance(value, self.background_data_model):
self._background_data = value
elif isinstance(value, dict):
self._background_data = self.background_data_model(**value)
else:
raise TypeError(
f"Cannot set background_data with an object of type {type(value)}"
)
except PydanticUserError as e:
raise NotImplementedError(
"BackgroundDetectAlgorithms must set their own background data model."
) from e
if value is None:
self._background_data = None
elif isinstance(value, self.background_data_model):
self._background_data = value
elif isinstance(value, dict):
self._background_data = self.background_data_model(**value)
else:
raise TypeError(
f"Cannot set background_data with an object of type {type(value)}"
)
@property
def settings(self) -> BaseModel:
def settings(self) -> SettingsType:
"""The statistics of the background image."""
return self._settings
@settings.setter
def settings(self, value: BaseModel | dict) -> None:
def settings(self, value: SettingsType | dict) -> None:
if isinstance(value, self.settings_data_model):
self._settings = value
elif isinstance(value, dict):
@ -186,7 +185,12 @@ class ColourChannelDetectSettings(BaseModel):
"""
class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
class ColourChannelDetectLUV(
BackgroundDetectAlgorithm[
ColourChannelDetectSettings,
ChannelDistributions,
]
):
"""Compare images with a known background in LUV colourspace.
This uses an LUV colour space checking only the mean and standard deviation of the
@ -194,8 +198,8 @@ class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
intuitive way.
"""
background_data_model: BaseModel = ChannelDistributions
settings_data_model: BaseModel = ColourChannelDetectSettings
background_data_model: type[ChannelDistributions] = ChannelDistributions
settings_data_model: type[ColourChannelDetectSettings] = ColourChannelDetectSettings
# These are the same as those used for ChannelDeviationLUV. More detail is
# provided there.
@ -209,6 +213,10 @@ class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
The image should be in LUV format, the output will be binary with the
same shape in the first two dimensions.
"""
if self.background_data is None:
raise RuntimeError(
"Cannot calculated background mask if no background is set."
)
# The ``[1:]`` selects only the U and V channels of the image.
# Only U and V are used as brightness (L) often changes as
# the height of the sample changes.
@ -240,7 +248,7 @@ class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
image_luv = cv2.cvtColor(image, cv2.COLOR_RGB2LUV)
mask = self.background_mask(image_luv)
return (1 - np.count_nonzero(mask) / np.prod(mask.shape)) * 100
return float(1 - np.count_nonzero(mask) / np.prod(mask.shape)) * 100
def image_is_sample(self, image: np.ndarray) -> tuple[bool, str]:
"""Label the current image as either background or sample.
@ -274,7 +282,12 @@ class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
)
class ChannelDeviationLUV(BackgroundDetectAlgorithm):
class ChannelDeviationLUV(
BackgroundDetectAlgorithm[
ColourChannelDetectSettings,
ChannelDistributions,
]
):
"""Compare the standard deviations of the LUV channels in a grid to background data.
Using an LUV colour space, each image is divided into an 8x8 grid of images.
@ -284,8 +297,8 @@ class ChannelDeviationLUV(BackgroundDetectAlgorithm):
# Note we don't use the means in this algorithm but we use the same channel
# distributions model
background_data_model: BaseModel = ChannelDistributions
settings_data_model: BaseModel = ColourChannelDetectSettings
background_data_model: type[ChannelDistributions] = ChannelDistributions
settings_data_model: type[ColourChannelDetectSettings] = ColourChannelDetectSettings
# Empirically, 0.5 seems to be approximate the standard deviation for a good image
# in L and V. U appears to be about 60% of this value. U is about 65% of V when

5
src/openflexure_microscope_server/logging.py
View file

@ -14,12 +14,13 @@ output to STDOUT/STDERR are captured by ``systemd``. These can be viewed by usin
import logging
import os
from logging.handlers import RotatingFileHandler
from typing import Optional
from fastapi import HTTPException
from fastapi.responses import PlainTextResponse
LOGGER = logging.getLogger(__name__)
OFM_LOG_FILE = None
OFM_LOG_FILE: Optional[str] = None
def configure_logging(log_folder: str) -> None:
@ -128,7 +129,7 @@ class OFMHandler(logging.Handler):
how many can be returned over HTTP.
"""
super().__init__(level=level)
self._log = []
self._log: list[str] = []
self._max_logs = max_logs
def append_record(self, record: logging.LogRecord) -> None:

11
src/openflexure_microscope_server/scan_directories.py
View file

@ -304,7 +304,14 @@ class ScanDirectoryManager:
else:
last_matching_scan = sorted(matching_scans)[-1]
# Get the first group from the regex, turn to int, and add 1
scan_num = int(scan_regex.match(last_matching_scan)[1]) + 1
scan_match = scan_regex.match(last_matching_scan)
if scan_match is None: # pragma: no cover
# Type narrow, we know it is a match but mypy doesn't. This code is
# Not reachable hence the no cover.
raise RuntimeError("Internal error: regex No longer matches")
scan_num = int(scan_match[1]) + 1
# Set a sensible limit for the number of scans of one name
# based on our zero padding.
@ -346,7 +353,7 @@ class ScanDirectoryManager:
shutil.rmtree(self.path_for(scan_name))
@requires_lock
def zip_scan(self, scan_name: str, final_version: bool = False) -> "ScanDirectory":
def zip_scan(self, scan_name: str, final_version: bool = False) -> str:
"""Zips any images from the scan not yet zipped, return full path to zip.
``final_version`` Set true to stitch all files not just the scan images

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

@ -13,10 +13,17 @@ from uvicorn.main import Server
import labthings_fastapi as lt
from labthings_fastapi.server import fallback
from labthings_fastapi.server.config_model import ThingServerConfig
from openflexure_microscope_server.things.camera import BaseCamera
from openflexure_microscope_server.utilities import load_patched_config
from ..logging import configure_logging, retrieve_log, retrieve_log_from_file
from ..logging import (
OFM_HANDLER,
configure_logging,
retrieve_log,
retrieve_log_from_file,
)
from .legacy_api import add_v2_endpoints
from .serve_static_files import add_static_files
@ -44,7 +51,9 @@ def set_shutdown_function(shutdown_function: Callable[[], None]) -> None:
shutdown_function()
original_handler(*args, **kwargs)
Server.handle_exit = handle_exit
# Ignore the MyPy doesn't want us monkey patching. We have to unless the
# FastAPI lifecycle is fixed.
Server.handle_exit = handle_exit # type: ignore[method-assign]
def customise_server(
@ -82,21 +91,24 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
log_config["loggers"]["uvicorn"]["propagate"] = True
log_config["loggers"]["uvicorn.access"]["propagate"] = True
# Create server and config vars before trying to configure so they are defined
# Create server and lt_config vars before trying to configure so they are defined
# if fallback is needed before they are set.
config = None
lt_config = None
server = None
try:
config = _full_config_from_args(args)
log_folder = config.pop("log_folder", "./openflexure/logs")
scans_folder = _get_scans_dir(config)
server = lt.ThingServer.from_config(config)
customise_server(server, log_folder, scans_folder)
lt_config, internal_config = _full_config_from_args(args)
server = lt.ThingServer.from_config(lt_config)
customise_server(
server, internal_config["log_folder"], internal_config["scans_folder"]
)
def shutdown_call() -> None:
try:
# Kill any mjpeg streams so that StreamingResponses close.
server.things["camera"].kill_mjpeg_streams()
camera_thing = server.things["camera"]
if not isinstance(camera_thing, BaseCamera):
raise RuntimeError("Camera thing is not a BaseCamera")
camera_thing.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
@ -121,10 +133,18 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
# presented in the fallback logs.
print(f"Error: {e}") # noqa: T201
print("Starting fallback server.") # noqa: T201
try:
log_history = OFM_HANDLER.log_history
except BaseException:
# If log history fails for any reason carry on.
log_history = None
app = fallback.app
app.labthings_config = config
app.labthings_server = server
app.labthings_error = e
app.set_context(
fallback.FallbackContext(
server=server, config=lt_config, error=e, log_history=log_history
)
)
uvicorn.run(
app,
host=args.host,
@ -135,14 +155,25 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
raise e
def _full_config_from_args(args: Namespace) -> dict:
def _full_config_from_args(args: Namespace) -> tuple[ThingServerConfig, dict[str, Any]]:
"""Load configuration from LabThings args allowing patching.
This returns the labthings ThingServerConfig model and a dictionary of the config
for the microscope.
This provides similar functionarlity to lt.cli.config_from_args except allows the
configuration file to specify a base config, and optionally patches.
"""
internal_config = {"log_folder": "./openflexure/logs", "scans_folder": None}
# If no config file specified let LabThings handle it.
if not args.config:
return lt.cli.config_from_args(args)
return lt.cli.config_from_args(args), internal_config
return load_patched_config(args.config)
patched_config = load_patched_config(args.config)
log_folder = patched_config.pop("log_folder", None)
if log_folder is not None:
internal_config["log_folder"] = log_folder
scans_folder = _get_scans_dir(patched_config)
if scans_folder is not None:
internal_config["log_folder"] = log_folder
return ThingServerConfig(**patched_config), internal_config

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

@ -6,6 +6,8 @@ from fastapi import Response
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera import BaseCamera
FAKE_ROUTES = [
"/api/v2/",
"/api/v2/streams/snapshot",
@ -38,7 +40,10 @@ 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()
camera_thing = thing_server.things["camera"]
if not isinstance(camera_thing, BaseCamera):
raise RuntimeError("Camera thing is not a BaseCamera")
blob = await camera_thing.lores_mjpeg_stream.grab_frame()
return JPEGResponse(blob)
@app.get("/api/v2/instrument/settings/name")

20
src/openflexure_microscope_server/stitching.py
View file

@ -12,8 +12,7 @@ import shlex
import signal
import subprocess
import threading
from io import TextIOWrapper
from typing import Any, Optional
from typing import IO, Any, Optional
import labthings_fastapi as lt
@ -87,7 +86,7 @@ class BaseStitcher:
self.min_overlap = round(overlap * 0.9, 2)
self.correlation_resize = float(correlation_resize)
self._mode = "all"
self._extra_args = []
self._extra_args: list[str] = []
@property
def command(self) -> list[str]:
@ -236,8 +235,8 @@ class FinalStitcher(BaseStitcher):
def _process_inputs(
self,
overlap: float,
correlation_resize: float,
overlap: Optional[float],
correlation_resize: Optional[float],
scan_data_dict: Optional[dict[str, Any]],
) -> tuple[float, float]:
"""Process inputs to ensure ``overlap`` and ``correlation_resize`` have values.
@ -296,13 +295,15 @@ class FinalStitcher(BaseStitcher):
# Run the command piping stdout into the process for reading and
# forwarding the stdrerr to stdout
process = subprocess.Popen(
process: subprocess.Popen[str] = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
text=True,
)
if process.stdout is None: # pragma: no cover - impossible with stdout=PIPE
raise RuntimeError("stdout pipe was not created")
# Stop opening pipe blocking writing to it
os.set_blocking(process.stdout.fileno(), False)
@ -320,8 +321,11 @@ class FinalStitcher(BaseStitcher):
f"Subprocess {STITCHING_CMD} errored with exit code {process.poll()}."
)
def _log_ongoing(self, process: subprocess.Popen) -> None:
def _log_ongoing(self, process: subprocess.Popen[str]) -> None:
"""Log the ongoing process unless it is cancelled."""
if process.stdout is None: # pragma: no cover - impossible with stdout=PIPE
raise RuntimeError("stdout pipe was not created")
# Poll returns None while running, will return the error code when finished
while process.poll() is None:
self.log_buffer(process.stdout)
@ -340,7 +344,7 @@ class FinalStitcher(BaseStitcher):
# Print everything in the buffer when program finishes
self.log_buffer(process.stdout)
def log_buffer(self, buffer: TextIOWrapper) -> None:
def log_buffer(self, buffer: IO[str]) -> None:
"""Log everything in the buffer at INFO level."""
# Use rstrip to remove newlines from each line, as logging provides its own
# new lines

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

@ -148,7 +148,7 @@ class CaptureInfo:
"""The information from a capture in a z_stack."""
buffer_id: int
position: dict[str, int]
position: Mapping[str, int]
sharpness: int
@property
@ -264,7 +264,7 @@ class JPEGSharpnessMonitor:
"""Clean up after context manager is closed."""
self.running = False
def focus_rel(self, dz: int, block_cancellation: int = False) -> tuple[int, int]:
def focus_rel(self, dz: int, block_cancellation: bool = False) -> tuple[int, int]:
"""Move the stage by dz, monitoring the position over time.
This performs exactly one move. Multiple calls of this method
@ -426,7 +426,7 @@ class AutofocusThing(lt.Thing):
while attempt < 10:
attempt += 1
if start == "centre":
self._stage.move_relative(x=0, y=0, z=-(backlash + dz / 2))
self._stage.move_relative(x=0, y=0, z=-int(backlash + dz / 2))
self._stage.move_relative(x=0, y=0, z=backlash)
# Always start centrally for future runs
@ -555,7 +555,7 @@ class AutofocusThing(lt.Thing):
stack_parameters: StackParams,
save_on_failure: bool = False,
check_turning_points: bool = True,
) -> tuple[bool, Optional[int]]:
) -> tuple[bool, int]:
"""Run a smart stack.
A smart stack captures images offset in z, testing whether the sharpest image
@ -612,7 +612,7 @@ class AutofocusThing(lt.Thing):
def reset_stack(
self,
initial_z_pos: list[int],
initial_z_pos: int,
autofocus_dz: int,
) -> None:
"""Return to the initial z position and run a looping autofocus.
@ -629,7 +629,7 @@ class AutofocusThing(lt.Thing):
def save_stack(
self,
sharpest_id: int,
captures: list[list],
captures: list[CaptureInfo],
stack_parameters: StackParams,
) -> int:
"""Save the required captures to disk.
@ -681,7 +681,7 @@ class AutofocusThing(lt.Thing):
# Move down by the height of the z stack, plus an overshoot
# Better to start too low and take too many images than too high and need to refocus
self._stage.move_relative(
z=-(
z=-int(
stack_parameters.steps_undershoot
+ stack_parameters.backlash_correction
+ stack_parameters.stack_z_range / 2
@ -689,7 +689,7 @@ class AutofocusThing(lt.Thing):
)
self._stage.move_relative(z=stack_parameters.backlash_correction)
captures = []
captures: list[CaptureInfo] = []
# Always check for focus using the the last `min_images_to_test` in the
# stack so check is fair
ims_to_check = slice(-stack_parameters.min_images_to_test, None)

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

@ -15,12 +15,11 @@ import tempfile
import time
from datetime import datetime
from types import TracebackType
from typing import Any, Literal, Mapping, Optional, Tuple
from typing import Any, Literal, Mapping, Optional, Self, Tuple
import numpy as np
import piexif
from PIL import Image
from pydantic import RootModel
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
@ -46,12 +45,6 @@ class PNGBlob(lt.blob.Blob):
media_type: str = "image/png"
class ArrayModel(RootModel):
"""A model for an array."""
root: NDArray
class CaptureError(RuntimeError):
"""An error trying to capture from a CameraThing."""
@ -66,7 +59,7 @@ class CameraMemoryBuffer:
However subclasses of BaseCamera can use this class to store other object types.
"""
_storage: dict[int, tuple[Any, Optional[dict]]]
_storage: dict[int, tuple[Any, Mapping[str, Any]]]
def __init__(self) -> None:
"""Create the buffer instance."""
@ -80,7 +73,7 @@ class CameraMemoryBuffer:
def add_image(
self,
image: Any,
metadata: Optional[Mapping[str, Any]] = None,
metadata: Mapping[str, Any],
buffer_max: int = 1,
) -> int:
"""Add an image to the Memory buffer.
@ -104,7 +97,7 @@ class CameraMemoryBuffer:
def get_image(
self, buffer_id: Optional[int] = None, remove: bool = True
) -> tuple[Any, Optional[dict]]:
) -> tuple[Any, Mapping[str, Any]]:
"""Return the image with the given id.
If no id is given the most recent image is returned. However, the
@ -187,7 +180,7 @@ class BaseCamera(lt.Thing):
}
self._detector_name = "Channel Deviations (LUV)"
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Open hardware connection when the Thing context manager is opened."""
raise NotImplementedError("CameraThings must define their own __enter__ method")
@ -211,12 +204,15 @@ class BaseCamera(lt.Thing):
@lt.action
def start_streaming(
self, main_resolution: tuple[int, int], buffer_count: int
self, main_resolution: tuple[int, int] = (800, 800), buffer_count: int = 1
) -> None:
"""Start (or stop and restart) the camera.
:param main_resolution: the resolution to use for the main stream.
:param buffer_count: number of images in the stream buffer.
Note that the default values for both parameters should be set appropriately
for the specific camera when defining a new Camera Thing.
"""
raise NotImplementedError(
"CameraThings must define their own start_streaming method"
@ -255,7 +251,7 @@ class BaseCamera(lt.Thing):
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 5,
) -> ArrayModel:
) -> NDArray:
"""Acquire one image from the camera and return as an array."""
raise NotImplementedError(
"CameraThings must define their own capture_array method"
@ -265,7 +261,7 @@ class BaseCamera(lt.Thing):
"""The downsampling factor when calling capture_downsampled_array."""
@lt.action
def capture_downsampled_array(self) -> ArrayModel:
def capture_downsampled_array(self) -> NDArray:
"""Acquire one image from the camera, downsample, and return as an array.
* The array is downsamples by the thing property `downsampled_array_factor`.
@ -279,7 +275,7 @@ class BaseCamera(lt.Thing):
@lt.action
def capture_jpeg(
self,
stream_name: str = "main",
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = None,
) -> JPEGBlob:
"""Acquire one image from the camera as a JPEG.
@ -334,7 +330,7 @@ class BaseCamera(lt.Thing):
def grab_as_array(
self,
stream_name: Literal["main", "lores"] = "main",
) -> ArrayModel:
) -> NDArray:
"""Acquire one image from the preview stream and return as an array.
It works like ``grab_jpeg`` but reliably handles broken streams. Prefer using
@ -369,9 +365,9 @@ class BaseCamera(lt.Thing):
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
stream_name: Literal["main", "lores", "full"],
wait: Optional[float] = None,
) -> Image:
) -> Image.Image:
"""Capture a PIL image from stream stream_name with timeout wait."""
raise NotImplementedError(
"CameraThings must define their own capture_image method"
@ -438,7 +434,7 @@ class BaseCamera(lt.Thing):
"""Clear all images in memory."""
self._memory_buffer.clear()
def _robust_image_capture(self) -> Tuple[Image, Mapping[str, Any]]:
def _robust_image_capture(self) -> Tuple[Image.Image, Mapping[str, Any]]:
"""Capture an image in memory and return it with metadata.
This robust capturing method attempts to capture the image five times
@ -520,8 +516,8 @@ class BaseCamera(lt.Thing):
def _save_capture(
self,
jpeg_path: str,
image: Image,
metadata: dict,
image: Image.Image,
metadata: Mapping[str, Any],
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Save the captured image and metadata to disk.
@ -533,7 +529,7 @@ class BaseCamera(lt.Thing):
nothing is returned on success
"""
if save_resolution is not None and image.size != save_resolution:
image = image.resize(save_resolution, Image.BOX)
image = image.resize(save_resolution, Image.Resampling.BOX)
try:
# Per PIL documentation,
# (https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg)
@ -544,7 +540,7 @@ class BaseCamera(lt.Thing):
# disabled, file size increases and quality is barely or not affected
image.save(jpeg_path, quality=95, subsampling=0)
try:
self._add_metadata_to_capture(jpeg_path, metadata)
self._add_metadata_to_capture(jpeg_path, dict(metadata))
except Exception:
# We need to capture any exception as there are many reasons metadata
# might not be added. We warn rather than log the error.
@ -591,7 +587,7 @@ class BaseCamera(lt.Thing):
return self._detector_name
@detector_name.setter
def detector_name(self, name: str) -> None:
def _set_detector_name(self, name: str) -> None:
"""Validate and set detector_name."""
if name not in self.background_detectors:
self.logger.warning(f"{name} is not a valid background detector name.")
@ -640,7 +636,7 @@ class BaseCamera(lt.Thing):
return data
@background_detector_data.setter
def background_detector_data(self, data: dict) -> None:
def _set_background_detector_data(self, data: dict) -> None:
"""Set the data for each detector. Only to be used as settings are loaded from disk.
Do not call over HTTP. This needs to be updated once LbaThings Settings can be

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

@ -11,7 +11,7 @@ from __future__ import annotations
import logging
from threading import Thread
from types import TracebackType
from typing import Literal, Optional
from typing import Literal, Optional, Self
import cv2
from PIL import Image
@ -39,7 +39,7 @@ class OpenCVCamera(BaseCamera):
self._capture_thread: Optional[Thread] = None
self._capture_enabled = False
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Start the capture thread when the Thing context manager is opened."""
self.cap = cv2.VideoCapture(self.camera_index)
self._capture_enabled = True
@ -59,7 +59,8 @@ class OpenCVCamera(BaseCamera):
"""
if self.stream_active:
self._capture_enabled = False
self._capture_thread.join()
if self._capture_thread is not None:
self._capture_thread.join()
self.cap.release()
@lt.property
@ -90,7 +91,7 @@ class OpenCVCamera(BaseCamera):
@lt.action
def capture_array(
self,
stream_name: Literal["main", "full"] = "full",
stream_name: Literal["main", "lores", "raw", "full"] = "full",
wait: Optional[float] = None,
) -> NDArray:
"""Acquire one image from the camera and return as an array.
@ -111,9 +112,9 @@ class OpenCVCamera(BaseCamera):
def capture_image(
self,
stream_name: Literal["main", "full"] = "main",
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = None,
) -> Image:
) -> Image.Image:
"""Acquire one image from the camera and return as a PIL image.
This function will produce a JPEG image.

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

@ -25,7 +25,7 @@ import time
from contextlib import contextmanager
from threading import RLock
from types import TracebackType
from typing import Annotated, Any, Iterator, Literal, Mapping, Optional
from typing import Annotated, Any, Iterator, Literal, Mapping, Optional, Self
import numpy as np
from picamera2 import Picamera2
@ -36,6 +36,7 @@ from pydantic import BaseModel, BeforeValidator
import labthings_fastapi as lt
from labthings_fastapi.exceptions import ServerNotRunningError
from labthings_fastapi.types.numpy import NDArray
from openflexure_microscope_server.background_detect import ChannelBlankError
from openflexure_microscope_server.ui import (
@ -45,7 +46,7 @@ from openflexure_microscope_server.ui import (
property_control_for,
)
from . import ArrayModel, BaseCamera
from . import BaseCamera
from . import picamera_recalibrate_utils as recalibrate_utils
from . import picamera_tuning_file_utils as tf_utils
@ -122,6 +123,9 @@ class StreamingPiCamera2(BaseCamera):
generalisation.
"""
tuning: dict = lt.setting(default_factory=dict, readonly=True)
"""The Raspberry PiCamera Tuning File JSON."""
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
@ -147,7 +151,7 @@ class StreamingPiCamera2(BaseCamera):
f"are {SUPPORTED_CAMS_SENSOR_INFO.keys()}."
)
self._sensor_info = SUPPORTED_CAMS_SENSOR_INFO[camera_board]
self._picamera_lock = None
self._picamera_lock = RLock()
self._picamera = None
# Load the tuning file for the specified sensor mode.
@ -159,10 +163,12 @@ class StreamingPiCamera2(BaseCamera):
# connected to the server if tuning is saved to disk.
try:
self.tuning = copy.deepcopy(self.default_tuning)
except ServerNotRunningError:
except ServerNotRunningError as e:
# 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
# a server. But we know this, so we ignore the error as long as the
# tuning data is set.
if "version" not in self.tuning:
raise RuntimeError("Tuning file could not be set.") from e
# Also set the colour gains based on the tuning. Set to _colour_gains to not
# trigger a ServerNotRunningError
@ -214,7 +220,7 @@ class StreamingPiCamera2(BaseCamera):
return self._analogue_gain
@analogue_gain.setter
def analogue_gain(self, value: float) -> None:
def _set_analogue_gain(self, value: float) -> None:
self._analogue_gain = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -234,7 +240,7 @@ class StreamingPiCamera2(BaseCamera):
return self._colour_gains
@colour_gains.setter
def colour_gains(self, value: tuple[float, float]) -> None:
def _set_colour_gains(self, value: tuple[float, float]) -> None:
self._colour_gains = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -258,7 +264,7 @@ class StreamingPiCamera2(BaseCamera):
return self._exposure_time
@exposure_time.setter
def exposure_time(self, value: int) -> None:
def _set_exposure_time(self, value: int) -> None:
self._exposure_time = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -282,7 +288,7 @@ class StreamingPiCamera2(BaseCamera):
"Sharpness": 1,
}
_sensor_modes = None
_sensor_modes: Optional[list[SensorMode]] = None
@lt.property
def sensor_modes(self) -> list[SensorMode]:
@ -302,7 +308,7 @@ class StreamingPiCamera2(BaseCamera):
return SensorModeSelector(**self._sensor_mode)
@sensor_mode.setter
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None:
def _set_sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None:
"""Change the sensor mode used."""
if new_mode is None:
self._sensor_mode = None
@ -323,9 +329,6 @@ class StreamingPiCamera2(BaseCamera):
with self._streaming_picamera() as cam:
return cam.sensor_resolution
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:
"""Acquire the picamera device and store it as ``self._picamera``.
@ -339,10 +342,7 @@ class StreamingPiCamera2(BaseCamera):
:raises PicameraModelError: If check_sensor_model is True and the real
camera sensor model doesn't match the expected sensor model.
"""
if self._picamera_lock is not None:
# Don't close the camera if it's in use
self._picamera_lock.acquire()
with tempfile.NamedTemporaryFile("w") as tuning_file:
with self._picamera_lock, tempfile.NamedTemporaryFile("w") as tuning_file:
json.dump(self.tuning, tuning_file)
tuning_file.flush() # but leave it open as closing it will delete it
os.environ["LIBCAMERA_RPI_TUNING_FILE"] = tuning_file.name
@ -363,6 +363,9 @@ class StreamingPiCamera2(BaseCamera):
camera_num=self._camera_num,
tuning=self.tuning,
)
if self._picamera is None:
# Type narrow (error if failure)
raise RuntimeError("Failed to start Picamera")
if check_sensor_model:
hw_sensor_model = self._picamera.camera_properties["Model"]
if hw_sensor_model != self._sensor_info.sensor_model:
@ -370,9 +373,8 @@ class StreamingPiCamera2(BaseCamera):
f"Wrong Picamera model. Expecting {self._sensor_info.sensor_model}, "
f"but found {hw_sensor_model}."
)
self._picamera_lock = RLock()
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Start streaming when the Thing context manager is opened.
This opens the picamera connection, initialises the camera, sets the
@ -533,7 +535,7 @@ class StreamingPiCamera2(BaseCamera):
self,
stream_name: Literal["main", "lores", "full"] = "main",
wait: Optional[float] = 0.9,
) -> Image:
) -> Image.Image:
"""Acquire one image from the camera and return it as a PIL Image.
If the ``stream_name`` parameter is ``main`` or ``lores``, it will be captured
@ -573,7 +575,7 @@ class StreamingPiCamera2(BaseCamera):
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 0.9,
) -> ArrayModel:
) -> NDArray:
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
@ -773,7 +775,8 @@ class StreamingPiCamera2(BaseCamera):
"""The calibration actions for both calibration wizard and settings panel."""
return [
action_button_for(
self.full_auto_calibrate,
self,
"full_auto_calibrate",
submit_label="Full Auto-Calibrate",
can_terminate=False,
requires_confirmation=True,
@ -791,13 +794,15 @@ class StreamingPiCamera2(BaseCamera):
"""The calibration actions that appear only in settings panel."""
return [
action_button_for(
self.auto_expose_from_minimum,
self,
"auto_expose_from_minimum",
submit_label="Auto Gain & Shutter Speed",
can_terminate=False,
button_primary=False,
),
action_button_for(
self.calibrate_lens_shading,
self,
"calibrate_lens_shading",
submit_label="Auto Flat Field Correction",
can_terminate=False,
button_primary=False,
@ -809,19 +814,22 @@ class StreamingPiCamera2(BaseCamera):
),
),
action_button_for(
self.flat_lens_shading,
self,
"flat_lens_shading",
submit_label="Disable Flat Field Correction",
can_terminate=False,
button_primary=False,
),
action_button_for(
self.flat_lens_shading_chrominance,
self,
"flat_lens_shading_chrominance",
submit_label="Disable Flat Field Chrominance",
can_terminate=False,
button_primary=False,
),
action_button_for(
self.reset_lens_shading,
self,
"reset_lens_shading",
submit_label="Reset Flat Field Correction",
can_terminate=False,
button_primary=False,
@ -856,7 +864,7 @@ class StreamingPiCamera2(BaseCamera):
]
@lt.property
def lens_shading_tables(self) -> Optional[tf_utils.LensShading]:
def lens_shading_tables(self) -> Optional[tf_utils.LensShadingModel]:
"""The current lens shading (i.e. flat-field correction).
Return the current lens shading correction, as three 2D lists each with
@ -869,19 +877,6 @@ class StreamingPiCamera2(BaseCamera):
"""
return tf_utils.get_lst(self.tuning)
@lens_shading_tables.setter
def lens_shading_tables(self, lst: tf_utils.LensShading) -> None:
"""Set the lens shading tables."""
with self._streaming_picamera(pause_stream=True):
self.tuning = tf_utils.set_lst(
self.tuning,
luminance=lst.luminance,
cr=lst.Cr,
cb=lst.Cb,
colour_temp=lst.colour_temp,
)
self._initialise_picamera()
@lt.action
def flat_lens_shading(self) -> None:
"""Disable flat-field correction.

18
src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py
View file

@ -51,7 +51,6 @@ import numpy as np
import picamera2
from picamera2 import Picamera2
from pydantic import BaseModel
from scipy.ndimage import zoom
LOGGER = logging.getLogger(__name__)
@ -322,18 +321,7 @@ def _get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray:
return np.reshape(np.array(grid), (12, 16))
def _upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray:
"""Zoom an image in the last two dimensions.
This is effectively the inverse operation of ``_get_16x12_grid``
"""
zoom_factors = [
1,
] + list(np.ceil(np.array(shape) / np.array(grids.shape[1:])))
return zoom(grids, zoom_factors, order=1)[:, : shape[0], : shape[1]]
def _downsampled_channels(channels: np.ndarray, blacklevel: int) -> list[np.ndarray]:
def _downsampled_channels(channels: np.ndarray, blacklevel: int) -> np.ndarray:
"""Generate a downsampled, un-normalised image from which to calculate the LST."""
channel_shape = np.array(channels.shape[1:])
lst_shape = np.array([12, 16])
@ -398,9 +386,7 @@ def _grids_from_lst(lum: np.ndarray, Cr: np.ndarray, Cb: np.ndarray) -> np.ndarr
return np.stack([B, G, G, R], axis=0)
def _raw_channels_from_camera(
camera: Picamera2, sensor_info: SensorInfo
) -> LensShadingTables:
def _raw_channels_from_camera(camera: Picamera2, sensor_info: SensorInfo) -> np.ndarray:
"""Acquire a raw image and return a 4xNxM array of the colour channels."""
if camera.started:
camera.stop_recording()

9
src/openflexure_microscope_server/things/camera/picamera_tuning_file_utils.py
View file

@ -20,9 +20,12 @@ CALIBRATED_COLOUR_TEMP = 5000
DEFAULT_COLOUR_TEMP = 1234
class LensShading(BaseModel):
class LensShadingModel(BaseModel):
"""A Pydantic model holding the lens shading tables.
Note this shouldn't be confused with the typehint for LensShadingTables in
recalibrate utils which is for the arrays.
PiCamera needs three numpy arrays for lens shading correction. Each array is
(12, 16) in size. The arrays are luminance, red-difference chroma (Cr), and
blue-difference chroma (Cb).
@ -165,7 +168,7 @@ def flatten_lst(tuning: dict, keep_luminance: bool = False) -> dict:
)
def get_lst(tuning: dict) -> LensShading:
def get_lst(tuning: dict) -> LensShadingModel:
"""Return the lens shading as a LenSading Base Model."""
# Note "alsc" is the Picamera2 term for "Automatic Lens Shading Correction"
alsc = find_tuning_algo(tuning, "rpi.alsc")
@ -175,7 +178,7 @@ def get_lst(tuning: dict) -> LensShading:
w, h = 16, 12
return [lin[w * i : w * (i + 1)] for i in range(h)]
return LensShading(
return LensShadingModel(
luminance=reshape_lst(alsc["luminance_lut"]),
Cr=reshape_lst(alsc["calibrations_Cr"][0]["table"]),
Cb=reshape_lst(alsc["calibrations_Cb"][0]["table"]),

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

@ -13,12 +13,13 @@ import logging
import time
from threading import Thread
from types import TracebackType
from typing import Literal, Optional
from typing import Literal, Optional, Self
import numpy as np
from PIL import Image, ImageFilter
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from openflexure_microscope_server.ui import (
ActionButton,
@ -28,11 +29,11 @@ from openflexure_microscope_server.ui import (
)
from ..stage.dummy import DummyStage
from . import ArrayModel, BaseCamera
from . import BaseCamera
LOGGER = logging.getLogger(__name__)
# The ratio between "motor" steps and pixels
# The ratio between "motor" steps and pixels in (x, y, z)
# higher related to a faster movement
RATIO = (2, 2, 0.2)
@ -195,17 +196,18 @@ class SimulatedCamera(BaseCamera):
self.canvas[top:bottom, left:right] -= sprite
def generate_image(self, pos: tuple[int, int, int]) -> Image:
def generate_image(self, pos: tuple[int, int, int]) -> Image.Image:
"""Generate an image with blobs based on supplied coordinates.
:param pos: a 3-item tuple containing the x,y,z coordinates of the 'stage'
"""
canvas_width, canvas_height, _ = self.canvas_shape
image_width, image_height, _ = self.shape
pos = tuple(x * s for x, s in zip(pos, RATIO, strict=True))
# Scale position by RATIO to get position in base image.
im_pos = tuple(x * ratio for x, ratio in zip(pos, RATIO, strict=True))
top_left = (
int(pos[0]) - image_width // 2 + self.sample_limits[0] // 2,
int(pos[1]) - image_height // 2 + self.sample_limits[1] // 2,
int(im_pos[0]) - image_width // 2 + self.sample_limits[0] // 2,
int(im_pos[1]) - image_height // 2 + self.sample_limits[1] // 2,
)
# Create index list with modulo rather than slicing to handle wrapping at the
# canvas edge.
@ -216,7 +218,7 @@ class SimulatedCamera(BaseCamera):
# Use npx to make each 1d index list 3D
focused_image = canvas[np.ix_(x_indices, y_indices, z_indices)]
image = fast_pil_blur(focused_image, sigma=np.abs(pos[2]) / 5)
image = fast_pil_blur(focused_image, sigma=np.abs(im_pos[2]) / 5)
if image.shape != self.shape:
raise ValueError(
@ -229,7 +231,7 @@ class SimulatedCamera(BaseCamera):
image[image > 255] = 255
return Image.fromarray(image.astype("uint8"))
def generate_frame(self) -> Image:
def generate_frame(self) -> Image.Image:
"""Generate a frame with blobs based on the stage coordinates."""
try:
pos = self._stage.instantaneous_position
@ -238,7 +240,7 @@ class SimulatedCamera(BaseCamera):
pos = {"x": 0, "y": 0, "z": 0}
return self.generate_image((pos["y"], pos["x"], pos["z"]))
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Start the capture thread when the Thing context manager is opened."""
self.start_streaming()
return self
@ -250,7 +252,7 @@ class SimulatedCamera(BaseCamera):
_traceback: Optional[TracebackType],
) -> None:
"""Close the capture thread when the Thing context manager is closed."""
if self.stream_active:
if self._capture_thread is not None and self._capture_thread.is_alive():
self._capture_enabled = False
self._capture_thread.join()
@ -299,7 +301,7 @@ class SimulatedCamera(BaseCamera):
try:
frame = self.generate_frame()
self.mjpeg_stream.add_frame(_frame2bytes(frame))
ds_frame = frame.resize((320, 240), resample=Image.NEAREST)
ds_frame = frame.resize((320, 240), resample=Image.Resampling.NEAREST)
self.lores_mjpeg_stream.add_frame(_frame2bytes(ds_frame))
except Exception as e:
@ -315,9 +317,9 @@ class SimulatedCamera(BaseCamera):
@lt.action
def capture_array(
self,
stream_name: Literal["main", "full"] = "full",
stream_name: Literal["main", "lores", "raw", "full"] = "full",
wait: Optional[float] = None,
) -> ArrayModel:
) -> NDArray:
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
@ -334,9 +336,9 @@ class SimulatedCamera(BaseCamera):
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
stream_name: Literal["main", "lores", "full"],
wait: Optional[float] = None,
) -> Image:
) -> Image.Image:
"""Capture to a PIL image. This is not exposed as a ThingAction.
It is used for capture to memory.
@ -384,7 +386,7 @@ class SimulatedCamera(BaseCamera):
"""The calibration actions for both calibration wizard and settings panel."""
return [
action_button_for(
self.full_auto_calibrate, submit_label="Full Auto-Calibrate"
self, "full_auto_calibrate", submit_label="Full Auto-Calibrate"
),
]
@ -392,8 +394,8 @@ class SimulatedCamera(BaseCamera):
def secondary_calibration_actions(self) -> list[ActionButton]:
"""The calibration actions that appear only in settings panel."""
return [
action_button_for(self.load_sample, submit_label="Load Sample"),
action_button_for(self.remove_sample, submit_label="Remove Sample"),
action_button_for(self, "load_sample", submit_label="Load Sample"),
action_button_for(self, "remove_sample", submit_label="Remove Sample"),
]
@lt.property
@ -402,7 +404,7 @@ class SimulatedCamera(BaseCamera):
return [property_control_for(self, "noise_level", label="Noise Level")]
def _frame2bytes(frame: Image) -> bytes:
def _frame2bytes(frame: Image.Image) -> bytes:
"""Convert frame to bytes."""
with io.BytesIO() as buf:
# Save in low quality for speed.

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

@ -10,19 +10,11 @@ This module is only intended to be called from the OpenFlexure Microscope
server, and depends on that server and its underlying LabThings library.
"""
import json
import time
from typing import (
Any,
Dict,
List,
Mapping,
NamedTuple,
Optional,
Tuple,
)
from typing import Any, List, Mapping, NamedTuple, Optional, Tuple, cast
import numpy as np
from fastapi import HTTPException
import labthings_fastapi as lt
from camera_stage_mapping.camera_stage_calibration_1d import (
@ -36,9 +28,6 @@ from labthings_fastapi.types.numpy import DenumpifyingDict
from .camera import BaseCamera
from .stage import BaseStage
CoordinateType = Tuple[float, float, float]
XYCoordinateType = Tuple[float, float]
class MoveHistory(NamedTuple):
"""A named tuple containing the position over time for a single move.
@ -50,7 +39,27 @@ class MoveHistory(NamedTuple):
"""
times: List[float]
stage_positions: List[CoordinateType]
stage_positions: List[tuple[int, int, int]]
def _array_to_stage_tuple(pos: np.ndarray) -> tuple[int, int, int]:
"""Convert a numpy array into a tuple of ints.
:param pos: Input position array must be 3 elements long.
:return: a tuple of 3 integers
:raises ValueError: If the array is not of length 3.
"""
pos_tuple = tuple(int(i) for i in pos)
if len(pos_tuple) == 3:
return cast(tuple[int, int, int], pos_tuple)
raise ValueError("Input array was not 3 elements long.")
def _serialise_numpy_in_dict(dict_with_numpy: dict) -> dict:
serialised = json.loads(DenumpifyingDict(dict_with_numpy).model_dump_json())
if not isinstance(serialised, dict):
raise TypeError(f"Expecting a dictionary to serialise not a {type(serialised)}")
return serialised
class RecordedMove:
@ -69,21 +78,24 @@ class RecordedMove:
be called whenever the instance is called.
"""
self._stage = stage
self._current_position: Optional[CoordinateType] = None
self._history: List[Tuple[float, Optional[CoordinateType]]] = []
self._current_position: Optional[tuple[int, int, int]] = None
self._history: List[Tuple[float, tuple[int, int, int]]] = []
def __call__(self, new_position: CoordinateType) -> None:
def __call__(self, new_position: np.ndarray) -> None:
"""Move to a new position, and record it."""
self._history.append((time.time(), self._current_position))
self._stage.move_to_xyz_position(xyz_pos=new_position)
self._current_position = new_position
self._history.append((time.time(), self._current_position))
new_stage_pos = _array_to_stage_tuple(new_position)
starting_pos = self._current_position
if starting_pos is not None:
self._history.append((time.time(), starting_pos))
self._stage.move_to_xyz_position(xyz_pos=new_stage_pos)
self._current_position = new_stage_pos
self._history.append((time.time(), new_stage_pos))
@property
def history(self) -> MoveHistory:
"""The history, as a numpy array of times and another of positions."""
times: List[float] = [t for t, p in self._history if p is not None]
positions: List[CoordinateType] = [p for t, p in self._history if p is not None]
times: List[float] = [t for t, p in self._history]
positions: List[tuple[int, int, int]] = [p for t, p in self._history]
return MoveHistory(times, positions)
def clear_history(self) -> None:
@ -91,25 +103,14 @@ class RecordedMove:
self._history = []
class CSMUncalibratedError(HTTPException):
"""An HTTP Exception raised if camera stage mapping data is needed but unavailable.
class CSMUncalibratedError(lt.exceptions.InvocationError):
"""An Exception raised if camera stage mapping data is needed but unavailable.
Camera Stage Mapping data is needed to convert from distances specified in fractions
of the field of view to distances in motor steps. This is used when clicking on the
live preview to move, or when performing a scan.
"""
def __init__(self) -> None:
"""Customise the default error code and message of HTTPException."""
HTTPException.__init__(
self,
503,
(
"The camera_stage_mapping calibration is not yet available. "
"This probably means you need to run the calibration routine."
),
)
class CameraStageMapper(lt.Thing):
"""A Thing to manage mapping between image and stage coordinates.
@ -121,8 +122,7 @@ class CameraStageMapper(lt.Thing):
_cam: BaseCamera = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
@lt.action
def calibrate_1d(self, direction: Tuple[float, float, float]) -> DenumpifyingDict:
def calibrate_1d(self, direction: Tuple[int, int, int]) -> dict:
"""Move a microscope's stage in 1D, and figure out the relationship with the camera."""
# Record positions and times for stage calibration
recorded_move = RecordedMove(self._stage)
@ -152,7 +152,7 @@ class CameraStageMapper(lt.Thing):
return result
@lt.action
def calibrate_xy(self) -> DenumpifyingDict:
def calibrate_xy(self) -> dict:
"""Move the microscope's stage in X and Y, to calibrate its relationship to the camera.
This performs two 1d calibrations in x and y, then combines their results.
@ -182,7 +182,7 @@ class CameraStageMapper(lt.Thing):
)
self.logger.info(f"CSM matrix is {csm_as_string}.")
data: Dict[str, dict] = {
data = {
"camera_stage_mapping_calibration": cal_xy,
"linear_calibration_x": cal_x,
"linear_calibration_y": cal_y,
@ -191,7 +191,8 @@ class CameraStageMapper(lt.Thing):
"downsampling": downsampling_factor,
}
self.last_calibration = DenumpifyingDict(data).model_dump()
data = _serialise_numpy_in_dict(data)
self.last_calibration = data
return data
@ -238,12 +239,14 @@ class CameraStageMapper(lt.Thing):
"""Whether the camera stage mapper needs calibrating."""
return self.image_to_stage_displacement_matrix is None
def assert_calibrated(self) -> None:
"""Raise an exception if the image_to_stage_displacement matrix is not set."""
def assert_calibration(self) -> List[List[float]]:
"""Return image_to_stage_displacement matrix or raise error if it's not set."""
if self.image_to_stage_displacement_matrix is None:
# Disable check of no message in raised exception as the message is explicitly
# added by CSMUncalibratedError
raise CSMUncalibratedError() # noqa: RSE102
raise CSMUncalibratedError(
"The camera_stage_mapping calibration is not yet available. "
"This probably means you need to run the calibration routine."
)
return self.image_to_stage_displacement_matrix
@lt.action
def move_in_image_coordinates(self, x: float, y: float) -> None:
@ -258,24 +261,26 @@ class CameraStageMapper(lt.Thing):
and ``y`` to the shorter one. Checking what shape your chosen toolkit reports for
an image usually helps resolve any ambiguity.
"""
self.assert_calibrated()
self._stage.move_relative(**self.convert_image_to_stage_coordinates(x=x, y=y))
self._stage.move_relative(
**self.convert_image_to_stage_coordinates(x=x, y=y),
block_cancellation=False,
)
@lt.action
def convert_image_to_stage_coordinates(
self, x: float, y: float, **_kwargs: float
) -> Mapping[str, int]:
"""Convert image coordinates to stage coordinates. Only x and y are returned."""
self.assert_calibrated()
return csm_img_to_stage(self.image_to_stage_displacement_matrix, x=x, y=y)
csm_matrix = self.assert_calibration()
return csm_img_to_stage(csm_matrix, x=x, y=y)
@lt.action
def convert_stage_to_image_coordinates(
self, x: int, y: int, **_kwargs: int
) -> Mapping[str, float]:
"""Convert stage coordinates to image coordinates. Only x and y are returned."""
self.assert_calibrated()
return csm_stage_to_img(self.image_to_stage_displacement_matrix, x=x, y=y)
csm_matrix = self.assert_calibration()
return csm_stage_to_img(csm_matrix, x=x, y=y)
@lt.property
def thing_state(self) -> Mapping[str, Any]:

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

@ -19,7 +19,6 @@ from typing import (
Mapping,
Optional,
ParamSpec,
Self,
TypeVar,
)
@ -35,7 +34,7 @@ from openflexure_microscope_server import scan_directories, scan_planners, stitc
# Things
from .autofocus import AutofocusThing, StackParams
from .camera import BaseCamera
from .camera_stage_mapping import CameraStageMapper
from .camera_stage_mapping import CameraStageMapper, CSMUncalibratedError
from .stage import BaseStage
T = TypeVar("T")
@ -69,8 +68,8 @@ class ScanNotRunningError(RuntimeError):
def _scan_running(
method: Callable[Concatenate[Self, P], T],
) -> Callable[Concatenate[Self, P], T]:
method: Callable[Concatenate["SmartScanThing", P], T],
) -> Callable[Concatenate["SmartScanThing", P], T]:
"""Decorate a method so that it will error if a scan is not running.
This decorator is used by all methods in SmartScanThing that are using
@ -79,7 +78,9 @@ def _scan_running(
the same time and released with the lock
"""
def scan_running_wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T:
def scan_running_wrapper(
self: "SmartScanThing", *args: P.args, **kwargs: P.kwargs
) -> T:
"""Only start the requested method if the scan is running."""
if self._scan_lock.locked():
return method(self, *args, **kwargs)
@ -116,13 +117,63 @@ class SmartScanThing(lt.Thing):
self._scan_dir_manager = scan_directories.ScanDirectoryManager(scans_folder)
self._scan_lock = threading.Lock()
# Variables set by the scan
self._latest_scan_name: Optional[str] = None
# Variables set by the scan
_stack_params: Optional[StackParams] = None
self._stack_params: Optional[StackParams] = None
self._ongoing_scan: Optional[scan_directories.ScanDirectory] = None
self._scan_data: Optional[scan_directories.ScanData] = None
self._preview_stitcher: Optional[stitching.PreviewStitcher] = None
@property
def stack_params(self) -> StackParams:
"""The parameters for z-stacking during the onging scan.
Only read this property is a scan is ongoing or it will raise an error.
"""
if self._stack_params is None:
raise RuntimeError("Cannot get stack parameters as they are not set.")
return self._stack_params
_ongoing_scan: Optional[scan_directories.ScanDirectory] = None
@property
def ongoing_scan(self) -> scan_directories.ScanDirectory:
"""The ScanDirectory object of the ongoing scan.
Only read this property is a scan is ongoing or it will raise an error.
"""
if self._ongoing_scan is None:
raise ScanNotRunningError("Cannot get ongoing scan if scan is not running.")
return self._ongoing_scan
_scan_data: Optional[scan_directories.ScanData] = None
@property
def scan_data(self) -> scan_directories.ScanData:
"""The ScanData object jolding information about the of the ongoing scan.
Only read this property is a scan is ongoing or it will raise an error.
"""
if self._scan_data is None:
raise ScanNotRunningError("Cannot get scan data if scan is not running.")
return self._scan_data
_preview_stitcher: Optional[stitching.PreviewStitcher] = None
@property
def preview_stitcher(self) -> stitching.PreviewStitcher:
"""The PreviewStitcher object for stitching live previews.
Only read this property is a scan is ongoing or it will raise an error.
"""
if self._preview_stitcher is None:
raise ScanNotRunningError(
"No preview stitcher agailable as scan is not running."
)
return self._preview_stitcher
_latest_scan_name: Optional[str] = None
@lt.property
def latest_scan_name(self) -> Optional[str]:
"""The name of the last scan to be started."""
return self._latest_scan_name
@lt.action
def sample_scan(self, scan_name: str = "") -> None:
@ -143,7 +194,7 @@ class SmartScanThing(lt.Thing):
try:
self._check_background_and_csm_set()
self._ongoing_scan = self._scan_dir_manager.new_scan_dir(scan_name)
self._latest_scan_name = self._ongoing_scan.name
self._latest_scan_name = self.ongoing_scan.name
self._autofocus.looping_autofocus(dz=self.autofocus_dz, start="centre")
self._run_scan()
except Exception as e:
@ -180,11 +231,7 @@ class SmartScanThing(lt.Thing):
Raise warning if not using background detect that scan will go on until max steps reached
"""
if self._csm.image_resolution is None:
raise RuntimeError(
"Camera-stage mapping is not calibrated. This is required before "
"scans can be carried out."
)
self._csm.assert_calibration()
if self.skip_background:
if not self._cam.background_detector_status.ready:
@ -200,11 +247,6 @@ class SmartScanThing(lt.Thing):
"of motion."
)
@lt.property
def latest_scan_name(self) -> Optional[str]:
"""The name of the last scan to be started."""
return self._latest_scan_name
@_scan_running
def _move_to_next_point(
self, next_point: tuple[int, int], z_estimate: Optional[int] = None
@ -230,7 +272,7 @@ class SmartScanThing(lt.Thing):
return (next_point[0], next_point[1], z_estimate)
@_scan_running
def _calc_displacement_from_test_image(self, overlap: int) -> tuple[int, int]:
def _calc_displacement_from_test_image(self, overlap: float) -> tuple[int, int]:
"""Take a test image and use camera stage mapping to calculate x and y displacement.
:param overlap: The desired overlap as a fraction of the image. i.e. 0.5 means
@ -238,9 +280,15 @@ class SmartScanThing(lt.Thing):
:returns: (dx, dy) - the x and y displacements in steps
"""
if (
self._csm.image_resolution is None
or self._csm.image_to_stage_displacement_matrix is None
):
raise CSMUncalibratedError("Camera stage mapping is not calibrated")
test_image = self._cam.grab_as_array()
test_image_res = list(test_image.shape)
csm_image_res = [int(i) for i in self._csm.image_resolution]
# If current stream width is different to csm calibration width,
@ -295,7 +343,7 @@ class SmartScanThing(lt.Thing):
# Fix scan parameters in case UI is updated during scan.
return scan_directories.ScanData(
scan_name=self._ongoing_scan.name,
scan_name=self.ongoing_scan.name,
starting_position=starting_position,
overlap=overlap,
max_dist=self.max_range,
@ -317,16 +365,16 @@ class SmartScanThing(lt.Thing):
Takes scan_result, a string that is either "success", "cancelled by user",
or the error that ended the scan.
"""
self._scan_data.set_final_data(result=scan_result)
self._ongoing_scan.save_scan_data(self._scan_data)
self.scan_data.set_final_data(result=scan_result)
self.ongoing_scan.save_scan_data(self.scan_data)
@_scan_running
def _manage_stitching_threads(self) -> None:
"""Manage the stitching threads, starting them if needed and not already running."""
# Assume 4 images means at least one offset in x and y, making the stitching
# well constrained.
if self._scan_data.image_count > 3 and not self._preview_stitcher.running:
self._preview_stitcher.start()
if self.scan_data.image_count > 3 and not self.preview_stitcher.running:
self.preview_stitcher.start()
@_scan_running
def _run_scan(self) -> None:
@ -339,16 +387,21 @@ class SmartScanThing(lt.Thing):
try:
self._cam.start_streaming(main_resolution=(3280, 2464))
self._scan_data = self._collect_scan_data()
self._ongoing_scan.save_scan_data(self._scan_data)
self.ongoing_scan.save_scan_data(self._scan_data)
images_dir = self.ongoing_scan.images_dir
if images_dir is None:
raise RuntimeError(
"Couldn't run scan, images directory was not created."
)
self._stack_params = self._autofocus.create_stack_params(
images_dir=self._ongoing_scan.images_dir,
images_dir=images_dir,
autofocus_dz=self.autofocus_dz,
save_resolution=self._scan_data.save_resolution,
save_resolution=self.scan_data.save_resolution,
)
self._preview_stitcher = stitching.PreviewStitcher(
self._ongoing_scan.images_dir,
overlap=self._scan_data.overlap,
correlation_resize=self._scan_data.correlation_resize,
images_dir,
overlap=self.scan_data.overlap,
correlation_resize=self.scan_data.correlation_resize,
)
# This is the main loop of the scan!
@ -392,9 +445,9 @@ class SmartScanThing(lt.Thing):
# have multiple starting positions, each of which will be visited before the
# scan can end.
planner_settings = {
"dx": self._scan_data.dx,
"dy": self._scan_data.dy,
"max_dist": self._scan_data.max_dist,
"dx": self.scan_data.dx,
"dy": self.scan_data.dy,
"max_dist": self.scan_data.max_dist,
}
route_planner = scan_planners.SmartSpiral(
initial_position=(self._stage.position["x"], self._stage.position["y"]),
@ -419,7 +472,7 @@ class SmartScanThing(lt.Thing):
capture_image = True
# If skipping background, take an image to check if current field of view is background
if self._scan_data.skip_background:
if self.scan_data.skip_background:
capture_image, bg_message = self._cam.image_is_sample()
if not capture_image:
@ -431,23 +484,23 @@ class SmartScanThing(lt.Thing):
continue
focused, focused_height = self._autofocus.run_smart_stack(
stack_parameters=self._stack_params,
save_on_failure=not self._scan_data.skip_background,
stack_parameters=self.stack_params,
save_on_failure=not self.scan_data.skip_background,
)
current_pos_xyz = (new_pos_xyz[0], new_pos_xyz[1], focused_height)
# An image was captured if we are focussed or we are not skipping background.
imaged = focused or not self._scan_data.skip_background
imaged = focused or not self.scan_data.skip_background
route_planner.mark_location_visited(
current_pos_xyz, imaged=imaged, focused=focused
)
# increment capture counter as thread has completed
self._scan_data.image_count += 1
self.scan_data.image_count += 1
# Add it to the incremental zip
self._ongoing_scan.zip_files()
self.ongoing_scan.zip_files()
@_scan_running
def _return_to_starting_position(self) -> None:
@ -455,7 +508,7 @@ class SmartScanThing(lt.Thing):
self.logger.info("Returning to starting position.")
if self._scan_data is not None:
self._stage.move_absolute(
**self._scan_data.starting_position, block_cancellation=True
**self.scan_data.starting_position, block_cancellation=True
)
@property
@ -463,28 +516,30 @@ class SmartScanThing(lt.Thing):
"""Return a metadata dict for ongoing scan to populate."""
if self._ongoing_scan is None:
return {}
return {"scan_name": self._ongoing_scan.name}
return {"scan_name": self.ongoing_scan.name}
@_scan_running
def _perform_final_stitch(self) -> None:
"""Update the scan zip and perform final stitch of the data."""
if self._scan_data.image_count <= 3:
if self.scan_data.image_count <= 3:
self.logger.info("Not performing a stitch as 3 or fewer images taken")
return
self._ongoing_scan.zip_files()
self.ongoing_scan.zip_files()
self.logger.info("Waiting for background processes to finish...")
# Actually check the sticher exists rather than using self.preview_sticher as
# this method can be called during exception handling.
if self._preview_stitcher is not None:
self._preview_stitcher.wait()
if self._scan_data.stitch_automatically:
if self.scan_data.stitch_automatically:
self.logger.info("Stitching final image (may take some time)...")
self.stitch_scan(
scan_name=self._ongoing_scan.name,
correlation_resize=self._scan_data.correlation_resize,
overlap=self._scan_data.overlap,
scan_name=self.ongoing_scan.name,
correlation_resize=self.scan_data.correlation_resize,
overlap=self.scan_data.overlap,
)
@lt.endpoint(
@ -540,7 +595,7 @@ class SmartScanThing(lt.Thing):
_scan_dir_manager.all_scans_info() directly as it will handle stopping the json
in any ongoing scans being read.
"""
ongoing_name = None if self._ongoing_scan is None else self._ongoing_scan.name
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.property
@ -555,7 +610,7 @@ class SmartScanThing(lt.Thing):
"""
return ScanListInfo(
scans=self._get_all_scan_info(),
ongoing=None if self._ongoing_scan is None else self._ongoing_scan.name,
ongoing=None if self._ongoing_scan is None else self.ongoing_scan.name,
)
@lt.endpoint(
@ -632,7 +687,7 @@ class SmartScanThing(lt.Thing):
This is a wrapper around scan manager's delete_scan that logs to the
things logger if there is a problem.
"""
if self._ongoing_scan is not None and scan_name == self._ongoing_scan.name:
if self._ongoing_scan is not None and scan_name == self.ongoing_scan.name:
self.logger.error("Attempted to delete ongoing scan.")
return False
try:
@ -648,7 +703,7 @@ class SmartScanThing(lt.Thing):
@property
def latest_preview_stitch_path(self) -> Optional[str]:
"""The path of the latest preview stitched image, or None if not available."""
if not self.latest_scan_name:
if self.latest_scan_name is None:
return None
return self._scan_dir_manager.get_file_path_from_img_dir(

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

@ -12,7 +12,7 @@ As the object will be used as a context manager create the hardware connection i
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any, Literal, Never
from typing import Any, Literal, overload
import labthings_fastapi as lt
@ -82,11 +82,19 @@ class BaseStage(lt.Thing):
moving: bool = lt.property(default=False, readonly=True)
"""Whether the stage is in motion."""
axis_inverted: Mapping[str, bool] = lt.setting(
axis_inverted: dict[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."""
@overload
def _apply_axis_direction(self, position: list[int] | tuple[int]) -> list[int]: ...
@overload
def _apply_axis_direction(
self, position: Mapping[str, int]
) -> Mapping[str, int]: ...
def _apply_axis_direction(
self, position: list[int] | tuple[int] | Mapping[str, int]
) -> list[int] | Mapping[str, int]:
@ -142,7 +150,7 @@ class BaseStage(lt.Thing):
def _hardware_move_relative(
self, block_cancellation: bool = False, **kwargs: int
) -> Never:
) -> None:
"""Make a relative move in the coordinate system used by the physical hardware.
Make sure to use and update ``self._hardware_position`` not ``self.position``.
@ -163,7 +171,7 @@ class BaseStage(lt.Thing):
self,
block_cancellation: bool = False,
**kwargs: int,
) -> Never:
) -> None:
"""Make a absolute move in the coordinate system used by the physical hardware.
Make sure to use and update ``self._hardware_position`` not ``self.position``.

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

@ -3,9 +3,8 @@
from __future__ import annotations
import time
from collections.abc import Mapping
from types import TracebackType
from typing import Any, Optional
from typing import Any, Optional, Self
import labthings_fastapi as lt
@ -36,9 +35,10 @@ class DummyStage(BaseStage):
self.step_time = step_time
self.instantaneous_position = self._hardware_position
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Register the stage position when the Thing context manager is opened."""
self.instantaneous_position = self._hardware_position
return self
def __exit__(
self,
@ -48,7 +48,7 @@ class DummyStage(BaseStage):
) -> None:
"""Nothing to do when the Thing context manager is closed."""
axis_inverted: Mapping[str, bool] = lt.setting(
axis_inverted: dict[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."""

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

@ -4,11 +4,10 @@ from __future__ import annotations
import threading
import time
from collections.abc import Mapping
from contextlib import contextmanager
from copy import copy
from types import TracebackType
from typing import Any, Iterator, Literal, Optional
from typing import Any, Iterator, Literal, Optional, Self
import semver
@ -33,7 +32,7 @@ class SangaboardThing(BaseStage):
def __init__(
self,
thing_server_interface: lt.ThingServerInterface,
port: str = None,
port: Optional[str] = None,
**kwargs: Any,
) -> None:
"""Initialise SangaboardThing.
@ -53,13 +52,14 @@ class SangaboardThing(BaseStage):
self._sangaboard_lock = threading.RLock()
super().__init__(thing_server_interface, **kwargs)
def __enter__(self) -> None:
def __enter__(self) -> Self:
"""Connect to the sangaboard when the Thing context manager is opened."""
self._sangaboard = sangaboard.Sangaboard(**self.sangaboard_kwargs)
with self.sangaboard() as sb:
sb.query("blocking_moves false")
self.check_firmware()
self.update_position()
return self
def __exit__(
self,
@ -80,7 +80,7 @@ class SangaboardThing(BaseStage):
with self._sangaboard_lock:
yield self._sangaboard
axis_inverted: Mapping[str, bool] = lt.setting(
axis_inverted: dict[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."""

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

@ -14,8 +14,9 @@ this is 10% of the expected motion is the measured axis.
"""
import time
from copy import copy
from threading import Lock
from typing import Any, Literal, Optional, overload
from typing import Any, Literal, Mapping, Optional, overload
import numpy as np
@ -48,20 +49,20 @@ class RomDataTracker:
def __init__(self) -> None:
"""Define useful data tracked throughout test."""
self.stage_coords: list[dict[str, int]] = []
self.offsets: list[dict[str, float]] = []
self.stage_coords: list[Mapping[str, int]] = []
self.offsets: list[Mapping[str, float]] = []
def record_movement(
self, current_pos: dict[str, int], offset: dict[str, float]
self, current_pos: Mapping[str, int], offset: Mapping[str, float]
) -> None:
"""Record the current position and the measured offset of the last move."""
self.stage_coords.append(current_pos)
self.offsets.append(offset)
@property
def final_position(self) -> dict[str, int]:
def final_position(self) -> Mapping[str, int]:
"""The last stage coordinate recorded."""
return self.stage_coords[-1].copy()
return copy(self.stage_coords[-1])
def fit_axis(self, axis: Literal["x", "y"]) -> np.poly1d:
"""Quadratic fit stage z against x or y and return poly1d of fit.
@ -78,8 +79,8 @@ class RomDataTracker:
def predict_z_displacement(
self,
axis: Literal["x", "y"],
stage_movement: dict[str, int],
stage_position: dict[str, int],
stage_movement: Mapping[str, int],
stage_position: Mapping[str, int],
) -> int:
"""Predict the z-displacement needed for a given movement.
@ -93,7 +94,7 @@ class RomDataTracker:
return int(z_dest - stage_position["z"])
def find_turning_point(self, axis: Literal["x", "y"]) -> dict[str, int]:
def find_turning_point(self, axis: Literal["x", "y"]) -> Mapping[str, int]:
"""Find the turing point from the recorded coordinates."""
fit_func = self.fit_axis(axis)
turning_loc = fit_func.deriv().roots[0]
@ -121,7 +122,9 @@ class RangeofMotionThing(lt.Thing):
_csm: CameraStageMapper = lt.thing_slot()
_stage: BaseStage = lt.thing_slot()
calibrated_range: Optional[list[int, int]] = lt.setting(default=None, readonly=True)
calibrated_range: Optional[tuple[int, int]] = lt.setting(
default=None, readonly=True
)
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
"""Initialise and create the lock."""
@ -131,7 +134,7 @@ class RangeofMotionThing(lt.Thing):
self._rom_data = RomDataTracker()
@lt.action
def perform_rom_test(self) -> dict[str, Any]:
def perform_rom_test(self) -> Mapping[str, Any]:
"""Measures the range of motion of the stage across the x and y axes.
:return: Results dictionary separated into keys of each axis and direction.
@ -173,7 +176,7 @@ class RangeofMotionThing(lt.Thing):
self.logger.info(
f"Range of motion is {step_range[0]} x {step_range[1]} steps"
)
self.calibrated_range = step_range
self.calibrated_range = (step_range[0], step_range[1])
finally:
self._lock.release()
return {
@ -324,7 +327,7 @@ class RangeofMotionThing(lt.Thing):
# if the distance is less than 1 big step away then move to it and exit
if abs(img_perc) < BIG_STEP:
self.logger.info(f"Estimated centre of {axis}-axis is {estimate[axis]}")
self._stage.move_absolute(**estimate)
self._stage.move_absolute(**estimate, block_cancellation=False)
# Note the second return, the direction, is meaningless here.
return True, 1
self.logger.info(
@ -353,7 +356,7 @@ class RangeofMotionThing(lt.Thing):
return (fov_perc / 100) * self._stream_resolution[img_index]
def _img_dir_from_stage_coords(
self, target: dict[str, int], axis: Literal["x", "y"]
self, target: Mapping[str, int], axis: Literal["x", "y"]
) -> Literal[1, -1]:
"""For a target location in stage coords, return the direction in image coordinates.
@ -369,7 +372,7 @@ class RangeofMotionThing(lt.Thing):
return -1 if target_im_coords[axis] < current_loc_im_coords[axis] else 1
def _distance_in_img_percentage(
self, target: dict[str, int], axis: Literal["x", "y"]
self, target: Mapping[str, int], axis: Literal["x", "y"]
) -> float:
"""For a target location in stage coords return the distance in percentage of FOV.
@ -394,7 +397,7 @@ class RangeofMotionThing(lt.Thing):
fov_perc: int,
axis: Literal["x", "y"],
direction: Literal[1, -1],
) -> dict[str, float]:
) -> Mapping[str, float]:
"""Return the dictionary for a move in image coordinates.
This dictionary can be passed directly to csm.move_in_image_coordinates
@ -546,11 +549,11 @@ class RangeofMotionThing(lt.Thing):
def _move_and_measure(
self,
movement: dict[str, float],
movement: Mapping[str, float],
perform_autofocus: bool = True,
max_autofocus_repeats: int = 0,
abs_min_offset: float = 0.0,
) -> dict[str, float]:
) -> Mapping[str, float]:
"""Move the stage and measure the offset between the two positions.
:param movement: A dictionary containing the distance to move in image coords.
@ -596,7 +599,7 @@ class RangeofMotionThing(lt.Thing):
return offset
def _offset_from(self, before_img: np.ndarray) -> dict[str, float]:
def _offset_from(self, before_img: np.ndarray) -> Mapping[str, float]:
"""Take an image and calculate the offset from an input image.
:param before_img: The image the offset should be calculated with respect to.
@ -623,11 +626,11 @@ class RangeofMotionThing(lt.Thing):
# let MyPy know with overloads typed to Literal[False] and Literal[True].
@overload
def _axis_from_movement_dict(
movement: dict[str, float], return_other: Literal[False]
movement: Mapping[str, float], return_other: Literal[False]
) -> str: ...
@overload
def _axis_from_movement_dict(
movement: dict[str, float], return_other: Literal[True]
movement: Mapping[str, float], return_other: Literal[True]
) -> tuple[str, str]: ...
@ -635,12 +638,12 @@ def _axis_from_movement_dict(
# because MyPy doesn't see that return other's default is `False` and use
# the `Literal[False]` overload.
@overload
def _axis_from_movement_dict(movement: dict[str, float]) -> str: ...
def _axis_from_movement_dict(movement: Mapping[str, float]) -> str: ...
# Finally The function.
def _axis_from_movement_dict(
movement: dict[str, float], return_other: bool = False
movement: Mapping[str, float], return_other: bool = False
) -> str | tuple[str, str]:
"""Return the axis that a given movement dictionary moves in.
@ -661,7 +664,7 @@ def _axis_from_movement_dict(
def _parasitic_motion_detected(
movement: dict[str, float], offset: dict[str, float]
movement: Mapping[str, float], offset: Mapping[str, float]
) -> bool:
"""Compare desired movement to measured offset, report if parasitic motion was detected.

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

@ -51,8 +51,8 @@ class OpenFlexureSystem(lt.Thing):
return UUID(self._microscope_id)
@microscope_id.setter
def microscope_id(self, uuid: UUID) -> None:
self._microscope_id = uuid
def _set_microscope_id(self, uuid: UUID) -> None:
self._microscope_id = str(uuid)
@lt.property
def hostname(self) -> str:
@ -98,6 +98,7 @@ class OpenFlexureSystem(lt.Thing):
SHUTDOWN_CMD,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
)
out, err = p.communicate()
@ -116,6 +117,7 @@ class OpenFlexureSystem(lt.Thing):
REBOOT_CMD,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
)
out, err = p.communicate()
return CommandOutput(output=out, error=err)

11
src/openflexure_microscope_server/ui.py
View file

@ -1,6 +1,6 @@
"""Functionality for communicating the required user interface for a thing."""
from typing import Any, Callable
from typing import Any
from pydantic import BaseModel
@ -47,14 +47,15 @@ class ActionButton(BaseModel):
"""The message to show on successful completion."""
def action_button_for(action: Callable[..., Any], **kwargs: Any) -> ActionButton:
def action_button_for(thing: lt.Thing, action_name: str, **kwargs: Any) -> ActionButton:
"""Create a ActionButton data for the specified Thing Action.
:param action: The thing action to create a button for.
:param thing: The instance of the thing that has the action.
:param action_name: The name of the action to create a button for.
:param kwargs: Any attribute of `ActionButton` except for ``thing`` or ``action``.
:return: An ActionButton (Pydantic Model) object with all the information the
webapp needs to create the action button.
"""
thing = action.args[0]
action_name = action.func.__name__
return ActionButton(thing=thing.name, action=action_name, **kwargs)

95
src/openflexure_microscope_server/utilities.py
View file

@ -8,15 +8,12 @@ import sys
import tomllib
from functools import wraps
from importlib.metadata import version
from threading import Thread
from typing import (
Any,
Callable,
Concatenate,
Literal,
Optional,
ParamSpec,
Self,
TypeAlias,
TypeVar,
overload,
@ -26,6 +23,7 @@ from pydantic import BaseModel
T = TypeVar("T")
P = ParamSpec("P")
LockableClass = TypeVar("LockableClass")
JSONScalar: TypeAlias = str | int | float | bool | None
JSONType: TypeAlias = dict[str, "JSONType"] | list["JSONType"] | JSONScalar
@ -50,15 +48,15 @@ def _is_lock_like(obj: Any) -> bool:
def requires_lock(
method: Callable[Concatenate[Self, P], T],
) -> Callable[Concatenate[Self, P], T]:
method: Callable[Concatenate[LockableClass, P], T],
) -> Callable[Concatenate[LockableClass, P], T]:
"""Decorate a class method so that it requires the class lock to run.
The class should have a reentrant lock with the name ``self._lock``.
"""
@wraps(method)
def wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T:
def wrapper(self: LockableClass, *args: P.args, **kwargs: P.kwargs) -> T:
# Confirm the object the method is attached to has a self._lock property
if not hasattr(self, "_lock"):
raise AttributeError(f"{self.__class__.__name__} has no '_lock' attribute")
@ -71,91 +69,6 @@ def requires_lock(
return wrapper
class ErrorCapturingThread(Thread):
"""Subclass of Thread that captures exceptions from the target function.
It wraps the target function in a try-except block.
Execution will stop with an unhandled exception, but the exception will not be raised
until the join method is called. When the join method is called, the exception is
called in the calling thread.
This allows for error handling to be performed by the calling thread at the time that
join is run.
"""
def __init__(
self,
group: Optional[Any] = None,
target: Optional[Callable[..., Any]] = None,
name: Optional[str] = None,
args: tuple[Any, ...] = (),
kwargs: Optional[dict[str, Any]] = None,
*,
daemon: Optional[bool] = None,
) -> None:
"""Initialise with the same arguments as Thread."""
# As all inputs are keywords we need to set the default values for args and kwargs:
if args is None:
args = ()
if kwargs is None:
kwargs = {}
# Make an empty list for any error.
# We are only ever going to have one or zero errors, this is an empty
# list as we need to pass the reference not the value to collect the
# error. If we could simply make a pointer we would have done that.
self._error_buffer = []
# Add the target function end error buffer to the thread to the start of
# the argument list so they are passed to _wrap_and_catch_errors()
args = (target, self._error_buffer, *args)
# Start the thread with _wrap_and_catch_errors() as the target
super().__init__(
group=group,
target=_wrap_and_catch_errors,
name=name,
args=args,
kwargs=kwargs,
daemon=daemon,
)
def join(self, timeout: Optional[float] = None) -> None:
"""Join when the thread is complete.
If the thread ended due to an unhandled exception, the exception will be raised
when this method is called.
"""
super().join(timeout)
# If there is an error in the error buffer clear the buffer and raise it
if self._error_buffer:
err = self._error_buffer[0]
self._error_buffer = []
raise err
def _wrap_and_catch_errors(
target: Callable[..., Any], error_buffer: list, *args: Any, **kwargs: Any
) -> None:
"""Run target function in a try-except block.
This function is designed only to be used by ErrorCapturingThread.
It will run a target function in a try-except block catching any exception.
If an exception is caught it is added to ``error_buffer``.
:param target: The target function to call
:param error_buffer: An empty list that is used for returning the exception from
the thread. It is a list to ensure it is passed by reference.
:param ``*args``: The arguments for the target function
:param ``**kwargs``: The Keyword arguments for the target function
"""
try:
target(*args, **kwargs)
except BaseException as e:
error_buffer.append(e)
# Compiled regular expressions for unsafe characters
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \
_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\\]")

42
tests/unit_tests/test_background_detectors.py
View file

@ -58,34 +58,38 @@ def test_bg_detect_base_class():
BackgroundDetectAlgorithm()
def test_partial_base_class(background_image):
"""Create a partial class to initialise the base class and test other methods.
This test is to check that if the necessary methods are not set, that an
appropriate error is raised.
"""
def test_partial_base_classes():
"""Create a partial classes and check they raise the correct errors."""
class BadAlgo1(BackgroundDetectAlgorithm):
"""Only has a settings model so it can initialise."""
"""Only has a settings model so it cannot initialise."""
settings_data_model: BaseModel = ColourChannelDetectSettings
bad_algo1 = BadAlgo1()
status = bad_algo1.status
assert not status.ready
# Check the settings dictionary can be validated as ``ColourChannelDetectSettings``
ColourChannelDetectSettings(**status.settings)
settings_data_model = ColourChannelDetectSettings
with pytest.raises(NotImplementedError):
# Should error on any dictionary input. This simulates loading settings from
# disk
bad_algo1.background_data = {"key": 1}
BadAlgo1()
class BadAlgo2(BackgroundDetectAlgorithm):
"""Only has a background model so it cannot initialise."""
background_data_model = ChannelDistributions
with pytest.raises(NotImplementedError):
bad_algo1.set_background(background_image)
BadAlgo2()
class BadAlgo3(BackgroundDetectAlgorithm):
"""Has both models, intalises by cannot run set_background or image_is_sample."""
settings_data_model = ColourChannelDetectSettings
background_data_model = ChannelDistributions
bad_algo3 = BadAlgo3()
with pytest.raises(NotImplementedError):
bad_algo1.image_is_sample(background_image)
bad_algo3.set_background(background_image)
with pytest.raises(NotImplementedError):
bad_algo3.image_is_sample(background_image)
def test_colour_channel_luv(background_image, sample_image):

28
tests/unit_tests/test_camera_buffer.py
View file

@ -33,7 +33,7 @@ def test_add_and_get_image():
"""Check images can be captured and retrieved."""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image)
buffer_id = mem_buf.add_image(misc_image, random_metadata())
returned_image, _ = mem_buf.get_image(buffer_id)
# It is the same image
assert misc_image is returned_image
@ -46,7 +46,7 @@ def test_add_and_get_image_twice():
"""Check images can be retrieved twice if remove flag set false."""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
buffer_id = mem_buf.add_image(misc_image)
buffer_id = mem_buf.add_image(misc_image, random_metadata())
returned_image, _ = mem_buf.get_image(buffer_id, remove=False)
# It is the same image
assert misc_image is returned_image
@ -62,7 +62,7 @@ def test_get_without_id():
"""Check images can be captured and retrieved without ID."""
mem_buf = CameraMemoryBuffer()
misc_image = random_image()
mem_buf.add_image(misc_image)
mem_buf.add_image(misc_image, random_metadata())
returned_image, _ = mem_buf.get_image()
# It is the same image
assert misc_image is returned_image
@ -76,8 +76,8 @@ def test_get_two_images():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, buffer_max=2)
buffer_id2 = mem_buf.add_image(misc_image2, buffer_max=2)
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2)
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2)
returned_image1, _ = mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
# It they the same images
@ -95,8 +95,8 @@ def test_get_two_images_without_setting_buffer_size():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1)
buffer_id2 = mem_buf.add_image(misc_image2)
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata())
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata())
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
returned_image2, _ = mem_buf.get_image(buffer_id2)
@ -110,10 +110,10 @@ def test_buffer_size_changing():
misc_image1 = random_image()
misc_image2 = random_image()
misc_image3 = random_image()
buffer_id1 = mem_buf.add_image(misc_image1, buffer_max=3)
buffer_id2 = mem_buf.add_image(misc_image2, buffer_max=3)
buffer_id1 = mem_buf.add_image(misc_image1, random_metadata(), buffer_max=3)
buffer_id2 = mem_buf.add_image(misc_image2, random_metadata(), buffer_max=3)
# Third capture doesn't set buffer size, so it will be reset
buffer_id3 = mem_buf.add_image(misc_image3)
buffer_id3 = mem_buf.add_image(misc_image3, random_metadata())
# As buffer size was reset, images 1 and 2 are deleted
with pytest.raises(NoImageInMemoryError):
mem_buf.get_image(buffer_id1)
@ -129,8 +129,8 @@ def test_capture_two_images_get_without_id():
mem_buf = CameraMemoryBuffer()
misc_image1 = random_image()
misc_image2 = random_image()
mem_buf.add_image(misc_image1, buffer_max=2)
mem_buf.add_image(misc_image2, buffer_max=2)
mem_buf.add_image(misc_image1, random_metadata(), buffer_max=2)
mem_buf.add_image(misc_image2, random_metadata(), buffer_max=2)
returned_image, _ = mem_buf.get_image()
# When buffer_id is not specified, the most recent image (image2) is expected to
# be retrieved
@ -148,7 +148,7 @@ def test_buffer_size_respected():
buffer_ids = []
for _i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, buffer_max=5)
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=5)
images.append(image)
buffer_ids.append(buffer_id)
@ -169,7 +169,7 @@ def test_clear_buffer():
buffer_ids = []
for _i in range(10):
image = random_image()
buffer_id = mem_buf.add_image(image, buffer_max=10)
buffer_id = mem_buf.add_image(image, random_metadata(), buffer_max=10)
images.append(image)
buffer_ids.append(buffer_id)

3
tests/unit_tests/test_legacy_api.py
View file

@ -6,13 +6,14 @@ from socket import gethostname
# Import as ofm server to attempt to minimise confusion with server as a var in other
# functions and also FastAPI `Server`.
from openflexure_microscope_server.server import legacy_api
from openflexure_microscope_server.things.camera import BaseCamera
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": mocker.Mock(spec=BaseCamera)}
mock_server.things["camera"].lores_mjpeg_stream.grab_frame = mocker.AsyncMock(
return_value="Mock Frame"
)

2
tests/unit_tests/test_picamera_tuning_files.py
View file

@ -111,7 +111,7 @@ def test_tuning_lst_tools(imx219_tuning, imx477_tuning):
assert not tf_utils.lst_calibrated(tuning)
lst_model = tf_utils.get_lst(tuning)
assert isinstance(lst_model, tf_utils.LensShading)
assert isinstance(lst_model, tf_utils.LensShadingModel)
# Check the tables are lists of lists
assert isinstance(lst_model.luminance, list)
assert isinstance(lst_model.luminance[0], list)

13
tests/unit_tests/test_server_cli.py
View file

@ -8,8 +8,9 @@ from fastapi import FastAPI
# Import as ofm server to attempt to minimise confusion with server as a var in other
# functions and also FastAPI `Server`.
from openflexure_microscope_server import server as ofm_server
from openflexure_microscope_server.things.camera import BaseCamera
from .test_server_config import FULL_CONFIG
from .test_server_config import SIM_CONFIG
def test_no_config():
@ -27,7 +28,7 @@ def test_successful_start(mocker, caplog):
mock_server = mocker.Mock()
# Create a mock for the camera so we can check the MJPEG streams are closed on
# shutdown.
mock_camera = mocker.Mock()
mock_camera = mocker.Mock(spec=BaseCamera)
mock_server.things = {"camera": mock_camera}
# Mock the LabThings function that returns the server so we have a mock server
mocker.patch(
@ -40,7 +41,7 @@ def test_successful_start(mocker, caplog):
mock_uvicorn_run = mocker.patch("openflexure_microscope_server.server.uvicorn.run")
# Run the mock CLI
ofm_server.serve_from_cli(["-c", FULL_CONFIG])
ofm_server.serve_from_cli(["-c", SIM_CONFIG])
# Check that the server was customised and the run
assert mock_customise.call_count == 1
@ -90,10 +91,10 @@ def test_failed_customise(mocker):
# Running the mock CLI will error
with pytest.raises(RuntimeError, match="Can't touch this"):
ofm_server.serve_from_cli(["-c", FULL_CONFIG])
ofm_server.serve_from_cli(["-c", SIM_CONFIG])
# But with the fallback flag uvicorn run will be run
ofm_server.serve_from_cli(["-c", FULL_CONFIG, "--fallback"])
ofm_server.serve_from_cli(["-c", SIM_CONFIG, "--fallback"])
assert mock_uvicorn_run.call_count == 1
# Get the fallback app passed to uvicorn tun
@ -101,4 +102,4 @@ def test_failed_customise(mocker):
# Check it really is a fastapi
assert isinstance(fallback_app, FastAPI)
# An that it has the error to display
assert str(fallback_app.labthings_error) == "Can't touch this"
assert str(fallback_app._context.error) == "Can't touch this"

10
tests/unit_tests/test_stage.py
View file

@ -3,7 +3,6 @@
import itertools
import pytest
from httpx import HTTPStatusError
from hypothesis import HealthCheck, given, settings
from hypothesis import strategies as st
@ -155,10 +154,8 @@ def test_direction_errors_local_and_http():
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:
with pytest.raises(lt.exceptions.ClientPropertyError):
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}
@ -166,10 +163,9 @@ def test_direction_errors_local_and_http():
# 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:
# But a FailedToInvokeActionError over HTTP
with pytest.raises(lt.exceptions.FailedToInvokeActionError):
stage_client.invert_axis_direction(axis="theta")
assert excinfo.value.response.status_code == 422
def _test_move_relative(dummy_stage, axis_inverted, path):