Implementing a Modular Master-Agent Telemetry & Diagnostic Framework in Python: Prime-Sentinel Command (PSC)

[ Prime Program (Central Controller) ]
        |                   |
        v                   v
[ Sentinel Node 001 ]  [ Sentinel Node 002 ]  ... [ Sentinel Node N ]
  (Self-Diagnostic)      (Self-Diagnostic)          (Self-Diagnostic)

# 作者：Ahmad Mateen Ishanzai 博士
# 框架：Prime-Sentinel命令（PSC）
# 架构：主代理集中式编排

import random
import time
from typing import Any, Dict, List

class SentinelProgram:
    """表示处理本地健康采样的自主边缘节点。"""

    def __init__(self, sentinel_id: str):
        self.sentinel_id = sentinel_id
        self.status = "INITIALIZED"

    def run_diagnostics(self) -> Dict[str, Any]:
        """执行诊断流程并生成遥测负载。"""
        print(f"[PSC-Sentinel-{self.sentinel_id}] 运行系统诊断...")
        time.sleep(1)

        # 模拟硬件采样
        cpu_load = round(random.uniform(10.0, 85.0), 2)
        memory_usage = round(random.uniform(30.0, 90.0), 2)
        network_status = "STABLE" if cpu_load < 80.0 else "DEGRADED"

        return {
            "sentinel_id": self.sentinel_id,
            "cpu_load_pct": cpu_load,
            "memory_usage_pct": memory_usage,
            "network_status": network_status,
        }

class PrimeProgram:
    """管理代理部署、流程和遥测采集的主控制器。"""

    def __init__(self, system_name: str = "Prime-Sentinel Command (PSC)"):
        self.system_name = system_name
        self.sentinels: List[SentinelProgram] = []

    def deploy_sentinels(self, count: int) -> None:
        """动态实例化并注册Sentinel代理节点。"""
        print(f"[{self.system_name}] 正在部署{count}个Sentinel单元...")
        for i in range(1, count + 1):
            sentinel_id = f"00{i}" if i < 10 else f"0{i}"
            self.sentinels.append(SentinelProgram(sentinel_id=sentinel_id))
        print(
            f"[{self.system_name}] {len(self.sentinels)}个Sentinels已成功连接。"
        )

    def execute_routine(self) -> None:
        """向所有已注册代理分发诊断扫描。"""
        print("=" * 60)
        print(f"[{self.system_name}] 正在执行系统健康检查")
        print("=" * 60)

        reports = [sentinel.run_diagnostics() for sentinel in self.sentinels]
        self._analyze_reports(reports)

    def _analyze_reports(self, reports: List[Dict[str, Any]]) -> None:
        """聚合并格式化接收到的代理遥测数据。"""
        print(f"\n--- {self.system_name} 遥测报告 ---")
        for report in reports:
            print(
                f"Sentinel {report['sentinel_id']} -> "
                f"CPU: {report['cpu_load_pct']}% | "
                f"RAM: {report['memory_usage_pct']}% | "
                f"状态: {report['network_status']}"
            )
        print("-" * 50)
        print(
            f"[{self.system_name}] 运行检查完成。所有单元状态正常。"
        )

if __name__ == "__main__":
    psc_system = PrimeProgram()
    psc_system.deploy_sentinels(count=3)
    psc_system.execute_routine()

[Prime-Sentinel Command (PSC)] 正在部署3个Sentinel单元...
[Prime-Sentinel Command (PSC)] 3个Sentinels已成功连接。
============================================================
[Prime-Sentinel Command (PSC)] 正在执行系统健康检查
============================================================
[PSC-Sentinel-001] 正在运行系统诊断...
[PSC-Sentinel-002] 正在运行系统诊断...
[PSC-Sentinel-003] 正在运行系统诊断...

--- Prime-Sentinel Command (PSC) 遥测报告 ---
Sentinel 001 -> CPU: 24.15% | RAM: 45.20% | 状态: STABLE
Sentinel 002 -> CPU: 68.90% | RAM: 72.10% | 状态: STABLE
Sentinel 003 -> CPU: 12.05% | RAM: 38.40% | 状态: STABLE
--------------------------------------------------
[Prime-Sentinel Command (PSC)] 运行检查完成。所有单元状态正常。