From b89449be777ceaa071a81d8d1807ed5e727afc0c Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sun, 13 Jul 2025 21:55:55 +0100 Subject: [PATCH] Add camera-stage-mapping compliant functions to BaseStage and BaseCamera. --- .../things/camera/__init__.py | 58 ++++++++ .../things/camera/opencv.py | 5 + .../things/camera/picamera.py | 12 +- .../things/camera/simulation.py | 7 + .../things/camera_stage_mapping.py | 135 ++++-------------- .../things/stage/__init__.py | 23 +++ 6 files changed, 130 insertions(+), 110 deletions(-) diff --git a/src/openflexure_microscope_server/things/camera/__init__.py b/src/openflexure_microscope_server/things/camera/__init__.py index 12301213..5c243683 100644 --- a/src/openflexure_microscope_server/things/camera/__init__.py +++ b/src/openflexure_microscope_server/things/camera/__init__.py @@ -9,7 +9,9 @@ See repository root for licensing information. from __future__ import annotations from typing import Literal, Optional, Tuple, Any import json +import time +import numpy as np from pydantic import RootModel from PIL import Image import piexif @@ -195,6 +197,13 @@ class BaseCamera(lt.Thing): "CameraThings must define their own stream_active method" ) + @lt.thing_action + def discard_frames(self) -> None: + """Discard frames so that the next frame captured is fresh.""" + raise NotImplementedError( + "CameraThings must define their own discard_frames method" + ) + @lt.thing_action def capture_array( self, @@ -206,6 +215,21 @@ class BaseCamera(lt.Thing): "CameraThings must define their own capture_array method" ) + downsampled_array_factor = lt.ThingProperty(int, 2) + """The downsampling factor when calling capture_downsampled_array.""" + + @lt.thing_action + 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`. + * The default capture array arguments are used. + + This method is provides the interface expected by the camera_stage_mapping. + """ + img = self.capture_array() + return downsample(self.downsampled_array_factor, img) + @lt.thing_action def capture_jpeg( self, @@ -415,6 +439,40 @@ class BaseCamera(lt.Thing): except Exception as e: raise IOError(f"An error occurred while saving {jpeg_path}") from e + settling_time = lt.ThingSetting(float, 0.2) + """The the settling time when calling the ``settle()`` method.""" + + @lt.thing_action + def settle(self) -> None: + """Sleep for the settling time, ready to provide a fresh frame. + + This method is provides the interface expected by the camera_stage_mapping. + """ + time.sleep(self.settling_time) + self.discard_frames() + CameraDependency = lt.deps.direct_thing_client_dependency(BaseCamera, "/camera/") RawCameraDependency = lt.deps.raw_thing_dependency(BaseCamera) + + +def downsample(factor: int, image: np.ndarray) -> np.ndarray: + """Downsample an image by taking the mean of each nxn region. + + This should be very efficient: + + * calculate each pixel as the mean of each ``factor * factor`` square without + interpolation. + * If the image is not an integer multiple of the resampling factor, discard + the left-over pixels to avoids odd edge effects and keep performance quick. + """ + if factor == 1: + return image + new_size = [d // factor for d in image.shape[:2]] + # First, we ensure we have something that's an integer multiple + # of `factor` + cropped = image[: new_size[0] * factor, : new_size[1] * factor, ...] + reshaped = cropped.reshape( + (new_size[0], factor, new_size[1], factor) + image.shape[2:] + ) + return reshaped.mean(axis=(1, 3)) diff --git a/src/openflexure_microscope_server/things/camera/opencv.py b/src/openflexure_microscope_server/things/camera/opencv.py index a5604432..7f543cf1 100644 --- a/src/openflexure_microscope_server/things/camera/opencv.py +++ b/src/openflexure_microscope_server/things/camera/opencv.py @@ -75,6 +75,11 @@ class OpenCVCamera(BaseCamera): ].tobytes() self.lores_mjpeg_stream.add_frame(jpeg_lores, portal) + @lt.thing_action + def discard_frames(self) -> None: + """Discard frames so that the next frame captured is fresh.""" + self.capture_array() + @lt.thing_action def capture_array( self, diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py index 0e3dd04f..c676b82d 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -240,12 +240,8 @@ class StreamingPiCamera2(BaseCamera): cam.set_controls({"ExposureTime": value + 1}) def _get_persistent_controls(self) -> dict: - discard_frames: int = 1 if self.streaming: - with self._streaming_picamera() as cam: - # Discard frames, so data is fresh - for i in range(discard_frames): - cam.capture_metadata() + self.discard_frames() return { "AeEnable": False, "AnalogueGain": self.analogue_gain, @@ -474,6 +470,12 @@ class StreamingPiCamera2(BaseCamera): # Adding a sleep to prevent camera getting confused by rapid commands time.sleep(0.2) + @lt.thing_action + def discard_frames(self) -> None: + """Discard frames so that the next frame captured is fresh.""" + with self._streaming_picamera() as cam: + cam.capture_metadata() + @lt.thing_action def capture_image( self, diff --git a/src/openflexure_microscope_server/things/camera/simulation.py b/src/openflexure_microscope_server/things/camera/simulation.py index ed2d7e14..73321082 100644 --- a/src/openflexure_microscope_server/things/camera/simulation.py +++ b/src/openflexure_microscope_server/things/camera/simulation.py @@ -187,6 +187,13 @@ class SimulatedCamera(BaseCamera): except Exception as e: logging.error(f"Failed to capture frame: {e}, retrying...") + @lt.thing_action + def discard_frames(self) -> None: + """Discard frames so that the next frame captured is fresh. + + There is nothing to do as this is a simulation! + """ + @lt.thing_action def capture_array( self, diff --git a/src/openflexure_microscope_server/things/camera_stage_mapping.py b/src/openflexure_microscope_server/things/camera_stage_mapping.py index 97aa5e25..f924ec3d 100644 --- a/src/openflexure_microscope_server/things/camera_stage_mapping.py +++ b/src/openflexure_microscope_server/things/camera_stage_mapping.py @@ -12,21 +12,17 @@ server, and depends on that server and its underlying LabThings library. import time from typing import ( - Annotated, Any, Callable, Dict, List, - Mapping, NamedTuple, Optional, - Sequence, Tuple, ) -from fastapi import Depends, HTTPException +from fastapi import HTTPException import numpy as np -from pydantic import BaseModel from camera_stage_mapping.camera_stage_calibration_1d import ( calibrate_backlash_1d, image_to_stage_displacement_from_1d, @@ -34,7 +30,7 @@ from camera_stage_mapping.camera_stage_calibration_1d import ( from camera_stage_mapping.exceptions import MappingError import labthings_fastapi as lt -from labthings_fastapi.types.numpy import NDArray, DenumpifyingDict +from labthings_fastapi.types.numpy import DenumpifyingDict from camera_stage_mapping.camera_stage_tracker import Tracker from .camera import CameraDependency as Camera @@ -44,89 +40,6 @@ CoordinateType = Tuple[float, float, float] XYCoordinateType = Tuple[float, float] -class HardwareInterfaceModel(BaseModel): - """A pydantic base model for a stage and camera interface. - - This code has been flagged as confusing and will be replaced soon, see #476. - """ - - move: Callable[[NDArray], None] - get_position: Callable[[], NDArray] - grab_image: Callable[[], NDArray] - settle: Callable[[], None] - grab_image_downsampling: float = 1 - - -def downsample(factor: int, image: np.ndarray) -> np.ndarray: - """Downsample an image by taking the mean of each nxn region. - - This should be very efficient: we calculate the mean of each - ``factor * factor`` square, no interpolation. If the image is - not an integer multiple of the resampling factor, we discard - the left-over pixels. This avoids odd edge effects and keeps - performance quick. - """ - if factor == 1: - return image - new_size = [d // factor for d in image.shape[:2]] - # First, we ensure we have something that's an integer multiple - # of `factor` - cropped = image[: new_size[0] * factor, : new_size[1] * factor, ...] - reshaped = cropped.reshape( - (new_size[0], factor, new_size[1], factor) + image.shape[2:] - ) - return reshaped.mean(axis=(1, 3)) - - -DEFAULT_SETTLING_TIME = 0.2 - - -def make_hardware_interface( - stage: Stage, camera: Camera, downsample_factor: int = 2 -) -> HardwareInterfaceModel: - """Construct the functions we need to interface with the hardware.""" - axes = stage.axis_names - - def pos2dict(pos: Sequence[float]) -> Mapping[str, float]: - return dict(zip(axes, pos)) - - def dict2pos(posd: Mapping[str, float]) -> Sequence[float]: - return tuple(posd[k] for k in axes if k in posd) - - def move(pos: CoordinateType) -> None: - current_pos = stage.position - new_pos = pos2dict(pos) - displacement = {k: new_pos[k] - current_pos[k] for k in new_pos.keys()} - stage.move_relative(**displacement) - - def get_position() -> CoordinateType: - return dict2pos(stage.position) - - def grab_image() -> np.ndarray: - img = camera.capture_array() - return downsample(downsample_factor, img) - - def settle() -> None: - time.sleep(DEFAULT_SETTLING_TIME) - try: - camera.capture_metadata # This discards frames on a picamera - except AttributeError: - pass # Don't raise an error for other cameras (may consider grabbing a frame) - - return HardwareInterfaceModel( - move=move, - get_position=get_position, - grab_image=grab_image, - settle=settle, - grab_image_downsampling=downsample_factor, - ) - - -HardwareInterfaceDep = Annotated[ - HardwareInterfaceModel, Depends(make_hardware_interface) -] - - class MoveHistory(NamedTuple): """A named tuple containing the position over time for a single move. @@ -158,7 +71,7 @@ class LoggingMoveWrapper: """ self._move_function: Callable = move_function self._current_position: Optional[CoordinateType] = None - self.clear_history() + self._history: List[Tuple[float, Optional[CoordinateType]]] = [] def __call__(self, new_position: CoordinateType, *args, **kwargs): """Move to a new position, and record it.""" @@ -176,7 +89,7 @@ class LoggingMoveWrapper: def clear_history(self): """Reset our history to be an empty list.""" - self._history: List[Tuple[float, Optional[CoordinateType]]] = [] + self._history = [] class CSMUncalibratedError(HTTPException): @@ -200,26 +113,34 @@ class CSMUncalibratedError(HTTPException): class CameraStageMapper(lt.Thing): - """A Thing to manage mapping between image and stage coordinates.""" + """A Thing to manage mapping between image and stage coordinates. + + To use this Thing, the stage must have axes named "x", "y", and "z", or must + override the ``get_xyz_position()`` and ``move_to_xyz_position()`` methods. + """ @lt.thing_action def calibrate_1d( self, - hw: HardwareInterfaceDep, + camera: Camera, stage: Stage, logger: lt.deps.InvocationLogger, direction: Tuple[float, float, float], ) -> DenumpifyingDict: """Move a microscope's stage in 1D, and figure out the relationship with the camera.""" # log positions and times for stage calibration - move = LoggingMoveWrapper(hw.move) - tracker = Tracker(hw.grab_image, hw.get_position, settle=hw.settle) + wrapped_move = LoggingMoveWrapper(stage.move_to_xyz_position) + tracker = Tracker( + camera.capture_downsampled_array, + stage.get_xyz_position, + settle=camera.settle, + ) direction_array: np.ndarray = np.array(direction) starting_position = stage.position try: result: dict = calibrate_backlash_1d( - tracker, move, direction_array, logger=logger + tracker, wrapped_move, direction_array, logger=logger ) except lt.exceptions.InvocationCancelledError as e: logger.info("User cancelled the camera stage mapping calibration") @@ -230,37 +151,41 @@ class CameraStageMapper(lt.Thing): logger.info("Returning to starting position due to failed calibration") stage.move_absolute(**starting_position, block_cancellation=True) raise e - result["move_history"] = move.history - result["image_resolution"] = hw.grab_image().shape[:2] + result["move_history"] = wrapped_move.history + result["image_resolution"] = camera.capture_downsampled_array().shape[:2] return result @lt.thing_action def calibrate_xy( - self, hw: HardwareInterfaceDep, stage: Stage, logger: lt.deps.InvocationLogger + self, camera: Camera, stage: Stage, logger: lt.deps.InvocationLogger ) -> DenumpifyingDict: """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. """ + downsampling_factor = camera.downsampled_array_factor # Calibrate y-axis first as it is more likely to fail. # The x-y difference is due to the camera aspect ratio, not the stage hardware. logger.info("Calibrating Y axis:") - cal_y: dict = self.calibrate_1d(hw, stage, logger, (0, 1, 0)) + cal_y: dict = self.calibrate_1d(camera, stage, logger, (0, 1, 0)) logger.info("Calibrating X axis:") - cal_x: dict = self.calibrate_1d(hw, stage, logger, (1, 0, 0)) + cal_x: dict = self.calibrate_1d(camera, stage, logger, (1, 0, 0)) logger.info("Calibration complete, updating metadata.") # Combine X and Y calibrations to make a 2D calibration cal_xy: dict = image_to_stage_displacement_from_1d([cal_x, cal_y]) # Correct the result for downsampling performed in the hardware interface # (this may be to speed up correlation, or to avoid debayering artifacts) - cal_xy["image_to_stage_displacement"] /= hw.grab_image_downsampling + cal_xy["image_to_stage_displacement"] /= downsampling_factor corrected_resolution = tuple( - r * hw.grab_image_downsampling for r in cal_x["image_resolution"] + r * downsampling_factor for r in cal_x["image_resolution"] ) csm_matrix = cal_xy["image_to_stage_displacement"] - csm_as_string = f"[{round(csm_matrix[0][0], 2)}, {round(csm_matrix[0][1], 2)},],[{round(csm_matrix[1][0], 2)}, {round(csm_matrix[1][1], 2)}]" + csm_as_string = ( + f"[{round(csm_matrix[0][0], 2)}, {round(csm_matrix[0][1], 2)},]," + f"[{round(csm_matrix[1][0], 2)}, {round(csm_matrix[1][1], 2)}]" + ) logger.info(f"CSM matrix is {csm_as_string}.") data: Dict[str, dict] = { @@ -269,7 +194,7 @@ class CameraStageMapper(lt.Thing): "linear_calibration_y": cal_y, "downsampled_image_resolution": cal_x["image_resolution"], "image_resolution": corrected_resolution, - "downsampling": hw.grab_image_downsampling, + "downsampling": downsampling_factor, } self.last_calibration = DenumpifyingDict(data).model_dump() diff --git a/src/openflexure_microscope_server/things/stage/__init__.py b/src/openflexure_microscope_server/things/stage/__init__.py index fb58cb65..ac9a9b2b 100644 --- a/src/openflexure_microscope_server/things/stage/__init__.py +++ b/src/openflexure_microscope_server/things/stage/__init__.py @@ -88,5 +88,28 @@ class BaseStage(lt.Thing): "StageThings must define their own set_zero_position method" ) + @lt.thing_action + def get_xyz_position(self) -> tuple[int, int, int]: + """Return a tuple containing (x, y, z) position. + + :raises KeyError: if this stage does not have axes named "x", "y", and "z". + + This method is provides the interface expected by the camera_stage_mapping. + """ + position_dict = self.position + return (position_dict["x"], position_dict["y"], position_dict["z"]) + + @lt.thing_action + def move_to_xyz_position( + self, cancel: lt.deps.CancelHook, xyz_pos: tuple[int, int, int] + ) -> None: + """Move to the location specified by an (x, y, z) tuple. + + :raises KeyError: if this stage does not have axes named "x", "y", and "z". + + This method is provides the interface expected by the camera_stage_mapping. + """ + self.move_absolute(cancel=cancel, x=xyz_pos[0], y=xyz_pos[1], z=xyz_pos[2]) + StageDependency = lt.deps.direct_thing_client_dependency(BaseStage, "/stage/")