机柜内部微环境监测：以太网温湿度传感器点位布局与采样策略设计

┌─────────────────────────────────────────────────────────┐
│ 机柜前视图（正视）                                      │
│                                                         │
│  ▲ 42U                                                │
│  │ ┌───────────────────────────────────────────────┐  │
│  │ │  交换机/路由器（高发热）                      │  │ 38~42℃
│  │ ├───────────────────────────────────────────────┤  │
│  │ │  空白区域（无设备/盲板缺失）                  │  │ 25~28℃
│  │ ├───────────────────────────────────────────────┤  │
│  │ │  服务器1U/2U（中发热）                       │  │ 32~36℃
│  │ ├───────────────────────────────────────────────┤  │
│  │ │  服务器/存储（高发热）                        │  │ 35~40℃
│  │ ├───────────────────────────────────────────────┤  │
│  │ │  配线架/PDU（低发热）                         │  │ 22~25℃
│ 0U └───────────────────────────────────────────────┘  │ 20~22℃
│                                                         │
│  温度梯度：底部→顶部 约 5~15℃ 温差                    │
│  热点集中区：机柜中上部（30U~42U）                     │
│  冷点集中区：机柜底部（0U~10U）                        │
└─────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│ 机柜侧剖面图（左视）                                       │
│                                                             │
│  前门(冷通道)              设备区              后门(热通道)  │
│  ◀──────────▶            ┌────────┐        ◀──────────▶  │
│                            │        │                      │
│         P3●                │ 设备   │                ●P4  │
│        (底部)              │        │              (后进风) │
│                            │        │                      │
│                            │        │                      │
│         P5●                │ 设备   │                      │
│        (前出风)            │        │                      │
│                            │        │                      │
│                            │ 设备   │                      │
│                            │        │                      │
│         P2●                │ 设备   │                      │
│        (中部)              │        │                      │
│                            │        │                      │
│                            │ 设备   │                      │
│                            │        │                      │
│         P1●                │ 设备   │                ●     │
│        (顶部)              │        │            (后上部) │
│                            └────────┘                      │
│                                                             │
│  ● = 传感器安装位置                                        │
│  P1~P3：基础3点布局（每柜必装）                            │
│  P4~P5：高密度增强点位                                      │
└─────────────────────────────────────────────────────────────┘

# installation_spec.py
"""传感器安装规范"""

INSTALLATION_RULES = {
    "positioning": {
        "avoid_airflow_shadow": "传感器不得安装在设备正前方挡风板后方（涡流死区）",
        "distance_from_heat_source": "距设备排风口>10cm，避免直接热辐射干扰",
        "cable_routing": "网线沿机柜立柱走线槽，不得横跨设备正面或阻挡通风孔",
        "orientation": "传感器感应面朝外（朝向通道），不朝设备面板"
    },
    "environmental": {
        "max_ambient": "传感器工作温度范围需覆盖-20~+80℃（机柜顶部可能超40℃）",
        "humidity_range": "0~100%RH，带防凝露涂层",
        "ingress_protection": "IP30以上（机柜内部无喷淋风险）"
    },
    "network": {
        "cable_type": "Cat5e/Cat6 UTP，最长100m",
        "poewhen": "优先PoE供电（802.3af），避免额外电源线",
        "switch_port": "每机柜列配置独立接入交换机，上联至汇聚"
    },
    "labeling": {
        "format": "R{row}-C{cabinet}-P{point}",
        "example": "R1-C03-P1 表示第1排第3号机柜顶部传感器",
        "mapping_table": "现场标签+CMDB同步"
    }
}

# adaptive_sampling.py
import time
from collections import deque

class AdaptiveSampler:
    """自适应采样策略——根据环境变化动态调整采样频率"""

    def __init__(self, sensor_id: str, base_interval: float = 30.0):
        self.sensor_id = sensor_id
        self.base_interval = base_interval
        self.current_interval = base_interval
        self.recent_readings = deque(maxlen=10)
        self.last_sample_time = 0
        self.consecutive_stable = 0

    def should_sample(self, current_time: float) -> bool:
        """判断当前时刻是否应该采样"""
        elapsed = current_time - self.last_sample_time
        return elapsed >= self.current_interval

    def on_sample(self, value: float, timestamp: float):
        """采样后更新策略"""
        self.recent_readings.append(value)
        self.last_sample_time = timestamp

        if len(self.recent_readings) < 3:
            return

        # 计算最近读数的变化率
        recent = list(self.recent_readings)[-3:]
        max_delta = max(abs(recent[i] - recent[i-1]) for i in range(1, len(recent)))

        if max_delta < 0.3:
            # 环境稳定，降低采样频率
            self.consecutive_stable += 1
            if self.consecutive_stable > 5:
                self.current_interval = min(self.base_interval * 2, 120.0)
        elif max_delta > 2.0:
            # 环境快速变化，提高采样频率
            self.consecutive_stable = 0
            self.current_interval = max(self.base_interval / 3, 5.0)
        else:
            # 正常变化，恢复基础频率
            self.consecutive_stable = 0
            self.current_interval = self.base_interval

# data_aggregation.py
from dataclasses import dataclass
from typing import List, Dict
import statistics

@dataclass
class AggregatedReading:
    sensor_id: str
    cabinet_id: str
    avg_value: float
    max_value: float
    min_value: float
    std_dev: float
    sample_count: int
    window_seconds: int

class CabinetDataAggregator:
    """机柜级数据聚合——将多点传感器数据汇总为机柜健康指标"""

    def __init__(self, cabinet_id: str, point_ids: List[str]):
        self.cabinet_id = cabinet_id
        self.point_ids = point_ids
        self.point_buffers: Dict[str, List[float]] = {pid: [] for pid in point_ids}

    def add_reading(self, point_id: str, value: float):
        if point_id in self.point_buffers:
            self.point_buffers[point_id].append(value)

    def aggregate_window(self, window_seconds: int = 300) -> dict:
        """5分钟窗口聚合"""
        result = {}

        all_values = []
        for pid, buf in self.point_buffers.items():
            if buf:
                result[pid] = {
                    "avg": statistics.mean(buf),
                    "max": max(buf),
                    "min": min(buf),
                    "count": len(buf)
                }
                all_values.extend(buf)
                buf.clear()

        # 机柜级综合指标
        if all_values:
            result["cabinet"] = {
                "avg_temp": statistics.mean(all_values),
                "max_temp": max(all_values),
                "min_temp": min(all_values),
                "gradient": self._calc_gradient(),
                "hotspot_count": sum(1 for v in all_values if v > 35),
                "cold_spot_count": sum(1 for v in all_values if v < 18)
            }

        return result

    def _calc_gradient(self) -> float:
        """计算垂直温度梯度（顶-底温差）"""
        top_buf = self.point_buffers.get(self.point_ids[0], [])
        bottom_buf = self.point_buffers.get(self.point_ids[-1], [])
        if top_buf and bottom_buf:
            return statistics.mean(top_buf) - statistics.mean(bottom_buf)
        return 0.0

    def detect_hotspot(self, threshold: float = 35.0) -> List[str]:
        """检测热点点位"""
        hotspots = []
        for pid, buf in self.point_buffers.items():
            if buf and statistics.mean(buf) > threshold:
                hotspots.append(pid)
        return hotspots

# sensor_network.py
import socket
import struct
import json
from typing import Dict, List

class EthernetSensorClient:
    """以太网温湿度传感器通信客户端"""

    def __init__(self, host: str, port: int = 502, timeout: float = 3.0):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock = None

    def connect(self) -> bool:
        try:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.sock.settimeout(self.timeout)
            self.sock.connect((self.host, self.port))
            return True
        except Exception:
            return False

    def read_modbus_tcp(self, unit_id: int, register: int, count: int = 2) -> List[int]:
        """读取Modbus TCP寄存器"""
        if not self.sock:
            self.connect()

        # 构造Modbus TCP请求帧
        transaction_id = 1
        protocol_id = 0
        length = 6
        pdu = struct.pack('>BBHH', unit_id, 0x03, register, count)

        header = struct.pack('>HHH', transaction_id, protocol_id, length)
        request = header + pdu

        self.sock.send(request)
        response = self.sock.recv(1024)

        # 解析响应
        if len(response) < 9:
            return []

        byte_count = response[8]
        values = []
        for i in range(byte_count // 2):
            offset = 9 + i * 2
            val = struct.unpack('>H', response[offset:offset+2])[0]
            values.append(val)

        return values

    def read_temperature_humidity(self) -> dict:
        """读取温湿度（假设寄存器映射：40001=温度×10，40002=湿度×10）"""
        try:
            regs = self.read_modbus_tcp(1, 40001, 2)
            if len(regs) >= 2:
                temp = regs[0] / 10.0
                humi = regs[1] / 10.0
                return {"temperature": temp, "humidity": humi, "status": "ok"}
        except Exception as e:
            pass
        return {"temperature": None, "humidity": None, "status": "error"}

    def read_http_json(self, endpoint: str = "/api/v1/sensor") -> dict:
        """HTTP API方式读取（部分传感器支持）"""
        import urllib.request
        url = f"http://{self.host}{endpoint}"
        try:
            req = urllib.request.Request(url)
            resp = urllib.request.urlopen(req, timeout=self.timeout)
            data = json.loads(resp.read().decode())
            return data
        except Exception:
            return {}

    def disconnect(self):
        if self.sock:
            self.sock.close()
            self.sock = None

class SensorPollingManager:
    """多传感器轮询管理"""

    def __init__(self, sensors: Dict[str, dict]):
        """
        sensors: {sensor_id: {"host": ip, "port": port, "cabinet": cabinet_id}}
        """
        self.sensors = sensors
        self.clients: Dict[str, EthernetSensorClient] = {}

    def poll_all(self) -> Dict[str, dict]:
        """轮询所有传感器"""
        results = {}
        for sid, cfg in self.sensors.items():
            if sid not in self.clients:
                self.clients[sid] = EthernetSensorClient(cfg["host"], cfg.get("port", 502))

            data = self.clients[sid].read_temperature_humidity()
            data["sensor_id"] = sid
            data["cabinet_id"] = cfg["cabinet"]
            data["timestamp"] = time.time()
            results[sid] = data

        return results

    def poll_by_cabinet(self, cabinet_id: str) -> Dict[str, dict]:
        """按机柜轮询"""
        results = {}
        for sid, cfg in self.sensors.items():
            if cfg.get("cabinet") == cabinet_id:
                if sid not in self.clients:
                    self.clients[sid] = EthernetSensorClient(cfg["host"], cfg.get("port", 502))
                data = self.clients[sid].read_temperature_humidity()
                data["sensor_id"] = sid
                data["timestamp"] = time.time()
                results[sid] = data
        return results

┌─────────────────────────────────────────────────────────────────────┐
│ 机柜列网络拓扑                                                    │
│                                                                     │
│  ┌──────────┐                                                     │
│  │ 汇聚交换机 │ ◀───── 光纤 ──────┐                                │
│  └────┬─────┘                    │                                │
│       │                          │                                │
│  ┌────▼────────┐      ┌──────────▼──────────┐                    │
│  │ 列头接入交换机│      │ 列头接入交换机        │     ...           │
│  │ (PoE)       │      │ (PoE)               │                    │
│  └────┬────────┘      └──────────┬──────────┘                    │
│       │                          │                                │
│  ┌────▼──────────────────────────▼──────────┐                    │
│  │ 机柜1    机柜2    机柜3    ...  机柜N    │                    │
│  │ ┌──┐ ┌──┐ ┌──┐         ┌──┐ │                    │
│  │ │S1│ │S1│ │S1│         │S1│ │  S=传感器           │
│  │ │S2│ │S2│ │S2│         │S2│ │                    │
│  │ │S3│ │S3│ │S3│         │S3│ │                    │
│  │ └──┘ └──┘ └──┘         └──┘ │                    │
│  └─────────────────────────────────────────────────┘                    │
│                                                                     │
│ 设计要点：                                                          │
│ · 每列配置1~2台PoE接入交换机（24口/48口）                           │
│ · 传感器通过Cat5e/Cat6网线直连，PoE供电                              │
│ · 汇聚交换机上联至监控服务器                                         │
│ · VLAN隔离：环境监控VLAN与业务VLAN分离                               │
└─────────────────────────────────────────────────────────────────────┘