From 4060c06a2ba24168452720ef26fda3aab4071538 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Wed, 17 Dec 2025 21:50:31 +0000 Subject: [PATCH] Assorted type fixes --- .../scan_directories.py | 2 +- .../things/autofocus.py | 2 +- .../things/camera/__init__.py | 8 ++-- .../things/camera/opencv.py | 2 +- .../things/camera/picamera.py | 2 +- .../things/camera/simulation.py | 8 ++-- .../things/camera_stage_mapping.py | 23 +++-------- .../things/smart_scan.py | 38 +++++++++++-------- 8 files changed, 41 insertions(+), 44 deletions(-) diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 8c0c4a5f..e0bee445 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -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 diff --git a/src/openflexure_microscope_server/things/autofocus.py b/src/openflexure_microscope_server/things/autofocus.py index d885c6b8..f46f7ee9 100644 --- a/src/openflexure_microscope_server/things/autofocus.py +++ b/src/openflexure_microscope_server/things/autofocus.py @@ -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 diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py index f65747ce..561d612f 100644 --- a/src/openflexure_microscope_server/things/camera/__init__.py +++ b/src/openflexure_microscope_server/things/camera/__init__.py @@ -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: diff --git a/src/openflexure_microscope_server/things/camera/opencv.py b/src/openflexure_microscope_server/things/camera/opencv.py index 3882730e..db8ee41f 100644 --- a/src/openflexure_microscope_server/things/camera/opencv.py +++ b/src/openflexure_microscope_server/things/camera/opencv.py @@ -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. diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py index cd7b50ad..de8389ad 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -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 diff --git a/src/openflexure_microscope_server/things/camera/simulation.py b/src/openflexure_microscope_server/things/camera/simulation.py index b7ceeb26..c825a4ce 100644 --- a/src/openflexure_microscope_server/things/camera/simulation.py +++ b/src/openflexure_microscope_server/things/camera/simulation.py @@ -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. diff --git a/src/openflexure_microscope_server/things/camera_stage_mapping.py b/src/openflexure_microscope_server/things/camera_stage_mapping.py index 6ca3ee7f..fd33d7af 100644 --- a/src/openflexure_microscope_server/things/camera_stage_mapping.py +++ b/src/openflexure_microscope_server/things/camera_stage_mapping.py @@ -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: diff --git a/src/openflexure_microscope_server/things/smart_scan.py b/src/openflexure_microscope_server/things/smart_scan.py index 45d9333d..06d2b6b6 100644 --- a/src/openflexure_microscope_server/things/smart_scan.py +++ b/src/openflexure_microscope_server/things/smart_scan.py @@ -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, )