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

Add ANN20X rules to ruff

Browse source
This commit is contained in:
Julian Stirling 2025-08-27 14:00:49 +01:00
parent d673196671
commit 4c46330959
20 changed files with 98 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -127,7 +127,7 @@ class StackParams:
"""
return self.min_images_to_test + 15
def slice_to_save(self, sharpest_index: int):
def slice_to_save(self, sharpest_index: int) -> slice:
"""Return the slice of images to save given the index of the sharpest image."""
images_each_side = (self.images_to_save - 1) // 2
return slice(
@ -234,7 +234,7 @@ class JPEGSharpnessMonitor:
running = False
async def monitor_sharpness(self):
async def monitor_sharpness(self) -> None:
"""Start monitoring the frame sizes."""
self.running = True
async for frame in self.camera.lores_mjpeg_stream.frame_async_generator():
@ -244,7 +244,7 @@ class JPEGSharpnessMonitor:
break
@contextmanager
def run(self):
def run(self) -> None:
"""Context manager, during which we will monitor sharpness from the camera."""
self.portal.start_task_soon(self.monitor_sharpness)
try:
@ -403,7 +403,7 @@ class AutofocusThing(lt.Thing):
sharpness_monitor: SharpnessMonitorDep,
dz: int = 2000,
start: Literal["centre", "base"] = "centre",
):
) -> tuple[list[float], list[float]]:
"""Repeatedly autofocus the stage until it looks focused.
This action will run the ``fast_autofocus`` action until it settles on a point

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

@ -133,7 +133,7 @@ class CameraMemoryBuffer:
"No image with matching id in memory to retrieve."
) from e
def clear(self):
def clear(self) -> None:
"""Clear all images from memory."""
self._storage.clear()
@ -208,7 +208,7 @@ class BaseCamera(lt.Thing):
"CameraThings must define their own start_streaming method"
)
def kill_mjpeg_streams(self):
def kill_mjpeg_streams(self) -> None:
"""Kill the streams now as the server is shutting down.
This is called when uvicorn gets the a shutdown signal. As this is called from

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

@ -65,7 +65,7 @@ class OpenCVCamera(BaseCamera):
return self._capture_thread.is_alive()
return False
def _capture_frames(self):
def _capture_frames(self) -> None:
portal = lt.get_blocking_portal(self)
while self._capture_enabled:
ret, frame = self.cap.read()

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

@ -72,7 +72,7 @@ class PicameraStreamOutput(Output):
_timestamp: Optional[int] = None,
_packet: Any = None,
_audio: bool = False,
):
) -> None:
"""Add a frame to the stream's ringbuffer."""
self.stream.add_frame(frame, self.portal)
@ -173,7 +173,7 @@ class StreamingPiCamera2(BaseCamera):
)
"""Whether the MJPEG stream is active."""
def save_settings(self):
def save_settings(self) -> None:
"""Override save_settings to ensure that camera properties don't recurse.
This method is run by any Thing when a ThingSetting is saved. However, the
@ -204,7 +204,7 @@ class StreamingPiCamera2(BaseCamera):
return self._analogue_gain
@analogue_gain.setter
def analogue_gain(self, value: float):
def analogue_gain(self, value: float) -> None:
self._analogue_gain = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -224,7 +224,7 @@ class StreamingPiCamera2(BaseCamera):
return self._colour_gains
@colour_gains.setter
def colour_gains(self, value: tuple[float, float]):
def colour_gains(self, value: tuple[float, float]) -> None:
self._colour_gains = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -248,7 +248,7 @@ class StreamingPiCamera2(BaseCamera):
return self._exposure_time
@exposure_time.setter
def exposure_time(self, value: int):
def exposure_time(self, value: int) -> None:
self._exposure_time = value
if self.streaming:
with self._streaming_picamera() as cam:
@ -292,7 +292,7 @@ class StreamingPiCamera2(BaseCamera):
return SensorModeSelector(**self._sensor_mode)
@sensor_mode.setter
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]):
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None:
"""Change the sensor mode used."""
if new_mode is None:
self._sensor_mode = None
@ -352,7 +352,7 @@ class StreamingPiCamera2(BaseCamera):
) from e
return None
def _initialise_picamera(self):
def _initialise_picamera(self) -> None:
"""Acquire the picamera device and store it as ``self._picamera``.
This duplicates logic in ``Picamera2.__init__`` to provide a tuning file that
@ -646,7 +646,7 @@ class StreamingPiCamera2(BaseCamera):
self,
target_white_level: int = 700,
percentile: float = 99.9,
):
) -> None:
"""Adjust exposure until a the target white level is reached.
Starting from the minimum exposure, gradually increase exposure until
@ -672,7 +672,7 @@ class StreamingPiCamera2(BaseCamera):
self,
method: Literal["percentile", "centre"] = "centre",
luminance_power: float = 1.0,
):
) -> None:
"""Correct the white balance of the image.
This calibration requires a neutral image, such that the 99th centile
@ -747,7 +747,7 @@ class StreamingPiCamera2(BaseCamera):
self._initialise_picamera()
@lt.thing_action
def reset_ccm(self):
def reset_ccm(self) -> None:
"""Overwrite the colour correction matrix in camera tuning with default values.
These values are from the Raspberry Pi Camera Algorithm and Tuning Guide, page

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

@ -8,7 +8,7 @@ See repository root for licensing information.
from __future__ import annotations
import logging
from typing import Literal, Optional
from typing import Literal, Optional, Mapping
from types import TracebackType
from threading import Thread
import time
@ -76,7 +76,7 @@ class SimulatedCamera(BaseCamera):
self.generate_blobs()
self.generate_canvas()
def generate_sprites(self):
def generate_sprites(self) -> None:
"""Generate sprites to populate the image."""
sprite_sizes = [5, 7, 10, 21, 36, 40]
self.sprites = []
@ -109,7 +109,7 @@ class SimulatedCamera(BaseCamera):
# Convert to uint8 and append to the list
self.sprites.append(sprite.astype(np.uint8))
def generate_blobs(self, n_blobs: int = 1000):
def generate_blobs(self, n_blobs: int = 1000) -> None:
"""Generate coordinates of blobs and their sizes.
A 1000x3 array is returned. Each row represents (x,y) coordinate
@ -125,7 +125,7 @@ class SimulatedCamera(BaseCamera):
self.blobs[:, 1] = RNG.uniform(w / 2, self.canvas_shape[1] - w / 2, n_blobs)
self.blobs[:, 2] = RNG.choice(len(self.sprites), n_blobs)
def generate_canvas(self):
def generate_canvas(self) -> None:
"""Generate a canvas.
Canvas is int16 so that random noise can be added to simulation image before
@ -145,7 +145,7 @@ class SimulatedCamera(BaseCamera):
self.canvas[self.canvas < 0] = 0
self.canvas[self.canvas > 255] = 255
def generate_image(self, pos: tuple[int, int, int]):
def generate_image(self, pos: tuple[int, int, int]) -> np.ndarray:
"""Generate an image with blobs based on supplied coordinates."""
canvas_width, canvas_height, _ = self.canvas_shape
image_width, image_height, _ = self.shape
@ -180,16 +180,16 @@ class SimulatedCamera(BaseCamera):
def attach_to_server(
self, server: lt.ThingServer, path: str, setting_storage_path: str
):
) -> None:
"""Wrap the attach_to_server method so the server instance can be stored.
Direct access to the server instance is needed to get the stage position while
maintaining the same public API as a real camera that doesn't need this access.
"""
self._server = server
return super().attach_to_server(server, path, setting_storage_path)
super().attach_to_server(server, path, setting_storage_path)
def get_stage_position(self):
def get_stage_position(self) -> Mapping[str, int]:
"""Return the stage position.
The simulation camera has access to the stage position so it can generate a
@ -199,7 +199,7 @@ class SimulatedCamera(BaseCamera):
self._stage = self._server.things["/stage/"]
return self._stage.instantaneous_position
def generate_frame(self):
def generate_frame(self) -> np.ndarray:
"""Generate a frame with blobs based on the stage coordinates."""
try:
pos = self.get_stage_position()
@ -256,7 +256,7 @@ class SimulatedCamera(BaseCamera):
noise_level = lt.ThingProperty(float, 2.0)
def _capture_frames(self):
def _capture_frames(self) -> None:
portal = lt.get_blocking_portal(self)
while self._capture_enabled:
time.sleep(self.frame_interval)
@ -312,14 +312,14 @@ class SimulatedCamera(BaseCamera):
return Image.fromarray(self.generate_frame())
@lt.thing_action
def remove_sample(self):
def remove_sample(self) -> None:
"""Show the simulated background with no sample."""
if not self._show_sample:
raise RuntimeError("Sample is already removed.")
self._show_sample = False
@lt.thing_action
def load_sample(self):
def load_sample(self) -> None:
"""Show the simulated sample."""
if self._show_sample:
raise RuntimeError("Sample is already in place.")

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

@ -18,6 +18,7 @@ from typing import (
NamedTuple,
Optional,
Tuple,
Mapping,
)
from fastapi import HTTPException
@ -85,7 +86,7 @@ class RecordedMove:
positions: List[CoordinateType] = [p for t, p in self._history if p is not None]
return MoveHistory(times, positions)
def clear_history(self):
def clear_history(self) -> None:
"""Reset our history to be an empty list."""
self._history = []
@ -241,7 +242,7 @@ class CameraStageMapper(lt.Thing):
return None
return self.last_calibration["image_resolution"]
def assert_calibrated(self):
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
@ -254,7 +255,7 @@ class CameraStageMapper(lt.Thing):
stage: Stage,
x: float,
y: float,
):
) -> None:
"""Move by a given number of pixels on the camera.
NB x and y here refer to what is usually understood to be the horizontal and
@ -273,7 +274,7 @@ class CameraStageMapper(lt.Thing):
stage.move_relative(x=relative_move[0], y=relative_move[1])
@lt.thing_property
def thing_state(self) -> dict[str, Any]:
def thing_state(self) -> Mapping[str, Any]:
"""Summary metadata describing the current state of the Thing."""
return {
k: getattr(self, k)

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

@ -38,8 +38,17 @@ CSMDep = lt.deps.direct_thing_client_dependency(
)
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
JPEGBlob = lt.blob.blob_type("image/jpeg")
ZipBlob = lt.blob.blob_type("application/zip")
class JPEGBlob(lt.blob.Blob):
"""A class representing a JPEG image as a LabThings FastAPI Blob."""
media_type: str = "image/jpeg"
class ZipBlob(lt.blob.Blob):
"""A class representing a Zip file as a LabThings FastAPI Blob."""
media_type: str = "application/zip"
class ScanNotRunningError(RuntimeError):
@ -116,7 +125,7 @@ class SmartScanThing(lt.Thing):
cam: CameraClient,
csm: CSMDep,
scan_name: str = "",
):
) -> None:
"""Move the stage to cover an area, taking images that can be tiled together.
The stage will move in a pattern that grows outwards from the starting point,
@ -171,7 +180,7 @@ class SmartScanThing(lt.Thing):
self._preview_stitcher = None
@_scan_running
def _check_background_and_csm_set(self):
def _check_background_and_csm_set(self) -> None:
"""Before starting a scan, check that background and camera-stage-mapping are set.
Raise error if:
@ -311,7 +320,7 @@ class SmartScanThing(lt.Thing):
)
@_scan_running
def _save_final_scan_data(self, scan_result: str):
def _save_final_scan_data(self, scan_result: str) -> None:
"""Update scan data JSON file with data only known at the end of the scan.
Takes scan_result, a string that is either "success", "cancelled by user",
@ -321,7 +330,7 @@ class SmartScanThing(lt.Thing):
self._ongoing_scan.save_scan_data(self._scan_data)
@_scan_running
def _manage_stitching_threads(self):
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.
@ -330,7 +339,7 @@ class SmartScanThing(lt.Thing):
self._preview_stitcher.start()
@_scan_running
def _run_scan(self):
def _run_scan(self) -> None:
"""Prepare and run the main scan, and perform final actions on completion.
The result (or exception) from the main scan loop determines whether the
@ -382,7 +391,7 @@ class SmartScanThing(lt.Thing):
self.purge_empty_scans(logger=self._scan_logger)
@_scan_running
def _main_scan_loop(self):
def _main_scan_loop(self) -> None:
"""Run the main loop of the scan.
This loop runs during a scan, until no more scan x,y positions
@ -449,7 +458,7 @@ class SmartScanThing(lt.Thing):
self._ongoing_scan.zip_files()
@_scan_running
def _return_to_starting_position(self):
def _return_to_starting_position(self) -> None:
"""Return to the initial scan position, if set."""
self._scan_logger.info("Returning to starting position.")
if self._scan_data is not None:
@ -458,7 +467,7 @@ class SmartScanThing(lt.Thing):
)
@_scan_running
def _perform_final_stitch(self):
def _perform_final_stitch(self) -> None:
"""Update the scan zip and perform final stitch of the data."""
if self._scan_data.image_count <= 3:
self._scan_logger.info("Not performing a stitch as 3 or fewer images taken")
@ -729,7 +738,7 @@ class SmartScanThing(lt.Thing):
def download_zip(
self,
scan_name: str,
):
) -> ZipBlob:
"""Return zip after including any files left until the end.
The zipfile is returned as a Blob.
@ -740,7 +749,7 @@ class SmartScanThing(lt.Thing):
@lt.thing_action
def stitch_all_scans(
self, logger: lt.deps.InvocationLogger, cancel: lt.deps.CancelHook
):
) -> None:
"""Check the list of scans, and stitch any that don't have a DZI associated with it.
:raises RuntimeError: if the microscope is currently running a scan

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

@ -11,7 +11,7 @@ As the object will be used as a context manager create the hardware connection i
from __future__ import annotations
from collections.abc import Sequence, Mapping
from typing import Literal
from typing import Literal, Never, Any
import labthings_fastapi as lt
@ -119,12 +119,12 @@ class BaseStage(lt.Thing):
)
@property
def thing_state(self):
def thing_state(self) -> Mapping[str, Any]:
"""Summary metadata describing the current state of the stage."""
return {"position": self.position}
@lt.thing_action
def invert_axis_direction(self, axis: Literal["x", "y", "z"]):
def invert_axis_direction(self, axis: Literal["x", "y", "z"]) -> None:
"""Invert the direction setting of the given axis.
:param axis: The axis name (x, y or z) to invert.
@ -143,7 +143,7 @@ class BaseStage(lt.Thing):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> None:
"""Make a relative move. Keyword arguments should be axis names."""
self._hardware_move_relative(
cancel=cancel,
@ -156,7 +156,7 @@ class BaseStage(lt.Thing):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> Never:
"""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``.
@ -171,7 +171,7 @@ class BaseStage(lt.Thing):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> None:
"""Make an absolute move. Keyword arguments should be axis names."""
self._hardware_move_absolute(
cancel=cancel,
@ -184,7 +184,7 @@ class BaseStage(lt.Thing):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> Never:
"""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``.
@ -194,7 +194,7 @@ class BaseStage(lt.Thing):
)
@lt.thing_action
def set_zero_position(self):
def set_zero_position(self) -> Never:
"""Make the current position zero in all axes.
This action does not move the stage, but resets the position to zero.

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

@ -55,7 +55,7 @@ class DummyStage(BaseStage):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> None:
"""Make a relative move. Keyword arguments should be axis names."""
displacement = [kwargs.get(k, 0) for k in self.axis_names]
self.moving = True
@ -93,7 +93,7 @@ class DummyStage(BaseStage):
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: int,
):
) -> None:
"""Make an absolute move. Keyword arguments should be axis names."""
displacement = {
axis: int(pos) - self._hardware_position[axis]
@ -105,7 +105,7 @@ class DummyStage(BaseStage):
)
@lt.thing_action
def set_zero_position(self):
def set_zero_position(self) -> None:
"""Make the current position zero in all axes.
This action does not move the stage, but resets the position to zero.

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

@ -51,7 +51,7 @@ class OpenFlexureSystem(lt.Thing):
return UUID(self._microscope_id)
@microscope_id.setter
def microscope_id(self, uuid: UUID):
def microscope_id(self, uuid: UUID) -> None:
# TODO make read only but still settable from disk
self._microscope_id = uuid