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

Remove dependencies from camera

Browse source
This commit is contained in:
Julian Stirling 2025-12-16 13:35:30 +00:00
parent 7038ecdf0b
commit 4b845fcf18
1 changed files with 14 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -10,7 +10,6 @@ from __future__ import annotations
import io import io
import json import json
import logging
import os import os
import tempfile import tempfile
import time import time
@ -34,8 +33,6 @@ from openflexure_microscope_server.background_detect import (
) )
from openflexure_microscope_server.ui import ActionButton, PropertyControl from openflexure_microscope_server.ui import ActionButton, PropertyControl
LOGGER = logging.getLogger(__name__)
class JPEGBlob(lt.blob.Blob): class JPEGBlob(lt.blob.Blob):
"""A class representing a JPEG image as a LabThings FastAPI Blob.""" """A class representing a JPEG image as a LabThings FastAPI Blob."""
@ -282,8 +279,6 @@ class BaseCamera(lt.Thing):
@lt.action @lt.action
def capture_jpeg( def capture_jpeg(
self, self,
metadata_getter: lt.deps.GetThingStates,
logger: lt.deps.InvocationLogger,
stream_name: str = "main", stream_name: str = "main",
wait: Optional[float] = None, wait: Optional[float] = None,
) -> JPEGBlob: ) -> JPEGBlob:
@ -292,9 +287,6 @@ class BaseCamera(lt.Thing):
This will use the internal capture image functionally of capture_image of This will use the internal capture image functionally of capture_image of
the specific camera being used. the specific camera being used.
:param metadata_getter: LabThings GetThingStates dependency, automatically
injected.
:param logger: LabThings InvocationLogger dependency, automatically injected.
:param stream_name: A stream name supported by this camera. :param stream_name: A stream name supported by this camera.
:param wait: (Optional, float) Set a timeout in seconds. If None it will :param wait: (Optional, float) Set a timeout in seconds. If None it will
use the default for the underlying camera. use the default for the underlying camera.
@ -305,13 +297,12 @@ class BaseCamera(lt.Thing):
img = self.capture_image(stream_name, wait) img = self.capture_image(stream_name, wait)
capture_metadata = self._capture_metadata(metadata_getter()) capture_metadata = self._capture_metadata()
self._save_capture( self._save_capture(
jpeg_path=jpeg_path, jpeg_path=jpeg_path,
image=img, image=img,
metadata=capture_metadata, metadata=capture_metadata,
logger=logger,
) )
return JPEGBlob.from_temporary_directory(directory, fname) return JPEGBlob.from_temporary_directory(directory, fname)
@ -390,70 +381,43 @@ class BaseCamera(lt.Thing):
def capture_and_save( def capture_and_save(
self, self,
jpeg_path: str, jpeg_path: str,
logger: lt.deps.InvocationLogger,
metadata_getter: lt.deps.GetThingStates,
save_resolution: Optional[Tuple[int, int]] = None, save_resolution: Optional[Tuple[int, int]] = None,
) -> 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 jpeg_path: The path to save the file to
:param logger: This should be injected automatically by Labthings FastAPI
when calling the action
:param metadata_getter: This should be injected automatically by Labthings
FastAPI when calling the action
:param save_resolution: can be set to resize the image before saving. By :param save_resolution: can be set to resize the image before saving. By
default this is None meaning that the image is saved at original resolution. default this is None meaning that the image is saved at original resolution.
""" """
image, capture_metadata = self._robust_image_capture( image, capture_metadata = self._robust_image_capture()
metadata_getter,
logger=logger,
)
self._save_capture( self._save_capture(jpeg_path, image, capture_metadata, save_resolution)
jpeg_path,
image,
capture_metadata,
logger,
save_resolution,
)
@lt.action @lt.action
def capture_to_memory( def capture_to_memory(self, buffer_max: int = 1) -> int:
self,
logger: lt.deps.InvocationLogger,
metadata_getter: lt.deps.GetThingStates,
buffer_max: int = 1,
) -> int:
"""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 Note that only one image is held in memory so this will overwrite any image
in memory. in memory.
:param logger: This should be injected automatically by Labthings FastAPI
when calling the action
:param metadata_getter: This should be injected automatically by Labthings
FastAPI when calling the action
:param buffer_max: The maximum number of images that should be in the buffer :param buffer_max: The maximum number of images that should be in the buffer
once this images is added. Default is 1. once this images is added. Default is 1.
:returns: the buffer id of the image captured :returns: the buffer id of the image captured
""" """
image, metadata = self._robust_image_capture(metadata_getter, logger) image, metadata = self._robust_image_capture()
return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max) return self._memory_buffer.add_image(image, metadata, buffer_max=buffer_max)
@lt.action @lt.action
def save_from_memory( def save_from_memory(
self, self,
jpeg_path: str, jpeg_path: str,
logger: lt.deps.InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None, save_resolution: Optional[Tuple[int, int]] = None,
buffer_id: Optional[int] = None, buffer_id: Optional[int] = None,
) -> None: ) -> None:
"""Save an image that has been captured to memory. """Save an image that has been captured to memory.
:param jpeg_path: The path to save the file to :param jpeg_path: The path to save the file to
:param logger: This should be injected automatically by Labthings FastAPI
when calling the action
:param save_resolution: can be set to resize the image before saving. By :param save_resolution: can be set to resize the image before saving. By
default this is None meaning that the image is saved at original default this is None meaning that the image is saved at original
resolution. resolution.
@ -466,7 +430,6 @@ class BaseCamera(lt.Thing):
jpeg_path=jpeg_path, jpeg_path=jpeg_path,
image=image, image=image,
metadata=metadata, metadata=metadata,
logger=logger,
save_resolution=save_resolution, save_resolution=save_resolution,
) )
@ -475,11 +438,7 @@ class BaseCamera(lt.Thing):
"""Clear all images in memory.""" """Clear all images in memory."""
self._memory_buffer.clear() self._memory_buffer.clear()
def _robust_image_capture( def _robust_image_capture(self) -> Tuple[Image, Mapping[str, Any]]:
self,
metadata_getter: lt.deps.GetThingStates,
logger: lt.deps.InvocationLogger,
) -> Tuple[Image, Mapping[str, Any]]:
"""Capture an image in memory and return it with metadata. """Capture an image in memory and return it with metadata.
This robust capturing method attempts to capture the image five times This robust capturing method attempts to capture the image five times
@ -491,22 +450,19 @@ class BaseCamera(lt.Thing):
""" """
for capture_attempts in range(5): for capture_attempts in range(5):
try: try:
metadata = metadata_getter() capture_metadata = self._capture_metadata()
capture_metadata = self._capture_metadata(metadata)
image = self.capture_image(stream_name="main", wait=5) image = self.capture_image(stream_name="main", wait=5)
return image, capture_metadata return image, capture_metadata
except TimeoutError: except TimeoutError:
logger.warning( self.logger.warning(
f"Attempt {capture_attempts + 1} to capture image timed out. Do you have enough RAM?" f"Attempt {capture_attempts + 1} to capture image timed out. Do you have enough RAM?"
) )
raise CaptureError("An error occurred while capturing after 5 attempts") raise CaptureError("An error occurred while capturing after 5 attempts")
def _capture_metadata( def _capture_metadata(self) -> dict:
self,
metadata: Mapping[str, Any],
) -> dict:
"""Return the metadata for a capture, from the thing states, time and known names.""" """Return the metadata for a capture, from the thing states, time and known names."""
metadata = self._thing_server_interface.get_thing_states()
current_time = datetime.now() current_time = datetime.now()
return { return {
"capture_time": current_time.timestamp(), "capture_time": current_time.timestamp(),
@ -566,12 +522,11 @@ class BaseCamera(lt.Thing):
jpeg_path: str, jpeg_path: str,
image: Image, image: Image,
metadata: dict, metadata: dict,
logger: lt.deps.InvocationLogger,
save_resolution: Optional[Tuple[int, int]] = None, save_resolution: Optional[Tuple[int, int]] = None,
) -> None: ) -> None:
"""Save the captured image and metadata to disk. """Save the captured image and metadata to disk.
A warning (via InvocationLogger) is raised if metadata is failed to be added A warning is logged if metadata cannot be added.
:raises IOError: if the file cannot be saved :raises IOError: if the file cannot be saved
@ -593,7 +548,7 @@ class BaseCamera(lt.Thing):
except Exception: except Exception:
# We need to capture any exception as there are many reasons metadata # We need to capture any exception as there are many reasons metadata
# might not be added. We warn rather than log the error. # might not be added. We warn rather than log the error.
logger.exception(f"Failed to add metadata to {jpeg_path}") self.logger.exception(f"Failed to add metadata to {jpeg_path}")
except Exception as e: except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e raise IOError(f"An error occurred while saving {jpeg_path}") from e
@ -639,7 +594,7 @@ class BaseCamera(lt.Thing):
def detector_name(self, name: str) -> None: def detector_name(self, name: str) -> None:
"""Validate and set detector_name.""" """Validate and set detector_name."""
if name not in self.background_detectors: if name not in self.background_detectors:
LOGGER.warning(f"{name} is not a valid background detector name.") self.logger.warning(f"{name} is not a valid background detector name.")
self._detector_name = name self._detector_name = name
@property @property
@ -697,7 +652,7 @@ class BaseCamera(lt.Thing):
obj.settings = instance_data["settings"] obj.settings = instance_data["settings"]
obj.background_data = instance_data["background_data"] obj.background_data = instance_data["background_data"]
else: else:
LOGGER.warning( self.logger.warning(
f"No background detector named {name}, settings will be discarded." f"No background detector named {name}, settings will be discarded."
) )