reachy2_sdk.media.camera_manager

  1"""Reachy Camera Manager module.
  2
  3Initialize the head and torso cameras if they are available.
  4"""
  5
  6import logging
  7from typing import Optional
  8
  9import grpc
 10from google.protobuf.empty_pb2 import Empty
 11from reachy2_sdk_api.video_pb2_grpc import VideoServiceStub
 12
 13from .camera import Camera, CameraType, DepthCamera
 14
 15
 16class CameraManager:
 17    """CameraManager class manages the available cameras on the robot.
 18
 19    Provides access to the robot's cameras, including teleoperation and depth cameras.
 20    It handles the initialization of cameras and offers methods to retrieve camera objects for use.
 21    """
 22
 23    def __init__(self, host: str, port: int) -> None:
 24        """Set up the camera manager module.
 25
 26        This initializes the gRPC channel for communicating with the camera service,
 27        sets up logging, and prepares the available cameras.
 28
 29        Args:
 30            host: The host address for the gRPC service.
 31            port: The port number for the gRPC service.
 32        """
 33        self._logger = logging.getLogger(__name__)
 34        self._grpc_video_channel = grpc.insecure_channel(f"{host}:{port}")
 35        self._host = host
 36
 37        self._video_stub = VideoServiceStub(self._grpc_video_channel)
 38
 39        self._teleop: Optional[Camera] = None
 40        self._depth: Optional[DepthCamera] = None
 41        self._setup_cameras()
 42
 43    def __repr__(self) -> str:
 44        """Clean representation of a reachy cameras."""
 45        s = "\n\t".join([str(cam) for cam in [self._depth, self._teleop] if cam is not None])
 46        return f"""<CameraManager intialized_cameras=\n\t{s}\n>"""
 47
 48    def _setup_cameras(self) -> None:
 49        """Initialize cameras based on availability.
 50
 51        This method retrieves the available cameras and sets
 52        up the teleop and depth cameras if they are found.
 53        """
 54        cams = self._video_stub.GetAvailableCameras(Empty())
 55        self._teleop = None
 56        self._depth = None
 57        if len(cams.camera_feat) == 0:
 58            self._logger.warning("There is no available camera.")
 59        else:
 60            self._logger.debug(cams.camera_feat)
 61            for c in cams.camera_feat:
 62                if c.name == CameraType.TELEOP.value:
 63                    self._logger.debug("Teleop Camera initialized.")
 64                    self._teleop = Camera(c, self._video_stub)
 65                elif c.name == CameraType.DEPTH.value:
 66                    self._logger.debug("Depth Camera initialized.")
 67                    self._depth = DepthCamera(c, self._video_stub)
 68                else:
 69                    self._logger.error(f"Camera {c.name} not defined")
 70
 71    def initialize_cameras(self) -> None:
 72        """Manually re-initialize cameras.
 73
 74        This method can be used to reinitialize the camera setup if changes occur
 75        or new cameras are connected.
 76        """
 77        self._setup_cameras()
 78
 79    @property
 80    def teleop(self) -> Optional[Camera]:
 81        """Retrieve the teleop camera.
 82
 83        Returns:
 84            The teleop Camera object if it is initialized; otherwise, logs an error
 85            and returns None.
 86        """
 87        if self._teleop is None:
 88            self._logger.error("Teleop camera is not initialized.")
 89            return None
 90
 91        return self._teleop
 92
 93    @property
 94    def depth(self) -> Optional[DepthCamera]:
 95        """Retrieve the depth camera.
 96
 97        Returns:
 98            The DepthCamera object if it is initialized; otherwise, logs an error
 99            and returns None.
100        """
101        if self._depth is None:
102            self._logger.error("Depth camera is not initialized.")
103            return None
104
105        return self._depth