pip install fastapi uvicorn

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
from datetime import datetime

app = FastAPI()

# 模拟硬件状态，实际应用中这会由后台脚本或数据库更新
# 可以使用全局变量、消息队列（如Redis Pub/Sub）或共享内存来管理状态
current_hardware_status = {"status": "UNKNOWN", "timestamp": datetime.now().isoformat()}

# 用于存储等待通知的客户端请求（更复杂的场景会用队列或Pub/Sub）
# 简单示例中，我们直接在生成器中检查状态
# 注意：此示例的全局变量方式不适合多进程或多实例部署，仅用于概念演示。
# 生产环境建议使用 Redis Pub/Sub 等机制。

async def update_hardware_status_externally(new_status: str):
    """模拟外部脚本更新硬件状态的函数"""
    global current_hardware_status
    current_hardware_status = {
        "status": new_status,
        "timestamp": datetime.now().isoformat()
    }
    print(f"Hardware status updated to: {new_status}")

async def sse_event_generator(request: Request):
    """
    SSE事件生成器。
    它会持续检查硬件状态是否更新，并在更新时发送数据。
    """
    last_sent_status = None
    while True:
        # 检查客户端是否断开连接
        if await request.is_disconnected():
            print("Client disconnected.")
            break

        global current_hardware_status
        # 如果当前状态与上次发送的状态不同，则发送新事件
        if current_hardware_status != last_sent_status:
            event_data = {
                "id": datetime.now().timestamp(), # 事件ID，用于客户端自动重连时定位
                "data": json.dumps(current_hardware_status) # 数据字段
            }
            # SSE数据格式：data: [your_json_data]\n\n
            # 也可以包含 event: [event_type]\n
            yield f"data: {event_data['data']}\n\n"
            last_sent_status = current_hardware_status.copy() # 更新上次发送的状态

        # 每隔一段时间检查一次状态，避免CPU空转
        await asyncio.sleep(1) # 1秒检查一次

@app.get("/hardware-status-stream")
async def hardware_status_stream(request: Request):
    """
    SSE端点，提供硬件状态的实时流。
    """
    return StreamingResponse(sse_event_generator(request), media_type="text/event-stream")

@app.post("/simulate-status-change")
async def simulate_status_change(status: str):
    """
    一个模拟的API，用于外部触发硬件状态更新。
    """
    await update_hardware_status_externally(status)
    return {"message": f"Hardware status simulated to: {status}"}

# 运行FastAPI应用: uvicorn main:app --reload

import React, { useState, useEffect } from 'react';

function HardwareStatusMonitor() {
  const [status, setStatus] = useState('连接中...');
  const [timestamp, setTimestamp] = useState('');
  const [error, setError] = useState(null);

  useEffect(() => {
    // 创建EventSource实例，指向FastAPI的SSE端点
    const eventSource = new EventSource('http://localhost:8000/hardware-status-stream');

    eventSource.onopen = () => {
      console.log('SSE 连接已建立。');
      setError(null); // 清除任何之前的错误
    };

    eventSource.onmessage = (event) => {
      // 接收到服务器推送的数据
      try {
        const data = JSON.parse(event.data);
        setStatus(data.status);
        setTimestamp(data.timestamp);
        console.log('接收到状态更新:', data);
      } catch (e) {
        console.error('解析SSE数据失败:', e);
        setError('数据解析错误');
      }
    };

    eventSource.onerror = (err) => {
      console.error('SSE 连接错误:', err);
      eventSource.close(); // 关闭当前连接
      setError('连接错误，尝试重连...');
      // 可以在这里实现更复杂的重连逻辑，EventSource默认会尝试重连
    };

    // 组件卸载时关闭EventSource连接
    return () => {
      console.log('关闭 SSE 连接。');
      eventSource.close();
    };
  }, []); // 空依赖数组表示只在组件挂载和卸载时执行

  return (
    <div>
      <h1>硬件状态实时监控</h1>
      {error && <p style={{ color: 'red' }}>错误: {error}</p>}
      <p>当前状态: <strong>{status}</strong></p>
      <p>更新时间: {timestamp}</p>
      <p>（尝试使用 `curl -X POST -d "status=ONLINE" http://localhost:8000/simulate-status-change` 模拟状态变化）</p>
    </div>
  );
}

export default HardwareStatusMonitor;