robot_cloud_system 是一个面向移动机器人云控场景的 Web + ROS 2 集成项目。系统以 Flask-SocketIO 作为 Web 服务入口,以 ROS 2 节点作为机器人能力接入层,以 SQLite 作为本地任务存储,提供以下核心能力:
项目当前更偏向“教学演示 / 原型验证 / 二次开发基础仓库”,不是完整生产级平台。
项目分为四层:
入口文件为 app.py,负责:
位于 service_nodes/,负责封装不同类型的机器人能力:
WebNode.py
速度控制、IMU 数据、电池状态MapServer.py
地图、位姿、雷达、导航点NavigationClient.py
导航目标发送、路径缓存、导航状态TaskQueueManager.py
任务数据库读写、状态历史管理TaskStateMachine.py
任务状态转换规则校验TrajectoryRecorder.py
轨迹记录服务SystemShow.py
系统状态和控制反馈位于 robot_dashboard/templates/:
base.html
公共布局与全局样式index.html
主控制台入口map.html
地图、路径、导航点相关 UItasks.html
任务管理页面system.html
系统信息与系统操作页面system_show.html
/system 路由使用的包装模板使用 SQLite,本地数据库文件路径来自环境变量 DATABASE_PATH。
主要表:
taskstask_status_historyrobot_cloud_system/
├─ app.py
├─ config.py
├─ requirements.txt
├─ .env.example
├─ maps/
├─ robot_dashboard/
│ ├─ static/
│ └─ templates/
├─ service_nodes/
├─ scripts/
│ └─ init_database.py
└─ docs/
目录职责:
maps/
示例地图文件,供前端地图展示和导航调试使用robot_dashboard/static/
浏览器静态资源robot_dashboard/templates/
页面模板与内联前端逻辑service_nodes/
ROS 2 业务节点scripts/
辅助脚本docs/
项目文档app.py 启动时的核心流程:
init_ros_nodes()WebNode、LidarProcessor、MapServer、NavigationClient、TrajectoryRecorder、TaskQueueManager、TaskStateMachine、SystemShowMultiThreadedExecutorspin()配置文件为 config.py,通过 .env 读取参数。
主要配置项:
FLASK_SECRET_KEY
Flask 密钥FLASK_HOST
监听地址FLASK_PORT
服务端口FLASK_DEBUG
调试模式DATABASE_PATH
SQLite 文件路径ROS_DISTRO
ROS 发行版标识app.pyapp.py 同时承担“页面路由 + API + WebSocket + ROS 初始化”四类职责。
主要能力:
当前特点:
MapServer.py职责:
/map/amcl_pose/odom/scan输出给前端的数据类型:
NavigationClient.py职责:
NavigateToPose 导航目标/plan 和 /local_plan典型调用场景:
app.py 组织目标点NavigationClient 发出 ROS 导航请求TaskQueueManager.py职责:
当前实现特点:
后续可演进方向:
SystemShow.py职责:
说明:
/
主页面/system
系统状态页面/debug/map
地图调试接口/debug/request_map
手动触发地图请求/api/robot_pose
获取机器人位姿/api/lidar_data
获取雷达数据/api/map_data
获取地图数据GET /api/tasks
获取任务列表POST /api/tasks
创建任务GET /api/tasks/<task_id>
获取任务详情PUT /api/tasks/<task_id>/status
更新任务状态GET /api/tasks/<task_id>/history
获取任务历史DELETE /api/tasks/<task_id>
删除任务GET /api/system/status
获取系统状态connectdisconnectupdate_battery_dataupdate_imu_dataupdate_lidar_datarequest_map_datacontrol_commandreal_control_commandadd_waypointdelete_waypointclear_waypointsrequest_robot_poserequest_lidar_datastart_nav_sequencestop_navigationtrajectory_record_commandget_waypointsget_navigation_pathstart_taskstop_taskcomplete_taskfail_tasksystem_reconnectsystem_restartsystem_factory_reset字段说明:
idnametypeprioritystatusparameterscreated_atupdated_at字段说明:
idtask_idstatustimestampmessageapp.pyapp.py 调用 WebNode.publish_cmd_vel()/cmd_velMapServer 接收 /mapMapServer 更新内部缓存request_map_data 请求app.py 将地图和位姿通过 Socket.IO 推送到页面/api/tasks 创建任务start_taskapp.py 从数据库读取任务task_status_historyapp.py 过大,后续适合按领域拆分app.py 拆成 routes/、socket_handlers/、services/第一次接手这个项目,建议按下面顺序阅读:
README.mddocs/architecture.mddocs/technical-reference.mdapp.pyservice_nodes/MapServer.pyservice_nodes/NavigationClient.pyservice_nodes/TaskQueueManager.py这样可以先建立整体模型,再进入具体实现。