基于SNMP的机柜温湿度传感器：设备轮询与异常告警开发笔记

┌──────────────────────────────────────────────────────────────────┐
│ 温湿度传感器私有MIB树（示例）                                   │
│                                                                  │
│  .1.3.6.1.4.1.{enterprise_id}.{product_id}                      │
│  │                                                               │
│  ├── 1.1  deviceInfo                                            │
│  │   ├── 1  deviceName      (OCTET STRING)                      │
│  │   ├── 2  deviceLocation  (OCTET STRING)                      │
│  │   ├── 3  deviceUptime    (Timeticks)                         │
│  │   └── 4  deviceStatus    (INTEGER: 1=ok, 2=warn, 3=error)    │
│  │                                                               │
│  ├── 1.2  sensorData                                            │
│  │   ├── 1  temperature    (INTEGER, 0.1°C精度)                 │
│  │   ├── 2  humidity       (INTEGER, 0.1%RH精度)                │
│  │   ├── 3  dewPoint       (INTEGER, 0.1°C精度)                 │
│  │   └── 4  sensorStatus   (INTEGER: 1=normal, 2=fault)         │
│  │                                                               │
│  ├── 1.3  thresholds                                             │
│  │   ├── 1  tempHighWarning  (INTEGER)                          │
│  │   ├── 2  tempHighCritical (INTEGER)                          │
│  │   ├── 3  tempLowWarning   (INTEGER)                          │
│  │   ├── 4  humiHighWarning  (INTEGER)                          │
│  │   └── 5  humiLowWarning   (INTEGER)                          │
│  │                                                               │
│  └── 1.4  traps                                                 │
│      ├── 1  tempThresholdExceeded (Trap OID)                    │
│      ├── 2  humiThresholdExceeded (Trap OID)                    │
│      └── 3  sensorCommunicationLost (Trap OID)                   │
└──────────────────────────────────────────────────────────────────┘

# snmp_mib_loader.py
"""SNMP MIB加载与OID映射"""

from typing import Dict

class SensorMIB:
    """温湿度传感器MIB定义"""

    # 企业OID前缀（示例，实际替换为厂家分配值）
    BASE_OID = ".1.3.6.1.4.1.12345.1"

    OID_MAP = {
        "device_name":        f"{BASE_OID}.1.1.1.0",
        "device_location":    f"{BASE_OID}.1.1.2.0",
        "device_uptime":      f"{BASE_OID}.1.1.3.0",
        "device_status":      f"{BASE_OID}.1.1.4.0",
        "temperature":        f"{BASE_OID}.1.2.1.0",
        "humidity":           f"{BASE_OID}.1.2.2.0",
        "dew_point":          f"{BASE_OID}.1.2.3.0",
        "sensor_status":      f"{BASE_OID}.1.2.4.0",
        "temp_high_warning":  f"{BASE_OID}.1.3.1.0",
        "temp_high_critical": f"{BASE_OID}.1.3.2.0",
        "temp_low_warning":   f"{BASE_OID}.1.3.3.0",
        "humi_high_warning":  f"{BASE_OID}.1.3.4.0",
        "humi_low_warning":   f"{BASE_OID}.1.3.5.0",
    }

    # 反向映射（OID→名称）
    REVERSE_MAP = {v: k for k, v in OID_MAP.items()}

    # Trap OID映射
    TRAP_OIDS = {
        f"{BASE_OID}.1.4.1.0": "temp_threshold_exceeded",
        f"{BASE_OID}.1.4.2.0": "humi_threshold_exceeded",
        f"{BASE_OID}.1.4.3.0": "sensor_comm_lost",
    }

    @classmethod
    def get_oid(cls, name: str) -> str:
        return cls.OID_MAP.get(name, "")

    @classmethod
    def get_name(cls, oid: str) -> str:
        return cls.REVERSE_MAP.get(oid, oid)

    @classmethod
    def get_table_oids(cls) -> list:
        """返回需要Walk的表型OID前缀"""
        return [
            f"{cls.BASE_OID}.1.2",  # sensorData表
        ]

# snmp_poller.py
from pysnmp.hlapi import *
from typing import Dict, Optional, List
import time

class SNMPSensorPoller:
    """SNMP温湿度传感器轮询器"""

    def __init__(self, community: str = "public", version: int = 2,
                 timeout: float = 2.0, retries: int = 2):
        self.community = community
        self.version = version  # 1 or 2
        self.timeout = timeout
        self.retries = retries

    def get_scalar(self, target_ip: str, oid: str, port: int = 161) -> Optional[dict]:
        """SNMP GET单个标量值"""
        try:
            iterator = getCmd(
                SnmpEngine(),
                CommunityData(self.community, mpModel=1 if self.version == 2 else 0),
                UdpTransportTarget((target_ip, port), timeout=self.timeout, retries=self.retries),
                ContextData(),
                ObjectType(ObjectIdentity(oid))
            )

            errorIndication, errorStatus, errorIndex, varBinds = next(iterator)

            if errorIndication:
                return {"error": str(errorIndication), "value": None}
            elif errorStatus:
                return {"error": f"{errorStatus.prettyPrint()} at {errorIndex}", "value": None}
            else:
                for varBind in varBinds:
                    oid_str = str(varBind[0])
                    value = varBind[1]
                    return {
                        "oid": oid_str,
                        "value": self._parse_value(value),
                        "raw": str(value)
                    }
        except Exception as e:
            return {"error": str(e), "value": None}

    def get_bulk(self, target_ip: str, oids: List[str], port: int = 161) -> Dict[str, dict]:
        """批量GET多个OID"""
        results = {}
        for oid in oids:
            result = self.get_scalar(target_ip, oid, port)
            name = SensorMIB.get_name(oid)
            results[name] = result
        return results

    def walk(self, target_ip: str, prefix_oid: str, port: int = 161) -> List[dict]:
        """SNMP Walk遍历表型数据"""
        results = []

        try:
            for errorIndication, errorStatus, errorIndex, varBinds in nextCmd(
                SnmpEngine(),
                CommunityData(self.community, mpModel=1 if self.version == 2 else 0),
                UdpTransportTarget((target_ip, port), timeout=self.timeout, retries=self.retries),
                ContextData(),
                ObjectType(ObjectIdentity(prefix_oid)),
                lexicographicMode=False
            ):
                if errorIndication or errorStatus:
                    break
                for varBind in varBinds:
                    results.append({
                        "oid": str(varBind[0]),
                        "value": self._parse_value(varBind[1]),
                        "raw": str(varBind[1])
                    })
        except Exception as e:
            pass

        return results

    def _parse_value(self, value) -> any:
        """解析SNMP返回值"""
        from pysnmp.proto.rfc1902 import Integer, Gauge32, OctetString, TimeTicks

        if isinstance(value, Integer) or isinstance(value, Gauge32):
            return int(value)
        elif isinstance(value, OctetString):
            return bytes(value).decode('utf-8', errors='replace')
        elif isinstance(value, TimeTicks):
            return int(value)  # 百分之一秒
        else:
            return str(value)

# multi_device_poller.py
import threading
import queue
import time
from typing import Dict, List
from snmp_poller import SNMPSensorPoller
from snmp_mib_loader import SensorMIB

class MultiDevicePoller:
    """多设备并发SNMP轮询管理器"""

    def __init__(self, devices: Dict[str, dict], poll_interval: float = 30.0):
        """
        devices: {device_id: {"ip": ip, "community": str, "version": int}}
        """
        self.devices = devices
        self.poll_interval = poll_interval
        self.poller = SNMPSensorPoller()
        self.result_queue = queue.Queue()
        self.latest_data: Dict[str, dict] = {}
        self.running = False

    def poll_single_device(self, device_id: str, config: dict) -> dict:
        """轮询单台设备"""
        ip = config["ip"]
        community = config.get("community", "public")
        version = config.get("version", 2)

        # 构造需要采集的OID列表
        oids = [
            SensorMIB.get_oid("temperature"),
            SensorMIB.get_oid("humidity"),
            SensorMIB.get_oid("dew_point"),
            SensorMIB.get_oid("sensor_status"),
            SensorMIB.get_oid("device_status"),
            SensorMIB.get_oid("device_uptime"),
        ]

        # 临时切换community
        self.poller.community = community
        self.poller.version = version

        start_time = time.time()
        results = self.poller.get_bulk(ip, oids)
        elapsed = (time.time() - start_time) * 1000

        # 解析温湿度值（注意0.1精度）
        parsed = {
            "device_id": device_id,
            "ip": ip,
            "poll_time_ms": elapsed,
            "timestamp": time.time(),
            "error": None
        }

        for name, result in results.items():
            if result.get("error"):
                parsed["error"] = result["error"]
                continue

            value = result.get("value")
            if name == "temperature" and value is not None:
                parsed["temperature"] = value / 10.0
            elif name == "humidity" and value is not None:
                parsed["humidity"] = value / 10.0
            elif name == "dew_point" and value is not None:
                parsed["dew_point"] = value / 10.0
            elif name == "sensor_status" and value is not None:
                parsed["sensor_status"] = value
            elif name == "device_status" and value is not None:
                parsed["device_status"] = value
            elif name == "device_uptime" and value is not None:
                parsed["uptime_seconds"] = value / 100  # TimeTicks→秒

        return parsed

    def poll_all_threaded(self) -> Dict[str, dict]:
        """多线程并发轮询所有设备"""
        results = {}
        threads = []
        lock = threading.Lock()

        def _worker(device_id, config):
            try:
                data = self.poll_single_device(device_id, config)
                with lock:
                    results[device_id] = data
            except Exception as e:
                with lock:
                    results[device_id] = {
                        "device_id": device_id,
                        "error": str(e),
                        "timestamp": time.time()
                    }

        for did, cfg in self.devices.items():
            t = threading.Thread(target=_worker, args=(did, cfg), daemon=True)
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=self.poller.timeout + 5)

        self.latest_data = results
        return results

    def start_daemon(self, callback=None):
        """启动后台轮询守护线程"""
        self.running = True

        def _loop():
            while self.running:
                start = time.time()
                data = self.poll_all_threaded()

                if callback:
                    callback(data)

                elapsed = time.time() - start
                sleep_time = max(0, self.poll_interval - elapsed)
                time.sleep(sleep_time)

        thread = threading.Thread(target=_loop, daemon=True)
        thread.start()
        return thread

# alert_engine.py
from dataclasses import dataclass, field
from typing import Dict, List, Callable
from enum import Enum
import time

class AlertSeverity(Enum):
    INFO = 1
    WARNING = 2
    CRITICAL = 3

class AlertState(Enum):
    CLEAR = 0
    ACTIVE = 1
    ACKNOWLEDGED = 2
    SUPPRESSED = 3

@dataclass
class AlertRule:
    rule_id: str
    device_id: str
    metric: str              # temperature / humidity / sensor_status
    condition: str           # > / < / == / !=
    threshold: float
    severity: AlertSeverity
    cooldown: float = 300    # 同一告警冷却时间（秒）
    consecutive_count: int = 3  # 连续N次满足条件才触发（防抖动）
    enabled: bool = True

@dataclass
class AlertEvent:
    rule_id: str
    device_id: str
    metric: str
    current_value: float
    threshold: float
    severity: AlertSeverity
    state: AlertState
    first_triggered: float
    last_triggered: float
    trigger_count: int
    message: str

class AlertEngine:
    """SNMP采集数据异常告警引擎"""

    def __init__(self):
        self.rules: Dict[str, AlertRule] = {}
        self.active_alerts: Dict[str, AlertEvent] = {}
        self.consecutive_hits: Dict[str, int] = {}  # rule_id → 连续命中计数
        self.last_alert_time: Dict[str, float] = {}
        self.callbacks: List[Callable] = []

    def add_rule(self, rule: AlertRule):
        self.rules[rule.rule_id] = rule

    def register_callback(self, callback: Callable):
        """注册告警通知回调（可对接短信/邮件/Webhook）"""
        self.callbacks.append(callback)

    def evaluate(self, poll_data: Dict[str, dict]):
        """评估一轮轮询数据"""
        for device_id, data in poll_data.items():
            if data.get("error"):
                # 设备不可达，触发通信异常告警
                self._handle_comm_failure(device_id, data["error"])
                continue

            for rule in self.rules.values():
                if not rule.enabled or rule.device_id != device_id:
                    continue

                current_value = data.get(rule.metric)
                if current_value is None:
                    continue

                matched = self._check_condition(
                    current_value, rule.condition, rule.threshold
                )

                rule_key = rule.rule_id
                if matched:
                    self.consecutive_hits[rule_key] = self.consecutive_hits.get(rule_key, 0) + 1

                    if self.consecutive_hits[rule_key] >= rule.consecutive_count:
                        self._trigger_alert(rule, device_id, current_value, data)
                else:
                    # 条件不满足，清零计数，检查是否需要清除告警
                    self.consecutive_hits[rule_key] = 0
                    if rule_key in self.active_alerts:
                        self._clear_alert(rule_key, current_value)

    def _check_condition(self, value: float, condition: str, threshold: float) -> bool:
        if condition == ">":
            return value > threshold
        elif condition == ">=":
            return value >= threshold
        elif condition == "<":
            return value < threshold
        elif condition == "<=":
            return value <= threshold
        elif condition == "==":
            return value == threshold
        elif condition == "!=":
            return value != threshold
        return False

    def _trigger_alert(self, rule: AlertRule, device_id: str,
                      current_value: float, data: dict):
        """触发告警"""
        now = time.time()
        rule_key = rule.rule_id

        # 冷却检查
        last = self.last_alert_time.get(rule_key, 0)
        if now - last < rule.cooldown and rule_key in self.active_alerts:
            return

        if rule_key in self.active_alerts:
            # 已有活跃告警，更新计数和时间
            self.active_alerts[rule_key].last_triggered = now
            self.active_alerts[rule_key].trigger_count += 1
            self.active_alerts[rule_key].current_value = current_value
        else:
            # 新告警
            event = AlertEvent(
                rule_id=rule_key,
                device_id=device_id,
                metric=rule.metric,
                current_value=current_value,
                threshold=rule.threshold,
                severity=rule.severity,
                state=AlertState.ACTIVE,
                first_triggered=now,
                last_triggered=now,
                trigger_count=1,
                message=self._format_message(rule, current_value, data)
            )
            self.active_alerts[rule_key] = event

            # 通知回调
            for cb in self.callbacks:
                try:
                    cb(event)
                except Exception:
                    pass

        self.last_alert_time[rule_key] = now

    def _clear_alert(self, rule_key: str, current_value: float):
        """清除告警"""
        if rule_key in self.active_alerts:
            event = self.active_alerts[rule_key]
            event.state = AlertState.CLEAR
            event.current_value = current_value

            # 通知清除
            for cb in self.callbacks:
                try:
                    cb(event)
                except Exception:
                    pass

            del self.active_alerts[rule_key]

    def _handle_comm_failure(self, device_id: str, error: str):
        """处理通信失败"""
        rule_key = f"comm_loss_{device_id}"
        now = time.time()

        if rule_key not in self.active_alerts:
            event = AlertEvent(
                rule_id=rule_key,
                device_id=device_id,
                metric="communication",
                current_value=0,
                threshold=0,
                severity=AlertSeverity.CRITICAL,
                state=AlertState.ACTIVE,
                first_triggered=now,
                last_triggered=now,
                trigger_count=1,
                message=f"设备{device_id}通信失败: {error}"
            )
            self.active_alerts[rule_key] = event

            for cb in self.callbacks:
                try:
                    cb(event)
                except Exception:
                    pass

    def _format_message(self, rule: AlertRule, value: float, data: dict) -> str:
        metric_name = {
            "temperature": "温度",
            "humidity": "湿度",
            "dew_point": "露点",
            "sensor_status": "传感器状态"
        }.get(rule.metric, rule.metric)

        return (f"机柜传感器 {data.get('ip', rule.device_id)} "
                f"{metric_name} {value}{'°C' if 'temp' in rule.metric else ('%RH' if 'humi' in rule.metric else '')} "
                f"{'超过' if rule.condition in ('>', '>=') else '低于'}阈值 {rule.threshold}")

    def get_active_alerts(self) -> List[AlertEvent]:
        return list(self.active_alerts.values())

    def acknowledge_alert(self, rule_id: str):
        if rule_id in self.active_alerts:
            self.active_alerts[rule_id].state = AlertState.ACKNOWLEDGED

# alert_rules_config.py
from alert_engine import AlertEngine, AlertRule, AlertSeverity

def load_default_rules(engine: AlertEngine, device_ids: list):
    """加载默认告警规则"""

    for did in device_ids:
        # 温度过高告警（Warning）
        engine.add_rule(AlertRule(
            rule_id=f"{did}_temp_high_warn",
            device_id=did,
            metric="temperature",
            condition=">",
            threshold=28.0,
            severity=AlertSeverity.WARNING,
            cooldown=300,
            consecutive_count=3
        ))

        # 温度过高告警（Critical）
        engine.add_rule(AlertRule(
            rule_id=f"{did}_temp_high_crit",
            device_id=did,
            metric="temperature",
            condition=">",
            threshold=35.0,
            severity=AlertSeverity.CRITICAL,
            cooldown=180,
            consecutive_count=2
        ))

        # 温度过低告警
        engine.add_rule(AlertRule(
            rule_id=f"{did}_temp_low_warn",
            device_id=did,
            metric="temperature",
            condition="<",
            threshold=18.0,
            severity=AlertSeverity.WARNING,
            cooldown=300,
            consecutive_count=3
        ))

        # 湿度过高告警
        engine.add_rule(AlertRule(
            rule_id=f"{did}_humi_high_warn",
            device_id=did,
            metric="humidity",
            condition=">",
            threshold=60.0,
            severity=AlertSeverity.WARNING,
            cooldown=600,
            consecutive_count=3
        ))

        # 湿度过低告警
        engine.add_rule(AlertRule(
            rule_id=f"{did}_humi_low_warn",
            device_id=did,
            metric="humidity",
            condition="<",
            threshold=30.0,
            severity=AlertSeverity.WARNING,
            cooldown=600,
            consecutive_count=3
        ))

        # 传感器故障告警
        engine.add_rule(AlertRule(
            rule_id=f"{did}_sensor_fault",
            device_id=did,
            metric="sensor_status",
            condition="!=",
            threshold=1,  # 1=normal
            severity=AlertSeverity.CRITICAL,
            cooldown=60,
            consecutive_count=2
        ))

# trap_receiver.py
from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import ntfrcv
from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.proto.api import v2c
from typing import Callable, List
import socket

class SNMPTrapReceiver:
    """SNMP Trap接收器"""

    def __init__(self, listen_ip: str = "0.0.0.0", listen_port: int = 162):
        self.listen_ip = listen_ip
        self.listen_port = listen_port
        self.snmp_engine = engine.SnmpEngine()
        self.callbacks: List[Callable] = []

    def register_callback(self, callback: Callable):
        self.callbacks.append(callback)

    def start(self):
        """启动Trap接收（阻塞）"""
        # 配置UDP传输
        config.addTransport(
            self.snmp_engine,
            udp.domainName + (1,),
            udp.UdpTransport().openServerMode((self.listen_ip, self.listen_port))
        )

        # 配置社区字符串
        config.addV1System(self.snmp_engine, 'trap_receiver', 'public')

        # 注册Trap处理回调
        ntfrcv.NotificationReceiver(self.snmp_engine, self._trap_handler)

        print(f"SNMP Trap接收器启动: {self.listen_ip}:{self.listen_port}")

        # 启动事件循环
        self.snmp_engine.transportDispatcher.jobStarted(1)
        try:
            self.snmp_engine.transportDispatcher.runDispatcher()
        except KeyboardInterrupt:
            self.snmp_engine.transportDispatcher.closeDispatcher()

    def _trap_handler(self, snmp_engine, state_reference, context_engine_id,
                      context_name, var_binds, cb_ctx):
        """Trap处理回调"""
        trap_data = {
            "source": None,
            "trap_oid": None,
            "variables": {},
            "timestamp": __import__('time').time()
        }

        for oid, value in var_binds:
            oid_str = str(oid)
            value_str = str(value)

            if oid_str == "1.3.6.1.6.3.1.1.4.1.0":  # snmpTrapOID
                trap_data["trap_oid"] = value_str
            else:
                trap_data["variables"][oid_str] = value_str

        # 尝试获取源IP
        if state_reference:
            transport_info = snmp_engine.msgAndPduDsp.getTransportInfo(state_reference)
            if transport_info:
                trap_data["source"] = transport_info[1][0]

        # 通知所有回调
        for cb in self.callbacks:
            try:
                cb(trap_data)
            except Exception:
                pass

┌──────────────────────────────────────────────────────────────────┐
│ Trap + 轮询协同架构                                             │
│                                                                  │
│  传感器 ──Trap(异常事件)──▶ Trap接收器 ──▶ 即时告警             │
│     │                                                            │
│     ├──SNMP GET(周期轮询)──▶ 轮询引擎 ──▶ 趋势分析              │
│     │                               │                            │
│     └──SNMP Walk(发现)──▶ MIB解析 ──▶ 设备注册                  │
│                                                                  │
│  设计原则：                                                       │
│  · Trap负责"即时通知"——异常事件秒级响应                          │
│  · 轮询负责"状态确认"——防止Trap丢失导致状态不一致               │
│  · 两者交叉验证——Trap触发后立即轮询确认当前值                   │
└──────────────────────────────────────────────────────────────────┘

# main_collector.py
"""SNMP机柜温湿度采集+告警主程序"""

import time
import json
from multi_device_poller import MultiDevicePoller
from alert_engine import AlertEngine, AlertSeverity
from alert_rules_config import load_default_rules
from trap_receiver import SNMPTrapReceiver

# 设备列表
DEVICES = {
    "R1-C01": {"ip": "192.168.10.101", "community": "public", "version": 2},
    "R1-C02": {"ip": "192.168.10.102", "community": "public", "version": 2},
    "R1-C03": {"ip": "192.168.10.103", "community": "public", "version": 2},
    "R2-C01": {"ip": "192.168.10.104", "community": "public", "version": 2},
    "R2-C02": {"ip": "192.168.10.105", "community": "public", "version": 2},
}

def alert_notification(alert):
    """告警通知处理"""
    severity_icon = {
        AlertSeverity.INFO: "ℹ️",
        AlertSeverity.WARNING: "⚠️",
        AlertSeverity.CRITICAL: "🔴"
    }
    icon = severity_icon.get(alert.severity, "•")

    print(f"\n{icon} 告警 [{alert.severity.name}]")
    print(f"  设备: {alert.device_id}")
    print(f"  指标: {alert.metric}")
    print(f"  当前值: {alert.current_value}")
    print(f"  阈值: {alert.threshold}")
    print(f"  消息: {alert.message}")
    print(f"  时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(alert.last_triggered))}")

    # 可扩展：发送邮件/短信/Webhook
    # send_email(alert)
    # send_webhook(alert)

def trap_handler(trap_data):
    """Trap事件处理"""
    print(f"\n📡 收到Trap: {trap_data['trap_oid']} from {trap_data['source']}")
    for oid, value in trap_data.get("variables", {}).items():
        print(f"  {oid} = {value}")

    # Trap触发后立即轮询确认
    # poller.poll_single_device(...)

def main():
    print("=" * 60)
    print("机柜温湿度SNMP监控系统启动")
    print("=" * 60)

    # 初始化轮询器
    poller = MultiDevicePoller(DEVICES, poll_interval=30.0)

    # 初始化告警引擎
    alert_engine = AlertEngine()
    load_default_rules(alert_engine, list(DEVICES.keys()))
    alert_engine.register_callback(alert_notification)

    # 回调：每轮轮询完成后评估告警
    def on_poll_complete(data):
        alert_engine.evaluate(data)

        # 打印健康摘要
        healthy = sum(1 for d in data.values() if not d.get("error"))
        total = len(data)
        print(f"\r[{time.strftime('%H:%M:%S')}] 轮询完成: {healthy}/{total} 设备正常", end="", flush=True)

    poller.start_daemon(callback=on_poll_complete)

    # 初始化Trap接收器（独立线程）
    trap_receiver = SNMPTrapReceiver(listen_ip="0.0.0.0", listen_port=162)
    trap_receiver.register_callback(trap_handler)

    trap_thread = threading.Thread(target=trap_receiver.start, daemon=True)
    trap_thread.start()

    print(f"\n已启动 {len(DEVICES)} 台设备监控，轮询间隔 30s")
    print("告警规则已加载，Trap接收器已启动")
    print("按 Ctrl+C 退出\n")

    # 主循环
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在停止...")
        poller.running = False
        print("已退出")

if __name__ == "__main__":
    import threading
    main()