老旧配电室动环升级：以太网温湿度传感器无布线改造方案

┌────────────────────────────────────────────────────────────────────┐
│  方案          │ 通信方式      │ 供电方式      │ 适用场景          │
├────────────────┼──────────────┼──────────────┼──────────────────┤
│ A 工业Wi-Fi    │ 802.11ac 双频│ PoE 注入器   │ 配电室有插座/PDU  │
│  以太网传感器  │ + WPA2-Enterprise│ 就近取电   │ 距离 AP ≤15m     │
│                │ Modbus TCP   │              │                  │
├────────────────┼──────────────┼──────────────┼──────────────────┤
│ B 无线网桥     │ 5GHz 点对点  │ 24V DC 外接  │ 跨防火分区/穿墙  │
│   透传         │ 透明串口     │ 柜内取DC12V  │ 距离 ≤500m       │
├────────────────┼──────────────┼──────────────┼──────────────────┤
│ C LoRa 扩频    │ 470MHz 私有  │ 内置电池     │ 完全无法取电      │
│   + 边缘汇聚   │ 星型组网     │ 3~5 年寿命   │ 传感器 ≤20 个    │
└────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│  配电室内                                                         │
│                                                                   │
│  ┌─────────┐  Wi-Fi    ┌──────────┐  5GHz 无线网桥   ┌────────┐ │
│  │ 传感器1 │◄─────────►│ 工业AP   │◄────────────────►│ 网桥   │ │
│  │ (高压区)│           │(导轨安装) │                   │ 远端   │ │
│  └─────────┘           └────┬─────┘                  └───┬────┘ │
│                             │                              │      │
│  ┌─────────┐  Wi-Fi         │ PoE 供电                     │      │
│  │ 传感器2 │◄───────────────┘                              │      │
│  │ (低压区)│                                              │      │
│  └─────────┘                                              ▼      │
│                                                           监控中心 │
│  ┌─────────┐  Wi-Fi                                      │      │
│  │ 传感器3 │◄─────────────────────────────────────────────►│ 服务器│
│  │ (电缆层)│          边缘网关(本地缓存+协议转换)           │      │
│  └─────────┘                                              └──────┘
└─────────────────────────────────────────────────────────────────┘

# 网桥链路质量监测（边缘网关侧）
import subprocess, re

def check_bridge_link(bridge_ip: str) -> dict:
    """检测 5GHz 网桥链路质量"""
    try:
        rtt = subprocess.check_output(
            f"ping -c 5 -W 2 {bridge_ip}", shell=True
        ).decode()
        loss = re.search(r'(\d+)% packet loss', rtt)
        avg = re.search(r'avg = [\d.]+/([\d.]+)/', rtt)
        return {
            'reachable': loss and int(loss.group(1)) < 20,
            'packet_loss': int(loss.group(1)) if loss else 100,
            'avg_rtt': float(avg.group(1)) if avg else 999,
            'quality': 'GOOD' if int(loss.group(1)) < 5 else 'DEGRADED'
        }
    except Exception:
        return {'reachable': False, 'quality': 'DOWN'}

供电路径优先级：

1. 柜内 PDU 插座（最优选）
   → PoE 注入器 → 传感器
   → 施工量：插拔即可，<5min/台

2. 柜内 DC12V/24V 辅助电源端子
   → DC-DC 降压 → PoE 注入器 → 传感器
   → 施工量：接线 15min/台

3. 独立 220V 防水插座（沿墙明装）
   → 沿踢脚线走线槽（不破坏墙体）
   → 施工量：30min/台

4. 内置锂电池（最后手段）
   → 充电周期 6 个月
   → 仅用于完全无法取电的死角

配电室传感器 VLAN 划分：
  VLAN 100 → 管理网（AP 管理、网关配置）
  VLAN 101 → 数据网（传感器 Modbus TCP 数据）
  VLAN 102 → 视频网（如有摄像头）

传感器 IP 分配：
  172.20.101.11 ~ 172.20.101.30 /24
  网关：172.20.101.1

import asyncio
from pymodbus.client import AsyncModbusTcpClient

class SensorPoller:
    def __init__(self, sensors: list):
        self.sensors = sensors

    async def poll_one(self, sensor: dict):
        try:
            client = AsyncModbusTcpClient(sensor['ip'], port=502)
            await client.connect()
            rr = await client.read_holding_registers(0, 2, slave=1)
            await client.close()
            if rr.isError():
                return {**sensor, 'error': 'read failed'}
            return {
                'ip': sensor['ip'],
                'zone': sensor['zone'],
                'humidity': rr.registers[0] / 10,
                'temperature': rr.registers[1] / 10,
                'ts': time.time()
            }
        except Exception as e:
            return {**sensor, 'error': str(e)}

    async def poll_all(self):
        tasks = [self.poll_one(s) for s in self.sensors]
        return await asyncio.gather(*tasks)

class EdgeBuffer:
    def __init__(self, max_points=10000):
        self.buffer = []
        self.max = max_points

    def push(self, data: dict):
        self.buffer.append(data)
        if len(self.buffer) > self.max:
            self.buffer.pop(0)

    def flush(self, upload_func) -> int:
        sent = 0
        for d in list(self.buffer):
            if upload_func(d):
                self.buffer.remove(d)
                sent += 1
        return sent