openflexure_microscope_server.things.camera.BaseCamera

class documentation

class BaseCamera(lt.Thing): (source)

Known subclasses: openflexure_microscope_server.things.camera.opencv.OpenCVCamera, openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

Constructor: BaseCamera()

View In Hierarchy

The base class for all cameras. All cameras must directly inherit from this class.

The connection to the camera hardware should be added to the __enter__ method not __init__ method of the subclass.

Method	`__enter__`	Open hardware connection when the Thing context manager is opened.
Method	`__exit__`	Close hardware connection when the Thing context manager is closed.
Method	`__init__`	Initialise the base camera, this creates the background detectors.
Method	`background_detector_data`	The data for each background detector, used to save to disk.
Method	`background_detector_data.setter`	Set the data for each detector. Only to be used as settings are loaded from disk.
Method	`capture_and_save`	Capture an image and save it to disk.
Method	`capture_array`	Acquire one image from the camera and return as an array.
Method	`capture_downsampled_array`	Acquire one image from the camera, downsample, and return as an array.
Method	`capture_image`	Capture a PIL image from stream stream_name with timeout wait.
Method	`capture_jpeg`	Acquire one image from the camera and return as a JPEG blob.
Method	`capture_to_memory`	Capture an image to memory. This can be saved later with `save_from_memory`.
Method	`clear_buffers`	Clear all images in memory.
Method	`detector_name`	The name of the active background selector.
Method	`detector_name.setter`	Validate and set detector_name.
Method	`discard_frames`	Discard frames so that the next frame captured is fresh.
Method	`grab_as_array`	Acquire one image from the preview stream and return as an array.
Method	`grab_jpeg`	Acquire one image from the preview stream and return as blob of JPEG data.
Method	`grab_jpeg_size`	Acquire one image from the preview stream and return its size.
Method	`image_is_sample`	Label the current image as either background or sample.
Method	`kill_mjpeg_streams`	Kill the streams now as the server is shutting down.
Method	`save_from_memory`	Save an image that has been captured to memory.
Method	`set_background`	Grab an image, and use its statistics to set the background.
Method	`settle`	Sleep for the settling time, ready to provide a fresh frame.
Method	`start_streaming`	Start (or stop and restart) the camera.
Method	`update_detector_settings`	Update the settings of the current detector.
Class Variable	`downsampled_array_factor`	The downsampling factor when calling capture_downsampled_array.
Class Variable	`lores_mjpeg_stream`	Undocumented
Class Variable	`mjpeg_stream`	Undocumented
Class Variable	`settling_time`	The settling time when calling the `settle()` method.
Instance Variable	`background_detectors`	Undocumented
Property	`active_detector`	The active background detector instance.
Property	`background_detector_status`	The status of the active detector for the UI.
Property	`manual_camera_settings`	The camera settings to expose as property controls in the settings panel.
Property	`primary_calibration_actions`	The calibration actions for both calibration wizard and settings panel.
Property	`secondary_calibration_actions`	The calibration actions that appear only in settings panel.
Property	`stream_active`	Whether the MJPEG stream is active.
Method	`_robust_image_capture`	Capture an image in memory and return it with metadata.
Method	`_save_capture`	Save the captured image and metadata to disk.
Class Variable	`_memory_buffer`	Undocumented
Instance Variable	`_detector_name`	Undocumented

def __enter__(self): (source) ¶

overridden in openflexure_microscope_server.things.camera.opencv.OpenCVCamera, openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

Open hardware connection when the Thing context manager is opened.

def __exit__(self, _exc_type, _exc_value, _traceback): (source) ¶

Close hardware connection when the Thing context manager is closed.

def __init__(self): (source) ¶

Initialise the base camera, this creates the background detectors.

This must be run by all child camera classes.

To add a new background detector to the server it must be added to the dictionary in this function. Configuration will be added at a later date.

@lt.thing_setting

def background_detector_data(self) -> dict: (source) ¶

The data for each background detector, used to save to disk.

@background_detector_data.setter

def background_detector_data(self, data: dict): (source) ¶

Set the data for each detector. Only to be used as settings are loaded from disk.

Do not call over HTTP. This needs to be updated once LbaThings Settings can be read-only over HTTP (#484).

@lt.thing_action

def capture_and_save(self, jpeg_path: str, logger: lt.deps.InvocationLogger, metadata_getter: lt.deps.GetThingStates, save_resolution: tuple[int, int] | None = None): (source) ¶

Capture an image and save it to disk.

Parameters
jpeg_path:`str`	The path to save the file to
logger:`lt.deps.InvocationLogger`	This should be injected automatically by Labthings FastAPI when calling the action
metadata_getter:`lt.deps.GetThingStates`	This should be injected automatically by Labthings FastAPI when calling the action
save_resolution:`tuple[int, int] \| None`	can be set to resize the image before saving. By default this is None meaning that the image is saved at original resolution.

@lt.thing_action

def capture_array(self, stream_name: Literal['main', 'lores', 'raw', 'full'] = 'main', wait: float | None = 5) -> ArrayModel: (source) ¶

Acquire one image from the camera and return as an array.

@lt.thing_action

def capture_downsampled_array(self) -> ArrayModel: (source) ¶

Acquire one image from the camera, downsample, and return as an array.

The array is downsamples by the thing property downsampled_array_factor.
The default capture array arguments are used.

This method provides the interface expected by the camera_stage_mapping.

def capture_image(self, stream_name: Literal['main', 'lores', 'raw'], wait: float | None) -> Image: (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

Capture a PIL image from stream stream_name with timeout wait.

@lt.thing_action

def capture_jpeg(self, metadata_getter: lt.deps.GetThingStates, resolution: Literal['lores', 'main', 'full'] = 'main', wait: float | None = 5) -> JPEGBlob: (source) ¶

Acquire one image from the camera and return as a JPEG blob.

@lt.thing_action

def capture_to_memory(self, logger: lt.deps.InvocationLogger, metadata_getter: lt.deps.GetThingStates, buffer_max: int = 1) -> int: (source) ¶

Capture an image to memory. This can be saved later with save_from_memory.

Note that only one image is held in memory so this will overwrite any image in memory.

Parameters
logger:`lt.deps.InvocationLogger`	This should be injected automatically by Labthings FastAPI when calling the action
metadata_getter:`lt.deps.GetThingStates`	This should be injected automatically by Labthings FastAPI when calling the action
buffer_max:`int`	The maximum number of images that should be in the buffer once this images is added. Default is 1.

Returns
`int`	the buffer id of the image captured

@lt.thing_action

def clear_buffers(self): (source) ¶

Clear all images in memory.

@lt.thing_setting

def detector_name(self) -> str: (source) ¶

The name of the active background selector.

@detector_name.setter

def detector_name(self, name: str): (source) ¶

Validate and set detector_name.

@lt.thing_action

def discard_frames(self): (source) ¶

Discard frames so that the next frame captured is fresh.

@lt.thing_action

def grab_as_array(self, portal: lt.deps.BlockingPortal, stream_name: Literal['main', 'lores'] = 'main') -> ArrayModel: (source) ¶

Acquire one image from the preview stream and return as an array.

It works like grab_jpeg but reliably handles broken streams. Prefer using this method over directly grabbing the frame and converting to a numpy array via PIL.

This differs from capture_array in that it does not pause the MJPEG preview stream.

@lt.thing_action

def grab_jpeg(self, portal: lt.deps.BlockingPortal, stream_name: Literal['main', 'lores'] = 'main') -> JPEGBlob: (source) ¶

Acquire one image from the preview stream and return as blob of JPEG data.

Note: in rare cases the JPEG stream may be broken. This can cause an OS error when loading the image. If loading with PIL, as long as the header data is complete, this error will not be raised until the data is accessed. Consider using grab_jpeg_as_array instead.

This differs from capture_jpeg in that it does not pause the MJPEG preview stream. Instead, we simply return the next frame from that stream (either "main" for the preview stream, or "lores" for the low resolution preview). No metadata is returned.

@lt.thing_action

def grab_jpeg_size(self, portal: lt.deps.BlockingPortal, stream_name: Literal['main', 'lores'] = 'main') -> int: (source) ¶

Acquire one image from the preview stream and return its size.

@lt.thing_action

def image_is_sample(self, portal: lt.deps.BlockingPortal) -> tuple[bool, str]: (source) ¶

Label the current image as either background or sample.

def kill_mjpeg_streams(self): (source) ¶

Kill the streams now as the server is shutting down.

This is called when uvicorn gets the a shutdown signal. As this is called from the event loop it cannot interact with the our ThingProperties or run self.mjpeg_stream.stop() as the portal cannot be called from this loop.

Instead we just set the _streaming value to False. This stops the async frame generator when the next frame notifies.

@lt.thing_action

def save_from_memory(self, jpeg_path: str, logger: lt.deps.InvocationLogger, save_resolution: tuple[int, int] | None = None, buffer_id: int | None = None): (source) ¶

Save an image that has been captured to memory.

Parameters
jpeg_path:`str`	The path to save the file to
logger:`lt.deps.InvocationLogger`	This should be injected automatically by Labthings FastAPI when calling the action
save_resolution:`tuple[int, int] \| None`	can be set to resize the image before saving. By default this is None meaning that the image is saved at original resolution.
buffer_id:`int \| None`	The buffer id of the image to save, this was returned by `capture_to_memory`

@lt.thing_action

def set_background(self, portal: lt.deps.BlockingPortal): (source) ¶

Grab an image, and use its statistics to set the background.

This should be run when the microscope is looking at an empty region, and will calculate the mean and standard deviation of the pixel values in the LUV colourspace. These values will then be used to compare future images to the distribution, to determine if each pixel is foreground or background.

@lt.thing_action

def settle(self): (source) ¶

Sleep for the settling time, ready to provide a fresh frame.

This function will sleep for the given time, and clear the buffer after sleeping. As such, the next frame captured from the camera after running this function will always be captured after settling.

This method provides the interface expected by the camera_stage_mapping.

@lt.thing_action

def start_streaming(self, main_resolution: tuple[int, int], buffer_count: int): (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

Start (or stop and restart) the camera.

Parameters
main_resolution:`tuple[int, int]`	the resolution to use for the main stream.
buffer_count:`int`	number of images in the stream buffer.

@lt.thing_action

def update_detector_settings(self, data): (source) ¶

Update the settings of the current detector.

This is an action not a setting/property as the data model depends on the selected detector. As such, it cannot be specified with the necessary precision to be included in a ThingDescription as a setting/property, while retaining enough useful information to communicate to the UI how it is set and read.

The information on how to read the settings is exposed in background_detector_status.

downsampled_array_factor = (source) ¶

The downsampling factor when calling capture_downsampled_array.

lores_mjpeg_stream = (source) ¶

Undocumented

mjpeg_stream = (source) ¶

Undocumented

settling_time = (source) ¶

The settling time when calling the settle() method.

background_detectors = (source) ¶

Undocumented

@property

active_detector: BackgroundDetectAlgorithm = (source) ¶

The active background detector instance.

@lt.thing_property

background_detector_status: BackgroundDetectorStatus = (source) ¶

The status of the active detector for the UI.

@lt.thing_property

manual_camera_settings: list[PropertyControl] = (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

The camera settings to expose as property controls in the settings panel.

@lt.thing_property

primary_calibration_actions: list[ActionButton] = (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2

The calibration actions for both calibration wizard and settings panel.

@lt.thing_property

secondary_calibration_actions: list[ActionButton] = (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2, openflexure_microscope_server.things.camera.simulation.SimulatedCamera

The calibration actions that appear only in settings panel.

@lt.thing_property

stream_active: bool = (source) ¶

Whether the MJPEG stream is active.

def _robust_image_capture(self, metadata_getter: lt.deps.GetThingStates, logger: lt.deps.InvocationLogger) -> Image: (source) ¶

Capture an image in memory and return it with metadata.

This robust capturing method attempts to capture the image five times each time with a 5 second timeout set.

Returns
`Image`	tuple with PIL Image, and dictionary of metadata.

Raises
`CaptureError`	if the capture fails for any reason

def _save_capture(self, jpeg_path: str, image: Image, metadata: dict, logger: lt.deps.InvocationLogger, save_resolution: tuple[int, int] | None = None): (source) ¶

Save the captured image and metadata to disk.

A warning (via InvocationLogger) is raised if metadata is failed to be added

nothing is returned on success

Raises
`IOError`	if the file cannot be saved

_memory_buffer = (source) ¶

Undocumented

_detector_name = (source) ¶

Undocumented