# -*- coding: utf-8 -*-
import json
import logging
import time
import threading
from paho.mqtt import client as mqtt_client
# MQTT 全局配置
MQTT_CONFIG = {
'broker': '172.19.64.31',
'port': 1883,
'username': 'mh',
'password': '123456',
'pub_topic': 'geek_pdu/{pdu_id}/publish',
'sub_topic': 'geek_pdu/{pdu_id}/subscribe',
}
# 全局 MQTT 客户端
_mqtt_client = None
_client_lock = threading.Lock()
_subscribed_pdus = set()
_subscribe_lock = threading.Lock()
def get_mqtt_client():
"""获取 MQTT 客户端单例"""
global _mqtt_client
if _mqtt_client is None:
with _client_lock:
if _mqtt_client is None:
_mqtt_client = _init_mqtt_client()
return _mqtt_client
def _ensure_subscribed(pdu_id, log_callback=None):
"""确保订阅了指定 PDU 的状态 Topic"""
client = get_mqtt_client()
if not client or not client.is_connected():
return False
with _subscribe_lock:
if pdu_id in _subscribed_pdus:
return True
sub_topic = MQTT_CONFIG['sub_topic'].replace('{pdu_id}', pdu_id)
client.subscribe(sub_topic)
_subscribed_pdus.add(pdu_id)
if log_callback:
log_callback(f"📡 订阅 PDU 状态: {sub_topic}")
logging.info(f"Subscribed to {sub_topic}")
return True
def _init_mqtt_client():
"""初始化 MQTT 客户端"""
client = mqtt_client.Client()
client.username_pw_set(MQTT_CONFIG['username'], MQTT_CONFIG['password'])
def on_connect(client, userdata, flags, rc):
if rc == 0:
logging.info("MQTT PDU client connected")
else:
logging.error(f"MQTT PDU client connect failed, rc={rc}")
def on_message(client, userdata, msg):
# 收到消息时只打印，不解析（避免刷屏）
try:
payload = msg.payload.decode()
topic = msg.topic
logging.debug(f"MQTT received: {topic} -> {payload}")
except Exception as e:
logging.error(f"MQTT on_message error: {e}")
def on_disconnect(client, userdata, rc):
logging.warning(f"MQTT PDU client disconnected, rc={rc}")
with _subscribe_lock:
_subscribed_pdus.clear()
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
try:
client.connect(MQTT_CONFIG['broker'], MQTT_CONFIG['port'], keepalive=60)
client.loop_start()
time.sleep(1)
except Exception as e:
logging.error(f"MQTT PDU client connect error: {e}")
return client
def _publish(pdu_id, msg_dict, log_callback=None):
"""发送 MQTT 消息"""
client = get_mqtt_client()
if not client or not client.is_connected():
error_msg = f"MQTT client not connected"
if log_callback:
log_callback(f"❌ {error_msg}")
return False, error_msg
pub_topic = MQTT_CONFIG['pub_topic'].replace('{pdu_id}', pdu_id)
msg = json.dumps(msg_dict)
result = client.publish(pub_topic, msg, qos=1)
if result[0] == 0:
if log_callback:
log_callback(f"✅ 发送命令到 PDU {pdu_id}: {msg}")
return True, ""
else:
error_msg = f"Send failed, code: {result[0]}"
if log_callback:
log_callback(f"❌ {error_msg}")
return False, error_msg
def pdu_power_off(pdu_id, log_callback=None):
"""发送 PDU 断电命令"""
_ensure_subscribed(pdu_id, log_callback)
return _publish(pdu_id, {'type': 'event', 'key': 0}, log_callback)
def pdu_power_on(pdu_id, log_callback=None):
"""发送 PDU 上电命令"""
_ensure_subscribed(pdu_id, log_callback)
return _publish(pdu_id, {'type': 'event', 'key': 1}, log_callback)
def pdu_power_cycle(pdu_id, delay=5, log_callback=None):
"""
PDU 断电后延时恢复（先断电，等待 delay 秒，再上电）
"""
if log_callback:
log_callback(f"执行 PDU 掉电循环 (ID: {pdu_id}, 断电等待 {delay} 秒)")
# 1. 断电
success, msg = pdu_power_off(pdu_id, log_callback)
if not success:
return False, msg
# 2. 等待
if log_callback:
log_callback(f"等待 {delay} 秒...")
time.sleep(delay)
# 3. 上电
success, msg = pdu_power_on(pdu_id, log_callback)
if not success:
return False, msg
return True, ""