实现获取机器人视频流接口

ruoyi-fastapi-backend/module_admin/controller/door_controller.py

@@ -100,6 +100,15 @@ async def get_door_status(request: Request, query_db: AsyncSession = Depends(get
     
     return ResponseUtil.error(msg=result[1])
 
+# 获取视频流地址
+@doorController.get('/video_uri/{robot_id}')
+async def get_video_uri(request: Request
+                        , robot_id: int
+                        , query_db: AsyncSession = Depends(get_db)
+                        ):
+    result = await DoorService.get_video_uri_service(robot_id, query_db)
+    return ResponseUtil.success(data=result)
+
 
 # @doorController.get(
 #     '/{id}', response_model=DoorModel, dependencies=[Depends(CheckUserInterfaceAuth('system:door:query'))]

ruoyi-fastapi-backend/module_admin/dao/info_dao.py

@@ -11,6 +11,26 @@ class InfoDao:
     """
     机器人信息模块数据库操作层
     """
+    
+    
+    @classmethod
+    async def get_video_uri(cls, robot_id: int, query_db: AsyncSession):
+        """根据机器人id获取视频流地址
+
+        Args:
+            robot_id (int): 机器人地址
+            query_db (AsyncSession): _description_
+        """
+        stmt = (
+            select(RobotInfo.video_uri)
+            .select_from(RobotInfo)
+            .where(RobotInfo.robot_id == robot_id)
+        )
+        
+        result = await query_db.execute(stmt)
+        
+        return result.scalars().first()
+    
     @classmethod
     async def refresh_robot_by_ids(cls, db: AsyncSession, robot_ids: str):
         id_list = map(int, robot_ids.strip(" ").split(','))

ruoyi-fastapi-backend/module_admin/service/door_service.py

@@ -4,6 +4,7 @@ from config.constant import CommonConstant
 from exceptions.exception import ServiceException
 from module_admin.entity.vo.common_vo import CrudResponseModel
 from module_admin.dao.door_dao import DoorDao
+from module_admin.dao.info_dao import InfoDao
 from module_admin.entity.vo.door_vo import DeleteDoorModel, DoorModel, DoorPageQueryModel
 from utils.common_util import CamelCaseUtil
 from utils.excel_util import ExcelUtil
@@ -15,6 +16,14 @@ class DoorService:
     门禁设备模块服务层
     """
     
+    @classmethod
+    async def get_video_uri_service(cls, robot_id: int , query_db: AsyncSession):
+        """获取机器人视频流
+        """
+        video_uri = await InfoDao.get_video_uri(robot_id, query_db)
+        return video_uri
+        
+    
     @classmethod
     async def get_door_status_service(cls, query_db: AsyncSession):
         """获取门禁状态