Merge branch 'camera-snapshot' into 'v3'

Provide an API endpoint for camera snapshots rather than relying on legacy.

Closes #557

See merge request openflexure/openflexure-microscope-server!619

src/openflexure_microscope_server/server/legacy_api.py

@@ -1,4 +1,4 @@
-"""Provide endpoints that mimic the v2 API for OpenFlexure Connect discoverability."""
+"""Provide endpoints that mimic the v2 API for legacy discoverability."""
 
 from socket import gethostname
 
@@ -22,12 +22,14 @@ class JPEGResponse(Response):
 
 
 def add_v2_endpoints(thing_server: lt.ThingServer) -> None:
-    """Add the v2 API endpoints for OpenFlexure Connect discoverability."""
+    """Add the v2 API endpoints for legacy discoverability.
+
+    Old versions of OpenFlexure Connect use the v2 endpoints to check for microscopes.
+    This provides these routes so the microscope can be discovered by old versions of
+    connect.
+    """
     app = thing_server.app
 
-    # The endpoints below fool OpenFlexure Connect into thinking we are a
-    # v2 microscope, so we show up correctly.
-    # This is necessary until Connect is rebuilt. See #557.
     @app.get("/routes")
     def routes_stub() -> dict[str, dict]:
         """Return a stub list of routes.

src/openflexure_microscope_server/things/camera/__init__.py

@@ -20,6 +20,7 @@ from typing import Any, Literal, Mapping, Optional, Self
 
 import numpy as np
 import piexif
+from fastapi import Response
 from PIL import Image
 from pydantic import BaseModel
 
@@ -259,6 +260,21 @@ class BaseCamera(OFMThing, ABC):
         """Close hardware connection when the Thing context manager is closed."""
         pass
 
+    @lt.endpoint(
+        "get",
+        "snapshot",
+        responses={
+            200: {
+                "description": "A snapshot of the microscope stream",
+                "content": {"image/jpeg": {}},
+            },
+        },
+    )
+    async def snapshot(self) -> Response:
+        """Return a snapshot from the microscope."""
+        jpeg_data = await self.lores_mjpeg_stream.grab_frame()
+        return Response(content=jpeg_data, media_type="image/jpeg")
+
     @property
     def focus_fom(self) -> int:
         """Return the focus figure of merit.

tests/unit_tests/test_simulated_camera.py

@@ -2,11 +2,13 @@
 
 import logging
 import time
+from io import BytesIO
 
 import numpy as np
 import pytest
 from hypothesis import given
 from hypothesis import strategies as st
+from PIL import Image
 from pydantic import ValidationError
 
 import labthings_fastapi as lt
@@ -44,6 +46,19 @@ def stage(test_env) -> lt.Thing:
     return test_env.get_thing_by_type(DummyStage)
 
 
+def test_snapshot(test_env) -> lt.Thing:
+    """Check the snapshot GET request returns an image."""
+    http_client = test_env.get_thing_client("camera").client
+    response = http_client.get("camera/snapshot")
+    response.raise_for_status()
+    assert response.headers["content-type"] == "image/jpeg"
+    image = Image.open(BytesIO(response.content))
+
+    assert image.format == "JPEG"
+    assert image.width == 320
+    assert image.height == 240
+
+
 def test_downsample_shape_2d():
     """Test downsampling for 2D array."""
     shape_2d = (100, 80)