Add endpoint for stream configs

src/isar/apis/api.py

@@ -17,6 +17,7 @@
 from pydantic import AnyHttpUrl
 
 from isar.apis.models.models import ControlMissionResponse, StartMissionResponse
+from isar.apis.robot_control.robot_controller import RobotController
 from isar.apis.schedule.scheduling_controller import SchedulingController
 from isar.apis.security.authentication import Authenticator
 from isar.config.configuration_error import ConfigurationError
@@ -34,12 +35,14 @@ def __init__(
         self,
         authenticator: Authenticator,
         scheduling_controller: SchedulingController,
+        robot_controller: RobotController,
         keyvault_client: Keyvault,
         port: int = settings.API_PORT,
         azure_ai_logging_enabled: bool = settings.LOG_HANDLER_APPLICATION_INSIGHTS_ENABLED,
     ) -> None:
         self.authenticator: Authenticator = authenticator
         self.scheduling_controller: SchedulingController = scheduling_controller
+        self.robot_controller: RobotController = robot_controller
         self.keyvault_client: Keyvault = keyvault_client
         self.host: str = "0.0.0.0"  # Locking uvicorn to use 0.0.0.0
         self.port: int = port
@@ -98,6 +101,8 @@ def _create_app(self) -> FastAPI:
 
         app.include_router(router=self._create_info_router())
 
+        app.include_router(router=self._create_media_control_router())
+
         return app
 
     def _create_scheduler_router(self) -> APIRouter:
@@ -284,6 +289,21 @@ def _create_info_router(self) -> APIRouter:
         )
 
         return router
+
+    def _create_media_control_router(self) -> APIRouter:
+        router: APIRouter = APIRouter(tags=["Media"])
+
+        authentication_dependency: Security = Security(self.authenticator.get_scheme())
+
+        router.add_api_route(
+            path="/media/media-stream-config",
+            endpoint=self.robot_controller.generate_media_stream_config,
+            methods=["GET"],
+            dependencies=[authentication_dependency],
+            summary="Generates a media stream connection config",
+        )
+
+        return router
 
     def _log_startup_message(self) -> None:
         address_format = "%s://%s:%d/docs"

src/isar/apis/robot_control/robot_controller.py

@@ -0,0 +1,41 @@
+import logging
+from typing import List
+
+from injector import inject
+
+from isar.apis.models.models import (
+    RobotInfoResponse,
+    TaskResponse,
+)
+from isar.config.settings import robot_settings, settings
+from isar.services.utilities.robot_utilities import RobotUtilities
+from robot_interface.models.mission.task import Task
+
+
+class RobotController:
+    @inject
+    def __init__(
+        self,
+        robot_utilities: RobotUtilities,
+    ):
+        self.robot_utilities: RobotUtilities = robot_utilities
+        self.logger = logging.getLogger("api")
+
+    def generate_media_stream_config(self):
+        return self.robot_utilities.generate_robot_media_config()
+
+    def get_info(self):
+        return RobotInfoResponse(
+            robot_package=settings.ROBOT_PACKAGE,
+            isar_id=settings.ISAR_ID,
+            robot_name=settings.ROBOT_NAME,
+            robot_map_name=settings.DEFAULT_MAP,
+            robot_capabilities=robot_settings.CAPABILITIES,
+        )
+
+    def _task_api_response(self, task: Task) -> TaskResponse:
+        steps: List[dict] = []
+        for step in task.steps:
+            steps.append({"id": step.id, "type": step.__class__.__name__})
+
+        return TaskResponse(id=task.id, tag_id=task.tag_id, steps=steps)

src/isar/modules.py

@@ -8,6 +8,7 @@
 
 from isar.apis.api import API
 from isar.apis.schedule.scheduling_controller import SchedulingController
+from isar.apis.robot_control.robot_controller import RobotController
 from isar.apis.security.authentication import Authenticator
 from isar.config.keyvault.keyvault_service import Keyvault
 from isar.config.settings import settings
@@ -18,6 +19,7 @@
 from isar.models.communication.queues.queues import Queues
 from isar.services.service_connections.request_handler import RequestHandler
 from isar.services.utilities.scheduling_utilities import SchedulingUtilities
+from isar.services.utilities.robot_utilities import RobotUtilities
 from isar.state_machine.state_machine import StateMachine
 from isar.storage.blob_storage import BlobStorage
 from isar.storage.local_storage import LocalStorage
@@ -35,9 +37,10 @@ def provide_api(
         self,
         authenticator: Authenticator,
         scheduling_controller: SchedulingController,
+        robot_controller: RobotController,
         keyvault: Keyvault,
     ) -> API:
-        return API(authenticator, scheduling_controller, keyvault)
+        return API(authenticator, scheduling_controller, robot_controller, keyvault)
 
     @provider
     @singleton
@@ -46,6 +49,14 @@ def provide_scheduling_controller(
         scheduling_utilities: SchedulingUtilities,
     ) -> SchedulingController:
         return SchedulingController(scheduling_utilities)
+
+    @provider
+    @singleton
+    def provide_robot_controller(
+        self,
+        robot_utilities: RobotUtilities,
+    ) -> SchedulingController:
+        return RobotController(robot_utilities)
 
 
 class AuthenticationModule(Module):
@@ -150,6 +161,13 @@ def provide_scheduling_utilities(
     ) -> SchedulingUtilities:
         return SchedulingUtilities(queues, mission_planner)
 
+class RobotUtilitiesModule(Module):
+    @provider
+    @singleton
+    def provide_robot_utilities(
+        self, robot: RobotInterface
+    ) -> RobotUtilities:
+        return RobotUtilities(robot)
 
 class ServiceModule(Module):
     @provider
@@ -194,6 +212,7 @@ def provide_task_selector(self) -> TaskSelectorInterface:
     "storage_slimm": (SlimmStorageModule, settings.STORAGE_SLIMM_ENABLED),
     "mqtt": (MqttModule, "required"),
     "utilities": (UtilitiesModule, "required"),
+    "robot_utilities": (RobotUtilitiesModule, "required"),
 }
 
 

src/isar/services/utilities/robot_utilities.py

@@ -0,0 +1,27 @@
+import logging
+
+from injector import inject
+
+from isar.config.settings import settings
+
+from isar.config.settings import settings
+
+from robot_interface.robot_interface import RobotInterface
+
+
+class RobotUtilities:
+    """
+    Contains utility functions for scheduling missions from the API. The class handles
+    required thread communication through queues to the state machine.
+    """
+
+    @inject
+    def __init__(
+        self,
+        robot: RobotInterface,
+    ):
+        self.robot: RobotInterface = robot
+        self.logger = logging.getLogger("api")
+
+    def generate_robot_media_config(self) -> str:
+        return self.robot.generate_media_config(settings.ISAR_ID, settings.ROBOT_NAME)

src/robot_interface/robot_interface.py

@@ -223,6 +223,19 @@ def initialize(self, params: InitializeParams) -> None:
 
         """
         raise NotImplementedError
+
+    @abstractmethod
+    def generate_media_config(self, isar_id: str, robot_name: str) -> str:
+        """
+        Generate a JSON containing the url and token needed to establish a media stream
+        connection to a robot. 
+
+        Returns
+        -------
+        str
+            JSON string containing the connection information for a media stream connection
+        """
+        raise NotImplementedError
 
     @abstractmethod
     def get_telemetry_publishers(