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

Add punctuation to docstrings

Browse source
This commit is contained in:
Julian Stirling 2025-07-10 02:03:02 +01:00
parent 4dc41bb008
commit 80beeea07b
34 changed files with 232 additions and 235 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,4 +1,4 @@
"""OpenFlexure Microscope Camera
"""OpenFlexure Microscope Camera.
This module defines the interface for cameras. Any compatible lt.Thing
should enabe the server to work.
@ -23,23 +23,23 @@ class JPEGBlob(lt.blob.Blob):
class PNGBlob(lt.blob.Blob):
"""A class representing a PNG image as a LabThings FastAPI Blob"""
"""A class representing a PNG image as a LabThings FastAPI Blob."""
media_type: str = "image/png"
class ArrayModel(RootModel):
"""A model for an array"""
"""A model for an array."""
root: NDArray
class CaptureError(RuntimeError):
"""An error trying to capture from a CameraThing"""
"""An error trying to capture from a CameraThing."""
class NoImageInMemoryError(RuntimeError):
"""An error called if no image in in memory when an method is called to use that image"""
"""An error called if no image in in memory when an method is called to use that image."""
class CameraMemoryBuffer:
@ -61,7 +61,7 @@ class CameraMemoryBuffer:
def add_image(
self, image: Any, metadata: Optional[dict] = None, buffer_max: int = 1
) -> int:
"""Add an image to the Memory buffer
"""Add an image to the Memory buffer.
This will add an image to the memory buffer. By default the buffer will
be cleared. To allow saving multiple images the buffer_max must be set
@ -114,7 +114,7 @@ class CameraMemoryBuffer:
) from e
def clear(self):
"""Clear all images from memory"""
"""Clear all images from memory."""
self._storage.clear()
def _create_space(self, buffer_max: int) -> None:
@ -140,7 +140,7 @@ class CameraMemoryBuffer:
class BaseCamera(lt.Thing):
"""The base class for all cameras. All cameras must directly inherit from this class"""
"""The base class for all cameras. All cameras must directly inherit from this class."""
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
@ -181,7 +181,7 @@ class BaseCamera(lt.Thing):
@lt.thing_property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active"""
"""Whether the MJPEG stream is active."""
raise NotImplementedError(
"CameraThings must define their own stream_active method"
)
@ -203,7 +203,7 @@ class BaseCamera(lt.Thing):
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 5,
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob"""
"""Acquire one image from the camera and return as a JPEG blob."""
raise NotImplementedError(
"CameraThings must define their own capture_jpeg method"
)
@ -214,7 +214,7 @@ class BaseCamera(lt.Thing):
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as an array
"""Acquire one image from the preview stream and return as an array.
This differs from ``capture_jpeg`` in that it does not pause the MJPEG
preview stream. Instead, we simply return the next frame from that
@ -233,7 +233,7 @@ class BaseCamera(lt.Thing):
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size"""
"""Acquire one image from the preview stream and return its size."""
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
@ -245,7 +245,7 @@ class BaseCamera(lt.Thing):
stream_name: Literal["main", "lores", "raw"],
wait: Optional[float],
) -> None:
"""Capture a PIL image from stream stream_name with timeout wait"""
"""Capture a PIL image from stream stream_name with timeout wait."""
raise NotImplementedError(
"CameraThings must define their own capture_image method"
)
@ -258,7 +258,7 @@ class BaseCamera(lt.Thing):
metadata_getter: lt.deps.GetThingStates,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Capture an image and save it to disk
"""Capture an image and save it to disk.
:param jpeg_path: The path to save the file to
:param logger: This should be injected automatically by Labthings FastAPI
@ -288,7 +288,7 @@ class BaseCamera(lt.Thing):
metadata_getter: lt.deps.GetThingStates,
buffer_max: int = 1,
) -> None:
"""Capture an image to memory. This can be saved later with ``save_from_memory``
"""Capture an image to memory. This can be saved later with ``save_from_memory``.
Note that only one image is held in memory so this will overwrite any image
in memory.
@ -336,7 +336,7 @@ class BaseCamera(lt.Thing):
@lt.thing_action
def clear_buffers(self) -> None:
"""Clear all images in memory"""
"""Clear all images in memory."""
self._memory_buffer.clear()
def _robust_image_capture(

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

@ -1,4 +1,4 @@
"""OpenFlexure Microscope OpenCV Camera
"""OpenFlexure Microscope OpenCV Camera.
This module defines a camera Thing that uses OpenCV's
``VideoCapture``.
@ -23,7 +23,7 @@ from . import BaseCamera, JPEGBlob
class OpenCVCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
"""A Thing representing an OpenCV camera."""
def __init__(self, camera_index: int = 0):
self.camera_index = camera_index
@ -45,7 +45,7 @@ class OpenCVCamera(BaseCamera):
@lt.thing_property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active"""
"""Whether the MJPEG stream is active."""
if self._capture_enabled and self._capture_thread:
return self._capture_thread.is_alive()
return False
@ -71,7 +71,7 @@ class OpenCVCamera(BaseCamera):
self,
resolution: Literal["main", "full"] = "full",
) -> NDArray:
"""Acquire one image from the camera and return as an array
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
It's likely to be highly inefficient - raw and/or uncompressed captures using
@ -91,7 +91,7 @@ class OpenCVCamera(BaseCamera):
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob
"""Acquire one image from the camera and return as a JPEG blob.
This function will produce a JPEG image.
"""

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

@ -40,10 +40,10 @@ from . import BaseCamera, JPEGBlob, ArrayModel
class PicameraStreamOutput(Output):
"""An Output class that sends frames to a stream"""
"""An Output class that sends frames to a stream."""
def __init__(self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal):
"""Create an output that puts frames in an MJPEGStream
"""Create an output that puts frames in an MJPEGStream.
We need to pass the stream object, and also the blocking portal, because
new frame notifications happen in the anyio event loop and frames are
@ -57,7 +57,7 @@ class PicameraStreamOutput(Output):
def outputframe(
self, frame, _keyframe=True, _timestamp=None, _packet=None, _audio=False
):
"""Add a frame to the stream's ringbuffer"""
"""Add a frame to the stream's ringbuffer."""
self.stream.add_frame(frame, self.portal)
@ -104,7 +104,7 @@ class LensShading(BaseModel):
class StreamingPiCamera2(BaseCamera):
"""A Thing that represents an OpenCV camera"""
"""A Thing that represents an OpenCV camera."""
def __init__(self, camera_num: int = 0):
self._setting_save_in_progress = False
@ -244,7 +244,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def sensor_modes(self) -> list[SensorMode]:
"""All the available modes the current sensor supports"""
"""All the available modes the current sensor supports."""
if not self._sensor_modes:
with self._streaming_picamera() as cam:
self._sensor_modes = cam.sensor_modes
@ -254,14 +254,14 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def sensor_mode(self) -> Optional[SensorModeSelector]:
"""The intended sensor mode of the camera"""
"""The intended sensor mode of the camera."""
if self._sensor_mode is None:
return None
return SensorModeSelector(**self._sensor_mode)
@sensor_mode.setter
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]):
"""Change the sensor mode used"""
"""Change the sensor mode used."""
if new_mode is None:
self._sensor_mode = None
elif isinstance(new_mode, SensorModeSelector):
@ -277,7 +277,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def sensor_resolution(self) -> Optional[tuple[int, int]]:
"""The native resolution of the camera's sensor"""
"""The native resolution of the camera's sensor."""
with self._streaming_picamera() as cam:
return cam.sensor_resolution
@ -324,7 +324,7 @@ class StreamingPiCamera2(BaseCamera):
@property
def streaming(self) -> bool:
"""True if the camera is streaming"""
"""True if the camera is streaming."""
return self._picamera is not None and self._picamera.started
@contextmanager
@ -432,7 +432,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_action
def stop_streaming(self, stop_web_stream: bool = True) -> None:
"""Stop the MJPEG stream"""
"""Stop the MJPEG stream."""
with self._streaming_picamera() as picam:
try:
picam.stop_recording() # This should also stop the extra lores encoder
@ -474,7 +474,7 @@ class StreamingPiCamera2(BaseCamera):
stream_name: Literal["main", "lores", "raw", "full"] = "main",
wait: Optional[float] = 0.9,
) -> ArrayModel:
"""Acquire one image from the camera and return as an array
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
It's likely to be highly inefficient - raw and/or uncompressed captures using
@ -496,7 +496,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def camera_configuration(self) -> Mapping:
"""The "configuration" dictionary of the picamera2 object
"""The "configuration" dictionary of the picamera2 object.
The "configuration" sets the resolution and format of the camera's streams.
Together with the "tuning" it determines how the sensor is configured and
@ -516,7 +516,7 @@ class StreamingPiCamera2(BaseCamera):
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 0.9,
) -> JPEGBlob:
"""Acquire one image from the camera as a JPEG
"""Acquire one image from the camera as a JPEG.
The JPEG will be acquired using ``Picamera2.capture_file``. If the
``resolution`` parameter is ``main`` or ``lores``, it will be captured
@ -567,7 +567,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def capture_metadata(self) -> dict:
"""Return the metadata from the camera"""
"""Return the metadata from the camera."""
with self._streaming_picamera() as cam:
return cam.capture_metadata()
@ -577,7 +577,7 @@ class StreamingPiCamera2(BaseCamera):
target_white_level: int = 700,
percentile: float = 99.9,
):
"""Adjust exposure until a the target white level is reached
"""Adjust exposure until a the target white level is reached.
Starting from the minimum exposure, gradually increase exposure until
the image reaches the specified white level.
@ -603,7 +603,7 @@ class StreamingPiCamera2(BaseCamera):
method: Literal["percentile", "centre"] = "centre",
luminance_power: float = 1.0,
):
"""Correct the white balance of the image
"""Correct the white balance of the image.
This calibration requires a neutral image, such that the 99th centile
of each colour channel should correspond to white. We calculate the
@ -711,7 +711,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_action
def full_auto_calibrate(self) -> None:
"""Perform a full auto-calibration
"""Perform a full auto-calibration.
This function will call the other calibration actions in sequence:
@ -729,7 +729,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_action
def flat_lens_shading(self) -> None:
"""Disable flat-field correction
"""Disable flat-field correction.
This method will set a completely flat lens shading table. It is not the
same as the default behaviour, which is to use an adaptive lens shading
@ -748,7 +748,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def lens_shading_tables(self) -> Optional[LensShading]:
"""The current lens shading (i.e. flat-field correction)
"""The current lens shading (i.e. flat-field correction).
Return the current lens shading correction, as three 2D lists each with
dimensions 16x12, if a static lens shading table is in use.
@ -770,7 +770,7 @@ class StreamingPiCamera2(BaseCamera):
return None
def reshape_lst(lin: list[float]) -> list[list[float]]:
"""Reshape the 192 element list into a 2D 16x12 list"""
"""Reshape the 192 element list into a 2D 16x12 list."""
w, h = 16, 12
return [lin[w * i : w * (i + 1)] for i in range(h)]
@ -782,7 +782,7 @@ class StreamingPiCamera2(BaseCamera):
@lens_shading_tables.setter
def lens_shading_tables(self, lst: LensShading) -> None:
"""Set the lens shading tables"""
"""Set the lens shading tables."""
with self._streaming_picamera(pause_stream=True):
recalibrate_utils.set_static_lst(
self.tuning,
@ -795,7 +795,7 @@ class StreamingPiCamera2(BaseCamera):
def correct_colour_gains_for_lens_shading(
self, colour_gains: tuple[float, float]
) -> tuple[float, float]:
"""Correct white balance gains for the effect of lens shading
"""Correct white balance gains for the effect of lens shading.
The white balance algorithm we use assumes the brightest pixels
should be white, and that the only thing affecting the colour of
@ -825,7 +825,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_action
def flat_lens_shading_chrominance(self) -> None:
"""Disable flat-field correction
"""Disable flat-field correction.
This method will set the chrominance of the lens shading table to be
flat, i.e. we'll correct vignetting of intensity, but not any change in
@ -840,7 +840,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_action
def reset_lens_shading(self) -> None:
"""Revert to default lens shading settings
"""Revert to default lens shading settings.
This method will restore the default "adaptive" lens shading method used
by the Raspberry Pi camera.
@ -851,7 +851,7 @@ class StreamingPiCamera2(BaseCamera):
@lt.thing_property
def lens_shading_is_static(self) -> bool:
"""Whether the lens shading is static
"""Whether the lens shading is static.
This property is true if the lens shading correction has been set to use
a static table (i.e. the number of automatic correction iterations is zero).

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

@ -1,4 +1,4 @@
"""Functions to set up a Raspberry Pi Camera v2 for scientific use
"""Functions to set up a Raspberry Pi Camera v2 for scientific use.
This module provides slower, simpler functions to set the
gain, exposure, and white balance of a Raspberry Pi camera, using
@ -52,7 +52,7 @@ LensShadingTables = tuple[np.ndarray, np.ndarray, np.ndarray]
def load_default_tuning(cam: Picamera2) -> dict:
"""Load the default tuning file for the camera
"""Load the default tuning file for the camera.
This will open and close the camera to determine its model. If you are
using a model that's supported by ``picamera2`` it should have a tuning
@ -76,7 +76,7 @@ def load_default_tuning(cam: Picamera2) -> dict:
def set_minimum_exposure(camera: Picamera2) -> None:
"""Enable manual exposure, with low gain and shutter speed
"""Enable manual exposure, with low gain and shutter speed.
We set exposure mode to manual, analog and digital gain
to 1, and shutter speed to the minimum (8us for Pi Camera v2)
@ -93,7 +93,7 @@ def set_minimum_exposure(camera: Picamera2) -> None:
class ExposureTest(BaseModel):
"""Record the results of testing the camera's current exposure settings"""
"""Record the results of testing the camera's current exposure settings."""
level: int
exposure_time: int
@ -101,7 +101,7 @@ class ExposureTest(BaseModel):
def test_exposure_settings(camera: Picamera2, percentile: float) -> ExposureTest:
"""Evaluate current exposure settings using a raw image
"""Evaluate current exposure settings using a raw image.
CAMERA SHOULD BE STARTED!
@ -136,7 +136,7 @@ def test_exposure_settings(camera: Picamera2, percentile: float) -> ExposureTest
def check_convergence(test: ExposureTest, target: int, tolerance: float) -> bool:
"""Check whether the brightness is within the specified target range"""
"""Check whether the brightness is within the specified target range."""
return abs(test.level - target) < target * tolerance
@ -321,7 +321,7 @@ def channels_from_bayer_array(bayer_array: np.ndarray) -> np.ndarray:
def get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray:
"""Compresses channel down to a 16x12 grid - from libcamera
"""Compresses channel down to a 16x12 grid - from libcamera.
This is taken from
https://git.linuxtv.org/libcamera.git/tree/utils/raspberrypi/ctt/ctt_alsc.py
@ -343,7 +343,7 @@ def get_16x12_grid(chan: np.ndarray, dx: int, dy: int) -> np.ndarray:
def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray:
"""Zoom an image in the last two dimensions
"""Zoom an image in the last two dimensions.
This is effectively the inverse operation of ``get_16x12_grid``
"""
@ -354,7 +354,7 @@ def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray:
def downsampled_channels(channels: np.ndarray, blacklevel=64) -> list[np.ndarray]:
"""Generate a downsampled, un-normalised image from which to calculate the LST
"""Generate a downsampled, un-normalised image from which to calculate the LST.
TODO: blacklevel probably ought to be determined from the camera...
"""
@ -382,7 +382,7 @@ def lst_from_channels(channels: np.ndarray) -> LensShadingTables:
def lst_from_grids(grids: np.ndarray) -> LensShadingTables:
"""Given 4 downsampled grids, generate the luminance and chrominance tables
"""Given 4 downsampled grids, generate the luminance and chrominance tables.
The grids are the 4 BAYER channels RGGB
@ -408,7 +408,7 @@ def lst_from_grids(grids: np.ndarray) -> LensShadingTables:
def grids_from_lst(lum: np.ndarray, Cr: np.ndarray, Cb: np.ndarray) -> np.ndarray:
"""Convert form luminance/chrominance dict to four RGGB channels
"""Convert form luminance/chrominance dict to four RGGB channels.
Note that these will be normalised - the maximum green value is always 1.
Also, note that the channels are BGGR, to be consistent with the
@ -462,13 +462,13 @@ def set_static_ccm(
def get_static_ccm(tuning: dict) -> None:
"""Get the ``rpi.ccm`` section of a camera tuning dict"""
"""Get the ``rpi.ccm`` section of a camera tuning dict."""
ccm = Picamera2.find_tuning_algo(tuning, "rpi.ccm")
return ccm["ccms"]
def lst_is_static(tuning: dict) -> bool:
"""Whether the lens shading table is set to static"""
"""Whether the lens shading table is set to static."""
alsc = Picamera2.find_tuning_algo(tuning, "rpi.alsc")
return alsc["n_iter"] == 0
@ -493,13 +493,13 @@ def set_static_geq(
def _geq_is_static(tuning: dict) -> bool:
"""Whether the green equalisation is set to static"""
"""Whether the green equalisation is set to static."""
geq = Picamera2.find_tuning_algo(tuning, "rpi.geq")
return geq["offset"] == 65535
def index_of_algorithm(algorithms: list[dict], algorithm: str) -> int:
"""Find the index of an algorithm's section in the tuning file"""
"""Find the index of an algorithm's section in the tuning file."""
for i, a in enumerate(algorithms):
if algorithm in a:
return i
@ -557,5 +557,5 @@ def recreate_camera_manager() -> None:
def as_flat_rounded_list(array: np.ndarray, round_to: int = 3) -> list[float]:
"""Flatten array, round, and then convert to list"""
"""Flatten array, round, and then convert to list."""
return np.reshape(array, -1).round(round_to).tolist()

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

@ -1,4 +1,4 @@
"""OpenFlexure Microscope OpenCV Camera
"""OpenFlexure Microscope OpenCV Camera.
This module defines a Thing that is responsible for using the stage and
camera together to perform an autofocus routine.
@ -30,7 +30,7 @@ RATIO = 0.2
class SimulatedCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
"""A Thing representing an OpenCV camera."""
_stage: Optional[BaseStage] = None
_server: Optional[lt.ThingServer] = None
@ -53,7 +53,7 @@ class SimulatedCamera(BaseCamera):
self.generate_canvas()
def generate_sprites(self):
"""Generate sprites to populate the image"""
"""Generate sprites to populate the image."""
self.sprites = []
black = np.zeros(self.glyph_shape, dtype=np.uint8)
x = np.arange(black.shape[0])
@ -65,7 +65,7 @@ class SimulatedCamera(BaseCamera):
self.sprites.append(sprite)
def generate_blobs(self, n_blobs: int = 1000):
"""Generate coordinates of blobs
"""Generate coordinates of blobs.
Blobs are characterised by X, Y, sprite
We also generate a KD tree to rapidly find blobs in an image
@ -78,7 +78,7 @@ class SimulatedCamera(BaseCamera):
self.blobs[:, 2] = rng.choice(len(self.sprites), n_blobs)
def generate_canvas(self):
"""Generate a blank canvas"""
"""Generate a blank canvas."""
self.canvas = np.zeros(self.canvas_shape, dtype=np.uint8)
self.canvas[...] = 255
w, h, _ = self.glyph_shape
@ -89,7 +89,7 @@ class SimulatedCamera(BaseCamera):
] -= self.sprites[int(sprite)]
def generate_image(self, pos: tuple[int, int, int]):
"""Generate an image with blobs based on supplied coordinates"""
"""Generate an image with blobs based on supplied coordinates."""
canvas_width, canvas_height, _ = self.canvas_shape
image_width, image_height, _ = self.shape
pos = tuple(x * RATIO for x in pos)
@ -124,7 +124,7 @@ class SimulatedCamera(BaseCamera):
return self._stage.instantaneous_position
def generate_frame(self):
"""Generate a frame with blobs based on the stage coordinates"""
"""Generate a frame with blobs based on the stage coordinates."""
try:
pos = self.get_stage_position()
except Exception as e:
@ -145,7 +145,7 @@ class SimulatedCamera(BaseCamera):
@lt.thing_property
def stream_active(self) -> bool:
"""Whether the MJPEG stream is active"""
"""Whether the MJPEG stream is active."""
if self._capture_enabled and self._capture_thread:
return self._capture_thread.is_alive()
return False
@ -170,7 +170,7 @@ class SimulatedCamera(BaseCamera):
self,
resolution: Literal["main", "full"] = "full",
) -> ArrayModel:
"""Acquire one image from the camera and return as an array
"""Acquire one image from the camera and return as an array.
This function will produce a nested list containing an uncompressed RGB image.
It's likely to be highly inefficient - raw and/or uncompressed captures using
@ -185,7 +185,7 @@ class SimulatedCamera(BaseCamera):
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob
"""Acquire one image from the camera and return as a JPEG blob.
This function will produce a JPEG image.
"""