"""
|
算法系统服务器
|
"""
|
import json
|
import logging
|
import os
|
import time
|
from typing import Dict, List, Optional, Any, Tuple
|
from flask import Flask, request, jsonify
|
|
from common.data_models import (
|
TaskData, AGVStatus, TaskAssignment, PlannedPath,
|
create_success_response, create_error_response, ResponseCode, to_dict, from_dict
|
)
|
from common.utils import load_path_mapping
|
from algorithm_system.models.agv_model import AGVModelManager
|
from algorithm_system.algorithms.task_allocation import TaskAllocationFactory
|
from algorithm_system.algorithms.path_planning import PathPlanningFactory
|
from common.api_client import RCSAPIClient
|
from algorithm_system.path_monitor import PathMonitorService
|
|
try:
|
from config.settings import (
|
ALGORITHM_SERVER_HOST, ALGORITHM_SERVER_PORT,
|
RCS_SERVER_HOST, RCS_SERVER_PORT,
|
MONITOR_POLLING_INTERVAL
|
)
|
except ImportError:
|
ALGORITHM_SERVER_HOST = "10.10.10.239"
|
ALGORITHM_SERVER_PORT = 8002
|
RCS_SERVER_HOST = "10.10.10.156"
|
RCS_SERVER_PORT = 8088
|
MONITOR_POLLING_INTERVAL = 5.0
|
logging.warning("无法从config.settings导入配置，使用默认值")
|
|
|
class AlgorithmServer:
|
"""算法系统服务器"""
|
|
def __init__(self, host: str = None, port: int = None, enable_path_monitor: bool = True,
|
monitor_interval: float = None):
|
"""初始化算法服务器"""
|
self.host = host or ALGORITHM_SERVER_HOST
|
self.port = port or ALGORITHM_SERVER_PORT
|
self.enable_path_monitor = enable_path_monitor
|
self.monitor_interval = monitor_interval or MONITOR_POLLING_INTERVAL
|
self.logger = logging.getLogger(__name__)
|
|
self.app = Flask(__name__)
|
|
self.path_mapping = load_path_mapping()
|
|
self.agv_manager = AGVModelManager(self.path_mapping)
|
|
self.rcs_client = RCSAPIClient()
|
|
self.task_allocation_algorithm = "LOAD_BALANCED" # 默认任务分配
|
self.path_planning_algorithm = "A_STAR" # 默认路径规划
|
|
self.path_monitor = None
|
if self.enable_path_monitor:
|
self.path_monitor = PathMonitorService(
|
rcs_host=RCS_SERVER_HOST,
|
rcs_port=RCS_SERVER_PORT,
|
poll_interval=self.monitor_interval,
|
path_algorithm=self.path_planning_algorithm,
|
auto_send_paths=True # 自动发送路径到RCS
|
)
|
self.logger.info(f"路径监控服务已初始化 - 轮询间隔: {self.monitor_interval}s, 自动发送路径: True")
|
|
self._setup_routes()
|
|
self.logger.info("算法系统服务器初始化完成")
|
|
def _setup_routes(self):
|
"""设置API路由"""
|
|
@self.app.route('/open/task/send/v1', methods=['POST'])
|
def task_send_endpoint():
|
"""任务下发接口"""
|
return self._handle_task_send_request()
|
|
@self.app.route('/open/path/plan/v1', methods=['POST'])
|
def path_plan_endpoint():
|
"""路径规划接口"""
|
return self._handle_path_planning_request()
|
|
@self.app.route('/open/path/batch/plan/v1', methods=['POST'])
|
def batch_path_plan_endpoint():
|
"""批量路径规划接口"""
|
return self._handle_batch_path_planning_request()
|
|
@self.app.route('/open/algorithm/config/v1', methods=['POST'])
|
def algorithm_config_endpoint():
|
"""算法配置接口"""
|
return self._handle_algorithm_config_request()
|
|
@self.app.route('/monitor/path/start/v1', methods=['POST'])
|
def start_path_monitor_endpoint():
|
"""启动路径监控服务接口"""
|
return self._handle_start_path_monitor_request()
|
|
@self.app.route('/monitor/path/stop/v1', methods=['POST'])
|
def stop_path_monitor_endpoint():
|
"""停止路径监控服务接口"""
|
return self._handle_stop_path_monitor_request()
|
|
@self.app.route('/monitor/path/status/v1', methods=['GET'])
|
def path_monitor_status_endpoint():
|
"""路径监控服务状态查询接口"""
|
return self._handle_path_monitor_status_request()
|
|
@self.app.route('/health', methods=['GET'])
|
def health_check():
|
"""健康检查接口"""
|
monitor_status = None
|
if self.path_monitor:
|
monitor_status = self.path_monitor.get_monitoring_status()
|
|
return jsonify({
|
"status": "healthy",
|
"service": "algorithm_system",
|
"timestamp": time.time(),
|
"path_mapping_loaded": len(self.path_mapping) > 0,
|
"algorithms": {
|
"task_allocation": self.task_allocation_algorithm,
|
"path_planning": self.path_planning_algorithm
|
},
|
"agv_count": len(self.agv_manager.get_all_agvs()),
|
"path_monitor": monitor_status,
|
"available_endpoints": [
|
"POST /open/task/send/v1 - 任务分配接口",
|
"POST /open/path/plan/v1 - 单路径规划接口",
|
"POST /open/path/batch/plan/v1 - 批量路径规划接口",
|
"POST /open/algorithm/config/v1 - 算法配置接口",
|
"POST /monitor/path/start/v1 - 启动路径监控服务",
|
"POST /monitor/path/stop/v1 - 停止路径监控服务",
|
"GET /monitor/path/status/v1 - 路径监控服务状态查询",
|
"GET /health - 健康检查接口"
|
]
|
})
|
|
@self.app.errorhandler(404)
|
def not_found(error):
|
"""404错误处理"""
|
return jsonify(to_dict(create_error_response(404, "接口不存在"))), 404
|
|
@self.app.errorhandler(500)
|
def internal_error(error):
|
"""500错误处理"""
|
self.logger.error(f"内部服务器错误: {error}")
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, "内部服务器错误"))), 500
|
|
def _handle_task_send_request(self) -> Dict[str, Any]:
|
"""处理任务下发请求"""
|
try:
|
if not request.is_json:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "请求数据格式错误")))
|
|
request_data = request.get_json()
|
if not request_data:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "请求数据为空")))
|
|
if isinstance(request_data, dict) and "tasks" in request_data:
|
task_data = request_data.get("tasks", [])
|
agv_status_data = request_data.get("agvStatus", [])
|
self.logger.info(f"收到结构化任务分配请求，任务数量: {len(task_data)}, AGV数量: {len(agv_status_data)}")
|
elif isinstance(request_data, list):
|
task_data = request_data
|
agv_status_data = []
|
self.logger.info(f"收到任务分配请求，任务数量: {len(task_data)}")
|
else:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "任务数据格式无效")))
|
|
if not task_data or not isinstance(task_data, list):
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "任务数据无效")))
|
|
tasks = []
|
for task_dict in task_data:
|
try:
|
task = TaskData(
|
taskId=task_dict.get("taskId", ""),
|
start=task_dict.get("start", ""),
|
end=task_dict.get("end", ""),
|
type=task_dict.get("type", "1"),
|
priority=task_dict.get("priority", 5)
|
)
|
tasks.append(task)
|
except Exception as e:
|
self.logger.warning(f"解析任务数据失败: {task_dict} - {e}")
|
continue
|
|
if not tasks:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "没有有效的任务数据")))
|
|
agv_status_list = []
|
if agv_status_data:
|
self.logger.info("使用请求中提供的AGV状态数据")
|
for agv_data in agv_status_data:
|
try:
|
agv_status = from_dict(AGVStatus, agv_data)
|
agv_status_list.append(agv_status)
|
except Exception as e:
|
self.logger.warning(f"解析请求中的AGV状态数据失败: {agv_data} - {e}")
|
continue
|
else:
|
self.logger.info("从RCS系统获取AGV状态数据")
|
agv_response = self.rcs_client.get_agv_status()
|
if agv_response.code == ResponseCode.SUCCESS and agv_response.data:
|
for agv_data in agv_response.data:
|
try:
|
agv_status = from_dict(AGVStatus, agv_data)
|
agv_status_list.append(agv_status)
|
except Exception as e:
|
self.logger.warning(f"解析RCS返回的AGV状态数据失败: {agv_data} - {e}")
|
continue
|
else:
|
error_msg = f"无法获取AGV状态: {agv_response.msg if agv_response else 'RCS系统连接失败'}"
|
self.logger.error(error_msg)
|
|
use_fallback_allocation = getattr(self, 'use_fallback_allocation', True)
|
|
if use_fallback_allocation:
|
self.logger.warning("使用轮询分配作为备用方案")
|
return self._simple_round_robin_allocation(tasks)
|
else:
|
return jsonify(to_dict(create_error_response(
|
ResponseCode.SERVER_ERROR,
|
error_msg
|
)))
|
|
if not agv_status_list:
|
return jsonify(to_dict(create_error_response(ResponseCode.NO_DATA, "没有可用的AGV状态数据")))
|
|
self.agv_manager.update_agv_data(agv_status_list)
|
self.logger.info(f"更新了 {len(agv_status_list)} 个AGV状态")
|
|
all_agvs = self.agv_manager.get_all_agvs()
|
self.logger.info(f"算法系统中当前有 {len(all_agvs)} 个AGV模型")
|
|
available_count = 0
|
for agv in all_agvs:
|
can_accept = agv.can_accept_task(5)
|
self.logger.debug(f"AGV {agv.agvId}: status={agv.status}, can_accept_task={can_accept}, "
|
f"is_overloaded={agv.is_overloaded()}, task_count={agv.current_task_count}")
|
if can_accept:
|
available_count += 1
|
|
self.logger.info(f"其中 {available_count} 个AGV可以接受任务")
|
|
if available_count == 0:
|
self.logger.warning("没有AGV可以接受任务！详细状态:")
|
for agv in all_agvs:
|
status_str = str(agv.status)
|
self.logger.warning(f" AGV {agv.agvId}: 状态={status_str}, 类型={type(agv.status)}, "
|
f"任务数={agv.current_task_count}/{agv.max_capacity}")
|
|
allocator = TaskAllocationFactory.create_allocator(
|
self.task_allocation_algorithm,
|
self.agv_manager
|
)
|
|
start_time = time.time()
|
assignments = allocator.allocate_tasks(tasks)
|
end_time = time.time()
|
|
allocation_time = (end_time - start_time) * 1000 # 转换为毫秒
|
|
assignment_data = [to_dict(assignment) for assignment in assignments]
|
|
self.logger.info(f"任务分配完成，分配了 {len(assignments)} 个任务，耗时: {allocation_time:.2f}ms")
|
|
response = create_success_response(assignment_data)
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"处理任务分配请求失败: {e}", exc_info=True)
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"服务器错误: {str(e)}")))
|
|
def _handle_path_planning_request(self) -> Dict[str, Any]:
|
"""处理路径规划请求"""
|
try:
|
# 验证请求数据
|
if not request.is_json:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "请求数据格式错误")))
|
|
request_data = request.get_json()
|
if not request_data:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "请求数据为空")))
|
|
agv_id = request_data.get("agvId")
|
start_code = request_data.get("start")
|
end_code = request_data.get("end")
|
constraints = request_data.get("constraints", [])
|
|
if not agv_id or not start_code or not end_code:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "缺少必要参数")))
|
|
self.logger.info(f"收到路径规划请求 - AGV: {agv_id}, 从 {start_code} 到 {end_code}")
|
|
# 创建路径规划器
|
path_planner = PathPlanningFactory.create_path_planner(
|
self.path_planning_algorithm,
|
self.path_mapping
|
)
|
|
# 执行路径规划
|
start_time = time.time()
|
planned_path = path_planner.plan_path(start_code, end_code, constraints)
|
end_time = time.time()
|
|
planning_time = (end_time - start_time) * 1000 # 转换为毫秒
|
|
if planned_path:
|
# 设置AGV ID
|
planned_path.agvId = agv_id
|
|
# 转换为响应格式
|
path_data = to_dict(planned_path)
|
|
self.logger.info(f"路径规划完成 - AGV: {agv_id}, 路径长度: {len(planned_path.codeList)}, 耗时: {planning_time:.2f}ms")
|
|
response = create_success_response(path_data)
|
return jsonify(to_dict(response))
|
else:
|
self.logger.warning(f"路径规划失败 - AGV: {agv_id}, 从 {start_code} 到 {end_code}")
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, "无法规划路径")))
|
|
except Exception as e:
|
self.logger.error(f"处理路径规划请求失败: {e}", exc_info=True)
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"服务器错误: {str(e)}")))
|
|
def _handle_batch_path_planning_request(self) -> Dict[str, Any]:
|
"""处理批量路径规划请求"""
|
try:
|
if not request.is_json:
|
request_data = {}
|
else:
|
request_data = request.get_json() or {}
|
|
constraints = request_data.get("constraints", [])
|
include_idle_agv = request_data.get("includeIdleAgv", False)
|
use_enhanced_planning = request_data.get("useEnhancedPlanning", True)
|
|
agv_status_from_request = (
|
request_data.get("agvStatusList", []) or
|
request_data.get("agvs", [])
|
)
|
|
self.logger.info(f"收到批量路径规划请求，包含空闲AGV: {include_idle_agv}, "
|
f"请求中AGV数量: {len(agv_status_from_request)}, "
|
f"使用增强规划: {use_enhanced_planning}")
|
|
agv_status_list = []
|
if agv_status_from_request:
|
self.logger.info("使用请求中提供的AGV状态数据进行路径规划")
|
for agv_data in agv_status_from_request:
|
try:
|
agv_status = from_dict(AGVStatus, agv_data)
|
agv_status_list.append(agv_status)
|
except Exception as e:
|
self.logger.warning(f"解析请求中的AGV状态数据失败: {agv_data} - {e}")
|
continue
|
else:
|
self.logger.info("从RCS系统获取AGV状态数据进行路径规划")
|
agv_response = self.rcs_client.get_agv_status()
|
if agv_response.code != ResponseCode.SUCCESS or not agv_response.data:
|
error_msg = f"无法获取AGV状态进行路径规划: {agv_response.msg if agv_response else 'RCS系统连接失败'}"
|
self.logger.error(error_msg)
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, error_msg)))
|
|
for agv_data in agv_response.data:
|
try:
|
agv_status = from_dict(AGVStatus, agv_data)
|
agv_status_list.append(agv_status)
|
except Exception as e:
|
self.logger.warning(f"解析RCS返回的AGV状态数据失败: {agv_data} - {e}")
|
continue
|
|
if not agv_status_list:
|
error_msg = "没有可用的AGV状态数据进行路径规划"
|
self.logger.error(error_msg)
|
return jsonify(to_dict(create_error_response(ResponseCode.NO_DATA, error_msg)))
|
|
# 更新AGV模型管理器
|
self.agv_manager.update_agv_data(agv_status_list)
|
|
self.logger.info(f"强制使用路径规划以确保包含新字段")
|
result = self._enhanced_batch_path_planning(agv_status_list, include_idle_agv, constraints)
|
|
response = create_success_response(result)
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"处理批量路径规划请求失败: {e}", exc_info=True)
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"服务器错误: {str(e)}")))
|
|
def _enhanced_batch_path_planning(self, agv_status_list: List[AGVStatus],
|
include_idle_agv: bool,
|
constraints: List[Tuple[int, int, float]]) -> Dict[str, Any]:
|
"""批量路径规划"""
|
from algorithm_system.algorithms.path_planning import PathPlanningFactory
|
|
# 创建批量路径规划器
|
batch_planner = PathPlanningFactory.create_batch_path_planner(
|
algorithm_type=self.path_planning_algorithm,
|
path_mapping=self.path_mapping
|
)
|
|
# 执行批量路径规划
|
result = batch_planner.plan_all_agv_paths(
|
agv_status_list=agv_status_list,
|
include_idle_agv=include_idle_agv,
|
constraints=constraints
|
)
|
|
# 提取plannedPaths作为最终返回数据
|
planned_paths = result.get('plannedPaths', [])
|
|
self.logger.info(f"批量路径规划完成 - 总AGV: {result['totalAgvs']}, "
|
f"执行任务: {result['executingTasksCount']}, "
|
f"规划路径: {result['plannedPathsCount']}, "
|
f"冲突检测: {result['conflictsDetected']}, "
|
f"总耗时: {result['totalPlanningTime']:.2f}ms")
|
|
# 只返回路径数组，不包含其他统计信息
|
return planned_paths
|
|
def _handle_algorithm_config_request(self) -> Dict[str, Any]:
|
"""
|
处理算法配置请求
|
|
Returns:
|
Dict: 响应数据
|
"""
|
try:
|
# 验证请求数据
|
if not request.is_json:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "请求数据格式错误")))
|
|
config_data = request.get_json()
|
if not config_data:
|
return jsonify(to_dict(create_error_response(ResponseCode.PARAM_EMPTY, "配置数据为空")))
|
|
# 更新算法配置
|
if "task_allocation_algorithm" in config_data:
|
self.task_allocation_algorithm = config_data["task_allocation_algorithm"]
|
self.logger.info(f"任务分配算法更新为: {self.task_allocation_algorithm}")
|
|
if "path_planning_algorithm" in config_data:
|
self.path_planning_algorithm = config_data["path_planning_algorithm"]
|
self.logger.info(f"路径规划算法更新为: {self.path_planning_algorithm}")
|
|
# 返回当前配置
|
current_config = {
|
"task_allocation_algorithm": self.task_allocation_algorithm,
|
"path_planning_algorithm": self.path_planning_algorithm
|
}
|
|
response = create_success_response(current_config)
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"处理算法配置请求失败: {e}", exc_info=True)
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"服务器错误: {str(e)}")))
|
|
def _simple_round_robin_allocation(self, tasks: List[TaskData]) -> Dict[str, Any]:
|
"""简单的轮询分配"""
|
assignments = []
|
|
# 尝试从RCS系统获取AGV数量
|
agv_count = 5 # 默认值
|
try:
|
agv_response = self.rcs_client.get_agv_status()
|
if agv_response.code == ResponseCode.SUCCESS and agv_response.data:
|
agv_count = len(agv_response.data)
|
self.logger.info(f"从RCS系统获取到AGV数量: {agv_count}")
|
else:
|
# 如果RCS返回无数据，使用配置的默认值
|
from config.settings import DEFAULT_AGV_COUNT
|
agv_count = DEFAULT_AGV_COUNT
|
self.logger.warning(f"无法从RCS获取AGV数量，使用默认值: {agv_count}")
|
except Exception as e:
|
# 如果完全无法连接RCS，使用配置的默认值
|
from config.settings import DEFAULT_AGV_COUNT
|
agv_count = DEFAULT_AGV_COUNT
|
self.logger.error(f"获取AGV数量失败，使用默认值: {agv_count} - {e}")
|
|
for i, task in enumerate(tasks):
|
agv_id = f"AGV_{10001 + (i % agv_count)}"
|
assignment = TaskAssignment(
|
taskId=task.taskId,
|
agvId=agv_id
|
)
|
assignments.append(assignment)
|
|
assignment_data = [to_dict(assignment) for assignment in assignments]
|
|
self.logger.info(f"使用简单轮询分配了 {len(assignments)} 个任务，AGV数量: {agv_count}")
|
|
response = create_success_response(assignment_data)
|
return jsonify(to_dict(response))
|
|
def _handle_start_path_monitor_request(self) -> Dict[str, Any]:
|
"""处理启动路径监控服务请求"""
|
try:
|
if not self.path_monitor:
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, "路径监控服务未初始化")))
|
|
if self.path_monitor.is_running:
|
return jsonify(to_dict(create_success_response({"message": "路径监控服务已在运行中"})))
|
|
# 启动路径监控服务
|
self.path_monitor.start_monitoring()
|
|
self.logger.info("路径监控服务已通过API启动")
|
|
response = create_success_response({
|
"message": "路径监控服务启动成功",
|
"monitor_status": self.path_monitor.get_monitoring_status()
|
})
|
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"启动路径监控服务失败: {e}")
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"启动路径监控服务失败: {str(e)}")))
|
|
def _handle_stop_path_monitor_request(self) -> Dict[str, Any]:
|
"""处理停止路径监控服务请求"""
|
try:
|
if not self.path_monitor:
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, "路径监控服务未初始化")))
|
|
if not self.path_monitor.is_running:
|
return jsonify(to_dict(create_success_response({"message": "路径监控服务未在运行"})))
|
|
# 停止路径监控服务
|
self.path_monitor.stop_monitoring()
|
|
self.logger.info("路径监控服务已通过API停止")
|
|
response = create_success_response({
|
"message": "路径监控服务停止成功",
|
"monitor_status": self.path_monitor.get_monitoring_status()
|
})
|
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"停止路径监控服务失败: {e}")
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"停止路径监控服务失败: {str(e)}")))
|
|
def _handle_path_monitor_status_request(self) -> Dict[str, Any]:
|
"""处理路径监控服务状态查询请求"""
|
try:
|
if not self.path_monitor:
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, "路径监控服务未初始化")))
|
|
monitor_status = self.path_monitor.get_monitoring_status()
|
|
response = create_success_response({
|
"monitor_status": monitor_status,
|
"description": "路径监控服务状态信息"
|
})
|
|
return jsonify(to_dict(response))
|
|
except Exception as e:
|
self.logger.error(f"获取路径监控服务状态失败: {e}")
|
return jsonify(to_dict(create_error_response(ResponseCode.SERVER_ERROR, f"获取路径监控服务状态失败: {str(e)}")))
|
|
def start_server(self):
|
"""启动算法服务器"""
|
self.logger.info(f"启动算法系统服务器: http://{self.host}:{self.port}")
|
|
try:
|
# 如果启用了路径监控服务，自动启动
|
if self.path_monitor and self.enable_path_monitor:
|
self.path_monitor.start_monitoring()
|
self.logger.info("路径监控服务已自动启动")
|
|
self.app.run(
|
host=self.host,
|
port=self.port,
|
debug=False,
|
threaded=True
|
)
|
except Exception as e:
|
self.logger.error(f"启动算法服务器失败: {e}")
|
# 确保在启动失败时停止路径监控服务
|
if self.path_monitor and self.path_monitor.is_running:
|
self.path_monitor.stop_monitoring()
|
raise
|
|
def stop_server(self):
|
"""停止算法服务器"""
|
self.logger.info("停止算法服务器")
|
|
# 停止路径监控服务
|
if self.path_monitor and self.path_monitor.is_running:
|
self.path_monitor.stop_monitoring()
|
self.logger.info("路径监控服务已停止")
|
|
|
def get_server_status(self) -> Dict[str, Any]:
|
"""获取服务器状态"""
|
monitor_status = None
|
if self.path_monitor:
|
monitor_status = self.path_monitor.get_monitoring_status()
|
|
return {
|
"host": self.host,
|
"port": self.port,
|
"path_mapping_loaded": len(self.path_mapping) > 0,
|
"agv_count": len(self.agv_manager.get_all_agvs()),
|
"algorithms": {
|
"task_allocation": self.task_allocation_algorithm,
|
"path_planning": self.path_planning_algorithm
|
},
|
"path_monitor": monitor_status,
|
"timestamp": time.time()
|
}
|
