Documentation of python magic class methods

This commit is contained in:
Julian Stirling 2025-07-10 16:44:18 +01:00
parent 864ca91e5c
commit e33fecaef0
13 changed files with 119 additions and 32 deletions

View file

@ -41,18 +41,19 @@ class CaptureError(RuntimeError):
class NoImageInMemoryError(RuntimeError):
"""An error called if no image in in memory when an method is called to use that image."""
"""An error called if no image is in memory when accessed."""
class CameraMemoryBuffer:
"""A class that holds images in memory. The images are by default PIL images.
However subclasses of BaseCamera can use this class to store other object types
However subclasses of BaseCamera can use this class to store other object types.
"""
_storage: dict[int, tuple[Any, Optional[dict]]]
def __init__(self):
"""Create the buffer instance."""
# This dictionary is the main store for data. Dictionaries are ordered since
# Python 3.6, so the order in the dictionary is the capture order
self._storage = {}
@ -142,16 +143,22 @@ class CameraMemoryBuffer:
class BaseCamera(lt.Thing):
"""The base class for all cameras. All cameras must directly inherit from this class."""
"""The base class for all cameras. All cameras must directly inherit from this class.
The connection to the camera hardware should be added to the ``__enter__`` method not
``__init__`` method of the subclass.
"""
mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
_memory_buffer = CameraMemoryBuffer()
def __enter__(self) -> None:
"""Open hardware connection when the Thing context manager is opened."""
raise NotImplementedError("CameraThings must define their own __enter__ method")
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
"""Close hardware connection when the Thing context manager is closed."""
raise NotImplementedError("CameraThings must define their own __exit__ method")
@lt.thing_action

View file

@ -23,14 +23,19 @@ from . import BaseCamera, JPEGBlob
class OpenCVCamera(BaseCamera):
"""A Thing representing an OpenCV camera."""
"""A Thing that provides and interface to an OpenCV Camera."""
def __init__(self, camera_index: int = 0):
"""Iniatilise the thing storing the index of the camera to use.
:param camera_index: The index of the camera to use for the microscope.
"""
self.camera_index = camera_index
self._capture_thread: Optional[Thread] = None
self._capture_enabled = False
def __enter__(self):
"""Start the capture thread when the Thing context manager is opened."""
self.cap = cv2.VideoCapture(self.camera_index)
self._capture_enabled = True
self._capture_thread = Thread(target=self._capture_frames)
@ -38,6 +43,10 @@ class OpenCVCamera(BaseCamera):
return self
def __exit__(self, _exc_type, _exc_value, _traceback):
"""Release the camera when the Thing context manager is closed.
Before releasing the camera the capture thread is closed.
"""
if self.stream_active:
self._capture_enabled = False
self._capture_thread.join()

View file

@ -104,9 +104,20 @@ class LensShading(BaseModel):
class StreamingPiCamera2(BaseCamera):
"""A Thing that represents an OpenCV camera."""
"""A Thing that provides and interface to the Raspberry Pi Camera.
Currently the Thing only supports the PiCamera v2 board. This needs
generalisation.
"""
def __init__(self, camera_num: int = 0):
"""Initialise the camera with the given camera number.
This makes no connection to the camera (except to get the default tuning file).
:param camera_num: The number of the camera. This should generally be left as 0
as most Raspberry Pi boards only support 1 camera.
"""
self._setting_save_in_progress = False
self.camera_num = camera_num
self.camera_configs: dict[str, dict] = {}
@ -324,6 +335,11 @@ class StreamingPiCamera2(BaseCamera):
self._picamera_lock = RLock()
def __enter__(self):
"""Start streaming when the Thing context manager is opened.
This opens the picamera connection, initialises the camera, sets the
sensor_modes property, and then starts the streams.
"""
self._initialise_picamera()
# populate sensor modes by reading the property
self.sensor_modes
@ -360,7 +376,7 @@ class StreamingPiCamera2(BaseCamera):
self.start_streaming()
def __exit__(self, exc_type, exc_value, traceback):
# Shut down the camera
"""Close the picamera connection when the Thing context manager is closed."""
self.stop_streaming()
with self._streaming_picamera() as cam:
cam.close()

View file

@ -30,7 +30,7 @@ RATIO = 0.2
class SimulatedCamera(BaseCamera):
"""A Thing representing an OpenCV camera."""
"""A Thing that simulates a camera for testing."""
_stage: Optional[BaseStage] = None
_server: Optional[lt.ThingServer] = None
@ -42,6 +42,16 @@ class SimulatedCamera(BaseCamera):
canvas_shape: tuple[int, int, int] = (3000, 4000, 3),
frame_interval: float = 0.1,
):
"""Initalise the simulated with settings for how images are generated.
:param shape: The shape (size) of the generated image.
:param glyph_shape: The size randomly positioned glyphs.
:param canvas_shape: The shape (size) of the canvas generated on initialisation
that images are cropped from. If this is too large the it uses resources,
but its size limits the range of motion of the simulation.
:param frame_interval: Nominally the time between frames on the MJPEG stream,
however the rate may be slower due to calculation time for focus.
"""
self.shape = shape
self.glyph_shape = glyph_shape
self.canvas_shape = canvas_shape
@ -143,12 +153,14 @@ class SimulatedCamera(BaseCamera):
return self.generate_image((pos["y"], pos["x"], pos["z"]))
def __enter__(self):
"""Start the capture thread when the Thing context manager is opened."""
self._capture_enabled = True
self._capture_thread = Thread(target=self._capture_frames)
self._capture_thread.start()
return self
def __exit__(self, _exc_type, _exc_value, _traceback):
"""Close the capture thread when the Thing context manager is closed."""
if self.stream_active:
self._capture_enabled = False
self._capture_thread.join()