🤖 第18课：远程监控

📌 远程监控概述

远程监控让运维人员随时掌握机器人状态，及时发现和处理异常：

📌 遥测数据采集

import json, time, random

class TelemetryCollector:
    """遥测数据采集器"""
    def __init__(self):
        self.metrics = {
            "battery": 95.0,
            "cpu_usage": 25.0,
            "memory_usage": 40.0,
            "temperature": 42.0,
            "nav_status": "idle",
            "position": {"x": 0, "y": 0, "floor": 1},
            "tasks_completed": 0,
            "uptime": 3600,
        }
        self.alert_rules = {
            "battery_low": {"metric": "battery", "threshold": 20, "operator": "<", "level": "critical"},
            "battery_medium": {"metric": "battery", "threshold": 40, "operator": "<", "level": "warning"},
            "cpu_high": {"metric": "cpu_usage", "threshold": 80, "operator": ">", "level": "warning"},
            "temp_high": {"metric": "temperature", "threshold": 60, "operator": ">", "level": "critical"},
            "memory_high": {"metric": "memory_usage", "threshold": 85, "operator": ">", "level": "warning"},
        }
        self.history = []

    def update(self):
        """模拟指标更新"""
        self.metrics["battery"] = max(0, self.metrics["battery"] - random.uniform(0.1, 0.5))
        self.metrics["cpu_usage"] = min(100, max(5, self.metrics["cpu_usage"] + random.uniform(-5, 5)))
        self.metrics["memory_usage"] = min(100, max(20, self.metrics["memory_usage"] + random.uniform(-2, 2)))
        self.metrics["temperature"] = max(30, min(70, self.metrics["temperature"] + random.uniform(-2, 2)))
        self.metrics["uptime"] += 1
        self.history.append(dict(self.metrics))

    def check_alerts(self):
        """检查告警"""
        alerts = []
        for rule_name, rule in self.alert_rules.items():
            value = self.metrics[rule["metric"]]
            threshold = rule["threshold"]
            triggered = value < threshold if rule["operator"] == "<" else value > threshold
            if triggered:
                alerts.append({
                    "rule": rule_name,
                    "metric": rule["metric"],
                    "value": value,
                    "threshold": threshold,
                    "level": rule["level"],
                })
        return alerts

    def get_snapshot(self):
        return dict(self.metrics)

telemetry = TelemetryCollector()
print("远程监控 - 遥测数据采集")
print("=" * 55)

random.seed(42)
for step in range(20):
    telemetry.update()
    alerts = telemetry.check_alerts()
    if alerts or step % 5 == 0:
        snap = telemetry.get_snapshot()
        print(f"\n步骤{step}: 电量{snap['battery']:.1f}% CPU{snap['cpu_usage']:.1f}% 温度{snap['temperature']:.1f}°C")
        for alert in alerts:
            icon = "🔴" if alert["level"] == "critical" else "🟡"
            print(f"  {icon} {alert['rule']}: {alert['metric']}={alert['value']:.1f} (阈值{alert['threshold']})")

print("\n✅ 遥测采集验证通过")

📌 远程控制

class RemoteController:
    """远程控制器"""
    def __init__(self):
        self.commands = []
        self.robot_state = "idle"
        self.position = [0, 0]
        self.authorized = False

    def authenticate(self, token):
        """认证"""
        self.authorized = token == "admin_token_123"
        return self.authorized

    def send_command(self, cmd_type, params=None):
        """发送远程命令"""
        if not self.authorized:
            return {"success": False, "error": "未认证"}

        cmd = {"type": cmd_type, "params": params or {}, "timestamp": time.time()}
        self.commands.append(cmd)

        if cmd_type == "goto":
            target = params.get("target", (0,0))
            self.robot_state = "navigating"
            self.position = list(target)
            return {"success": True, "message": f"导航到{target}"}
        elif cmd_type == "stop":
            self.robot_state = "idle"
            return {"success": True, "message": "已停止"}
        elif cmd_type == "return_home":
            self.robot_state = "returning"
            self.position = [0, 0]
            return {"success": True, "message": "返回充电桩"}
        elif cmd_type == "restart":
            self.robot_state = "idle"
            return {"success": True, "message": "重启完成"}
        elif cmd_type == "screenshot":
            return {"success": True, "message": "截图已保存"}
        return {"success": False, "error": "未知命令"}

import time
ctrl = RemoteController()
print("远程控制模拟")
print("=" * 55)

ctrl.authenticate("admin_token_123")
commands = [
    ("goto", {"target": (5, 3)}),
    ("stop", {}),
    ("goto", {"target": (10, 8)}),
    ("screenshot", {}),
    ("return_home", {}),
    ("restart", {}),
]
for cmd_type, params in commands:
    result = ctrl.send_command(cmd_type, params)
    print(f"  {'✅' if result['success'] else '❌'} {cmd_type}: {result['message']}")

print(f"\n命令历史: {len(ctrl.commands)}条")
print("✅ 远程控制验证通过")

📌 告警管理

class AlertManager:
    """告警管理器"""
    def __init__(self):
        self.alerts = []
        self.escalation_rules = {
            "critical": {"notify": ["管理员","主管"], "timeout": 300, "action": "自动处理"},
            "warning": {"notify": ["管理员"], "timeout": 1800, "action": "记录日志"},
            "info": {"notify": [], "timeout": 3600, "action": "记录日志"},
        }
        self.auto_actions = {
            "battery_low": "自动返回充电",
            "temp_high": "降低运行频率",
            "stuck": "尝试脱困",
            "network_lost": "切换离线模式",
        }

    def create_alert(self, alert_type, level, message, auto_handle=True):
        """创建告警"""
        alert = {
            "type": alert_type, "level": level, "message": message,
            "timestamp": time.time(), "handled": False, "auto_handled": False
        }
        self.alerts.append(alert)

        rule = self.escalation_rules.get(level, self.escalation_rules["info"])
        print(f"  {'🔴' if level=='critical' else '🟡' if level=='warning' else 'ℹ️'} [{level}] {message}")
        print(f"    通知: {', '.join(rule['notify'])}")

        if auto_handle and alert_type in self.auto_actions:
            action = self.auto_actions[alert_type]
            alert["auto_handled"] = True
            print(f"    自动处理: {action}")
        
        return alert

import time
am = AlertManager()
print("告警管理模拟")
print("=" * 55)

am.create_alert("battery_low", "critical", "电量仅剩15%，即将关机")
am.create_alert("temp_high", "warning", "CPU温度达到65°C")
am.create_alert("stuck", "critical", "机器人被困，无法移动")
am.create_alert("network_lost", "warning", "网络连接丢失")
am.create_alert("task_delay", "info", "任务T003预计延迟5分钟")

handled = sum(1 for a in am.alerts if a["auto_handled"])
print(f"\n告警统计: 总计{len(am.alerts)}个, 自动处理{handled}个")
print("✅ 告警管理验证通过")

📌 监控架构

机器人 → 采集Agent → MQTT/Kafka → 时序数据库
                                         ↓
                              Grafana仪表盘 ← 告警引擎
                              远程控制台   ← 日志服务

📌 练习

📌 成就