实现获取门禁状态接口

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

@@ -90,6 +90,16 @@ async def delete_system_door(request: Request, ids: str, query_db: AsyncSession
 
 # 视频流地址
 
+# 获取设备状态
+@doorController.get('/door_status',)
+async def get_door_status(request: Request, query_db: AsyncSession = Depends(get_db)):
+    result = await DoorService.get_door_status_service(query_db)
+    
+    if result[0]:
+        return ResponseUtil.success(data=result[1])
+    
+    return ResponseUtil.error(msg=result[1])
+
 
 # @doorController.get(
 #     '/{id}', response_model=DoorModel, dependencies=[Depends(CheckUserInterfaceAuth('system:door:query'))]

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

@@ -9,6 +9,17 @@ class DoorDao:
     """
     门禁设备模块数据库操作层
     """
+    
+    @classmethod
+    async def get_door_door_status(cls, db: AsyncSession):
+        """获取门禁状态
+        """
+        stmt = (
+            select(Door.indexCode, Door.name).select_from(Door).where(Door.permission=="1")
+        )
+        result = await db.execute(stmt)
+        
+        return result.mappings().all()
 
     @classmethod
     async def get_door_index_code_list(cls, db: AsyncSession, permission: str = None):

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

@@ -7,12 +7,44 @@ from module_admin.dao.door_dao import DoorDao
 from module_admin.entity.vo.door_vo import DeleteDoorModel, DoorModel, DoorPageQueryModel
 from utils.common_util import CamelCaseUtil
 from utils.excel_util import ExcelUtil
+from utils.haikang_util import HaikangUtil
 
 
 class DoorService:
     """
     门禁设备模块服务层
     """
+    
+    @classmethod
+    async def get_door_status_service(cls, query_db: AsyncSession):
+        """获取门禁状态
+
+        Args:
+            query_db (AsyncSession): orm对象
+        Returns:
+            _type_: _description_
+        """
+        
+        door_indexcode_name = await DoorDao.get_door_door_status(query_db)
+        
+        t_d = dict()
+        index_code_list = list()
+        for t in door_indexcode_name:
+            t_d[t["indexCode"]] = t["name"]
+            index_code_list.append(t["indexCode"])
+        
+        door_online_status_result = await HaikangUtil.door_online_status(index_code_list)
+        
+        if door_online_status_result[0]:
+            back = list()
+            for door_online_status in door_online_status_result[1]['list']:
+                back.append({
+                    "name": t_d.get(door_online_status['indexCode'], None),
+                    "online": door_online_status['online']
+                })
+            
+            return True, back
+        return door_online_status_result
 
     @classmethod
     async def get_door_list_services(