Improve __enter__ typing for things

src/openflexure_microscope_server/things/camera/__init__.py

@@ -15,7 +15,7 @@ import tempfile
 import time
 from datetime import datetime
 from types import TracebackType
-from typing import Any, Literal, Mapping, Optional, Tuple
+from typing import Any, Literal, Mapping, Optional, Self, Tuple
 
 import numpy as np
 import piexif
@@ -180,7 +180,7 @@ class BaseCamera(lt.Thing):
         }
         self._detector_name = "Channel Deviations (LUV)"
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Open hardware connection when the Thing context manager is opened."""
         raise NotImplementedError("CameraThings must define their own __enter__ method")
 

src/openflexure_microscope_server/things/camera/opencv.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import logging
 from threading import Thread
 from types import TracebackType
-from typing import Literal, Optional
+from typing import Literal, Optional, Self
 
 import cv2
 from PIL import Image
@@ -39,7 +39,7 @@ class OpenCVCamera(BaseCamera):
         self._capture_thread: Optional[Thread] = None
         self._capture_enabled = False
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Start the capture thread when the Thing context manager is opened."""
         self.cap = cv2.VideoCapture(self.camera_index)
         self._capture_enabled = True
@@ -59,7 +59,8 @@ class OpenCVCamera(BaseCamera):
         """
         if self.stream_active:
             self._capture_enabled = False
-            self._capture_thread.join()
+            if self._capture_thread is not None:
+                self._capture_thread.join()
         self.cap.release()
 
     @lt.property

src/openflexure_microscope_server/things/camera/picamera.py

@@ -25,7 +25,7 @@ import time
 from contextlib import contextmanager
 from threading import RLock
 from types import TracebackType
-from typing import Annotated, Any, Iterator, Literal, Mapping, Optional
+from typing import Annotated, Any, Iterator, Literal, Mapping, Optional, Self
 
 import numpy as np
 from picamera2 import Picamera2
@@ -148,7 +148,7 @@ class StreamingPiCamera2(BaseCamera):
                 f"are {SUPPORTED_CAMS_SENSOR_INFO.keys()}."
             )
         self._sensor_info = SUPPORTED_CAMS_SENSOR_INFO[camera_board]
-        self._picamera_lock = None
+        self._picamera_lock = RLock()
         self._picamera = None
 
         # Load the tuning file for the specified sensor mode.
@@ -340,10 +340,7 @@ class StreamingPiCamera2(BaseCamera):
         :raises PicameraModelError: If check_sensor_model is True and the real
             camera sensor model doesn't match the expected sensor model.
         """
-        if self._picamera_lock is not None:
+        with self._picamera_lock, tempfile.NamedTemporaryFile("w") as tuning_file:
-            # Don't close the camera if it's in use
-            self._picamera_lock.acquire()
-        with tempfile.NamedTemporaryFile("w") as tuning_file:
             json.dump(self.tuning, tuning_file)
             tuning_file.flush()  # but leave it open as closing it will delete it
             os.environ["LIBCAMERA_RPI_TUNING_FILE"] = tuning_file.name
@@ -371,9 +368,8 @@ class StreamingPiCamera2(BaseCamera):
                         f"Wrong Picamera model. Expecting {self._sensor_info.sensor_model}, "
                         f"but found {hw_sensor_model}."
                     )
-        self._picamera_lock = RLock()
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Start streaming when the Thing context manager is opened.
 
         This opens the picamera connection, initialises the camera, sets the

src/openflexure_microscope_server/things/camera/simulation.py

@@ -13,7 +13,7 @@ import logging
 import time
 from threading import Thread
 from types import TracebackType
-from typing import Literal, Optional
+from typing import Literal, Optional, Self
 
 import numpy as np
 from PIL import Image, ImageFilter
@@ -239,7 +239,7 @@ class SimulatedCamera(BaseCamera):
             pos = {"x": 0, "y": 0, "z": 0}
         return self.generate_image((pos["y"], pos["x"], pos["z"]))
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Start the capture thread when the Thing context manager is opened."""
         self.start_streaming()
         return self

src/openflexure_microscope_server/things/stage/dummy.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import time
 from collections.abc import Mapping
 from types import TracebackType
-from typing import Any, Optional
+from typing import Any, Optional, Self
 
 import labthings_fastapi as lt
 
@@ -36,9 +36,10 @@ class DummyStage(BaseStage):
         self.step_time = step_time
         self.instantaneous_position = self._hardware_position
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Register the stage position when the Thing context manager is opened."""
         self.instantaneous_position = self._hardware_position
+        return self
 
     def __exit__(
         self,

src/openflexure_microscope_server/things/stage/sangaboard.py

@@ -8,7 +8,7 @@ from collections.abc import Mapping
 from contextlib import contextmanager
 from copy import copy
 from types import TracebackType
-from typing import Any, Iterator, Literal, Optional
+from typing import Any, Iterator, Literal, Optional, Self
 
 import semver
 
@@ -33,7 +33,7 @@ class SangaboardThing(BaseStage):
     def __init__(
         self,
         thing_server_interface: lt.ThingServerInterface,
-        port: str = None,
+        port: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
         """Initialise SangaboardThing.
@@ -53,13 +53,14 @@ class SangaboardThing(BaseStage):
         self._sangaboard_lock = threading.RLock()
         super().__init__(thing_server_interface, **kwargs)
 
-    def __enter__(self) -> None:
+    def __enter__(self) -> Self:
         """Connect to the sangaboard when the Thing context manager is opened."""
         self._sangaboard = sangaboard.Sangaboard(**self.sangaboard_kwargs)
         with self.sangaboard() as sb:
             sb.query("blocking_moves false")
         self.check_firmware()
         self.update_position()
+        return self
 
     def __exit__(
         self,