元器件车间产线监控：TCP/IP以太网温湿度传感器时间同步与数据校准

传感器内部时间源：
  → RTC（实时时钟芯片）：精度 ±20ppm，月偏差约 52 秒
  → 无 RTC（靠 MCU 定时器）：掉电即归零
  → NTP 客户端：部分高端型号支持，但默认未启用

实际偏差来源：
  1. 晶振温漂：车间温度 25→45℃，晶振频率偏移 ±50ppm
  2. 断电重启：RTC 电池耗尽或缺失，重启后时间归零
  3. 固件 Bug：时间溢出（32 位秒计数器 2038 年问题）
  4. 手动设置不一致：安装时人工设置时间，人为误差

正常情况：传感器通过 SNTP 协议从 NTP 服务器同步时间（每 24h 一次）
异常情况：采集前置机通过 Modbus TCP 写寄存器强制校准（兜底）

┌─────────────────────┐
                    │   车间 NTP 服务器    │
                    │  (192.168.10.1)     │
                    │   stratum 2         │
                    └──────────┬──────────┘
                               │ NTP/SNTP
              ┌────────────────┼────────────────┐
              │                │                │
     ┌────────▼───────┐ ┌─────▼────────┐ ┌────▼───────────┐
     │ 传感器群组 A     │ │ 传感器群组 B   │ │ 传感器群组 C    │
     │ (回流焊区)      │ │ (贴片区)      │ │ (仓储区)       │
     │ 12 台           │ │ 24 台         │ │ 8 台           │
     │ 同步间隔: 24h   │ │ 同步间隔: 24h │ │ 同步间隔: 24h  │
     │ 偏移随机: ±300s │ │ 偏移随机: ±.. │ │ 偏移随机: ±... │
     └────────────────┘ └──────────────┘ └────────────────┘

  关键设计：
  1. 所有传感器从同一个 NTP 服务器同步（保证一致性）
  2. 同步间隔 24h（RTC 精度 ±20ppm，24h 偏差约 1.7s，可接受）
  3. 随机偏移 ±300s（避免 60 台同时发 SNTP 请求）
  4. NTP 服务器本身从 GPS/北斗或上游 stratum 1 同步

/* SNTP 时间同步任务 */

#include "lwip/apps/sntp.h"
#include "lwip/dns.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include <time.h>

#define SNTP_SYNC_INTERVAL_S  86400    // 24 小时同步一次
#define SNTP_RETRY_INTERVAL_S 60       // 同步失败 60s 后重试
#define SNTP_SERVER "192.168.10.1"     // 车间 NTP 服务器

static bool g_time_synced = false;
static time_t g_last_sync_time = 0;
static int32_t g_time_offset = 0;      // 软件时间偏移补偿（ppm 校准）

/* SNTP 同步完成回调 */
static void sntp_sync_callback(struct timeval *tv)
{
    if (tv) {
        g_time_synced = true;
        g_last_sync_time = tv->tv_sec;
        struct tm timeinfo;
        localtime_r(&tv->tv_sec, &timeinfo);
        LWIP_LOG("SNTP sync success: %04d-%02d-%02d %02d:%02d:%02d",
                 timeinfo.tm_year + 1900, timeinfo.tm_mon + 1,
                 timeinfo.tm_mday, timeinfo.tm_hour,
                 timeinfo.tm_min, timeinfo.tm_sec);
    } else {
        LWIP_LOG("SNTP sync failed");
    }
}

/* 初始化 SNTP */
static void sntp_init_task(void)
{
    // 设置时区（中国 UTC+8）
    setenv("TZ", "CST-8", 1);
    tzset();
    
    // 初始化 SNTP（LwIP 内置 SNTP 客户端）
    sntp_setoperatingmode(SNTP_OPMODE_POLL);
    sntp_setservername(0, SNTP_SERVER);
    sntp_set_sync_interval(SNTP_SYNC_INTERVAL_S * 1000);
    sntp_set_sync_callback(sntp_sync_callback);
    sntp_init();
    
    LWIP_LOG("SNTP initialized, server=%s, interval=%ds",
             SNTP_SERVER, SNTP_SYNC_INTERVAL_S);
}

/* 获取当前时间戳（带软件补偿） */
uint64_t get_timestamp_ms(void)
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    
    // 应用软件补偿（晶振温漂修正）
    int64_t adjusted = (int64_t)tv.tv_sec * 1000000LL + tv.tv_usec;
    adjusted += g_time_offset * (adjusted / 1000000);  // ppm 补偿
    
    return (uint64_t)(adjusted / 1000);  // 返回毫秒时间戳
}

/* SNTP 监控任务 */
void time_sync_task(void *pvParameters)
{
    sntp_init_task();
    
    TickType_t last_sync_attempt = 0;
    uint32_t retry_count = 0;
    
    while (1) {
        time_t now;
        time(&now);
        
        // 检查是否需要同步
        if (!g_time_synced || (now - g_last_sync_time) > SNTP_SYNC_INTERVAL_S) {
            TickType_t tick = xTaskGetTickCount();
            if (tick - last_sync_attempt > pdMS_TO_TICKS(SNTP_RETRY_INTERVAL_S * 1000)) {
                retry_count++;
                
                // 指数退避（防止 NTP 服务器被频繁请求）
                uint32_t delay_s = SNTP_RETRY_INTERVAL_S;
                if (retry_count > 1) {
                    delay_s = SNTP_RETRY_INTERVAL_S * (1 << (retry_count - 1));
                    delay_s = (delay_s > 3600) ? 3600 : delay_s;  // 最大 1h
                }
                
                LWIP_LOG("SNTP retry #%u, next attempt in %us", retry_count, delay_s);
                last_sync_attempt = tick;
                
                // 触发重新同步
                sntp_stop();
                sntp_init();
            }
        } else {
            retry_count = 0;  // 同步成功，重置重试计数
        }
        
        // 每小时检查 RTC 偏差（与 NTP 服务器对比）
        static TickType_t last_check = 0;
        TickType_t tick = xTaskGetTickCount();
        if (tick - last_check > pdMS_TO_TICKS(3600000)) {  // 1 小时
            last_check = tick;
            if (g_time_synced) {
                struct timeval tv;
                gettimeofday(&tv, NULL);
                time_t drift = tv.tv_sec - g_last_sync_time;
                LWIP_LOG("Time since last sync: %lds", drift);
                
                // 如果偏差超过 5s，立即触发重新同步
                if (llabs((long long)drift) > 5) {
                    LWIP_LOG("Time drift > 5s, forcing re-sync");
                    sntp_stop();
                    sntp_init();
                }
            }
        }
        
        vTaskDelay(pdMS_TO_TICKS(10000));  // 10s 检查一次
    }
}

/* Modbus TCP 时间寄存器映射（供上位机读取/写入） */
// 寄存器地址映射：
// 40001-40002: Unix 时间戳（32 位，低 16 位 + 高 16 位）
// 40003: 时间同步状态（0=未同步, 1=已同步, 2=同步中）
// 40004: 上次同步时间戳
// 40005: 晶振补偿值（ppm × 100，有符号）

int modbus_read_time_registers(uint16_t reg_addr, uint16_t *value)
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    uint32_t timestamp = (uint32_t)tv.tv_sec;
    
    switch (reg_addr) {
        case 0:  // 时间戳低 16 位
            *value = timestamp & 0xFFFF;
            return 0;
        case 1:  // 时间戳高 16 位
            *value = (timestamp >> 16) & 0xFFFF;
            return 0;
        case 2:  // 同步状态
            *value = g_time_synced ? 1 : 0;
            return 0;
        case 3:  // 上次同步时间低 16 位
            *value = ((uint32_t)g_last_sync_time) & 0xFFFF;
            return 0;
        case 4:  // 上次同步时间高 16 位
            *value = ((uint32_t)g_last_sync_time >> 16) & 0xFFFF;
            return 0;
        case 5:  // 晶振补偿 ppm × 100
            *value = (uint16_t)(g_time_offset & 0xFFFF);
            return 0;
        default:
            return -1;
    }
}

int modbus_write_time_registers(uint16_t reg_addr, uint16_t value)
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    
    switch (reg_addr) {
        case 0:  // 写入时间戳低 16 位（配合寄存器 1 使用）
        case 1:  // 写入时间戳高 16 位
            // 上位机先写低 16 位，再写高 16 位，最后写寄存器 2 触发生效
            // 这里只缓存，不立即生效
            return 0;
        case 2:  // 触发时间设置（写入 0xAAAA 确认）
            if (value == 0xAAAA) {
                // 从缓存的时间戳设置系统时间
                uint32_t new_time = 0;  // 从寄存器 0/1 读取
                tv.tv_sec = new_time;
                tv.tv_usec = 0;
                settimeofday(&tv, NULL);
                g_time_synced = true;
                g_last_sync_time = new_time;
                LWIP_LOG("Time set via Modbus: %u", new_time);
                return 0;
            }
            return -1;
        case 5:  // 设置晶振补偿
            g_time_offset = (int16_t)value;
            return 0;
        default:
            return -1;
    }
}

import asyncio
import struct
import time
import socket
from datetime import datetime, timezone
from typing import List, Dict, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SensorTimeManager:
    """传感器时间管理：SNTP 监控 + Modbus TCP 强制校准"""
    
    def __init__(self, ntp_server: str = "192.168.10.1",
                 sync_interval: int = 86400,
                 drift_threshold: float = 5.0):  # 偏差超过 5s 触发告警
        self.ntp_server = ntp_server
        self.sync_interval = sync_interval
        self.drift_threshold = drift_threshold
        self.sensor_times: Dict[str, Dict] = {}
    
    def get_ntp_time(self) -> Optional[float]:
        """从 NTP 服务器获取时间（简化版 SNTP 客户端）"""
        try:
            # NTP 数据包格式（48 字节）
            ntp_packet = bytearray(48)
            ntp_packet[0] = 0x1B  # LI=0, VN=3, Mode=3 (Client)
            
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            sock.settimeout(3.0)
            sock.sendto(ntp_packet, (self.ntp_server, 123))
            response, _ = sock.recvfrom(48)
            sock.close()
            
            # 解析 NTP 时间戳（从 1900-01-01 起的秒数）
            ntp_timestamp = struct.unpack('>II', response[40:48])[0]
            # 转换为 Unix 时间戳（从 1970-01-01 起）
            unix_timestamp = ntp_timestamp - 2208988800
            
            return float(unix_timestamp)
            
        except Exception as e:
            logger.error(f"NTP query failed: {e}")
            return None
    
    async def read_sensor_time(self, ip: str, port: int = 502,
                               slave_id: int = 1) -> Optional[float]:
        """通过 Modbus TCP 读取传感器时间"""
        try:
            reader, writer = await asyncio.wait_for(
                asyncio.open_connection(ip, port), timeout=3.0
            )
            
            # 读取寄存器 0-1（Unix 时间戳 32 位）
            tid = int(time.time()) & 0xFFFF
            mbap = struct.pack('>HHHB', tid, 0, 6, slave_id)
            pdu = struct.pack('>BHH', 0x03, 0, 2)  # 读寄存器 0, 1
            request = mbap + pdu
            
            writer.write(request)
            await writer.drain()
            
            response = await asyncio.wait_for(
                reader.read(256), timeout=2.0
            )
            
            writer.close()
            await writer.wait_closed()
            
            if len(response) < 11:
                return None
            
            # 解析时间戳
            reg0, reg1 = struct.unpack('>HH', response[9:13])
            sensor_timestamp = (reg1 << 16) | reg0
            
            return float(sensor_timestamp)
            
        except Exception as e:
            logger.error(f"Read time from {ip} failed: {e}")
            return None
    
    async def write_sensor_time(self, ip: str, timestamp: float,
                                 port: int = 502, slave_id: int = 1) -> bool:
        """通过 Modbus TCP 写入传感器时间"""
        try:
            reader, writer = await asyncio.wait_for(
                asyncio.open_connection(ip, port), timeout=3.0
            )
            
            ts_int = int(timestamp)
            reg0 = ts_int & 0xFFFF
            reg1 = (ts_int >> 16) & 0xFFFF
            
            # 写入寄存器 0（低 16 位）
            tid = int(time.time()) & 0xFFFF
            mbap = struct.pack('>HHHB', tid, 0, 6, slave_id)
            pdu = struct.pack('>BHH', 0x06, 0, reg0)  # 写单个寄存器
            request = mbap + pdu
            writer.write(request)
            await writer.drain()
            await asyncio.wait_for(reader.read(256), timeout=2.0)
            
            # 写入寄存器 1（高 16 位）
            tid = (tid + 1) & 0xFFFF
            mbap = struct.pack('>HHHB', tid, 0, 6, slave_id)
            pdu = struct.pack('>BHH', 0x06, 1, reg1)
            request = mbap + pdu
            writer.write(request)
            await writer.drain()
            await asyncio.wait_for(reader.read(256), timeout=2.0)
            
            # 写入寄存器 2（触发时间生效）
            tid = (tid + 1) & 0xFFFF
            mbap = struct.pack('>HHHB', tid, 0, 6, slave_id)
            pdu = struct.pack('>BHH', 0x06, 2, 0xAAAA)
            request = mbap + pdu
            writer.write(request)
            await writer.drain()
            await asyncio.wait_for(reader.read(256), timeout=2.0)
            
            writer.close()
            await writer.wait_closed()
            
            logger.info(f"{ip}: time set to {datetime.fromtimestamp(ts_int)}")
            return True
            
        except Exception as e:
            logger.error(f"Write time to {ip} failed: {e}")
            return False
    
    async def check_and_calibrate(self, sensors: List[str]):
        """检查所有传感器时间偏差并校准"""
        # 获取 NTP 参考时间
        ntp_time = self.get_ntp_time()
        if not ntp_time:
            logger.error("Cannot get NTP time, skipping calibration")
            return
        
        logger.info(f"NTP reference time: {datetime.fromtimestamp(ntp_time)}")
        
        # 并发检查所有传感器
        tasks = [self.read_sensor_time(ip) for ip in sensors]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for ip, result in zip(sensors, results):
            if isinstance(result, Exception) or result is None:
                logger.warning(f"{ip}: failed to read time")
                continue
            
            drift = abs(result - ntp_time)
            self.sensor_times[ip] = {
                'sensor_time': result,
                'ntp_time': ntp_time,
                'drift': drift,
                'last_check': time.time()
            }
            
            if drift > self.drift_threshold:
                logger.warning(f"{ip}: time drift {drift:.1f}s, calibrating...")
                success = await self.write_sensor_time(ip, ntp_time)
                if success:
                    logger.info(f"{ip}: calibrated successfully")
                else:
                    logger.error(f"{ip}: calibration failed")
            else:
                logger.debug(f"{ip}: drift {drift:.1f}s (OK)")
    
    async def run_periodic_check(self, sensors: List[str], interval: int = 3600):
        """定期检查和校准"""
        while True:
            logger.info(f"Starting time check for {len(sensors)} sensors...")
            await self.check_and_calibrate(sensors)
            await asyncio.sleep(interval)


# 使用示例
async def main():
    sensors = [f"192.168.100.{i}" for i in range(1, 61)]  # 60 台传感器
    
    manager = SensorTimeManager(
        ntp_server="192.168.10.1",
        sync_interval=86400,
        drift_threshold=5.0
    )
    
    # 立即执行一次校准
    await manager.check_and_calibrate(sensors)
    
    # 然后每小时检查一次
    await manager.run_periodic_check(sensors, interval=3600)


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

温度漂移来源：
  1. 自热效应：传感器电路自身发热（PoE 供电芯片、MCU），约 0.2~0.5℃
  2. 太阳辐射/设备热辐射：车间设备散热导致局部温度偏高
  3. 安装位置：靠近热源（如交换机、电源）导致测量偏差
  4. 长期老化：敏感元件性能衰减，年漂移约 0.1~0.3℃

湿度漂移来源：
  1. 电容式湿度传感器老化：敏感膜吸湿后恢复不完全（"记忆效应"）
  2. 化学污染：车间助焊剂挥发物附着在传感器表面
  3. 温度交叉敏感：湿度传感器对温度变化的响应误差
  4. 长期高湿暴露：>80%RH 长期暴露导致传感器"饱和"

校准周期：
  → 关键区域（回流焊区）：每 3 个月
  → 一般区域（贴片区）：每 6 个月
  → 仓储区：每 12 个月

校准方法：
  1. 将传感器与标准器（Fluke 754 或同等精度）放入恒温恒湿箱
  2. 设置温度点：15℃、25℃、35℃、45℃
  3. 设置湿度点：30%RH、50%RH、70%RH
  4. 每个点稳定 30 分钟后记录传感器读数和标准器读数
  5. 计算补偿系数（线性拟合或查表法）

@dataclass
class CalibrationParams:
    """校准参数"""
    temp_offset: float = 0.0        # 温度偏移
    temp_gain: float = 1.0          # 温度增益
    temp_poly: List[float] = None   # 温度多项式系数 [a0, a1, a2, ...]
    hum_offset: float = 0.0         # 湿度偏移
    hum_gain: float = 1.0           # 湿度增益
    hum_poly: List[float] = None    # 湿度多项式系数
    last_cal: float = 0.0           # 上次校准时间戳
    cal_valid_days: int = 180       # 校准有效期（天）

class SensorCalibrator:
    """传感器数据校准"""
    
    def __init__(self, params: Dict[str, CalibrationParams]):
        self.params = params
    
    def is_cal_valid(self, ip: str) -> bool:
        """检查校准是否在有效期内"""
        if ip not in self.params:
            return False
        p = self.params[ip]
        age_days = (time.time() - p.last_cal) / 86400
        return age_days < p.cal_valid_days
    
    def compensate_temperature(self, ip: str, raw_temp: float) -> float:
        """温度补偿"""
        if ip not in self.params:
            return raw_temp
        
        p = self.params[ip]
        
        # 线性补偿
        compensated = (raw_temp + p.temp_offset) * p.temp_gain
        
        # 多项式补偿（如果配置了）
        if p.temp_poly:
            x = raw_temp
            result = 0.0
            for i, coeff in enumerate(p.temp_poly):
                result += coeff * (x ** i)
            compensated = result
        
        return round(compensated, 2)
    
    def compensate_humidity(self, ip: str, raw_hum: float, 
                             compensated_temp: float) -> float:
        """湿度补偿（含温度交叉补偿）"""
        if ip not in self.params:
            return raw_hum
        
        p = self.params[ip]
        
        # 基础补偿
        compensated = (raw_hum + p.hum_offset) * p.hum_gain
        
        # 温度交叉补偿（湿度传感器受温度影响）
        # 典型：每偏离 25℃ 约 ±0.5%RH
        temp_deviation = compensated_temp - 25.0
        cross_compensation = -0.02 * temp_deviation  # -0.02%RH/℃
        compensated += cross_compensation
        
        # 多项式补偿
        if p.hum_poly:
            x = raw_hum
            result = 0.0
            for i, coeff in enumerate(p.hum_poly):
                result += coeff * (x ** i)
            compensated = result
        
        # 限制范围
        compensated = max(0.0, min(100.0, compensated))
        
        return round(compensated, 2)
    
    def calibrate_reading(self, ip: str, raw_temp: float, 
                          raw_hum: float) -> Tuple[float, float]:
        """完整校准流程"""
        if not self.is_cal_valid(ip):
            logger.warning(f"{ip}: calibration expired, using raw values")
            return raw_temp, raw_hum
        
        temp = self.compensate_temperature(ip, raw_temp)
        hum = self.compensate_humidity(ip, raw_hum, temp)
        
        return temp, hum


# 校准参数配置示例
cal_params = {
    "192.168.100.50": CalibrationParams(
        temp_offset=-0.3,      # 偏低 0.3℃
        temp_gain=1.005,       # 增益 0.5%
        hum_offset=1.2,        # 偏低 1.2%RH
        hum_gain=0.998,        # 增益 -0.2%
        last_cal=time.time(),
        cal_valid_days=180
    ),
    "192.168.100.51": CalibrationParams(
        temp_offset=0.2,
        temp_gain=1.002,
        hum_offset=-0.8,
        hum_gain=1.003,
        last_cal=time.time(),
        cal_valid_days=180
    ),
}

问题：同一区域多台传感器读数不一致

排查步骤：
  1. 检查时间同步：所有传感器时间戳是否一致
  2. 检查安装位置：是否有的靠近热源/冷源
  3. 检查校准状态：是否都做了校准
  4. 检查传感器健康度：是否有漂移超限

一致性判定标准：
  → 同一区域（10m 范围内）温度偏差应 < 0.5℃
  → 湿度偏差应 < 3%RH
  → 超出范围则标记异常，触发校准或维护流程

async def consistency_check(sensors_data: Dict[str, Tuple[float, float]],
                            temp_tolerance: float = 0.5,
                            hum_tolerance: float = 3.0) -> Dict:
    """多节点一致性检查"""
    results = {}
    
    if len(sensors_data) < 2:
        return results
    
    # 计算平均值
    temps = [t for t, h in sensors_data.values()]
    hums = [h for t, h in sensors_data.values()]
    
    avg_temp = sum(temps) / len(temps)
    avg_hum = sum(hums) / len(hums)
    
    for ip, (temp, hum) in sensors_data.items():
        temp_dev = abs(temp - avg_temp)
        hum_dev = abs(hum - avg_hum)
        
        results[ip] = {
            'temp': temp,
            'hum': hum,
            'temp_deviation': temp_dev,
            'hum_deviation': hum_dev,
            'temp_consistent': temp_dev <= temp_tolerance,
            'hum_consistent': hum_dev <= hum_tolerance,
            'status': 'OK' if (temp_dev <= temp_tolerance and 
                               hum_dev <= hum_tolerance) else 'INCONSISTENT'
        }
        
        if results[ip]['status'] != 'OK':
            logger.warning(f"{ip}: inconsistent - "
                         f"temp_dev={temp_dev:.2f}℃, hum_dev={hum_dev:.2f}%RH")
    
    return results

关键原则：时间戳在数据采集的源头打，而不是在接收端打

原因：
  → 网络传输延迟（局域网 1~10ms，跨网段可能 50~200ms）
  → 采集端处理延迟（队列等待、CPU 繁忙）
  → 如果接收端打时间戳，所有延迟都会累积到数据时间偏差中

实现方式：
  → 传感器在采样完成的时刻记录本地时间戳
  → 时间戳随数据一起上报（Modbus TCP 寄存器或 UDP 报文）
  → 采集端直接使用传感器上报的时间戳

推荐格式：Unix 时间戳（秒或毫秒）

Modbus TCP 寄存器映射：
  寄存器 10-11: 采样时间戳（秒，32 位）
  寄存器 12: 采样时间戳毫秒部分（0-999）
  寄存器 13-14: 传感器启动运行时间（秒，32 位，用于无 RTC 场景）

UDP 报文格式（32 字节）：
  Byte 0-1:   帧头 0xAA55
  Byte 2-5:   设备 ID
  Byte 6-9:   时间戳（Unix 秒）
  Byte 10-11: 毫秒
  Byte 12-15: 序列号
  Byte 16-17: 温度（×10）
  Byte 18-19: 湿度（×10）
  Byte 20-21: 状态字
  Byte 22-23: 电源电压（×100）
  Byte 24-25: CRC16

def align_timestamp(sensor_ts: float, server_ts: float, 
                    sensor_sync_status: bool) -> float:
    """
    时间对齐策略：
    - 传感器已同步：使用传感器时间戳
    - 传感器未同步：使用服务器时间戳，但标记质量
    """
    if sensor_sync_status:
        return sensor_ts
    else:
        # 未同步，使用服务器时间，但降低数据质量标记
        return server_ts

def calculate_transit_delay(sensor_ts: float, receive_ts: float) -> float:
    """计算网络传输延迟"""
    return receive_ts - sensor_ts