我将为您创建一个完整的机器人充电管理系统。以下是各个模块的代码：
1. 项目结构
robot_charging_system/
├── main.py # 主程序入口
├── robot.py # 机器人控制模块
├── battery_manager.py # 电池管理模块
├── charging_station.py # 充电站模块
├── scheduler.py # 任务调度模块
├── monitor.py # 监控模块
├── utils.py # 工具函数
├── config.py # 配置文件
├── test_robot.py # 测试模块
├── README.md # 使用说明
└── knowledge_cards.md # 核心知识点卡片
2. 主要代码文件
main.py - 主程序入口
"""
机器人充电管理系统主程序
功能：模拟机器人智能充电管理，根据电量自动控制充电行为
基于智能机器人导论课程的传感器、决策和控制理论
"""
import sys
import time
from datetime import datetime, timedelta
from robot import Robot
from battery_manager import BatteryManager
from charging_station import ChargingStation
from scheduler import TaskScheduler
from monitor import SystemMonitor
from utils import setup_logging, print_welcome, get_user_input
def main():
"""主函数 - 机器人充电管理系统入口"""
# 初始化系统
logger = setup_logging()
print_welcome()
try:
# 创建系统组件
robot = Robot("R001", "清洁机器人")
battery_manager = BatteryManager()
charging_station = ChargingStation()
scheduler = TaskScheduler()
monitor = SystemMonitor()
print("🤖 机器人充电管理系统启动成功！")
print("=" * 50)
# 主循环
while True:
# 显示当前状态
display_system_status(robot, battery_manager, charging_station)
# 获取用户输入或传感器数据
choice = get_user_action()
if choice == 'quit':
break
elif choice == 'auto':
# 自动模式：模拟传感器读数
auto_mode_simulation(robot, battery_manager, charging_station, scheduler)
elif choice == 'manual':
# 手动模式：用户输入电量
manual_mode_operation(robot, battery_manager, charging_station, scheduler)
elif choice == 'status':
# 查看详细状态
show_detailed_status(robot, battery_manager, charging_station, monitor)
elif choice == 'test':
# 运行测试
run_system_test(robot, battery_manager, charging_station)
# 检查系统健康状态
monitor.check_system_health(robot, battery_manager)
time.sleep(1) # 防止过快循环
except KeyboardInterrupt:
print("\n\n🛑 系统被用户中断")
except Exception as e:
print(f"\n❌ 系统运行出错：{e}")
finally:
print("\n👋 感谢使用机器人充电管理系统！")
cleanup_resources()
def display_system_status(robot, battery_manager, charging_station):
"""显示系统当前状态"""
print(f"\n📊 系统状态 [{datetime.now().strftime('%H:%M:%S')}]")
print("-" * 40)
# 机器人状态
robot_status = robot.get_status()
print(f"🤖 机器人: {robot_status['name']} ({robot_status['id']})")
print(f"📍 位置: {robot_status['position']}")
print(f"🔋 电量: {robot_status['battery_level']}%")
print(f"⚡ 状态: {robot_status['state']}")
# 充电站状态
station_status = charging_station.get_status()
print(f"🔌 充电站: {'可用' if station_status['available'] else '占用'}")
print(f"⏱️ 上次充电: {station_status['last_charging_time']}")
def get_user_action():
"""获取用户操作选择"""
print("\n请选择操作模式:")
print("1. 自动模拟模式 (auto)")
print("2. 手动输入模式 (manual)")
print("3. 查看详细状态 (status)")
print("4. 运行系统测试 (test)")
print("5. 退出系统 (quit)")
while True:
choice = input("\n请输入选择 (auto/manual/status/test/quit): ").strip().lower()
if choice in ['auto', 'manual', 'status', 'test', 'quit']:
return choice
else:
print("❌ 无效选择，请重新输入")
def auto_mode_simulation(robot, battery_manager, charging_station, scheduler):
"""自动模拟模式"""
print("\n🤖 进入自动模拟模式...")
# 模拟电量消耗
battery_manager.simulate_battery_drain(robot)
# 获取当前电量
current_battery = battery_manager.get_battery_level(robot.id)
# 决策：是否需要充电
decision = make_charging_decision(current_battery, robot, charging_station)
# 执行决策
execute_decision(decision, robot, charging_station, scheduler)
# 模拟时间流逝
time.sleep(2)
def manual_mode_operation(robot, battery_manager, charging_station, scheduler):
"""手动操作模式"""
print("\n📝 进入手动输入模式...")
try:
# 获取用户输入的电量
battery_input = input("请输入当前电量百分比 (0-100): ").strip()
battery_level = float(battery_input)
if not (0 <= battery_level <= 100):
print("❌ 电量值必须在0-100之间")
return
# 设置机器人电量
battery_manager.set_battery_level(robot.id, battery_level)
# 获取当前电量
current_battery = battery_manager.get_battery_level(robot.id)
# 决策：是否需要充电
decision = make_charging_decision(current_battery, robot, charging_station)
# 执行决策
execute_decision(decision, robot, charging_station, scheduler)
except ValueError:
print("❌ 请输入有效的数字")
except Exception as e:
print(f"❌ 输入处理出错：{e}")
def make_charging_decision(battery_level, robot, charging_station):
"""
智能决策：是否需要充电
基于智能机器人导论的决策理论
"""
print(f"\n🧠 AI决策分析中... 当前电量: {battery_level}%")
# 决策规则（基于专家系统）
if battery_level < 20:
if charging_station.is_available():
return {
'action': 'charge',
'reason': f'电量过低({battery_level}%)，前往充电站充电',
'priority': 'high',
'urgency': 'immediate'
}
else:
return {
'action': 'wait',
'reason': f'电量过低({battery_level}%)，但充电站被占用，等待中',
'priority': 'high',
'urgency': 'delayed'
}
elif battery_level < 50:
return {
'action': 'monitor',
'reason': f'电量中等({battery_level}%)，继续工作并密切监控',
'priority': 'medium',
'urgency': 'none'
}
else:
return {
'action': 'work',
'reason': f'电量充足({battery_level}%)，继续正常工作',
'priority': 'low',
'urgency': 'none'
}
def execute_decision(decision, robot, charging_station, scheduler):
"""执行决策"""
action = decision['action']
reason = decision['reason']
print(f"\n⚡ 决策结果: {reason}")
if action == 'charge':
print("🚀 机器人正在前往充电站...")
success = charging_station.request_charging(robot.id)
if success:
robot.start_charging()
print("✅ 成功连接到充电站，开始充电")
# 模拟充电过程
simulate_charging_process(robot, charging_station)
else:
print("❌ 无法连接到充电站")
elif action == 'wait':
print("⏳ 机器人在当前位置等待充电站空闲")
robot.set_state("等待充电")
elif action == 'monitor':
print("👀 机器人继续工作，监控系统会密切关注电量变化")
robot.set_state("工作中(监控电量)")
elif action == 'work':
print("💪 机器人继续正常工作")
robot.set_state("工作中")
# 分配新任务
new_task = scheduler.get_next_task()
if new_task:
print(f"📋 新任务分配: {new_task}")
def simulate_charging_process(robot, charging_station):
"""模拟充电过程"""
print("🔋 充电进行中...")
# 模拟充电时间
for i in range(1, 6):
time.sleep(1)
progress = i * 20
battery_level = min(100, robot.battery_level + 15) # 每次增加15%
robot.update_battery(battery_level)
print(f" 充电进度: {progress}% | 当前电量: {battery_level}%")
# 充电完成
robot.stop_charging()
charging_station.release_charging(robot.id)
print("✅ 充电完成，机器人恢复工作状态")
def show_detailed_status(robot, battery_manager, charging_station, monitor):
"""显示详细系统状态"""
print("\n" + "="*50)
print("📋 详细系统状态报告")
print("="*50)
# 机器人详细信息
robot.show_detailed_status()
# 电池详细信息
battery_manager.show_battery_details(robot.id)
# 充电站详细信息
charging_station.show_station_details()
# 系统监控信息
monitor.show_monitoring_report()
def run_system_test(robot, battery_manager, charging_station):
"""运行系统测试"""
print("\n🧪 开始系统测试...")
test_cases = [
("低电量测试", 15),
("临界电量测试", 20),
("中等电量测试", 50),
("充足电量测试", 80),
("满电量测试", 100)
]
for test_name, battery_level in test_cases:
print(f"\n--- {test_name} ---")
battery_manager.set_battery_level(robot.id, battery_level)
current_battery = battery_manager.get_battery_level(robot.id)
decision = make_charging_decision(current_battery, robot, charging_station)
print(f"电量: {current_battery}% -> 决策: {decision['action']}")
print(f"原因: {decision['reason']}")
print("\n✅ 测试完成")
def cleanup_resources():
"""清理系统资源"""
print("🧹 清理系统资源...")
if __name__ == "__main__":
main()
robot.py - 机器人控制模块
"""
机器人控制模块
实现机器人的基本行为和状态管理
基于智能机器人导论的机器人学基础理论
"""
import time
import uuid
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime
class RobotState(Enum):
"""机器人状态枚举"""
IDLE = "待机"
WORKING = "工作中"
CHARGING = "充电中"
MOVING = "移动中"
WAITING = "等待中"
ERROR = "故障"
class RobotType(Enum):
"""机器人类型枚举"""
CLEANER = "清洁机器人"
DELIVERY = "配送机器人"
SECURITY = "安防机器人"
GUIDANCE = "导览机器人"
@dataclass
class Position:
"""位置信息数据类"""
x: float
y: float
z: float = 0.0
area: str = "未知区域"
def __str__(self):
return f"({self.x}, {self.y}, {self.z}) @ {self.area}"
class Robot:
"""机器人基类 - 实现智能机器人的基本功能"""
def __init__(self, robot_id: str, name: str, robot_type: RobotType = RobotType.CLEANER):
"""
初始化机器人
Args:
robot_id: 机器人唯一标识
name: 机器人名称
robot_type: 机器人类型
"""
self.id = robot_id
self.name = name
self.type = robot_type
# 状态属性
self.state = RobotState.IDLE
self.battery_level = 100.0 # 初始电量100%
self.position = Position(0.0, 0.0, 0.0, "充电站")
self.speed = 1.0 # m/s
# 任务相关
self.current_task = None
self.task_queue = []
self.completed_tasks = []
# 传感器数据（模拟）
self.sensors = {
'battery_sensor': True,
'proximity_sensor': True,
'position_sensor': True,
'obstacle_sensor': True
}
# 时间戳
self.created_time = datetime.now()
self.last_update_time = datetime.now()
self.last_charging_time = None
print(f"🤖 机器人 {name}({robot_id}) 初始化完成")
def get_status(self) -> Dict:
"""获取机器人状态信息"""
return {
'id': self.id,
'name': self.name,
'type': self.type.value,
'battery_level': round(self.battery_level, 1),
'state': self.state.value,
'position': str(self.position),
'current_task': self.current_task,
'tasks_queued': len(self.task_queue),
'tasks_completed': len(self.completed_tasks)
}
def set_state(self, new_state: RobotState):
"""设置机器人状态"""
old_state = self.state
self.state = new_state
self.last_update_time = datetime.now()
print(f"🔄 机器人状态变更: {old_state.value} -> {new_state.value}")
# 状态变更时的特殊处理
if new_state == RobotState.CHARGING:
self.last_charging_time = datetime.now()
def update_battery(self, new_level: float):
"""更新电池电量"""
if not (0 <= new_level <= 100):
raise ValueError("电量值必须在0-100之间")
old_level = self.battery_level
self.battery_level = new_level
self.last_update_time = datetime.now()
# 电量告警
if old_level > 20 and new_level <= 20:
print(f"⚠️ 电量告警: {old_level}% -> {new_level}%，请及时充电！")
elif new_level == 100:
print("✅ 电池已充满")
def start_charging(self):
"""开始充电"""
if self.state == RobotState.CHARGING:
print("ℹ️ 机器人已在充电中")
return
self.set_state(RobotState.CHARGING)
self.position = Position(0.0, 0.0, 0.0, "充电站")
print(f"🔌 {self.name} 开始充电")
def stop_charging(self):
"""停止充电"""
if self.state != RobotState.CHARGING:
print("ℹ️ 机器人未在充电")
return
self.set_state(RobotState.IDLE)
print(f"🔋 {self.name} 充电结束")
def move_to(self, target_position: Position) -> bool:
"""移动到指定位置"""
if self.state == RobotState.CHARGING:
print("❌ 充电中的机器人无法移动")
return False
if self.battery_level < 10:
print("❌ 电量不足，无法执行移动任务")
return False
print(f"🚀 {self.name} 从 {self.position} 移动到 {target_position}")
# 模拟移动过程
self.set_state(RobotState.MOVING)
# 计算移动距离和时间
distance = self._calculate_distance(self.position, target_position)
move_time = distance / self.speed
print(f" 距离: {distance:.2f}m, 预计时间: {move_time:.1f}s")
# 模拟电量消耗
battery_consumption = distance * 0.5 # 每米消耗0.5%电量
new_battery = max(0, self.battery_level - battery_consumption)
self.update_battery(new_battery)
# 更新位置
self.position = target_position
self.set_state(RobotState.WORKING)
print(f"✅ 移动完成，当前电量: {self.battery_level:.1f}%")
return True
def perform_task(self, task_description: str) -> bool:
"""执行任务"""
if self.state == RobotState.CHARGING:
print("❌ 充电中的机器人无法执行任务")
return False
if self.battery_level < 15:
print("❌ 电量不足，无法执行任务")
return False
print(f"📋 {self.name} 开始执行任务: {task_description}")
self.set_state(RobotState.WORKING)
self.current_task = task_description
# 模拟任务执行
task_duration = 5 # 模拟任务耗时
for i in range(task_duration):
time.sleep(0.5)
# 任务执行中消耗电量
self.update_battery(max(0, self.battery_level - 1))
print(f" 任务进度: {(i+1)*20}% | 电量: {self.battery_level:.1f}%")
# 任务完成
self.completed_tasks.append({
'task': task_description,
'completed_time': datetime.now(),
'battery_used': 5 # 任务消耗5%电量
})
self.current_task = None
self.set_state(RobotState.IDLE)
print(f"✅ 任务完成: {task_description}")
return True
def _calculate_distance(self, pos1: Position, pos2: Position) -> float:
"""计算两点间距离"""
import math
dx = pos2.x - pos1.x
dy = pos2.y - pos1.y
dz = pos2.z - pos1.z
return math.sqrt(dx**2 + dy**2 + dz**2)
def show_detailed_status(self):
"""显示详细状态信息"""
print(f"\n🤖 机器人详细信息:")
print(f" ID: {self.id}")
print(f" 名称: {self.name}")
print(f" 类型: {self.type.value}")
print(f" 当前状态: {self.state.value}")
print(f" 电池电量: {self.battery_level:.1f}%")
print(f" 当前位置: {self.position}")
print(f" 移动速度: {self.speed} m/s")
print(f" 当前任务: {self.current_task or '无'}")
print(f" 队列任务数: {len(self.task_queue)}")
print(f" 完成任务数: {len(self.completed_tasks)}")
print(f" 创建时间: {self.created_time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f" 最后更新: {self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f" 最后充电: {self.last_charging_time.strftime('%Y-%m-%d %H:%M:%S') if self.last_charging_time else '从未充电'}")
class CleanRobot(Robot):
"""清洁机器人子类"""
def __init__(self, robot_id: str, name: str):
super().__init__(robot_id, name, RobotType.CLEANER)
self.clean_area = 0.0 # 已清洁面积
self.clean_mode = "自动清洁"
def clean_area_task(self, area_name: str, area_size: float):
"""清洁指定区域"""
task_desc = f"清洁{area_name}({area_size}㎡)"
success = self.perform_task(task_desc)
if success:
self.clean_area += area_size
print(f"🧹 累计清洁面积: {self.clean_area:.1f}㎡")
return success
battery_manager.py - 电池管理模块
"""
电池管理模块
实现电池状态监控、电量管理和充放电控制
基于智能机器人导论的能源管理理论
"""
import time
import threading
from datetime import datetime, timedelta
from enum import Enum
from typing import Dict, List, Optional
from dataclasses import dataclass
class BatteryStatus(Enum):
"""电池状态枚举"""
NORMAL = "正常"
LOW = "低电量"
CRITICAL = "严重低电量"
CHARGING = "充电中"
FULL = "已充满"
FAULT = "故障"
class BatteryType(Enum):
"""电池类型枚举"""
LITHIUM = "锂电池"
LEAD_ACID = "铅酸电池"
NIMH = "镍氢电池"
@dataclass
class BatteryInfo:
"""电池信息数据类"""
capacity: float # 电池容量 (Wh)
voltage: float # 额定电压 (V)
current: float # 当前电流 (A)
temperature: float = 25.0 # 温度 (°C)
cycle_count: int = 0 # 充放电循环次数
health_status: str = "良好" # 健康状态
class BatteryManager:
"""电池管理器 - 智能电池管理系统"""
def __init__(self):
"""初始化电池管理器"""
# 机器人电池数据存储
self.robot_batteries: Dict[str, Dict] = {}
# 电池配置
self.low_battery_threshold = 20.0 # 低电量阈值
self.critical_threshold = 10.0 # 严重低电量阈值
self.full_charge_threshold = 95.0 # 充满电阈值
# 电池物理参数（模拟）
self.battery_config = {
'capacity': 50.0, # 50Wh
'voltage': 24.0, # 24V
'discharge_rate': 0.1, # 每小时自放电率
'charge_rate': 0.3, # 每小时充电速率
'temperature_range': (0, 45) # 工作温度范围
}
# 监控线程
self.monitoring_active = False
self.monitor_thread = None
print("🔋 电池管理系统初始化完成")
def register_robot_battery(self, robot_id: str, initial_level: float = 100.0):
"""
注册机器人电池
Args:
robot_id: 机器人ID
initial_level: 初始电量百分比
"""
if not (0 <= initial_level <= 100):
raise ValueError("初始电量必须在0-100之间")
battery_info = BatteryInfo(
capacity=self.battery_config['capacity'],
voltage=self.battery_config['voltage'],
current=0.0,
temperature=25.0
)
self.robot_batteries[robot_id] = {
'level': initial_level,
'status': self._determine_battery_status(initial_level),
'info': battery_info,
'last_updated': datetime.now(),
'charging_start_time': None,
'discharge_start_time': None,
'total_discharged': 0.0, # 累计放电量
'total_charged': 0.0 # 累计充电量
}
print(f"✅ 已为机器人 {robot_id} 注册电池，初始电量: {initial_level}%")
def get_battery_level(self, robot_id: str) -> float:
"""
获取机器人电池电量
Args:
robot_id: 机器人ID
Returns:
float: 电量百分比
"""
if robot_id not in self.robot_batteries:
raise ValueError(f"未找到机器人 {robot_id} 的电池信息")
return self.robot_batteries[robot_id]['level']
def set_battery_level(self, robot_id: str, level: float):
"""
设置机器人电池电量
Args:
robot_id: 机器人ID
level: 电量百分比
"""
if robot_id not in self.robot_batteries:
self.register_robot_battery(robot_id, level)
return
if not (0 <= level <= 100):
raise ValueError("电量值必须在0-100之间")
old_level = self.robot_batteries[robot_id]['level']
self.robot_batteries[robot_id]['level'] = level
self.robot_batteries[robot_id]['status'] = self._determine_battery_status(level)
关注我，有更多实用程序等着你！
)
