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(thing_server_interface)

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_name`	The name of the active background selector.
Method	`background_detector_name.setter`	Validate and set background_detector_name.
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 as a JPEG.
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	`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.
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.
Property	`background_detector`	The active background detector instance.
Property	`calibration_required`	Whether the camera needs calibrating.
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.
Property	`thing_state`	Empty metadata dict for subclasses to populate.
Method	`_add_metadata_to_capture`	Add the EXIF metadata for a JPEG image.
Method	`_capture_metadata`	Return the metadata for a capture, from the thing states, time and known names.
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	`_all_background_detectors`	Undocumented
Class Variable	`_memory_buffer`	Undocumented
Instance Variable	`_background_detector_name`	Undocumented
Instance Variable	`_default_background_detector`	Undocumented

def __enter__(self) -> 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: type[BaseException], _exc_value: BaseException | None, _traceback: TracebackType | None): (source) ¶

Close hardware connection when the Thing context manager is closed.

def __init__(self, thing_server_interface: lt.ThingServerInterface): (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.setting

def background_detector_name(self) -> str | None: (source) ¶

The name of the active background selector.

@background_detector_name.setter

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

Validate and set background_detector_name.

@lt.action

def capture_and_save(self, jpeg_path: str, 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
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.action

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

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

@lt.action

def capture_downsampled_array(self) -> NDArray: (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', 'full'], wait: float | None = None) -> Image.Image: (source) ¶

Capture a PIL image from stream stream_name with timeout wait.

@lt.action

def capture_jpeg(self, stream_name: Literal['main', 'lores', 'full'] = 'main', wait: float | None = None) -> JPEGBlob: (source) ¶

Acquire one image from the camera as a JPEG.

This will use the internal capture image functionally of capture_image of the specific camera being used.

Parameters
stream_name:`Literal['main', 'lores', 'full']`	A stream name supported by this camera.
wait:`float \| None`	(Optional, float) Set a timeout in seconds. If None it will use the default for the underlying camera.

Returns
`JPEGBlob`	Undocumented

@lt.action

def capture_to_memory(self, 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
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.action

def clear_buffers(self): (source) ¶

Clear all images in memory.

@lt.action

def discard_frames(self): (source) ¶

Discard frames so that the next frame captured is fresh.

@lt.action

def grab_as_array(self, stream_name: Literal['main', 'lores'] = 'main') -> NDArray: (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.action

def grab_jpeg(self, 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.action

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

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

@lt.action

def image_is_sample(self) -> 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.action

def save_from_memory(self, jpeg_path: str, 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
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.action

def set_background(self): (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.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.action

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

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

Start (or stop and restart) the camera.

Note that the default values for both parameters should be set appropriately for the specific camera when defining a new Camera Thing.

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

downsampled_array_factor: int = (source) ¶

The downsampling factor when calling capture_downsampled_array.

lores_mjpeg_stream = (source) ¶

Undocumented

mjpeg_stream = (source) ¶

Undocumented

settling_time: float = (source) ¶

The settling time when calling the settle() method.

@property

background_detector: BackgroundDetectAlgorithm | None = (source) ¶

The active background detector instance.

@lt.property

calibration_required: bool = (source) ¶

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

Whether the camera needs calibrating.

This always returns False in BaseCamera. It should be reimplemented by child classes if calibration is required.

@lt.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.property

primary_calibration_actions: list[ActionButton] = (source) ¶

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

The calibration actions for both calibration wizard and settings panel.

@lt.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.property

stream_active: bool = (source) ¶

Whether the MJPEG stream is active.

@property

thing_state: Mapping[str, Any] = (source) ¶

overridden in openflexure_microscope_server.things.camera.picamera.StreamingPiCamera2

Empty metadata dict for subclasses to populate.

def _add_metadata_to_capture(self, jpeg_path: str, capture_metadata: dict): (source) ¶

Add the EXIF metadata for a JPEG image.

This adds: - UserComment (JSON-encoded metadata from the Things) - Capture time (DateTimeOriginal, DateTimeDigitized, 0th DateTime) - Camera Make and Model

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

Return the metadata for a capture, from the thing states, time and known names.

def _robust_image_capture(self) -> tuple[Image.Image, Mapping[str, Any]]: (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
`tuple[Image.Image, Mapping[str, Any]]`	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.Image, metadata: Mapping[str, Any], save_resolution: tuple[int, int] | None = None): (source) ¶

Save the captured image and metadata to disk.

A warning is logged if metadata cannot be added.

nothing is returned on success

Raises
`IOError`	if the file cannot be saved

_all_background_detectors: Mapping[str, BackgroundDetectAlgorithm] = (source) ¶

Undocumented

_memory_buffer = (source) ¶

Undocumented

_background_detector_name: str | None = (source) ¶

Undocumented

_default_background_detector: str = (source) ¶

Undocumented