Merge branch 'opencv-camera-switcher' into 'v3'
Add functionality to switch OpenCV camera without restarting the server. See merge request openflexure/openflexure-microscope-server!487
This commit is contained in:
Julian Stirling
commit
ae8285ed28
4 changed files with 158 additions and 19 deletions
|
|
@ -19,7 +19,9 @@ from PIL import Image
|
|||
import labthings_fastapi as lt
|
||||
from labthings_fastapi.types.numpy import NDArray
|
||||
|
||||
from . import BaseCamera
|
||||
from openflexure_microscope_server.ui import PropertyControl, property_control_for
|
||||
|
||||
from . import BaseCamera, opencv_utils
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -27,25 +29,31 @@ LOGGER = logging.getLogger(__name__)
|
|||
class OpenCVCamera(BaseCamera):
|
||||
"""A Thing that provides and interface to an OpenCV Camera."""
|
||||
|
||||
def __init__(
|
||||
self, thing_server_interface: lt.ThingServerInterface, camera_index: int = 0
|
||||
) -> None:
|
||||
def __init__(self, thing_server_interface: lt.ThingServerInterface) -> None:
|
||||
"""Iniatilise the thing storing the index of the camera to use.
|
||||
|
||||
:param camera_index: The index of the camera to use for the microscope.
|
||||
:param thing_server_interface: The thing server interface to be passed to to
|
||||
the parent class.
|
||||
"""
|
||||
super().__init__(thing_server_interface)
|
||||
self.camera_index = camera_index
|
||||
|
||||
self.cameras: dict[str, int] = {}
|
||||
self._capture_thread: Optional[Thread] = None
|
||||
self._capture_enabled = False
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
"""Start the capture thread when the Thing context manager is opened."""
|
||||
super().__enter__()
|
||||
self.cap = cv2.VideoCapture(self.camera_index)
|
||||
self._capture_enabled = True
|
||||
self._capture_thread = Thread(target=self._capture_frames)
|
||||
self._capture_thread.start()
|
||||
all_camera_ids = opencv_utils.find_all_cameras()
|
||||
if not all_camera_ids:
|
||||
raise RuntimeError("No cameras available.")
|
||||
self.cameras = opencv_utils.identify_cameras(all_camera_ids)
|
||||
if self.camera_name not in self.cameras:
|
||||
if self.camera_name:
|
||||
self.logger.warning(f"{self.camera_name} not found.")
|
||||
self._camera_name = next(iter(self.cameras))
|
||||
|
||||
self._start_stream()
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
|
|
@ -58,12 +66,49 @@ class OpenCVCamera(BaseCamera):
|
|||
|
||||
Before releasing the camera the capture thread is closed.
|
||||
"""
|
||||
self._stop_stream()
|
||||
super().__exit__(exc_type, exc_value, traceback)
|
||||
|
||||
_camera_name = ""
|
||||
|
||||
@lt.setting
|
||||
def camera_name(self) -> str:
|
||||
"""The name of the camera."""
|
||||
return self._camera_name
|
||||
|
||||
@camera_name.setter
|
||||
def _set_camera_name(self, value: str) -> None:
|
||||
"""Set the name of the camera."""
|
||||
if not self.cameras:
|
||||
# Don't try to validate if cameras dict is empty, just set the value.
|
||||
# As this is the startup behaviour, on __enter__ we check if the
|
||||
# initial camera is valid and that at least 1 camera exists.
|
||||
self._camera_name = value
|
||||
# Return so we don't try to start the stream
|
||||
return
|
||||
self._camera_name = value
|
||||
if value not in self.cameras:
|
||||
raise ValueError(f"{value} is not a valid camera name.")
|
||||
self._start_stream()
|
||||
|
||||
def _start_stream(self) -> None:
|
||||
"""Start the camera stream or restart if running."""
|
||||
if self.stream_active:
|
||||
self._stop_stream()
|
||||
self.cap = cv2.VideoCapture(
|
||||
self.cameras[self.camera_name], opencv_utils.BACKEND
|
||||
)
|
||||
self._capture_enabled = True
|
||||
self._capture_thread = Thread(target=self._capture_frames)
|
||||
self._capture_thread.start()
|
||||
|
||||
def _stop_stream(self) -> None:
|
||||
"""Stop the camera stream."""
|
||||
if self.stream_active:
|
||||
self._capture_enabled = False
|
||||
if self._capture_thread is not None:
|
||||
self._capture_thread.join()
|
||||
self.cap.release()
|
||||
super().__exit__(exc_type, exc_value, traceback)
|
||||
|
||||
@lt.property
|
||||
def stream_active(self) -> bool:
|
||||
|
|
@ -76,7 +121,7 @@ class OpenCVCamera(BaseCamera):
|
|||
while self._capture_enabled:
|
||||
ret, frame = self.cap.read()
|
||||
if not ret:
|
||||
LOGGER.error(f"Failed to capture frame from camera {self.camera_index}")
|
||||
LOGGER.error("Failed to capture frame from camera.")
|
||||
break
|
||||
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
|
||||
self.mjpeg_stream.add_frame(jpeg)
|
||||
|
|
@ -107,9 +152,7 @@ class OpenCVCamera(BaseCamera):
|
|||
LOGGER.warning(f"OpenCV camera doesn't respect {stream_name=}")
|
||||
ret, frame = self.cap.read()
|
||||
if not ret:
|
||||
raise RuntimeError(
|
||||
f"Failed to capture frame from camera {self.camera_index}"
|
||||
)
|
||||
raise RuntimeError("Failed to capture frame from camera.")
|
||||
return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
||||
|
||||
def capture_image(
|
||||
|
|
@ -125,3 +168,19 @@ class OpenCVCamera(BaseCamera):
|
|||
LOGGER.warning("OpenCV camera has no wait option. Use None.")
|
||||
LOGGER.warning(f"OpenCV camera doesn't respect {stream_name=}")
|
||||
return Image.fromarray(self.capture_array())
|
||||
|
||||
@lt.property
|
||||
def manual_camera_settings(self) -> list[PropertyControl]:
|
||||
"""The camera settings to expose as property controls in the settings panel.
|
||||
|
||||
The options for the camera selector are populated with camera names once the
|
||||
server starts and available cameras are have been detected.
|
||||
"""
|
||||
return [
|
||||
property_control_for(
|
||||
self,
|
||||
"camera_name",
|
||||
label="Camera",
|
||||
options={key: key for key in self.cameras},
|
||||
),
|
||||
]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,80 @@
|
|||
"""Utilities for managing CV2 cameras."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from typing import Callable, Optional
|
||||
|
||||
import cv2
|
||||
|
||||
try:
|
||||
from cv2_enumerate_cameras.camera_info import CameraInfo
|
||||
|
||||
enumerate_cameras: Optional[Callable[[int], list[CameraInfo]]]
|
||||
# MyPy thinks enumerate_cameras is being redefined. We just needed to set the type
|
||||
# before defining.
|
||||
from cv2_enumerate_cameras import enumerate_cameras # type: ignore[no-redef]
|
||||
except ModuleNotFoundError:
|
||||
enumerate_cameras = None
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
BACKEND = cv2.CAP_DSHOW
|
||||
elif sys.platform.startswith("linux"):
|
||||
BACKEND = cv2.CAP_V4L2
|
||||
elif sys.platform == "darwin":
|
||||
BACKEND = cv2.CAP_AVFOUNDATION
|
||||
else:
|
||||
raise RuntimeError(f"Unsupported platform {sys.platform}")
|
||||
|
||||
# Max cameras is set to balance finding all cameras with not taking too long to start up.
|
||||
# Note that due to extra camera modes in Linux 1 camera can take up multiple camera numbers.
|
||||
MAX_CAMERAS = 12
|
||||
|
||||
|
||||
def find_all_cameras() -> list[int]:
|
||||
"""Find all accessible USB cameras on the device."""
|
||||
available = []
|
||||
for i in range(MAX_CAMERAS):
|
||||
cap = cv2.VideoCapture(i, BACKEND)
|
||||
|
||||
if cap.isOpened():
|
||||
available.append(i)
|
||||
cap.release()
|
||||
return available
|
||||
|
||||
|
||||
def identify_cameras(camera_ids: list[int]) -> dict[str, int]:
|
||||
"""For a list of camera IDs return a dictionary of name -> ID."""
|
||||
# When first creating the mapping with default names it goes from
|
||||
# id -> camera name. This makes it easy to replace the name (based on a fixed camera
|
||||
# id as the key) if the name is then found.
|
||||
# Before returning this is swapped to be a mapping from camera name -> id. As this
|
||||
# is what is needed to switch the cameras based on a name.
|
||||
name_dict = {n: f"Unknown Camera {n}" for n in camera_ids}
|
||||
|
||||
# enumerate cameras works for all backends if it is installed:
|
||||
if enumerate_cameras is not None:
|
||||
for camera_info in enumerate_cameras(BACKEND):
|
||||
if camera_info.index in name_dict:
|
||||
name_dict[camera_info.index] = camera_info.name
|
||||
elif BACKEND == cv2.CAP_V4L2:
|
||||
# If Linux, try to read video4linux name as a fallback option.
|
||||
for camera_id in camera_ids:
|
||||
try:
|
||||
if not isinstance(camera_id, int):
|
||||
raise TypeError("Camera ID must be an integer.")
|
||||
# Linux only so just use direct path.
|
||||
path = f"/sys/class/video4linux/video{camera_id}/name"
|
||||
with open(path, "r") as f_obj:
|
||||
name_dict[camera_id] = f_obj.read().strip()
|
||||
except IOError:
|
||||
pass
|
||||
else:
|
||||
LOGGER.warning(
|
||||
"Cannot determine camera names. Please install the Optional `manual` "
|
||||
"dependencies."
|
||||
)
|
||||
|
||||
# Swap order for return
|
||||
return {name: cam_id for cam_id, name in name_dict.items()}
|
||||
Loading…
Add table
Add a link
Reference in a new issue