Formal Protocol for camera interface

I've now split the base camera code into a runtime-checkable Protocol
and a concrete base class (which is optional). This removes the requirement
for cameras to subclass `openflexure_microscope.things.camera.Camera`
and instead uses duck typing that should
work both statically and dynamically.

As part of this, I've moved the commonly-used dependencies
on the camera inside `openflexure_microscope.things.camera` for consistency, and to make it easy to update them if it changes in the future.

src/openflexure_microscope_server/things/auto_recentre_stage.py

@@ -5,12 +5,10 @@ from labthings_fastapi.thing import Thing
 from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
 from labthings_fastapi.decorators import thing_action
 from .stage import Stage
-from .camera import Camera
 from openflexure_microscope_server.things.autofocus import AutofocusThing
 from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
 
 StageDep = direct_thing_client_dependency(Stage, "/stage/")
-CamDep = direct_thing_client_dependency(Camera, "/camera/")
 CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
 AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
 

src/openflexure_microscope_server/things/autofocus.py

@@ -19,14 +19,13 @@ from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
 from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
 from labthings_fastapi.decorators import thing_action
 from labthings_fastapi.types.numpy import NDArray
-from .camera import Camera as CameraThing
+from .camera import RawCameraDependency as Camera
+from .camera import CameraDependency as WrappedCamera
 from .stage import Stage as StageThing
 import numpy as np
 from pydantic import BaseModel
 
 Stage = direct_thing_client_dependency(StageThing, "/stage/")
-Camera = raw_thing_dependency(CameraThing)
-WrappedCamera = direct_thing_client_dependency(CameraThing, "/camera/")
 
 ### Autofocus utilities
 

src/openflexure_microscope_server/things/camera/__init__.py

@@ -7,54 +7,89 @@ See repository root for licensing information.
 """
 from __future__ import annotations
 import logging
-from typing import Literal
+from typing import Literal, Protocol, TypeAlias, runtime_checkable
 
 from labthings_fastapi.thing import Thing
-from labthings_fastapi.decorators import thing_action, thing_property
+from labthings_fastapi.decorators import thing_action
 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.outputs.mjpeg_stream import MJPEGStreamDescriptor
 from labthings_fastapi.outputs.blob import blob_type
 from labthings_fastapi.types.numpy import NDArray
 
 
-JPEGBlob = blob_type("image/jpeg")
+JPEGBlob: TypeAlias = blob_type("image/jpeg")
 
-class CameraMeta:
+@runtime_checkable
-    def __subclasscheck__(cls, C):
+class CameraProtocol(Protocol):
-        """Override Python's default behaviour and use duck typing"""
-        print(f"Checking if {C} is a camera...")
-        for action in ["snap_image", "capture_array", "capture_jpeg", "grab_jpeg", "grab_jpeg_size"]:
-            a = getattr(C, action, None)
-            if not callable(a):
-                return False
-        for prop in ["stream_active"]:
-            if not hasattr(C, prop):
-                return False
-        for k in ["mjpeg_stream", "lores_mjpeg_stream"]:
-            stream = getattr(C, k)
-            if not isinstance(stream, MJPEGStreamDescriptor):
-                return False
-        return True
-
-    def __instancecheck__(cls, I):
-        return cls.__subclasscheck__(I)
-
-
-class Camera(Thing):
-    __metaclass__ = CameraMeta
     """A Thing representing a camera"""
 
+    def __enter__(self) -> None:
+        ...
+    
+    def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
+        ...
+
+    @property
+    def stream_active(self) -> bool:
+        "Whether the MJPEG stream is active"
+        ...
+
+    def snap_image(self) -> NDArray:
+        """Acquire one image from the camera."""
+        ...
+
+    def capture_array(
+        self,
+        resolution: Literal["lores", "main", "full"] = "main",
+    ) -> NDArray:
+        ...
+    
+    def capture_jpeg(
+        self,
+        metadata_getter: GetThingStates,
+        resolution: Literal["lores", "main", "full"] = "main",
+    ) -> JPEGBlob:
+        """Acquire one image from the camera and return as a JPEG blob"""
+        ...
+
+    def grab_jpeg(
+        self,
+        portal: BlockingPortal,
+        stream_name: Literal["main", "lores"] = "main",
+    ) -> JPEGBlob:
+        """Acquire one image from the preview stream and return as an array
+
+        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.
+        """
+        ...
+
+    def grab_jpeg_size(
+        self,
+        portal: BlockingPortal,
+        stream_name: Literal["main", "lores"] = "main",
+    ) -> int:
+        """Acquire one image from the preview stream and return its size"""
+        ...
+
+class BaseCamera(Thing):
+    """A Thing representing a camera
+    
+    This is a concrete base class for `Thing`s implementing the `CameraProtocol`.
+    It provides the stream descriptors and actions to grab from the stream.
+    """
+
     def __enter__(self):
         raise NotImplementedError("Subclasses must implement __enter__")
     
     def __exit__(self, _exc_type, _exc_value, _traceback):
         raise NotImplementedError("Subclasses must implement __exit__")
 
-    @thing_property
-    def stream_active(self) -> bool:
-        "Whether the MJPEG stream is active"
-        raise NotImplementedError("Subclasses must implement stream_active")
     mjpeg_stream = MJPEGStreamDescriptor()
     lores_mjpeg_stream = MJPEGStreamDescriptor()
 
@@ -67,31 +102,6 @@ class Camera(Thing):
         """
         return self.capture_array()
 
-    @thing_action
-    def capture_array(
-        self,
-        resolution: Literal["lores", "main", "full"] = "main",
-    ) -> NDArray:
-        """Acquire one image from the camera and return as an array
-
-        This function will produce a nested list containing an uncompressed RGB image.
-        It's likely to be highly inefficient - raw and/or uncompressed captures using
-        binary image formats will be added in due course.
-        """
-        raise NotImplementedError("Subclasses must implement capture_array")
-    
-    @thing_action
-    def capture_jpeg(
-        self,
-        metadata_getter: GetThingStates,
-        resolution: Literal["lores", "main", "full"] = "main",
-    ) -> JPEGBlob:
-        """Acquire one image from the camera and return as a JPEG blob
-
-        This function will produce a JPEG image.
-        """
-        raise NotImplementedError("Subclasses must implement capture_jpeg")
-
     @thing_action
     def grab_jpeg(
         self,
@@ -125,3 +135,6 @@ class Camera(Thing):
         )
         return portal.call(stream.next_frame_size)
 
+
+CameraDependency = direct_thing_client_dependency(CameraProtocol, "/camera/")
+RawCameraDependency = raw_thing_dependency(CameraProtocol)

src/openflexure_microscope_server/things/camera/opencv.py

@@ -21,10 +21,10 @@ from labthings_fastapi.dependencies.metadata import GetThingStates
 from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
 from labthings_fastapi.types.numpy import NDArray
 
-from . import Camera, JPEGBlob
+from . import BaseCamera, JPEGBlob
 
 
-class OpenCVCamera(Camera):
+class OpenCVCamera(BaseCamera):
     """A Thing representing an OpenCV camera"""
     def __init__(self, camera_index: int=0):
         self.camera_index = camera_index

src/openflexure_microscope_server/things/camera/simulation.py

@@ -24,11 +24,11 @@ from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
 from labthings_fastapi.types.numpy import NDArray
 from labthings_fastapi.server import ThingServer
 
-from . import Camera, JPEGBlob
+from . import BaseCamera, JPEGBlob
 from ..stage import Stage
 
 
-class SimulatedCamera(Camera):
+class SimulatedCamera(BaseCamera):
     """A Thing representing an OpenCV camera"""
     shape = (600, 800, 3)
     glyph_shape = (51, 51, 3)

src/openflexure_microscope_server/things/camera_stage_mapping.py

@@ -22,7 +22,7 @@ from camera_stage_mapping.camera_stage_calibration_1d import (
     image_to_stage_displacement_from_1d,
 )
 from camera_stage_mapping.camera_stage_tracker import Tracker
-from .camera import Camera as CameraThing
+from .camera import CameraDependency as Camera
 from .stage import Stage as StageThing
 
 from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
@@ -31,7 +31,6 @@ from labthings_fastapi.types.numpy import NDArray, denumpify, DenumpifyingDict
 from labthings_fastapi.decorators import thing_action, thing_property
 from labthings_fastapi.thing import Thing
 
-Camera = direct_thing_client_dependency(CameraThing, "/camera/")
 Stage = direct_thing_client_dependency(StageThing, "/stage/")
 
 CoordinateType = Tuple[float, float, float]

src/openflexure_microscope_server/things/smart_scan.py

@@ -28,14 +28,13 @@ 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
-from .camera import Camera
+from .camera import CameraDependency as CamDep
 from .stage import Stage
 from openflexure_microscope_server.things.autofocus import AutofocusThing
 from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
 from openflexure_microscope_server.things.auto_recentre_stage import RecentringThing
 
 
-CamDep = direct_thing_client_dependency(Camera, "/camera/")
 StageDep = direct_thing_client_dependency(Stage, "/stage/")
 CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
 AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")