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

Assorted type fixes

Browse source
This commit is contained in:
Julian Stirling 2025-12-17 21:50:31 +00:00
parent 472aa03822
commit 4060c06a2b
8 changed files with 41 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/openflexure_microscope_server/scan_directories.py
View file

@ -346,7 +346,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

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

@ -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

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

@ -204,7 +204,7 @@ 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.
@ -364,7 +364,7 @@ class BaseCamera(lt.Thing):
self,
stream_name: Literal["main", "lores", "raw"],
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"
@ -431,7 +431,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
@ -513,7 +513,7 @@ class BaseCamera(lt.Thing):
def _save_capture(
self,
jpeg_path: str,
image: Image,
image: Image.Image,
metadata: dict,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:

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

@ -113,7 +113,7 @@ class OpenCVCamera(BaseCamera):
self,
stream_name: Literal["main", "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.

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

@ -534,7 +534,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

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

@ -196,7 +196,7 @@ 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'
@ -230,7 +230,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
@ -337,7 +337,7 @@ class SimulatedCamera(BaseCamera):
self,
stream_name: Literal["main", "lores", "raw"],
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.
@ -403,7 +403,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.

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

@ -22,7 +22,6 @@ from typing import (
)
import numpy as np
from fastapi import HTTPException
import labthings_fastapi as lt
from camera_stage_mapping.camera_stage_calibration_1d import (
@ -91,25 +90,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.
@ -241,9 +229,10 @@ class CameraStageMapper(lt.Thing):
def assert_calibrated(self) -> None:
"""Raise an exception if the image_to_stage_displacement matrix is 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."
)
@lt.action
def move_in_image_coordinates(self, x: float, y: float) -> None:

38
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)
@ -230,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_calibrated()
if self.skip_background:
if not self._cam.background_detector_status.ready:
@ -275,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
@ -283,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,
@ -363,7 +366,7 @@ class SmartScanThing(lt.Thing):
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.ongoing_scan.save_scan_data(self.scan_data)
@_scan_running
def _manage_stitching_threads(self) -> None:
@ -385,13 +388,18 @@ class SmartScanThing(lt.Thing):
self._cam.start_streaming(main_resolution=(3280, 2464))
self._scan_data = self._collect_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,
)
self._preview_stitcher = stitching.PreviewStitcher(
self.ongoing_scan.images_dir,
images_dir,
overlap=self.scan_data.overlap,
correlation_resize=self.scan_data.correlation_resize,
)
@ -476,7 +484,7 @@ class SmartScanThing(lt.Thing):
continue
focused, focused_height = self._autofocus.run_smart_stack(
stack_parameters=self._stack_params,
stack_parameters=self.stack_params,
save_on_failure=not self.scan_data.skip_background,
)