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

Merge branch 'new-thing-setting-syntax' into 'v3'

Browse source 
Update Thing Settings and import statments for LabThings-FastAPI 0.0.10

Closes #302 and #362

See merge request openflexure/openflexure-microscope-server!296
This commit is contained in:
Julian Stirling 2025-07-07 18:29:01 +00:00
commit ea785cb844
21 changed files with 568 additions and 699 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -30,7 +30,7 @@ var/
.installed.cfg
*.egg
pip-wheel-metadata/
/openflexure/
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.

2
pyproject.toml
View file

@ -15,7 +15,7 @@ classifiers = [
"Operating System :: OS Independent",
]
dependencies = [
"labthings-fastapi[server] == 0.0.9",
"labthings-fastapi == 0.0.10",
"sangaboard",
"camera-stage-mapping ~= 0.1.10",
"opencv-python ~= 4.11.0",

14
src/openflexure_microscope_server/server/__init__.py
View file

@ -3,14 +3,16 @@ from __future__ import annotations
from typing import Optional
from copy import copy
from labthings_fastapi.server import cli, ThingServer
import labthings_fastapi as lt
import uvicorn
from .serve_static_files import add_static_files
from .legacy_api import add_v2_endpoints
from ..logging import configure_logging, retrieve_log, retrieve_log_from_file
def customise_server(server: ThingServer, log_folder: str, scans_folder: Optional[str]):
def customise_server(
server: lt.ThingServer, log_folder: str, scans_folder: Optional[str]
):
"""Customise the server with additional endpoints, etc."""
configure_logging(log_folder)
add_v2_endpoints(server)
@ -39,7 +41,7 @@ def _get_scans_dir(config: dict) -> Optional[str]:
def serve_from_cli(argv: Optional[list[str]] = None):
"""Start the server from the command line"""
args = cli.parse_args(argv)
args = lt.cli.parse_args(argv)
log_config = copy(uvicorn.config.LOGGING_CONFIG)
log_config["loggers"]["uvicorn"]["propagate"] = True
@ -50,10 +52,10 @@ def serve_from_cli(argv: Optional[list[str]] = None):
config = None
server = None
try:
config = cli.config_from_args(args)
config = lt.cli.config_from_args(args)
log_folder = config.get("log_folder", "./openflexure/logs")
scans_folder = _get_scans_dir(config)
server = cli.server_from_config(config)
server = lt.cli.server_from_config(config)
customise_server(server, log_folder, scans_folder)
uvicorn.run(
server.app,
@ -68,7 +70,7 @@ def serve_from_cli(argv: Optional[list[str]] = None):
print(f"Error: {e}")
fallback_server = "labthings_fastapi.server.fallback:app"
print(f"Starting fallback server {fallback_server}.")
app = cli.object_reference_to_object(fallback_server)
app = lt.cli.object_reference_to_object(fallback_server)
app.labthings_config = config
app.labthings_server = server
app.labthings_error = e

4
src/openflexure_microscope_server/server/legacy_api.py
View file

@ -1,9 +1,9 @@
from . import ThingServer
import labthings_fastapi as lt
from fastapi import Response
from socket import gethostname
def add_v2_endpoints(thing_server: ThingServer):
def add_v2_endpoints(thing_server: lt.ThingServer):
app = thing_server.app
# TODO: update openflexure connect to make this unnecessary!!

15
src/openflexure_microscope_server/things/auto_recentre_stage.py
View file

@ -1,19 +1,20 @@
import numpy as np
import logging
from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
from labthings_fastapi.decorators import thing_action
import labthings_fastapi as lt
from .stage import StageDependency as StageDep
from openflexure_microscope_server.things.autofocus import AutofocusThing
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
CSMDep = lt.deps.direct_thing_client_dependency(
CameraStageMapper, "/camera_stage_mapping/"
)
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
class RecentringThing(Thing):
@thing_action
class RecentringThing(lt.Thing):
@lt.thing_action
def recentre(
self,
autofocus: AutofocusDep,

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

@ -17,9 +17,7 @@ from fastapi import Depends
import numpy as np
from pydantic import BaseModel
from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from labthings_fastapi.decorators import thing_action, thing_property
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from .camera import RawCameraDependency as Camera
@ -189,7 +187,7 @@ class SharpnessDataArrays(BaseModel):
class JPEGSharpnessMonitor:
def __init__(self, stage: Stage, camera: Camera, portal: BlockingPortal):
def __init__(self, stage: Stage, camera: Camera, portal: lt.deps.BlockingPortal):
self.camera = camera
self.stage = stage
self.portal = portal
@ -287,14 +285,14 @@ class JPEGSharpnessMonitor:
SharpnessMonitorDep = Annotated[JPEGSharpnessMonitor, Depends()]
class AutofocusThing(Thing):
class AutofocusThing(lt.Thing):
"""The Thing concerned with combinations of z axis movements and the camera.
Actions here involve moving a stage in z, and using the camera to either
capture images (generally, z-stacking) and measuring the sharpness of the
field of view to assess focus (autofocus and testing the success of a z-stack)"""
@thing_action
@lt.thing_action
def fast_autofocus(
self,
sharpness_monitor: SharpnessMonitorDep,
@ -324,7 +322,7 @@ class AutofocusThing(Thing):
# Return all focus data
return sharpness_monitor.data_dict()
@thing_action
@lt.thing_action
def z_move_and_measure_sharpness(
self,
sharpness_monitor: SharpnessMonitorDep,
@ -350,7 +348,7 @@ class AutofocusThing(Thing):
sharpness_monitor.focus_rel(current_dz)
return sharpness_monitor.data_dict()
@thing_action
@lt.thing_action
def looping_autofocus(
self,
stage: Stage,
@ -396,39 +394,38 @@ class AutofocusThing(Thing):
stage.move_absolute(z=peak_height)
return heights.tolist(), sizes.tolist()
@thing_property
def stack_images_to_save(self) -> int:
"""The number of images to capture and save in a stack
Defaults to 1 unless you need to see either side of focus"""
return self.thing_settings.get("stack_images_to_save", 1)
stack_images_to_save = lt.ThingSetting(
initial_value=1,
model=int,
description="""The number of images to save in a stack.
@stack_images_to_save.setter
def stack_images_to_save(self, value: int) -> None:
self.thing_settings["stack_images_to_save"] = value
Defaults to 1 unless you need to see either side of focus""",
)
@thing_property
def stack_min_images_to_test(self) -> int:
"""The number of images to test for successful focusing in a stack
Defaults to 9, which balances reliability and speed"""
return self.thing_settings.get("stack_min_images_to_test", 9)
stack_min_images_to_test = lt.ThingSetting(
initial_value=9,
model=int,
description="""The minimum number of images to capture in a stack.
@stack_min_images_to_test.setter
def stack_min_images_to_test(self, value: int) -> None:
self.thing_settings["stack_min_images_to_test"] = value
This many images are captures and tested for focus, if the focus
is not central enough more images may be captured. After new images
are captured the number sets the number of images used for checking
if focus is central.
@thing_property
def stack_dz(self) -> int:
"""Space in steps between images in a z-stack
Suggested is 50 for 60-100x
100 for 40x
200 for 20x"""
return self.thing_settings.get("stack_dz", 50)
Defaults to 9 which balances reliability and speed
""",
)
@stack_dz.setter
def stack_dz(self, value: int) -> None:
self.thing_settings["stack_dz"] = value
stack_dz = lt.ThingSetting(
initial_value=50,
model=int,
description="""Space in steps between images in a z-stack
Suggested is 50 for 60-100x
100 for 40x
200 for 20x""",
)
@thing_action
@lt.thing_action
def run_smart_stack(
self,
cam: WrappedCamera,

69
src/openflexure_microscope_server/things/background_detect.py
View file

@ -5,8 +5,7 @@ from PIL import Image
from pydantic import BaseModel
from scipy.stats import norm
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
import labthings_fastapi as lt
from .camera import CameraDependency as CamDep
@ -16,39 +15,45 @@ class ChannelDistributions(BaseModel):
colorspace: str = "LUV"
class BackgroundDetectThing(Thing):
@thing_property
class BackgroundDetectThing(lt.Thing):
# Requires a getter and a setter to support being a BaseModel but being
# saved to file as a dict
_background_distributions: Optional[ChannelDistributions] = None
@lt.thing_setting
def background_distributions(self) -> Optional[ChannelDistributions]:
"""The statistics of the background image"""
bd = self.thing_settings.get("background_distributions", None)
if bd:
return ChannelDistributions(**bd)
return None
bd = self._background_distributions
if bd is None:
return None
return ChannelDistributions(**bd)
@background_distributions.setter
def background_distributions(self, value: Optional[ChannelDistributions]) -> None:
try:
self.thing_settings["background_distributions"] = value.model_dump()
except AttributeError:
self.thing_settings["background_distributions"] = None
def background_distributions(
self, value: Optional[ChannelDistributions | dict]
) -> None:
if value is None:
self._background_distributions = None
elif isinstance(value, ChannelDistributions):
self._background_distributions = value.model_dump()
elif isinstance(value, dict):
self._background_distributions = value
else:
raise TypeError(
f"Cannot set background_distributions with an object of type {type(value)}"
)
@thing_property
def tolerance(self) -> float:
"""How many standard deviations to allow for the background"""
return self.thing_settings.get("tolerance", 7)
tolerance = lt.ThingSetting(
initial_value=7.0,
model=float,
description="How many standard deviations to allow for the background",
)
@tolerance.setter
def tolerance(self, value: float) -> None:
self.thing_settings["tolerance"] = value
@thing_property
def fraction(self) -> float:
"""How much of the image needs to be not background to label as sample"""
return self.thing_settings.get("fraction", 25)
@fraction.setter
def fraction(self, value: float) -> None:
self.thing_settings["fraction"] = value
fraction = lt.ThingSetting(
initial_value=25.0,
model=float,
description="How much of the image needs to be not background to label as sample",
)
def background_mask(self, image: np.ndarray) -> np.ndarray:
"""Calculate a binary image, showing whether each pixel is background
@ -71,7 +76,7 @@ class BackgroundDetectThing(Thing):
axis=2,
)
@thing_action
@lt.thing_action
def background_fraction(self, cam: CamDep) -> float:
"""Determine what fraction of the current image is background
@ -89,7 +94,7 @@ class BackgroundDetectThing(Thing):
mask = self.background_mask(current_image_luv)
return np.count_nonzero(mask) / np.prod(mask.shape) * 100
@thing_action
@lt.thing_action
def image_is_sample(self, cam: CamDep) -> bool:
"""Label the current image as either background or sample"""
b_fraction = self.background_fraction(cam)
@ -97,7 +102,7 @@ class BackgroundDetectThing(Thing):
return (100 - b_fraction) > fraction_threshold
@thing_action
@lt.thing_action
def set_background(self, cam: CamDep):
"""Grab an image, and use its statistics to set the background

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

@ -1,6 +1,6 @@
"""OpenFlexure Microscope Camera
This module defines the interface for cameras. Any compatible Thing
This module defines the interface for cameras. Any compatible lt.Thing
should enabe the server to work.
See repository root for licensing information.
@ -14,23 +14,15 @@ from pydantic import RootModel
from PIL import Image
import piexif
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
from labthings_fastapi.dependencies.raw_thing import raw_thing_dependency
from labthings_fastapi.dependencies.invocation import InvocationLogger
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
from labthings_fastapi.outputs.blob import Blob
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
class JPEGBlob(Blob):
class JPEGBlob(lt.blob.Blob):
media_type: str = "image/jpeg"
class PNGBlob(Blob):
class PNGBlob(lt.blob.Blob):
"""A class representing a PNG image as a LabThings FastAPI Blob"""
media_type: str = "image/png"
@ -154,11 +146,11 @@ class CameraMemoryBuffer:
del self._storage[key]
class BaseCamera(Thing):
class BaseCamera(lt.Thing):
"""The base class for all cameras. All cameras must directly inherit from this class"""
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
def __enter__(self) -> None:
@ -167,7 +159,7 @@ class BaseCamera(Thing):
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
raise NotImplementedError("CameraThings must define their own __exit__ method")
@thing_action
@lt.thing_action
def start_streaming(
self, main_resolution: tuple[int, int], buffer_count: int
) -> None:
@ -177,14 +169,14 @@ class BaseCamera(Thing):
"CameraThings must define their own start_streaming method"
)
@thing_property
@lt.thing_property
def stream_active(self) -> bool:
"Whether the MJPEG stream is active"
raise NotImplementedError(
"CameraThings must define their own stream_active method"
)
@thing_action
@lt.thing_action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
@ -194,10 +186,10 @@ class BaseCamera(Thing):
"CameraThings must define their own capture_array method"
)
@thing_action
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: GetThingStates,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 5,
) -> JPEGBlob:
@ -206,10 +198,10 @@ class BaseCamera(Thing):
"CameraThings must define their own capture_jpeg method"
)
@thing_action
@lt.thing_action
def grab_jpeg(
self,
portal: BlockingPortal,
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> JPEGBlob:
"""Acquire one image from the preview stream and return as an array
@ -225,10 +217,10 @@ class BaseCamera(Thing):
frame = portal.call(stream.grab_frame)
return JPEGBlob.from_bytes(frame)
@thing_action
@lt.thing_action
def grab_jpeg_size(
self,
portal: BlockingPortal,
portal: lt.deps.BlockingPortal,
stream_name: Literal["main", "lores"] = "main",
) -> int:
"""Acquire one image from the preview stream and return its size"""
@ -237,7 +229,7 @@ class BaseCamera(Thing):
)
return portal.call(stream.next_frame_size)
@thing_action
@lt.thing_action
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"],
@ -248,12 +240,12 @@ class BaseCamera(Thing):
"CameraThings must define their own capture_image method"
)
@thing_action
@lt.thing_action
def capture_and_save(
self,
jpeg_path: str,
logger: InvocationLogger,
metadata_getter: GetThingStates,
logger: lt.deps.InvocationLogger,
metadata_getter: lt.deps.GetThingStates,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Capture an image and save it to disk
@ -279,11 +271,11 @@ class BaseCamera(Thing):
save_resolution,
)
@thing_action
@lt.thing_action
def capture_to_memory(
self,
logger: InvocationLogger,
metadata_getter: GetThingStates,
logger: lt.deps.InvocationLogger,
metadata_getter: lt.deps.GetThingStates,
buffer_max: int = 1,
) -> None:
"""
@ -304,11 +296,11 @@ class BaseCamera(Thing):
image, metadata = self._robust_image_capture(metadata_getter, logger)
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
@thing_action
@lt.thing_action
def save_from_memory(
self,
jpeg_path: str,
logger: InvocationLogger,
logger: lt.deps.InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None,
buffer_id: Optional[int] = None,
) -> None:
@ -333,15 +325,15 @@ class BaseCamera(Thing):
save_resolution=save_resolution,
)
@thing_action
@lt.thing_action
def clear_buffers(self) -> None:
"""Clear all images in memory"""
self._memory_buffer.clear()
def _robust_image_capture(
self,
metadata_getter: GetThingStates,
logger: InvocationLogger,
metadata_getter: lt.deps.GetThingStates,
logger: lt.deps.InvocationLogger,
) -> Image:
"""Capture an image in memory and return it with metadata
CaptureError raised if the capture fails for any reason
@ -366,7 +358,7 @@ class BaseCamera(Thing):
jpeg_path: str,
image: Image,
metadata: dict,
logger: InvocationLogger,
logger: lt.deps.InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None,
) -> None:
"""Saving the captured image and metadata to disk
@ -399,5 +391,5 @@ class BaseCamera(Thing):
raise IOError(f"An error occurred while saving {jpeg_path}") from e
CameraDependency = direct_thing_client_dependency(BaseCamera, "/camera/")
RawCameraDependency = raw_thing_dependency(BaseCamera)
CameraDependency = lt.deps.direct_thing_client_dependency(BaseCamera, "/camera/")
RawCameraDependency = lt.deps.raw_thing_dependency(BaseCamera)

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

@ -16,9 +16,7 @@ from threading import Thread
import cv2
import piexif
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray
from . import BaseCamera, JPEGBlob
@ -45,7 +43,7 @@ class OpenCVCamera(BaseCamera):
self._capture_thread.join()
self.cap.release()
@thing_property
@lt.thing_property
def stream_active(self) -> bool:
"Whether the MJPEG stream is active"
if self._capture_enabled and self._capture_thread:
@ -53,7 +51,7 @@ class OpenCVCamera(BaseCamera):
return False
def _capture_frames(self):
portal = get_blocking_portal(self)
portal = lt.get_blocking_portal(self)
while self._capture_enabled:
ret, frame = self.cap.read()
if not ret:
@ -68,7 +66,7 @@ class OpenCVCamera(BaseCamera):
].tobytes()
self.lores_mjpeg_stream.add_frame(jpeg_lores, portal)
@thing_action
@lt.thing_action
def capture_array(
self,
resolution: Literal["main", "full"] = "full",
@ -87,10 +85,10 @@ class OpenCVCamera(BaseCamera):
)
return frame
@thing_action
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: GetThingStates,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob

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

@ -16,59 +16,34 @@ https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf
"""
from __future__ import annotations
from typing import Annotated, Iterator, Literal, Mapping, Optional
from datetime import datetime
import json
import logging
import os
import tempfile
import time
from tempfile import TemporaryDirectory
from contextlib import contextmanager
from threading import RLock
from pydantic import BaseModel, BeforeValidator
from labthings_fastapi.descriptors.property import PropertyDescriptor
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStream
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
from typing import Annotated, Any, Iterator, Literal, Mapping, Optional
from contextlib import contextmanager
import piexif
from threading import RLock
import picamera2
import numpy as np
from picamera2 import Picamera2
from picamera2.encoders import MJPEGEncoder
from picamera2.outputs import Output
import numpy as np
import labthings_fastapi as lt
from labthings_fastapi.exceptions import NotConnectedToServerError
from . import picamera_recalibrate_utils as recalibrate_utils
from . import BaseCamera, JPEGBlob, ArrayModel
class PicameraControl(PropertyDescriptor):
def __init__(
self, control_name: str, model: type = float, description: Optional[str] = None
):
"""A property descriptor controlling a picamera control"""
PropertyDescriptor.__init__(
self, model, observable=False, description=description
)
self.control_name = control_name
def _getter(self, obj: StreamingPiCamera2):
with obj.picamera() as cam:
return cam.capture_metadata()[self.control_name]
def _setter(self, obj: StreamingPiCamera2, value: Any):
with obj.picamera() as cam:
cam.set_controls({self.control_name: value})
class PicameraStreamOutput(Output):
"""An Output class that sends frames to a stream"""
def __init__(self, stream: MJPEGStream, portal: BlockingPortal):
def __init__(self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal):
"""Create an output that puts frames in an MJPEGStream
We need to pass the stream object, and also the blocking portal, because
@ -134,98 +109,37 @@ class StreamingPiCamera2(BaseCamera):
"""A Thing that represents an OpenCV camera"""
def __init__(self, camera_num: int = 0):
self._setting_save_in_progress = False
self.camera_num = camera_num
self.camera_configs: dict[str, dict] = {}
# Note persistent controls will be updated with settings, in __enter__.
self.persistent_controls = {
"AeEnable": False,
"AnalogueGain": 1.0,
"AwbEnable": False,
"Brightness": 0,
"ColourGains": (1, 1),
"Contrast": 1,
"ExposureTime": 0,
"Saturation": 1,
"Sharpness": 1,
}
self.persistent_control_tolerances = {
"ExposureTime": 30,
}
def update_persistent_controls(self, discard_frames: int = 1):
"""Update the persistent controls dict from the camera
Query the camera and update the value of `persistent_controls` to
match the current state of the camera.
There is a work-around here, that will suppress small updates. There
appears to be a bug in the camera code that causes a slight drift in
`ExposureTime` each time the camera is reinitialised: this can
add up over time, particularly if the camera is reconfigured many
times. To get around this, we look in `self.persistent_control_tolerances`
and only update `self.persistent_controls` if the change is greater than
this tolerance.
"""
with self.picamera() as cam:
for i in range(discard_frames):
# Discard frames, so data is fresh
cam.capture_metadata()
for key, value in cam.capture_metadata().items():
if key not in self.persistent_controls:
# Skip keys that are not persistent controls
continue
if self._control_value_within_tolerance(key, value):
logging.debug(
"Ignoring a small change in persistent control %s from %s to "
"%s while updating persistent controls.",
key,
self.persistent_controls[key],
value,
)
else:
self.persistent_controls[key] = value
self.thing_settings.update(self.persistent_controls)
def _control_value_within_tolerance(self, key: str, value: float) -> bool:
"""Return True if the updated value for a persistent control is within
tolerance of current value. This allows ignoring small changes
"""
if key not in self.persistent_control_tolerances:
# Not tolerance range set, so it is not within tolerance
return False
difference = np.abs(self.persistent_controls[key] - value)
return difference < self.persistent_control_tolerances[key]
def settings_to_persistent_controls(self):
"""Update the persistent controls dict from the settings dict
NB this must be called **after** self.thing_settings is initialised,
i.e. during or after `__enter__`.
"""
self._picamera_lock = None
self._picamera = None
logging.info("Starting & reconfiguring camera to populate sensor_modes.")
with Picamera2(camera_num=self.camera_num) as cam:
self.default_tuning = recalibrate_utils.load_default_tuning(cam)
logging.info("Done reading sensor modes & default tuning.")
# Set tuning to default tuning. This will be overwritten when the Thing is
# connects to the server if tuning is saved to disk.
try:
pc = self.thing_settings["persistent_controls"]
except KeyError:
return # If there are no saved settings, use defaults
for k in self.persistent_controls:
try:
self.persistent_controls[k] = pc[k]
except KeyError:
pass # If controls are missing, leave at default
self.tuning = self.default_tuning
except NotConnectedToServerError:
# This will throw an error after setting as we are not connected to
# a server. But we know this, so we ignore the error.
pass
stream_resolution = PropertyDescriptor(
stream_resolution = lt.ThingProperty(
tuple[int, int],
initial_value=(820, 616),
description="Resolution to use for the MJPEG stream",
)
mjpeg_bitrate = PropertyDescriptor(
mjpeg_bitrate = lt.ThingProperty(
Optional[int],
initial_value=100000000,
description="Bitrate for MJPEG stream (None for default)",
)
stream_active = PropertyDescriptor(
stream_active = lt.ThingProperty(
bool,
initial_value=False,
description="Whether the MJPEG stream is active",
@ -233,144 +147,205 @@ class StreamingPiCamera2(BaseCamera):
readonly=True,
)
analogue_gain = PicameraControl("AnalogueGain", float)
colour_gains = PicameraControl("ColourGains", tuple[float, float])
def save_settings(self):
"""Override save_settings to ensure that camera properties don't recurse.
exposure_time = PicameraControl(
"ExposureTime", int, description="The exposure time in microseconds"
)
This method is run by any Thing when a ThingSetting is saved. However, the
method reads the thing_setting. As reading the thing setting talks to the
camera and calls save_settings if the value is not as expected, this could
cause recursion. Aslo this means that saving one setting causes all others
to be read each time.
"""
try:
self._setting_save_in_progress = True
super().save_settings()
finally:
self._setting_save_in_progress = False
## Persistent controls! These are settings
_analogue_gain: float = 1.0
@lt.thing_setting
def analogue_gain(self) -> float:
if not self._setting_save_in_progress and self.streaming:
with self._streaming_picamera() as cam:
cam_value = cam.capture_metadata()["AnalogueGain"]
if cam_value != self._analogue_gain:
self._analogue_gain = cam_value
self.save_settings()
return self._analogue_gain
@analogue_gain.setter
def analogue_gain(self, value: float):
self._analogue_gain = value
if self.streaming:
with self._streaming_picamera() as cam:
cam.set_controls({"AnalogueGain": value})
_colour_gains: tuple[float, float] = (1.0, 1.0)
@lt.thing_setting
def colour_gains(self) -> tuple[float, float]:
if not self._setting_save_in_progress and self.streaming:
with self._streaming_picamera() as cam:
cam_value = cam.capture_metadata()["ColourGains"]
if cam_value != self._colour_gains:
self._colour_gains = cam_value
self.save_settings()
return self._colour_gains
@colour_gains.setter
def colour_gains(self, value: tuple[float, float]):
self._colour_gains = value
if self.streaming:
with self._streaming_picamera() as cam:
cam.set_controls({"ColourGains": value})
_exposure_time: int = 0
@lt.thing_setting
def exposure_time(self) -> int:
if not self._setting_save_in_progress and self.streaming:
with self._streaming_picamera() as cam:
cam_value = cam.capture_metadata()["ExposureTime"]
if cam_value != self._exposure_time:
self._exposure_time = cam_value
self.save_settings()
return self._exposure_time
@exposure_time.setter
def exposure_time(self, value: int):
"""
Custom setter for the above exposure_time PicameraControl.
_exposure_time = value
if self.streaming:
with self._streaming_picamera() as cam:
# Note: This set a value 1 higher than requested as picamera2 always
# sets a lower value than requested, even if the requested is allowed
cam.set_controls({"ExposureTime": value + 1})
This is overriding the standard setter for a PicameraControl, as
the value needs to be adjusted before setting to behave as expected.
See comment within the function for more detail.
"""
with self.picamera() as cam:
# Note: This set a value 1 higher than requested as picamera2 always sets
# a lower value than requested, even if the requested is allowed
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()
return {
"AeEnable": False,
"AnalogueGain": self.analogue_gain,
"AwbEnable": False,
"Brightness": 0,
"ColourGains": self.colour_gains,
"Contrast": 1,
"ExposureTime": self.exposure_time,
"Saturation": 1,
"Sharpness": 1,
}
_sensor_modes = None
@thing_property
@lt.thing_property
def sensor_modes(self) -> list[SensorMode]:
"""All the available modes the current sensor supports"""
if not self._sensor_modes:
with self.picamera() as cam:
with self._streaming_picamera() as cam:
self._sensor_modes = cam.sensor_modes
return self._sensor_modes
@thing_property
_sensor_mode: Optional[dict] = None
@lt.thing_property
def sensor_mode(self) -> Optional[SensorModeSelector]:
"""The intended sensor mode of the camera"""
return self.thing_settings["sensor_mode"]
if self._sensor_mode is None:
return None
return SensorModeSelector(**self._sensor_mode)
@sensor_mode.setter
def sensor_mode(self, new_mode: Optional[SensorModeSelector]):
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]):
"""Change the sensor mode used"""
if isinstance(new_mode, SensorModeSelector):
new_mode = new_mode.model_dump()
with self.picamera(pause_stream=True):
self.thing_settings["sensor_mode"] = new_mode
@thing_property
def sensor_resolution(self) -> tuple[int, int]:
if new_mode is None:
self._sensor_mode = None
elif isinstance(new_mode, SensorModeSelector):
self._sensor_mode = new_mode.model_dump()
elif isinstance(new_mode, dict):
self._sensor_mode = new_mode
# By pausing the stream on when accessing, streaming_picamera
# self._sensor_mode will be read and set when the stream restarts
# after the context manager closes.
with self._streaming_picamera(pause_stream=True):
pass
@lt.thing_property
def sensor_resolution(self) -> Optional[tuple[int, int]]:
"""The native resolution of the camera's sensor"""
with self.picamera() as cam:
with self._streaming_picamera() as cam:
return cam.sensor_resolution
tuning = PropertyDescriptor(Optional[dict], None, readonly=True)
tuning = lt.ThingSetting(Optional[dict], None, readonly=True)
def settings_to_properties(self):
"""Set the values of properties based on the settings dict"""
try:
props = self.thing_settings["properties"]
except KeyError:
return
for k, v in props.items():
setattr(self, k, v)
def _initialise_picamera(self):
"""Acquire the picamera device and store it as `self._picamera`.
def properties_to_settings(self):
"""Save certain properties to the settings dictionary"""
props = {}
for k in ["mjpeg_bitrate", "stream_resolution"]:
props[k] = getattr(self, k)
self.thing_settings["properties"] = props
def initialise_tuning(self):
"""Read the tuning from the settings, or load default tuning
NB this relies on `self.thing_settings` and `self.default_tuning`
so will fail if it's run before those are populated in `__enter__`.
This duplicates logic in `Picamera2.__init__` to provide a tuning file that
will be read when the camera system initialises.
"""
if "tuning" in self.thing_settings:
self.tuning = self.thing_settings["tuning"].dict
else:
logging.info("Did not find tuning in settings, reading from camera...")
self.tuning = self.default_tuning
def initialise_picamera(self):
"""Acquire the picamera device and store it as `self._picamera`"""
if hasattr(self, "_picamera_lock"):
if self._picamera_lock is not None:
# Don't close the camera if it's in use
self._picamera_lock.acquire()
with tempfile.NamedTemporaryFile("w") as tuning_file:
# This duplicates logic in `Picamera2.__init__` to provide a tuning file
# that will be read when the camera system initialises.
# This is a necessary work-around until `picamera2` better supports
# reinitialisation of the camera with new tuning.
json.dump(self.tuning, tuning_file)
tuning_file.flush() # but leave it open as closing it will delete it
os.environ["LIBCAMERA_RPI_TUNING_FILE"] = tuning_file.name
# NB even though we've put the tuning file in the environment, we will
# need to specify the filename in the `Picamera2` initialiser as otherwise
# it will be overwritten with None.
if hasattr(self, "_picamera") and self._picamera:
print("Closing picamera object for reinitialisation")
if self._picamera is not None:
logging.info("Closing picamera object for reinitialisation")
logging.info(
"Camera object already exists, closing for reinitialisation"
)
self._picamera.close()
print("closed, deleting picamera")
logging.info("Picamera closed, deleting picamera")
del self._picamera
recalibrate_utils.recreate_camera_manager()
print("[re]creating Picamera2 object")
self._picamera = picamera2.Picamera2(
logging.info("Creating new Picamera2 object")
# Specify tuning file otherwise it will be overwritten with None.
self._picamera = Picamera2(
camera_num=self.camera_num,
tuning=self.tuning,
)
self._picamera_lock = RLock()
def __enter__(self):
self.populate_default_tuning()
self.initialise_tuning()
self.initialise_picamera()
self._initialise_picamera()
# populate sensor modes by reading the property
self.sensor_modes
self.settings_to_persistent_controls()
self.settings_to_properties()
self.start_streaming()
return self
@contextmanager
def picamera(self, pause_stream=False) -> Iterator[Picamera2]:
"""Return the underlying `Picamera2` instance, optionally pausing the stream.
@property
def streaming(self) -> bool:
"""True if the camera is streaming"""
return self._picamera is not None and self._picamera.started
If pause_stream is True (default is False), we will stop the MJPEG stream
before yielding control of the camera, and restart afterwards. If you make
changes to the camera settings, these may be ignored when the stream is
restarted: you may nened to call `update_persistent_controls()` to ensure
your changes persist after the stream restarts.
@contextmanager
def _streaming_picamera(self, pause_stream=False) -> Iterator[Picamera2]:
"""Lock access to picamera and return the underlying `Picamera2` instance.
Optionally the stream can be paused to allow updating the camera settings.
:param pause_stream: If False the `Picamera2` instance is simply yielded.
If True:
* Stop the MJPEG Stream
* Yield the `Picamera2` instance for function calling the context manager to
make changes.
* On closing of the context manager the stream will restart.
"""
already_streaming = self.stream_active
with self._picamera_lock:
if pause_stream and already_streaming:
self.update_persistent_controls()
self.stop_streaming(stop_web_stream=False)
try:
yield self._picamera
@ -378,36 +353,19 @@ class StreamingPiCamera2(BaseCamera):
if pause_stream and already_streaming:
self.start_streaming()
def populate_default_tuning(self):
"""Sensor modes are enumerated and stored, once, on start-up (`__enter__`).
This opens and closes the camera - must be run before the camera is
initialised.
"""
logging.info("Starting & reconfiguring camera to populate sensor_modes.")
with Picamera2(camera_num=self.camera_num) as cam:
self.default_tuning = recalibrate_utils.load_default_tuning(cam)
logging.info("Done reading sensor modes & default tuning.")
def __exit__(self, exc_type, exc_value, traceback):
# Allow key controls to persist across restarts
self.update_persistent_controls()
self.thing_settings["persistent_controls"] = self.persistent_controls
self.thing_settings["tuning"] = self.tuning
self.properties_to_settings()
self.thing_settings.write_to_file()
# Shut down the camera
self.stop_streaming()
with self.picamera() as cam:
with self._streaming_picamera() as cam:
cam.close()
del self._picamera
@thing_action
@lt.thing_action
def start_streaming(
self, main_resolution: tuple[int, int] = (820, 616), buffer_count: int = 6
) -> None:
"""
Start the MJPEG stream
Start the MJPEG stream. This is where persistent controls are sent to camera.
Sets the camera resolutions based on input parameters, and sets the low-res
resolution to (320, 240). Note: (320, 240) is a standard from the Pi Camera
@ -424,7 +382,7 @@ class StreamingPiCamera2(BaseCamera):
buffer_count: the number of frames to hold in the buffer. Higher uses more memory,
lower may cause dropped frames. Value must be between 1 and 8, Defaults to 6.
"""
controls = self._get_persistent_controls()
# Buffer count can't be negative, zero, or too high.
if buffer_count < 1 or buffer_count > 8:
# 8 is slightly arbitrary. 6 is the PiCamera default for video
@ -434,7 +392,7 @@ class StreamingPiCamera2(BaseCamera):
f"Can't set a buffer count of {buffer_count}. "
"Buffer count must be an integer from 1-8"
)
with self.picamera() as picam:
with self._streaming_picamera() as picam:
try:
if picam.started:
picam.stop()
@ -442,8 +400,8 @@ class StreamingPiCamera2(BaseCamera):
stream_config = picam.create_video_configuration(
main={"size": main_resolution},
lores={"size": (320, 240), "format": "YUV420"},
sensor=self.thing_settings.get("sensor_mode", None),
controls=self.persistent_controls,
sensor=self._sensor_mode,
controls=controls,
)
stream_config["buffer_count"] = buffer_count
picam.configure(stream_config)
@ -453,7 +411,7 @@ class StreamingPiCamera2(BaseCamera):
MJPEGEncoder(self.mjpeg_bitrate),
PicameraStreamOutput(
self.mjpeg_stream,
get_blocking_portal(self),
lt.get_blocking_portal(self),
),
name=stream_name,
)
@ -461,7 +419,7 @@ class StreamingPiCamera2(BaseCamera):
MJPEGEncoder(100000000),
PicameraStreamOutput(
self.lores_mjpeg_stream,
get_blocking_portal(self),
lt.get_blocking_portal(self),
),
name="lores",
)
@ -474,12 +432,12 @@ class StreamingPiCamera2(BaseCamera):
"Started MJPEG stream at %s on port %s", self.stream_resolution, 1
)
@thing_action
@lt.thing_action
def stop_streaming(self, stop_web_stream: bool = True) -> None:
"""
Stop the MJPEG stream
"""
with self.picamera() as picam:
with self._streaming_picamera() as picam:
try:
picam.stop_recording() # This should also stop the extra lores encoder
except Exception as e:
@ -495,7 +453,7 @@ class StreamingPiCamera2(BaseCamera):
# Adding a sleep to prevent camera getting confused by rapid commands
time.sleep(0.2)
@thing_action
@lt.thing_action
def capture_image(
self,
stream_name: Literal["main", "lores", "raw"] = "main",
@ -510,10 +468,10 @@ class StreamingPiCamera2(BaseCamera):
A TimeoutError is raised if this time is exceeded during capture.
Default = 0.9s, lower than the 1s timeout default in picamera yaml settings
"""
with self.picamera() as cam:
with self._streaming_picamera() as cam:
return cam.capture_image(stream_name, wait=wait)
@thing_action
@lt.thing_action
def capture_array(
self,
stream_name: Literal["main", "lores", "raw", "full"] = "main",
@ -534,13 +492,13 @@ class StreamingPiCamera2(BaseCamera):
# This was slower than capture_image for our use case, but directly returning
# an image as an array is still a useful feature
if stream_name == "full":
with self.picamera(pause_stream=True) as picam2:
with self._streaming_picamera(pause_stream=True) as picam2:
capture_config = picam2.create_still_configuration()
return picam2.switch_mode_and_capture_array(capture_config, wait=wait)
with self.picamera() as cam:
with self._streaming_picamera() as cam:
return cam.capture_array(stream_name, wait=wait)
@thing_property
@lt.thing_property
def camera_configuration(self) -> Mapping:
"""The "configuration" dictionary of the picamera2 object
@ -552,13 +510,13 @@ class StreamingPiCamera2(BaseCamera):
this property refers to whatever configuration is currently in force -
usually the one used for the preview stream.
"""
with self.picamera() as cam:
with self._streaming_picamera() as cam:
return cam.camera_configuration()
@thing_action
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: GetThingStates,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["lores", "main", "full"] = "main",
wait: Optional[float] = 0.9,
) -> JPEGBlob:
@ -582,20 +540,20 @@ class StreamingPiCamera2(BaseCamera):
bypass this, you must use a raw capture.
"""
fname = datetime.now().strftime("%Y-%m-%d-%H%M%S.jpeg")
folder = TemporaryDirectory()
folder = tempfile.TemporaryDirectory()
path = os.path.join(folder.name, fname)
config = self.camera_configuration
# Low-res and main streams are running already - so we don't need
# to reconfigure for these
if resolution in ("lores", "main") and config[resolution]:
with self.picamera() as cam:
with self._streaming_picamera() as cam:
cam.capture_file(path, name=resolution, format="jpeg", wait=wait)
else:
if resolution != "full":
logging.warning(
f"There was no {resolution} stream, capturing full resolution"
)
with self.picamera(pause_stream=True) as cam:
with self._streaming_picamera(pause_stream=True) as cam:
logging.info("Reconfiguring camera for full resolution capture")
cam.configure(cam.create_still_configuration())
cam.start()
@ -611,13 +569,13 @@ class StreamingPiCamera2(BaseCamera):
piexif.insert(piexif.dump(exif_dict), path)
return JPEGBlob.from_temporary_directory(folder, fname)
@thing_property
@lt.thing_property
def capture_metadata(self) -> dict:
"""Return the metadata from the camera"""
with self.picamera() as cam:
with self._streaming_picamera() as cam:
return cam.capture_metadata()
@thing_action
@lt.thing_action
def auto_expose_from_minimum(
self,
target_white_level: int = 700,
@ -635,15 +593,14 @@ class StreamingPiCamera2(BaseCamera):
calculating the brightest pixel, a percentile is used rather than the
maximum in order to be robust to a small number of noisy/bright pixels.
"""
with self.picamera(pause_stream=True) as cam:
with self._streaming_picamera(pause_stream=True) as cam:
recalibrate_utils.adjust_shutter_and_gain_from_raw(
cam,
target_white_level=target_white_level,
percentile=percentile,
)
self.update_persistent_controls()
@thing_action
@lt.thing_action
def calibrate_white_balance(
self,
method: Literal["percentile", "centre"] = "centre",
@ -660,7 +617,7 @@ class StreamingPiCamera2(BaseCamera):
If `method` is `"centre"`, we will correct the mean of the central 10%
of the image.
"""
with self.picamera(pause_stream=True) as cam:
with self._streaming_picamera(pause_stream=True) as cam:
if self.lens_shading_is_static:
lst: LensShading = self.lens_shading_tables
recalibrate_utils.adjust_white_balance_from_raw(
@ -676,9 +633,8 @@ class StreamingPiCamera2(BaseCamera):
recalibrate_utils.adjust_white_balance_from_raw(
cam, percentile=99, method=method
)
self.update_persistent_controls()
@thing_action
@lt.thing_action
def calibrate_lens_shading(self) -> None:
"""Take an image and use it for flat-field correction.
@ -687,31 +643,40 @@ class StreamingPiCamera2(BaseCamera):
one. This uses the camera's "tuning" file to correct the preview and
the processed images. It should not affect raw images.
"""
with self.picamera(pause_stream=True) as cam:
with self._streaming_picamera(pause_stream=True) as cam:
# Suppress lint warning that L, Cr, and Cb are not lowercase, as these are
# the standard mathematical terms for:
# luminance (L), red-difference chroma (Cr), and blue-difference chroma
# (Cb).
L, Cr, Cb = recalibrate_utils.lst_from_camera(cam) # noqa: N806
recalibrate_utils.set_static_lst(self.tuning, L, Cr, Cb)
self.initialise_picamera()
self._initialise_picamera()
@thing_property
@lt.thing_property
def colour_correction_matrix(
self,
) -> tuple[float, float, float, float, float, float, float, float, float]:
"""An alias for `colour_correction_matrix` to fit the micromanager API"""
return self.thing_settings.get(
"colour_correction_matrix",
tuple(recalibrate_utils.get_static_ccm(self.tuning)[0]["ccm"]),
)
"""The `colour_correction_matrix` from the tuning file.
This is broken out into its own property for convenience and compatibility with
the micromanager API
Ir is a 9 value tuple used to specify the 3x3 matrix that the GPU pipeline uses
to convert from the camera R,G,B vector to the standard R,G,B.
See page Raspberry Pi Camera Algorithm and Tuning Guide, page 45.
"""
return tuple(recalibrate_utils.get_static_ccm(self.tuning)[0]["ccm"])
@colour_correction_matrix.setter # type: ignore
def colour_correction_matrix(self, value) -> None:
self.thing_settings["colour_correction_matrix"] = value
self.calibrate_colour_correction(value)
recalibrate_utils.set_static_ccm(self.tuning, value)
@thing_action
if self._picamera is not None:
with self._streaming_picamera(pause_stream=True):
self._initialise_picamera()
@lt.thing_action
def reset_ccm(self):
"""
Overwrite the colour correction matrix in camera tuning with default values.
@ -719,7 +684,7 @@ class StreamingPiCamera2(BaseCamera):
These values are from the Raspberry Pi Camera Algorithm and Tuning Guide, page
45.
"""
# This is flattened 3x3 matrix. See `calibrate_colour_correction`
# This is flattened 3x3 matrix. See `colour_correction_matrix`
col_corr_matrix = [
1.80439,
-0.73699,
@ -733,26 +698,7 @@ class StreamingPiCamera2(BaseCamera):
]
self.colour_correction_matrix = col_corr_matrix
@thing_action
def calibrate_colour_correction(
self,
col_corr_matrix: tuple[
float, float, float, float, float, float, float, float, float
],
) -> None:
"""Overwrite the colour correction matrix in camera tuning
col_corr_matrix: This is a 9 value tuple used to specify the 3x3
matrix that the GPU pipeline uses to convert from the camera R,G,B vector
to the standard R,G,B.
See page Raspberry Pi Camera Algorithm and Tuning Guide, page 45.
"""
with self.picamera(pause_stream=True):
recalibrate_utils.set_static_ccm(self.tuning, col_corr_matrix)
self.initialise_picamera()
@thing_action
@lt.thing_action
def set_static_green_equalisation(self, offset: int = 65535) -> None:
"""Set the green equalisation to a static value.
@ -763,11 +709,11 @@ class StreamingPiCamera2(BaseCamera):
A value of 0 here does nothing, a value of 65535 is maximum correction.
"""
with self.picamera(pause_stream=True):
with self._streaming_picamera(pause_stream=True):
recalibrate_utils.set_static_geq(self.tuning, offset)
self.initialise_picamera()
self._initialise_picamera()
@thing_action
@lt.thing_action
def full_auto_calibrate(self) -> None:
"""Perform a full auto-calibration
@ -785,7 +731,7 @@ class StreamingPiCamera2(BaseCamera):
self.calibrate_lens_shading()
self.calibrate_white_balance()
@thing_action
@lt.thing_action
def flat_lens_shading(self) -> None:
"""Disable flat-field correction
@ -796,15 +742,15 @@ class StreamingPiCamera2(BaseCamera):
This flat table is used to take an image wth no lens shading so that the
correct lens shading table can be calibrated.
"""
with self.picamera(pause_stream=True):
with self._streaming_picamera(pause_stream=True):
# Generate and array of ones of the correct size for each channel
flat_array = np.ones((12, 16))
recalibrate_utils.set_static_lst(
self.tuning, flat_array, flat_array, flat_array
)
self.initialise_picamera()
self._initialise_picamera()
@thing_property
@lt.thing_property
def lens_shading_tables(self) -> Optional[LensShading]:
"""The current lens shading (i.e. flat-field correction)
@ -841,14 +787,14 @@ class StreamingPiCamera2(BaseCamera):
@lens_shading_tables.setter
def lens_shading_tables(self, lst: LensShading) -> None:
"""Set the lens shading tables"""
with self.picamera(pause_stream=True):
with self._streaming_picamera(pause_stream=True):
recalibrate_utils.set_static_lst(
self.tuning,
luminance=lst.luminance,
cr=lst.Cr,
cb=lst.Cb,
)
self.initialise_picamera()
self._initialise_picamera()
def correct_colour_gains_for_lens_shading(
self, colour_gains: tuple[float, float]
@ -881,7 +827,7 @@ class StreamingPiCamera2(BaseCamera):
float(gain_b / np.max(lst.Cb)),
)
@thing_action
@lt.thing_action
def flat_lens_shading_chrominance(self) -> None:
"""Disable flat-field correction
@ -889,25 +835,25 @@ class StreamingPiCamera2(BaseCamera):
flat, i.e. we'll correct vignetting of intensity, but not any change in
colour across the image.
"""
with self.picamera(pause_stream=True):
with self._streaming_picamera(pause_stream=True):
alsc = Picamera2.find_tuning_algo(self.tuning, "rpi.alsc")
luminance = alsc["luminance_lut"]
flat = np.ones((12, 16))
recalibrate_utils.set_static_lst(self.tuning, luminance, flat, flat)
self.initialise_picamera()
self._initialise_picamera()
@thing_action
@lt.thing_action
def reset_lens_shading(self) -> None:
"""Revert to default lens shading settings
This method will restore the default "adaptive" lens shading method used
by the Raspberry Pi camera.
"""
with self.picamera(pause_stream=True):
with self._streaming_picamera(pause_stream=True):
recalibrate_utils.copy_alsc_section(self.default_tuning, self.tuning)
self.initialise_picamera()
self._initialise_picamera()
@thing_property
@lt.thing_property
def lens_shading_is_static(self) -> bool:
"""Whether the lens shading is static

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

@ -19,10 +19,7 @@ import numpy as np
import piexif
from scipy.ndimage import gaussian_filter
from labthings_fastapi.utilities import get_blocking_portal
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.server import ThingServer
import labthings_fastapi as lt
from . import BaseCamera, JPEGBlob, ArrayModel
from ..stage import BaseStage
@ -36,7 +33,7 @@ class SimulatedCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
_stage: Optional[BaseStage] = None
_server: Optional[ThingServer] = None
_server: Optional[lt.ThingServer] = None
def __init__(
self,
@ -115,9 +112,11 @@ class SimulatedCamera(BaseCamera):
)
return image
def attach_to_server(self, server: ThingServer, path: str):
def attach_to_server(
self, server: lt.ThingServer, path: str, setting_storage_path: str
):
self._server = server
return super().attach_to_server(server, path)
return super().attach_to_server(server, path, setting_storage_path)
def get_stage_position(self):
if not self._stage and self._server:
@ -144,7 +143,7 @@ class SimulatedCamera(BaseCamera):
self._capture_enabled = False
self._capture_thread.join()
@thing_property
@lt.thing_property
def stream_active(self) -> bool:
"Whether the MJPEG stream is active"
if self._capture_enabled and self._capture_thread:
@ -152,7 +151,7 @@ class SimulatedCamera(BaseCamera):
return False
def _capture_frames(self):
portal = get_blocking_portal(self)
portal = lt.get_blocking_portal(self)
while self._capture_enabled:
time.sleep(self.frame_interval)
try:
@ -166,7 +165,7 @@ class SimulatedCamera(BaseCamera):
except Exception as e:
logging.error(f"Failed to capture frame: {e}, retrying...")
@thing_action
@lt.thing_action
def capture_array(
self,
resolution: Literal["main", "full"] = "full",
@ -180,10 +179,10 @@ class SimulatedCamera(BaseCamera):
logging.warning(f"Simulation camera doen't respect {resolution} setting")
return self.generate_frame()
@thing_action
@lt.thing_action
def capture_jpeg(
self,
metadata_getter: GetThingStates,
metadata_getter: lt.deps.GetThingStates,
resolution: Literal["main", "full"] = "main",
) -> JPEGBlob:
"""Acquire one image from the camera and return as a JPEG blob

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

@ -33,13 +33,10 @@ from camera_stage_mapping.camera_stage_calibration_1d import (
image_to_stage_displacement_from_1d,
)
from camera_stage_mapping.exceptions import MappingError
from labthings_fastapi.dependencies.invocation import (
InvocationCancelledError,
InvocationLogger,
)
from labthings_fastapi.types.numpy import NDArray, denumpify, DenumpifyingDict
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.thing import Thing
import labthings_fastapi as lt
from labthings_fastapi.types.numpy import NDArray, DenumpifyingDict
from camera_stage_mapping.camera_stage_tracker import Tracker
from .camera import CameraDependency as Camera
from .stage import StageDependency as Stage
@ -178,21 +175,15 @@ class CSMUncalibratedError(HTTPException):
)
class CameraStageMapper(Thing):
class CameraStageMapper(lt.Thing):
"""A Thing to manage mapping between image and stage coordinates"""
def __enter__(self):
pass
def __exit__(self, exc_type, exc_value, traceback):
self.thing_settings.write_to_file()
@thing_action
@lt.thing_action
def calibrate_1d(
self,
hw: HardwareInterfaceDep,
stage: Stage,
logger: InvocationLogger,
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"""
@ -207,7 +198,7 @@ class CameraStageMapper(Thing):
result: dict = calibrate_backlash_1d(
tracker, move, direction_array, logger=logger
)
except InvocationCancelledError as e:
except lt.exceptions.InvocationCancelledError as e:
logger.info("User cancelled the camera stage mapping calibration")
logger.info("Returning to starting position")
stage.move_absolute(**starting_position, block_cancellation=True)
@ -220,9 +211,9 @@ class CameraStageMapper(Thing):
result["image_resolution"] = hw.grab_image().shape[:2]
return result
@thing_action
@lt.thing_action
def calibrate_xy(
self, hw: HardwareInterfaceDep, stage: Stage, logger: InvocationLogger
self, hw: HardwareInterfaceDep, stage: Stage, logger: lt.deps.InvocationLogger
) -> DenumpifyingDict:
"""Move the microscope's stage in X and Y, to calibrate its relationship to the camera
@ -244,8 +235,6 @@ class CameraStageMapper(Thing):
corrected_resolution = tuple(
r * hw.grab_image_downsampling for r in cal_x["image_resolution"]
)
self.thing_settings.update(denumpify(cal_xy))
self.thing_settings["image_resolution"] = corrected_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)}]"
@ -260,11 +249,11 @@ class CameraStageMapper(Thing):
"downsampling": hw.grab_image_downsampling,
}
self.thing_settings["last_calibration"] = DenumpifyingDict(data).model_dump()
self.last_calibration = DenumpifyingDict(data).model_dump()
return data
@thing_property
@lt.thing_property
def image_to_stage_displacement_matrix(
self,
) -> Optional[List[List[float]]]: # 2x2 integer array
@ -285,15 +274,26 @@ class CameraStageMapper(Thing):
)
```
"""
displacement_matrix = self.thing_settings.get("image_to_stage_displacement")
if not displacement_matrix:
if self.last_calibration is None:
return None
displacement_matrix = self.last_calibration["camera_stage_mapping_calibration"][
"image_to_stage_displacement"
]
return np.array(displacement_matrix).tolist()
@thing_property
last_calibration = lt.ThingSetting(
initial_value=None,
model=Optional[dict],
readonly=True,
description="The most recent CSM calibration",
)
@lt.thing_property
def image_resolution(self) -> Optional[Tuple[float, float]]:
"""The image size used to calibrate the image_to_stage_displacement_matrix"""
return self.thing_settings.get("image_resolution", None)
if self.last_calibration is None:
return None
return self.last_calibration["image_resolution"]
def assert_calibrated(self):
"""Raise an exception if the image_to_stage_displacement matrix is not set"""
@ -302,12 +302,7 @@ class CameraStageMapper(Thing):
# added by CSMUncalibratedError
raise CSMUncalibratedError() # noqa: RSE102
@thing_property
def last_calibration(self) -> Optional[Dict]:
"""The results of the last calibration that was run"""
return self.thing_settings.get("last_calibration", None)
@thing_action
@lt.thing_action
def move_in_image_coordinates(
self,
stage: Stage,
@ -331,7 +326,7 @@ class CameraStageMapper(Thing):
)
stage.move_relative(x=relative_move[0], y=relative_move[1])
@thing_property
@lt.thing_property
def thing_state(self) -> dict[str, Any]:
"""Summary metadata describing the current state of the Thing"""
return {

95
src/openflexure_microscope_server/things/settings_manager.py
View file

@ -11,19 +11,14 @@ from socket import gethostname
from typing import Annotated, Any, MutableMapping, Optional, Sequence
from uuid import UUID, uuid4
from fastapi import Depends, HTTPException, Request
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.thing_server import find_thing_server
from labthings_fastapi.server import ThingServer
from labthings_fastapi.dependencies.invocation import InvocationLogger
import labthings_fastapi as lt
def thing_server_from_request(request: Request) -> ThingServer:
return find_thing_server(request.app)
def thing_server_from_request(request: Request) -> lt.ThingServer:
return lt.find_thing_server(request.app)
ThingServerDep = Annotated[ThingServer, Depends(thing_server_from_request)]
ThingServerDep = Annotated[lt.ThingServer, Depends(thing_server_from_request)]
def recursive_update(old_dict: MutableMapping, update: Mapping):
@ -48,13 +43,20 @@ def nested_dict_get(data: dict, key: Sequence[str], create=False) -> Any:
return subdict
class SettingsManager(Thing):
@thing_property
def external_metadata(self) -> Mapping:
"""External metadata stored in the server's settings"""
return self.thing_settings.get("external_metadata", {})
class SettingsManager(lt.Thing):
external_metadata = lt.ThingSetting(
initial_value={},
model=Mapping,
description="External metadata stored in the server's settings",
)
@thing_action
external_metadata_in_state = lt.ThingSetting(
initial_value=[],
model=Sequence[str],
description='A list of strings that are included in the "state" metadata',
)
@lt.thing_action
def update_external_metadata(
self, data: Mapping, key: Optional[str] = None
) -> None:
@ -75,9 +77,9 @@ class SettingsManager(Thing):
if key:
subdict = nested_dict_get(subdict, key.split("/"), create=True)
recursive_update(subdict, data)
self.thing_settings["external_metadata"] = metadata
self.external_metadata = metadata
@thing_action
@lt.thing_action
def delete_external_metadata(self, key: str) -> None:
"""Delete a key from the stored metadata.
@ -93,35 +95,32 @@ class SettingsManager(Thing):
raise HTTPException(
status_code=404, detail="The specified key '{key}' was not found"
)
self.thing_settings["external_metadata"] = metadata
self.external_metadata = metadata
@thing_property
_microscope_id: Optional[str] = None
@lt.thing_setting
def microscope_id(self) -> UUID:
"""A unique identifier for this microscope"""
if "microscope_id" not in self.thing_settings:
self.thing_settings["microscope_id"] = str(uuid4())
return UUID(self.thing_settings["microscope_id"])
if self._microscope_id is None:
self._microscope_id = str(uuid4())
return UUID(self._microscope_id)
@thing_property
@microscope_id.setter
def microscope_id(self, uuid: UUID):
# TODO make read only but still settable from disk
self._microscope_id = uuid
@lt.thing_property
def hostname(self) -> str:
"""The hostname of the microscope, as reported by its operating system."""
return gethostname()
@thing_action
def get_things_state(self, metadata_getter: GetThingStates) -> Mapping:
@lt.thing_action
def get_things_state(self, metadata_getter: lt.deps.GetThingStates) -> Mapping:
"""Metadata summarising the current state of all Things in the server"""
return metadata_getter()
@thing_property
def external_metadata_in_state(self) -> Sequence[str]:
"""A list of strings that are included in the "state" metadata"""
return self.thing_settings.get("external_metadata_in_state", [])
@external_metadata_in_state.setter
def external_metadata_in_state(self, keys: Sequence[str]):
"""Set the keys from external metadata that are returned in state"""
self.thing_settings["external_metadata_in_state"] = keys
@property
def thing_state(self) -> Mapping:
state = {
@ -138,29 +137,3 @@ class SettingsManager(Thing):
except KeyError:
continue
return state
@thing_action
def save_all_thing_settings(
self, thing_server: ThingServerDep, logger: InvocationLogger
) -> None:
"""Ensure all the Things sync their settings to disk.
Normally, each Thing has a `thing_settings` attribute that can be
used to save settings. That dictionary is saved to a JSON file
when the server shuts down cleanly.
This action causes all the `thing_settings` dictionaries to be
saved to files immediately, meaning that settings will not be
lost in the event of a server crash, power outage, etc.
"""
for name, thing in thing_server.things.items():
try:
if thing._labthings_thing_settings:
thing._labthings_thing_settings.write_to_file()
logger.info(f"Wrote {name} settings to disk.")
except PermissionError:
logger.warning(
f"Could not write {name} settings to disk: permission error."
)
except FileNotFoundError:
logger.warning(f"Could not write {name} settings, folder not found")

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

@ -11,16 +11,7 @@ from fastapi.responses import FileResponse
import numpy as np
from PIL import Image
from labthings_fastapi.thing import Thing
from labthings_fastapi.dependencies.metadata import GetThingStates
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
from labthings_fastapi.dependencies.invocation import (
CancelHook,
InvocationLogger,
InvocationCancelledError,
)
from labthings_fastapi.decorators import thing_action, thing_property, fastapi_endpoint
from labthings_fastapi.outputs.blob import blob_type
import labthings_fastapi as lt
from openflexure_microscope_server.utilities import ErrorCapturingThread
from openflexure_microscope_server import scan_directories
@ -33,14 +24,16 @@ from .background_detect import BackgroundDetectThing
from .camera import CameraDependency as CamDep
from .stage import StageDependency as StageDep
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
BackgroundDep = direct_thing_client_dependency(
CSMDep = lt.deps.direct_thing_client_dependency(
CameraStageMapper, "/camera_stage_mapping/"
)
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
BackgroundDep = lt.deps.direct_thing_client_dependency(
BackgroundDetectThing, "/background_detect/"
)
JPEGBlob = blob_type("image/jpeg")
ZipBlob = blob_type("application/zip")
JPEGBlob = lt.blob.blob_type("image/jpeg")
ZipBlob = lt.blob.blob_type("application/zip")
SCAN_DATA_FILENAME = "scan_data.json"
STITCHING_CMD = "openflexure-stitch"
@ -71,7 +64,7 @@ def _scan_running(method):
return scan_running_wrapper
class SmartScanThing(Thing):
class SmartScanThing(lt.Thing):
def __init__(self, scans_folder):
self._scan_dir_manager = scan_directories.ScanDirectoryManager(scans_folder)
self._preview_stitch_popen = None
@ -82,17 +75,17 @@ class SmartScanThing(Thing):
self._latest_scan_name: Optional[str] = None
# Scan logger is the invocation logger labthings-fastapi creates
# when the `sample_scan` thing_action is called. It is saved as
# when the `sample_scan` lt.thing_action is called. It is saved as
# private class variable along with many others here.
# Access to these variables requires a scan to be running,
# any method that calls these should be decorrected with
# @_scan_running
self._scan_logger: Optional[InvocationLogger] = None
self._cancel: Optional[CancelHook] = None
self._scan_logger: Optional[lt.deps.InvocationLogger] = None
self._cancel: Optional[lt.deps.CancelHook] = None
self._autofocus: Optional[AutofocusDep] = None
self._stage: Optional[StageDep] = None
self._cam: Optional[CamDep] = None
self._metadata_getter: Optional[GetThingStates] = None
self._metadata_getter: Optional[lt.deps.GetThingStates] = None
self._csm: Optional[CSMDep] = None
self._background_detect: Optional[BackgroundDep] = None
self._ongoing_scan: Optional[scan_directories.ScanDirectory] = None
@ -102,15 +95,15 @@ class SmartScanThing(Thing):
# TODO Scan data is a dict during refactoring, should become a dataclass
self._scan_data: Optional[dict] = None
@thing_action
@lt.thing_action
def sample_scan(
self,
cancel: CancelHook,
logger: InvocationLogger,
cancel: lt.deps.CancelHook,
logger: lt.deps.InvocationLogger,
autofocus: AutofocusDep,
stage: StageDep,
cam: CamDep,
metadata_getter: GetThingStates,
metadata_getter: lt.deps.GetThingStates,
csm: CSMDep,
background_detect: BackgroundDep,
scan_name: str = "",
@ -205,7 +198,7 @@ class SmartScanThing(Thing):
"of motion."
)
@thing_property
@lt.thing_property
def latest_scan_name(self) -> Optional[str]:
"""The name of the last scan to be started."""
return self._latest_scan_name
@ -414,7 +407,7 @@ class SmartScanThing(Thing):
self._main_scan_loop()
self._update_scan_data_json(scan_result="success")
except InvocationCancelledError:
except lt.exceptions.InvocationCancelledError:
scan_successful = False
# Reset the cancel event so it can be thrown again
self._cancel.clear()
@ -567,7 +560,7 @@ class SmartScanThing(Thing):
except SubprocessError as e:
self._scan_logger.error(f"Stitching failed: {e}", exc_info=e)
@fastapi_endpoint(
@lt.fastapi_endpoint(
"get",
"scans/stitched_thumbnail.jpg",
responses={
@ -590,74 +583,56 @@ class SmartScanThing(Thing):
raise HTTPException(404, "File not found")
return FileResponse(preview_path)
@thing_property
def save_resolution(self) -> tuple[int, int]:
"""A tuple of the image resolution to capture. Should be in a
4:3 aspect ratio"""
return self.thing_settings.get("save_resolution", ((1640, 1232)))
save_resolution = lt.ThingSetting(
initial_value=(1640, 1232),
model=tuple[int, int],
description=("A tuple of the image resolution to capture."),
)
@save_resolution.setter
def save_resolution(self, value: tuple[int, int]) -> None:
self.thing_settings["save_resolution"] = value
max_range = lt.ThingSetting(
initial_value=45000,
model=int,
description=(
"The maximum distance from the centre of the scan before we break in steps"
),
)
@thing_property
def max_range(self) -> int:
"""The maximum distance from the centre of the scan before we break in steps"""
return self.thing_settings.get("max_range", 45000)
stitch_tiff = lt.ThingSetting(
initial_value=False,
model=bool,
description="Whether or not to also produce a pyramidal tiff",
)
@max_range.setter
def max_range(self, value: int) -> None:
self.thing_settings["max_range"] = value
skip_background = lt.ThingSetting(
initial_value=True,
model=bool,
description="""Whether to detect and skip empty fields of view
@thing_property
def stitch_tiff(self) -> bool:
"""Whether or not to also produce a pyramidal tiff"""
return self.thing_settings.get("stitch_tiff", False)
This uses the settings from the `background_detect` Thing.""",
)
@stitch_tiff.setter
def stitch_tiff(self, value: bool) -> None:
self.thing_settings["stitch_tiff"] = value
autofocus_dz = lt.ThingSetting(
initial_value=1000,
model=int,
description="The z distance to perform an autofocus in steps",
)
@thing_property
def skip_background(self) -> bool:
"""Whether to detect and skip empty fields of view
overlap = lt.ThingSetting(
initial_value=0.45,
model=float,
description="The fraction (0-1) that adjacent images should overlap in x or y",
)
This uses the settings from the `background_detect` Thing.
"""
return self.thing_settings.get("skip_background", True)
stitch_automatically = lt.ThingSetting(
initial_value=True,
model=bool,
description=(
"Whether to run a final stitch at the end of the scan (assuming scan "
"success)"
),
)
@skip_background.setter
def skip_background(self, value: bool) -> None:
self.thing_settings["skip_background"] = value
@thing_property
def autofocus_dz(self) -> int:
"""The z distance to perform an autofocus in steps"""
return self.thing_settings.get("autofocus_dz", 1000)
@autofocus_dz.setter
def autofocus_dz(self, value: int) -> None:
self.thing_settings["autofocus_dz"] = value
@thing_property
def overlap(self) -> float:
"""The fraction (0-1) that adjacent images should overlap in x or y"""
return self.thing_settings.get("overlap", 0.45)
@overlap.setter
def overlap(self, value: float) -> None:
self.thing_settings["overlap"] = value
@thing_property
def stitch_automatically(self) -> bool:
"""Whether to run a final stitch at the end of the scan (assuming scan success)"""
return self.thing_settings.get("stitch_automatically", True)
@stitch_automatically.setter
def stitch_automatically(self, value: bool) -> None:
self.thing_settings["stitch_automatically"] = value
@thing_property
@lt.thing_property
def scans(self) -> list[scan_directories.ScanInfo]:
"""All the available scans
@ -669,7 +644,7 @@ class SmartScanThing(Thing):
"""
return self._scan_dir_manager.all_scans_info()
@fastapi_endpoint(
@lt.fastapi_endpoint(
"get",
"get_stitch/{scan_name}",
responses={
@ -693,7 +668,7 @@ class SmartScanThing(Thing):
raise HTTPException(404, "File not found")
return FileResponse(stitch_path)
@fastapi_endpoint(
@lt.fastapi_endpoint(
"delete",
"scans/{scan_name}",
responses={
@ -701,7 +676,7 @@ class SmartScanThing(Thing):
400: {"description": "An error occurred while trying to delete scan"},
},
)
def delete_scan(self, scan_name: str, logger: InvocationLogger) -> None:
def delete_scan(self, scan_name: str, logger: lt.deps.InvocationLogger) -> None:
"""Delete the folder for the specified scan.
This endpoint allows scans to be deleted from disk.
@ -715,11 +690,11 @@ class SmartScanThing(Thing):
if not deleted_scan_success:
raise HTTPException(400, "Couldn't delete scan, check log for details")
@fastapi_endpoint(
@lt.fastapi_endpoint(
"delete",
"scans",
)
def delete_all_scans(self, logger: InvocationLogger) -> None:
def delete_all_scans(self, logger: lt.deps.InvocationLogger) -> None:
"""Delete all the scans on the microscope
**This will irreversibly remove all scanned data from the
@ -729,8 +704,8 @@ class SmartScanThing(Thing):
for scan_name in self._scan_dir_manager.all_scans:
self._delete_scan(scan_name, logger)
@thing_action
def purge_empty_scans(self, logger: InvocationLogger) -> None:
@lt.thing_action
def purge_empty_scans(self, logger: lt.deps.InvocationLogger) -> None:
"""
Delete all scan folders containing no images at the top level
"""
@ -740,7 +715,7 @@ class SmartScanThing(Thing):
if scan_info.number_of_images == 0:
self._delete_scan(scan_info.name, logger)
def _delete_scan(self, scan_name, logger: InvocationLogger) -> bool:
def _delete_scan(self, scan_name, logger: lt.deps.InvocationLogger) -> bool:
"""
A wrapper around scan manager's delete_scan that logs to the invocation logger
"""
@ -765,7 +740,7 @@ class SmartScanThing(Thing):
scan_name=self.latest_scan_name, filename="preview.jpg", check_exists=True
)
@thing_property
@lt.thing_property
def latest_preview_stitch_time(self) -> Optional[float]:
"""The modification time of the latest preview image, to allow live updating
@ -780,7 +755,7 @@ class SmartScanThing(Thing):
return None
return os.path.getmtime(self.latest_preview_stitch_path)
@fastapi_endpoint(
@lt.fastapi_endpoint(
"get",
"latest_preview_stitch.jpg",
responses={
@ -849,8 +824,8 @@ class SmartScanThing(Thing):
def run_subprocess(
self,
logger: InvocationLogger,
cancel: CancelHook,
logger: lt.deps.InvocationLogger,
cancel: lt.deps.CancelHook,
cmd: list[str],
) -> CompletedProcess:
"""
@ -886,7 +861,7 @@ class SmartScanThing(Thing):
# Note that using cancel.sleep allows the InvocationCancelledError to
# be thrown
cancel.sleep(0.2)
except InvocationCancelledError as e:
except lt.exceptions.InvocationCancelledError as e:
logger.info("Stitching cancelled by user")
process.kill()
raise e
@ -899,18 +874,18 @@ class SmartScanThing(Thing):
else:
raise ChildProcessError(f"Subprocess {cmd[0]} exited with an error.")
@thing_action
@lt.thing_action
def stitch_scan(
self,
logger: InvocationLogger,
cancel: CancelHook,
logger: lt.deps.InvocationLogger,
cancel: lt.deps.CancelHook,
scan_name: str,
stitch_resize: Optional[float] = None,
overlap: float = 0.0,
) -> None:
"""Generate a stitched image based on stage position metadata
Note that as this is a thing_action it needs the logger passed as
Note that as this is a lt.thing_action it needs the logger passed as
a variable if called from another thing action
"""
json_fpath = self._scan_dir_manager.get_file_path_from_img_dir(
@ -982,12 +957,12 @@ class SmartScanThing(Thing):
self._scan_dir_manager.img_dir_for(scan_name),
],
)
except InvocationCancelledError:
except lt.exceptions.InvocationCancelledError:
# Sleep for 1 second just to allow invocation logs to pass to user.
time.sleep(1)
pass
@thing_action
@lt.thing_action
def download_zip(
self,
scan_name: str,

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

@ -2,14 +2,10 @@ from __future__ import annotations
from typing import TypeAlias
from collections.abc import Sequence, Mapping
from labthings_fastapi.descriptors.property import PropertyDescriptor
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
from labthings_fastapi.dependencies.invocation import CancelHook
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
import labthings_fastapi as lt
class BaseStage(Thing):
class BaseStage(lt.Thing):
"""A base stage class for OpenFlexure translation stages
This can't be used directly but should reduce boilerplate code when
@ -20,12 +16,12 @@ class BaseStage(Thing):
_axis_names = ("x", "y", "z")
@thing_property
@lt.thing_property
def axis_names(self) -> Sequence[str]:
"""The names of the stage's axes, in order."""
return self._axis_names
position = PropertyDescriptor(
position = lt.ThingProperty(
Mapping[str, int],
dict.fromkeys(_axis_names, 0),
description="Current position of the stage",
@ -33,7 +29,7 @@ class BaseStage(Thing):
observable=True,
)
moving = PropertyDescriptor(
moving = lt.ThingProperty(
bool,
False,
description="Whether the stage is in motion",
@ -46,10 +42,10 @@ class BaseStage(Thing):
"""Summary metadata describing the current state of the stage"""
return {"position": self.position}
@thing_action
@lt.thing_action
def move_relative(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
):
@ -58,10 +54,10 @@ class BaseStage(Thing):
"StageThings must define their own move_relative method"
)
@thing_action
@lt.thing_action
def move_absolute(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
):
@ -70,7 +66,7 @@ class BaseStage(Thing):
"StageThings must define their own move_absolute method"
)
@thing_action
@lt.thing_action
def set_zero_position(self):
"""Make the current position zero in all axes
@ -83,4 +79,6 @@ class BaseStage(Thing):
)
StageDependency: TypeAlias = direct_thing_client_dependency(BaseStage, "/stage/")
StageDependency: TypeAlias = lt.deps.direct_thing_client_dependency(
BaseStage, "/stage/"
)

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

@ -1,12 +1,10 @@
from __future__ import annotations
from labthings_fastapi.decorators import thing_action
from labthings_fastapi.dependencies.invocation import (
CancelHook,
InvocationCancelledError,
)
from collections.abc import Mapping
import time
import labthings_fastapi as lt
from . import BaseStage
@ -27,10 +25,10 @@ class DummyStage(BaseStage):
def __exit__(self, _exc_type, _exc_value, _traceback):
pass
@thing_action
@lt.thing_action
def move_relative(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
):
@ -53,7 +51,7 @@ class DummyStage(BaseStage):
for k, v in zip(self.axis_names, displacement)
}
fraction_complete = 1.0
except InvocationCancelledError as e:
except lt.exceptions.InvocationCancelledError as e:
# If the move has been cancelled, stop it but don't handle the exception.
# We need the exception to propagate in order to stop any calling tasks,
# and to mark the invocation as "cancelled" rather than stopped.
@ -66,10 +64,10 @@ class DummyStage(BaseStage):
}
self.instantaneous_position = self.position
@thing_action
@lt.thing_action
def move_absolute(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
):
@ -83,7 +81,7 @@ class DummyStage(BaseStage):
cancel, block_cancellation=block_cancellation, **displacement
)
@thing_action
@lt.thing_action
def set_zero_position(self):
"""Make the current position zero in all axes

20
src/openflexure_microscope_server/things/stage/sangaboard.py
View file

@ -7,11 +7,7 @@ from contextlib import contextmanager
from collections.abc import Mapping
import sangaboard
from labthings_fastapi.decorators import thing_action
from labthings_fastapi.dependencies.invocation import (
CancelHook,
InvocationCancelledError,
)
import labthings_fastapi as lt
from . import BaseStage
@ -57,10 +53,10 @@ class SangaboardThing(BaseStage):
with self.sangaboard() as sb:
self.position = dict(zip(self.axis_names, sb.position))
@thing_action
@lt.thing_action
def move_relative(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
) -> None:
@ -75,7 +71,7 @@ class SangaboardThing(BaseStage):
else:
while sb.query("moving?") == "true":
cancel.sleep(0.1)
except InvocationCancelledError as e:
except lt.exceptions.InvocationCancelledError as e:
# If the move has been cancelled, stop it but don't handle the exception.
# We need the exception to propagate in order to stop any calling tasks,
# and to mark the invocation as "cancelled" rather than stopped.
@ -85,10 +81,10 @@ class SangaboardThing(BaseStage):
self.moving = False
self.update_position()
@thing_action
@lt.thing_action
def move_absolute(
self,
cancel: CancelHook,
cancel: lt.deps.CancelHook,
block_cancellation: bool = False,
**kwargs: Mapping[str, int],
) -> None:
@ -104,7 +100,7 @@ class SangaboardThing(BaseStage):
cancel, block_cancellation=block_cancellation, **displacement
)
@thing_action
@lt.thing_action
def set_zero_position(self) -> None:
"""Make the current position zero in all axes
@ -116,7 +112,7 @@ class SangaboardThing(BaseStage):
sb.zero_position()
self.update_position()
@thing_action
@lt.thing_action
def flash_led(
self,
number_of_flashes: int = 10,

11
src/openflexure_microscope_server/things/system_control.py
View file

@ -6,8 +6,7 @@ This module defines a Thing that can shut down or restart the host computer.
import subprocess
import os
from labthings_fastapi.thing import Thing
from labthings_fastapi.decorators import thing_action, thing_property
import labthings_fastapi as lt
from pydantic import BaseModel
@ -16,12 +15,12 @@ class CommandOutput(BaseModel):
error: str
class SystemControlThing(Thing):
class SystemControlThing(lt.Thing):
"""
Attempt to shutdown the device
"""
@thing_action
@lt.thing_action
def shutdown(self) -> CommandOutput:
"""
Attempt to shutdown the device
@ -35,14 +34,14 @@ class SystemControlThing(Thing):
out, err = p.communicate()
return CommandOutput(output=out, error=err)
@thing_property
@lt.thing_property
def is_raspberrypi() -> bool:
"""
Checks if we are running on a Raspberry Pi.
"""
return os.path.exists("/usr/bin/raspi-config")
@thing_action
@lt.thing_action
def reboot(self) -> CommandOutput:
"""Attempt to reboot the device"""
p = subprocess.Popen(

18
tests/test_dummy_server.py
View file

@ -3,13 +3,13 @@ import os
import tempfile
from fastapi.testclient import TestClient
from labthings_fastapi.client import ThingClient
from PIL import Image
import numpy as np
import piexif
import pytest
from openflexure_microscope_server.server import ThingServer
import labthings_fastapi as lt
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
from openflexure_microscope_server.things.stage.dummy import DummyStage
from openflexure_microscope_server.things.autofocus import AutofocusThing
@ -22,7 +22,7 @@ camera_stage_mapping.DEFAULT_SETTLING_TIME = 0 # skip the settling time for tes
@pytest.fixture
def thing_server():
temp_folder = tempfile.TemporaryDirectory()
server = ThingServer(settings_folder=temp_folder.name)
server = lt.ThingServer(settings_folder=temp_folder.name)
server.add_thing(
SimulatedCamera(
shape=(240, 320, 3), canvas_shape=(960, 1240, 3), frame_interval=0.01
@ -52,18 +52,18 @@ def slower_client(thing_server):
def test_autofocus(slower_client):
client = slower_client
autofocus = ThingClient.from_url("/autofocus/", client)
autofocus = lt.ThingClient.from_url("/autofocus/", client)
_ = autofocus.fast_autofocus()
def test_grab_jpeg(client):
camera = ThingClient.from_url("/camera/", client)
camera = lt.ThingClient.from_url("/camera/", client)
blob = camera.grab_jpeg()
_image = Image.open(blob.open())
def test_capture_jpeg_metadata(client):
camera = ThingClient.from_url("/camera/", client)
camera = lt.ThingClient.from_url("/camera/", client)
blob = camera.capture_jpeg()
image = Image.open(blob.open())
exif_dict = piexif.load(image.info["exif"])
@ -73,7 +73,7 @@ def test_capture_jpeg_metadata(client):
def test_stage(client):
stage = ThingClient.from_url("/stage/", client)
stage = lt.ThingClient.from_url("/stage/", client)
start = stage.position
move = {"x": 1, "y": 2, "z": 3}
stage.move_relative(**move)
@ -87,12 +87,12 @@ def test_stage(client):
def test_capture_array(client):
camera = ThingClient.from_url("/camera/", client)
camera = lt.ThingClient.from_url("/camera/", client)
array = np.asarray(camera.capture_array())
assert array.shape == (240, 320, 3)
# Currently this fails, not yet sure why.
def test_camera_stage_mapping_calibration(client):
camera_stage_mapping = ThingClient.from_url("/camera_stage_mapping/", client)
camera_stage_mapping = lt.ThingClient.from_url("/camera_stage_mapping/", client)
camera_stage_mapping.calibrate_xy()

4
tests/test_smart_scan.py
View file

@ -256,7 +256,9 @@ def test_outer_scan_wo_sample_skip():
"""Test setup and teardown of the scan."""
def _set_skip_background(mock_ss_thing):
mock_ss_thing.thing_settings["skip_background"] = False
# As the Thing is not connected to a server we set the setting via the internal
# __dict__ to avoid triggering property emits that require a server
mock_ss_thing.__dict__["skip_background"] = False
mock_ss_thing, exec_info = _run_only_outer_scan(_set_skip_background)

9
webapp/src/components/tabContentComponents/settingsContent.vue
View file

@ -58,11 +58,6 @@
</tabIcon>
</li>
</ul>
<action-button
thing="settings"
action="save_all_thing_settings"
submit-label="Save All Settings"
/>
</div>
<div class="view-component uk-width-expand uk-padding-small">
<tabContent
@ -118,7 +113,6 @@ import stageSettings from "./settingsComponents/stageSettings.vue";
// Import generic components
import tabIcon from "../genericComponents/tabIcon";
import tabContent from "../genericComponents/tabContent";
import ActionButton from "../labThingsComponents/actionButton.vue";
// Export main app
export default {
@ -131,8 +125,7 @@ export default {
CSMSettings,
appSettings,
tabIcon,
tabContent,
ActionButton
tabContent
},
data: function() {