边缘网关 + 以太网温湿度传感器：本地化环境数据采集架构案例

┌─────────────────────────────────────────────────────────────────────────┐
│  设备层（以太网温湿度传感器）                                           │
│  • RJ45 接口，IEEE 802.3af PoE 供电                                   │
│  • 支持 Modbus TCP / HTTP API / MQTT 三种协议                          │
│  • 采样周期 5~60s 可配                                                │
│  • 内置温湿度校准系数，支持远程校准                                    │
│  • 每台独立 IP，双网卡冗余（部分高端型号）                             │
└───────────────────────────┬─────────────────────────────────────────────┘
                            │ TCP/IP（VLAN 隔离）
┌───────────────────────────▼─────────────────────────────────────────────┐
│  边缘网关层（每库房 1~2 台）                                            │
│  • 硬件：ARM Cortex-A55 四核 / 2GB RAM / 32GB eMMC                    │
│  • 系统：Debian 12 + Docker                                           │
│  • 功能：                                                               │
│    ├── 多协议并发采集（Modbus TCP / SNMP / HTTP）                      │
│    ├── 数据预处理（去抖、插值、加权聚合）                              │
│    ├── 本地时序库（InfluxDB，保留 90 天热数据）                        │
│    ├── 联动规则引擎（断网也能执行启停逻辑）                            │
│    ├── 断网缓存（SQLite WAL 模式，支持 30 天补传）                     │
│    └── 安全：TLS 双向认证 + 防火墙白名单                               │
└───────────────────────────┬─────────────────────────────────────────────┘
                            │ 加密上送（MQTT over TLS / HTTPS）
┌───────────────────────────▼─────────────────────────────────────────────┐
│  中心层（监控中心 / 云平台）                                           │
│  • 接收网关聚合数据（非原始逐点数据，降低带宽 80%）                    │
│  • 组态画面展示 + 告警推送 + 历史报表                                  │
│  • 配置下发（阈值、采样周期、联动规则）                                │
│  • 审计日志存储（永久）                                                │
└─────────────────────────────────────────────────────────────────────────┘

┌──────────────┐
                        │  核心交换机   │
                        └──────┬───────┘
                               │ 光纤/网线
                        ┌──────▼───────┐
                        │ 边缘网关      │ ← Docker 容器组
                        │ (库房A)      │   ├── collector (采集)
                        │              │   ├── influxdb  (存储)
                        │ 172.20.10.1  │   ├── rules     (联动)
                        └──┬───┬───┬──┘   └── uploader  (上送)
                           │   │   │
              ┌────────────┘   │   └────────────┐
              ▼                ▼                ▼
        ┌──────────┐    ┌──────────┐     ┌──────────┐
        │ 传感器01  │    │ 传感器02  │     │ 传感器08  │
        │ 172.20.10│    │ 172.20.10│     │ 172.20.10│
        │   .11    │    │   .12    │ ... │   .18    │
        └──────────┘    └──────────┘     └──────────┘

import asyncio
from pymodbus.client import AsyncModbusTcpClient
import time

class EdgeCollector:
    def __init__(self, sensors: list, concurrency=16):
        self.sensors = sensors
        self.semaphore = asyncio.Semaphore(concurrency)
        self.results = {}

    async def fetch_one(self, sensor: dict):
        async with self.semaphore:
            client = AsyncModbusTcpClient(
                sensor['ip'], port=502,
                timeout=3, retries=2
            )
            try:
                await client.connect()
                rr = await client.read_holding_registers(
                    sensor['reg_base'], 2, slave=sensor.get('slave', 1)
                )
                if not rr.isError():
                    self.results[sensor['id']] = {
                        'temp': rr.registers[0] / 10,
                        'hum': rr.registers[1] / 10,
                        'ts': time.time(),
                        'status': 'OK'
                    }
                else:
                    self.results[sensor['id']]['status'] = 'READ_ERR'
            except Exception as e:
                self.results[sensor['id']] = {
                    'status': 'CONN_ERR', 'err': str(e)
                }
            finally:
                await client.close()

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

class LocalProcessor:
    def __init__(self, zone_weights: dict):
        self.weights = zone_weights

    def aggregate(self, raw: dict) -> dict:
        valid = {k: v for k, v in raw.items()
                 if v.get('status') == 'OK'}
        if not valid:
            return {'error': 'no valid data'}

        total_w = sum(self.weights.get(k, 1.0) for k in valid)
        avg_temp = sum(v['temp'] * self.weights.get(k, 1.0)
                       for k, v in valid.items()) / total_w
        avg_hum = sum(v['hum'] * self.weights.get(k, 1.0)
                      for k, v in valid.items()) / total_w
        max_hum = max(v['hum'] for v in valid.values())
        min_hum = min(v['hum'] for v in valid.values())

        return {
            'zone_avg_temp': round(avg_temp, 2),
            'zone_avg_hum': round(avg_hum, 2),
            'zone_hum_gradient': round(max_hum - min_hum, 2),
            'point_count': len(valid),
            'ts': time.time()
        }

import sqlite3, json

class OfflineBuffer:
    def __init__(self, db_path='/data/offline.db'):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute('''CREATE TABLE IF NOT EXISTS cache (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            payload TEXT, ts REAL, uploaded INTEGER DEFAULT 0
        )''')
        self.conn.commit()

    def store(self, data: dict):
        self.conn.execute(
            'INSERT INTO cache (payload, ts) VALUES (?, ?)',
            (json.dumps(data), time.time())
        )
        self.conn.commit()

    def pending(self, limit=500):
        cur = self.conn.execute(
            'SELECT id, payload FROM cache WHERE uploaded=0 ORDER BY ts LIMIT ?',
            (limit,)
        )
        return [(row[0], json.loads(row[1])) for row in cur.fetchall()]

    def mark_uploaded(self, ids: list):
        self.conn.executemany(
            'UPDATE cache SET uploaded=1 WHERE id=?',
            [(i,) for i in ids]
        )
        self.conn.commit()

# docker-compose.yml
version: '3.8'
services:
  collector:
    build: ./collector
    network_mode: host
    restart: always
    volumes:
      - /data/config:/app/config

  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    volumes:
      - /data/influxdb:/var/lib/influxdb2
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=${INFLUX_PWD}
      - DOCKER_INFLUXDB_INIT_ORG=archive
      - DOCKER_INFLUXDB_INIT_BUCKET=env_data

  rules_engine:
    build: ./rules
    network_mode: host
    restart: always
    depends_on:
      - influxdb

  uploader:
    build: ./uploader
    network_mode: host
    restart: always
    environment:
      - UPLOAD_URL=${CENTER_URL}
      - UPLOAD_TOKEN=${UPLOAD_TOKEN}