服务器机房动环开发：POE以太网温湿度传感器UDP报文解析与服务器入库实战

// 设备端UDP上报帧（18字节）
typedef struct __attribute__((packed)) {
    uint16_t  header;        // 0x55AA 帧头
    uint8_t   dev_addr;      // 设备地址（1-247）
    uint8_t   msg_type;      // 0x01=周期遥测 0x02=事件Trap
    uint16_t  seq;           // 序列号，单调递增，用于去重和丢包检测
    uint32_t  timestamp;     // 采集时间戳（Unix epoch，设备RTC）
    int16_t   temperature;   // 温度 ×100，范围 -40.00 ~ +85.00℃
    uint16_t  humidity;      // 湿度 ×10，范围 0.0 ~ 100.0%
    uint16_t  voltage;       // PoE供电电压 ×100，范围 0.00 ~ 60.00V
    uint8_t   status;        // 状态位：bit0=传感器故障 bit1=低电压 bit2=越限
    uint16_t  crc16;         // CRC-16/Modbus，覆盖前16字节
} udp_frame_t;  // 总计18字节

#!/usr/bin/env python3
"""
动环系统UDP接收服务 - POE以太网温湿度传感器
"""
import asyncio
import struct
import socket
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, Optional
import crc16


@dataclass
class SensorReading:
    dev_addr: int
    msg_type: int
    seq: int
    timestamp: int
    temperature: float
    humidity: float
    voltage: float
    status: int
    source_ip: str
    recv_time: float = field(default_factory=lambda: asyncio.get_event_loop().time())


class UDPFrameParser:
    """UDP报文解析器"""
    
    FRAME_SIZE = 18
    HEADER_MAGIC = 0x55AA
    
    @classmethod
    def parse(cls, data: bytes, addr: tuple) -> Optional[SensorReading]:
        """解析UDP帧，失败返回None"""
        if len(data) != cls.FRAME_SIZE:
            return None
        
        # 解包前16字节（不含CRC）
        try:
            header, dev_addr, msg_type, seq = struct.unpack(">HBBH", data[:6])
        except struct.error:
            return None
        
        if header != cls.HEADER_MAGIC:
            return None
        
        # CRC校验
        crc_calc = crc16.modbus(data[:16])
        crc_recv = struct.unpack("<H", data[16:18])[0]
        if crc_calc != crc_recv:
            return None
        
        # 解包数据字段
        ts, temp_raw, hum_raw, volt_raw, status = struct.unpack(">IhHHH", data[6:16])
        
        # 缩放转换
        temperature = temp_raw / 100.0
        humidity = hum_raw / 10.0
        voltage = volt_raw / 100.0
        
        # 合理性校验
        if not (-40.0 <= temperature <= 85.0):
            return None
        if not (0.0 <= humidity <= 100.0):
            return None
        if not (0.0 <= voltage <= 60.0):
            return None
        
        return SensorReading(
            dev_addr=dev_addr,
            msg_type=msg_type,
            seq=seq,
            timestamp=ts,
            temperature=temperature,
            humidity=humidity,
            voltage=voltage,
            status=status,
            source_ip=addr[0],
        )


class Deduplicator:
    """序列号去重器，处理UDP重复包"""
    
    def __init__(self, window_size: int = 65536, ttl: float = 300.0):
        self._window: Dict[int, Dict[int, float]] = {}  # dev_addr -> {seq: recv_time}
        self._window_size = window_size
        self._ttl = ttl
    
    def is_duplicate(self, dev_addr: int, seq: int, now: float) -> bool:
        """检查是否为重复包"""
        # 清理过期条目
        if dev_addr in self._window:
            self._window[dev_addr] = {
                s: t for s, t in self._window[dev_addr].items()
                if now - t < self._ttl
            }
        else:
            self._window[dev_addr] = {}
        
        dev_window = self._window[dev_addr]
        
        # 检查重复
        if seq in dev_window:
            return True
        
        # 检查序列号回绕（uint16_t）
        # 如果当前seq比窗口中最大的小很多，可能是回绕后的新包
        if dev_window:
            max_seq = max(dev_window.keys())
            if seq < max_seq and (max_seq - seq) > 32768:
                # 回绕后新包，清理旧窗口
                dev_window.clear()
        
        dev_window[seq] = now
        return False


class UDPServerProtocol(asyncio.DatagramProtocol):
    """UDP服务器协议实现"""
    
    def __init__(self, parser: UDPFrameParser, dedup: Deduplicator, 
                 handler: callable):
        self.parser = parser
        self.dedup = dedup
        self.handler = handler
        self._stats = {
            "received": 0,
            "parsed": 0,
            "duplicates": 0,
            "crc_errors": 0,
            "invalid": 0,
        }
    
    def datagram_received(self, data: bytes, addr: tuple):
        self._stats["received"] += 1
        
        reading = self.parser.parse(data, addr)
        if reading is None:
            self._stats["invalid"] += 1
            return
        
        # 去重
        now = asyncio.get_event_loop().time()
        if self.dedup.is_duplicate(reading.dev_addr, reading.seq, now):
            self._stats["duplicates"] += 1
            return
        
        self._stats["parsed"] += 1
        asyncio.create_task(self.handler(reading))
    
    def error_received(self, exc: Exception):
        print(f"UDP错误: {exc}")
    
    def get_stats(self) -> dict:
        return self._stats.copy()


async def start_udp_server(host: str, port: int, handler: callable) -> None:
    """启动UDP服务器"""
    parser = UDPFrameParser()
    dedup = Deduplicator()
    protocol = UDPServerProtocol(parser, dedup, handler)
    
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        lambda: protocol,
        local_addr=(host, port),
    )
    
    print(f"UDP Server 监听 {host}:{port}")
    try:
        await asyncio.Event().wait()
    finally:
        transport.close()

# CRC-16/Modbus实现
def crc16_modbus(data: bytes) -> int:
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 0x0001:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc & 0xFFFF

from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS
import asyncio
from datetime import datetime, timezone


class InfluxDBWriter:
    """InfluxDB异步写入器"""
    
    def __init__(self, url: str, token: str, org: str, bucket: str):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(write_options=ASYNCHRONOUS)
        self.bucket = bucket
        self.org = org
        self._queue = asyncio.Queue(maxsize=50000)
        self._running = False
        self._batch_size = 500
        self._flush_interval = 5.0  # 秒
    
    async def start(self):
        """启动后台写入任务"""
        self._running = True
        asyncio.create_task(self._batch_writer())
    
    async def stop(self):
        """停止写入任务"""
        self._running = False
        # 刷出剩余数据
        await self._flush()
        self.write_api.close()
        self.client.close()
    
    async def write(self, reading: SensorReading):
        """写入单条数据"""
        try:
            self._queue.put_nowait(reading)
        except asyncio.QueueFull:
            # 队列满时丢弃最旧数据，记录告警
            print(f"警告: 写入队列满，丢弃设备{reading.dev_addr}的数据")
            # 可选：写入本地文件兜底
    
    async def _batch_writer(self):
        """后台批量写入"""
        batch = []
        last_flush = asyncio.get_event_loop().time()
        
        while self._running:
            try:
                # 等待数据，超时则检查是否需要flush
                timeout = self._flush_interval - (asyncio.get_event_loop().time() - last_flush)
                if timeout > 0:
                    reading = await asyncio.wait_for(self._queue.get(), timeout=timeout)
                    batch.append(reading)
                
                # 达到批量大小或超时，执行flush
                now = asyncio.get_event_loop().time()
                if len(batch) >= self._batch_size or (now - last_flush) >= self._flush_interval:
                    if batch:
                        await self._flush_batch(batch)
                        batch.clear()
                        last_flush = now
                        
            except asyncio.TimeoutError:
                # 超时，flush当前批次
                if batch:
                    await self._flush_batch(batch)
                    batch.clear()
                    last_flush = asyncio.get_event_loop().time()
            except Exception as e:
                print(f"批量写入异常: {e}")
                await asyncio.sleep(1)
    
    async def _flush_batch(self, batch: list):
        """刷出一批数据到InfluxDB"""
        points = []
        for r in batch:
            # 主测量值
            point = Point("env_sensor") \
                .tag("device", f"sensor_{r.dev_addr:03d}") \
                .tag("msg_type", "trap" if r.msg_type == 0x02 else "telemetry") \
                .tag("source_ip", r.source_ip) \
                .field("temperature", r.temperature) \
                .field("humidity", r.humidity) \
                .field("voltage", r.voltage) \
                .field("status", r.status) \
                .time(datetime.fromtimestamp(r.timestamp, tz=timezone.utc), 
                      WritePrecision.S)
            points.append(point)
        
        try:
            self.write_api.write(bucket=self.bucket, org=self.org, record=points)
        except Exception as e:
            print(f"InfluxDB写入失败: {e}")
            # 写入失败，重新入队或落本地文件
            await self._fallback(batch)
    
    async def _fallback(self, batch: list):
        """写入失败时的本地兜底"""
        import json
        try:
            with open("influx_fallback.jsonl", "a") as f:
                for r in batch:
                    f.write(json.dumps({
                        "dev_addr": r.dev_addr,
                        "timestamp": r.timestamp,
                        "temperature": r.temperature,
                        "humidity": r.humidity,
                        "voltage": r.voltage,
                        "status": r.status,
                    }) + "\n")
        except Exception as e:
            print(f"兜底文件写入失败: {e}")

-- 查询某设备最近1小时的温度
SELECT mean(temperature) FROM env_sensor 
WHERE device='sensor_001' AND time > now() - 1h 
GROUP BY time(1m)

-- 查询所有设备当前电压状态
SELECT last(voltage) FROM env_sensor 
WHERE time > now() - 5m 
GROUP BY device

-- 查询Trap事件
SELECT * FROM env_sensor 
WHERE msg_type='trap' AND time > now() - 1h

#!/usr/bin/env python3
"""
动环系统主程序 - UDP接收 + 解析 + 入库
"""
import asyncio
import signal
import sys


async def handle_reading(reading: SensorReading, writer: InfluxDBWriter):
    """处理解析后的数据"""
    # 打印日志
    msg_type_str = "Trap" if reading.msg_type == 0x02 else "遥测"
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg_type_str} "
          f"设备{reading.dev_addr:03d} "
          f"T={reading.temperature:.1f}℃ "
          f"H={reading.humidity:.1f}% "
          f"V={reading.voltage:.2f}V "
          f"Status=0x{reading.status:02X}")
    
    # 写入InfluxDB
    await writer.write(reading)


async def main():
    # InfluxDB配置
    influx_url = "http://localhost:8086"
    influx_token = "your-token"
    influx_org = "your-org"
    influx_bucket = "env_monitor"
    
    # 创建写入器
    writer = InfluxDBWriter(influx_url, influx_token, influx_org, influx_bucket)
    await writer.start()
    
    # 启动UDP服务器
    udp_task = asyncio.create_task(
        start_udp_server("0.0.0.0", 9000, 
                         lambda r: handle_reading(r, writer))
    )
    
    # 优雅退出
    loop = asyncio.get_running_loop()
    stop_event = asyncio.Event()
    
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop_event.set)
    
    await stop_event.wait()
    
    print("正在关闭...")
    udp_task.cancel()
    await writer.stop()
    print("已关闭")


if __name__ == "__main__":
    asyncio.run(main())

# 暴露Prometheus指标
from prometheus_client import Counter, Gauge, Histogram

UDP_RECEIVED = Counter('udp_received_total', 'UDP数据报接收总数', ['device'])
UDP_PARSED = Counter('udp_parsed_total', 'UDP解析成功数', ['device'])
UDP_DUPLICATES = Counter('udp_duplicates_total', 'UDP重复包数', ['device'])
UDP_CRC_ERRORS = Counter('udp_crc_errors_total', 'UDP CRC错误数', ['device'])
INFLUX_WRITE_LATENCY = Histogram('influx_write_latency_seconds', 'InfluxDB写入延迟')
INFLUX_QUEUE_SIZE = Gauge('influx_queue_size', 'InfluxDB写入队列长度')

# 过滤特定设备的UDP包
udp.port == 9000 && ip.src == 10.10.21.11

# 查看UDP重传
udp.analysis.retransmission

# 导出原始字节
Follow → UDP Stream → Show data as Hex Dump

sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024)  # 1MB