用Python脚本实现山特UPS状态监控与智能自动化

山特UPS作为企业级电源保护设备，其自带的WinPower软件虽然功能完善，但缺乏灵活性和可编程性。对于技术爱好者而言，通过串口直接与UPS通信，不仅能获取更丰富的实时数据，还能实现高度定制化的自动化流程。本文将带你从零开始构建一个基于Python的UPS监控系统，突破官方软件的限制。

1. 硬件准备与环境搭建

在开始编码前，我们需要确保硬件连接正确。山特C3K系列UPS配备RS-232串口，但现代计算机通常已不再配备这种传统接口。这里有几个关键注意事项：

• USB转串口适配器选择 ：建议使用FTDI芯片的转换器（如FT232RL），稳定性远优于廉价方案

• 线序确认 ：必须使用直连线而非交叉线，接线方式为：

  UPS端	电脑端
  TX	RX
  RX	TX
  GND	GND

• 串口参数配置 ：

  serial_config = {
      'port': 'COM3',  # 根据实际端口调整
      'baudrate': 2400,
      'bytesize': 8,
      'parity': 'N',
      'stopbits': 1,
      'timeout': 1
  }
  

提示：在Linux系统下，串口设备通常为 /dev/ttyUSB0 形式，需要确保用户有读写权限（可执行 sudo chmod 666 /dev/ttyUSB0 ）

安装必要的Python库：

pip install pyserial schedule python-dotenv

2. 通讯协议解析与基础交互

通过逆向分析WinPower的通信过程，我们发现山特UPS采用基于ASCII字符的简单协议。核心命令包括：

• Q1 ：获取基础状态（输入/输出电压、频率等）
• Q6 ：扩展状态查询（包含电池电压和温度）
• WA ：负载百分比
• WC ：功率数据（有功/视在功率）

典型交互流程如下：

import serial

def query_ups(command):
    with serial.Serial(**serial_config) as ser:
        ser.write(f"{command}\r".encode())
        response = ser.readline().decode().strip()
        return response

# 示例：获取Q6数据
q6_data = query_ups("Q6")
print(f"Raw Q6 response: {q6_data}")

响应数据通常采用固定格式：

( 235.6 236.0 220.0 006 50.0 2.30 28.5 00000000

各字段含义可通过以下映射表理解：

位置	示例值	含义	单位
1	235.6	输入电压	V
2	236.0	输入故障电压	V
3	220.0	输出电压	V
4	006	负载百分比	%
5	50.0	市电频率	Hz
6	2.30	电池电压	V/节
7	28.5	电池温度	°C
8	00000000	状态码	-

3. 构建完整的监控类

为实现更结构化的访问，我们可以创建一个 UPSMonitor 类：

class UPSMonitor:
    def __init__(self, port):
        self.serial_config = {
            'port': port,
            'baudrate': 2400,
            'timeout': 1
        }
        
    def _send_command(self, cmd):
        try:
            with serial.Serial(**self.serial_config) as ser:
                ser.write(f"{cmd}\r".encode())
                return ser.readline().decode().strip()
        except serial.SerialException as e:
            print(f"Serial error: {e}")
            return None

    def get_status(self):
        """获取完整状态信息"""
        q6 = self._parse_q6(self._send_command("Q6"))
        wa = self._parse_wa(self._send_command("WA"))
        wc = self._parse_wc(self._send_command("WC"))
        
        return {
            **q6,
            'load_percentage': wa['load_percentage'],
            'active_power': wc['active_power'],
            'apparent_power': wc['apparent_power'],
            'power_factor': round(wc['active_power'] / wc['apparent_power'], 2)
        }
    
    def _parse_q6(self, response):
        # 实现Q6响应解析
        pass
    
    def _parse_wa(self, response):
        # 实现WA响应解析
        pass
    
    def _parse_wc(self, response):
        # 实现WC响应解析
        pass

4. 高级应用场景实现

4.1 与Home Assistant集成

通过MQTT协议将UPS数据接入智能家居系统：

import paho.mqtt.client as mqtt

class UPS2MQTT(UPSMonitor):
    def __init__(self, port, mqtt_config):
        super().__init__(port)
        self.mqtt_client = mqtt.Client()
        self.mqtt_client.connect(**mqtt_config)
        
    def publish_status(self):
        status = self.get_status()
        for key, value in status.items():
            self.mqtt_client.publish(f"ups/{key}", value)

4.2 自动化关机保护

实现低电量自动安全关机逻辑：

import os
import time

class AutoShutdown(UPSMonitor):
    def __init__(self, port, threshold=20):
        super().__init__(port)
        self.threshold = threshold
        
    def run(self):
        while True:
            status = self.get_status()
            if status['battery_voltage'] < self.threshold:
                self._safe_shutdown()
            time.sleep(60)
            
    def _safe_shutdown(self):
        print("Initiating safe shutdown sequence...")
        # 执行应用关闭脚本
        os.system("/path/to/backup_script.sh")
        # 关机
        os.system("shutdown -h now")

4.3 Prometheus监控集成

创建适合Prometheus拉取的metrics端点：

from prometheus_client import start_http_server, Gauge

class UPSMetrics(UPSMonitor):
    def __init__(self, port, prom_port=8000):
        super().__init__(port)
        self.metrics = {
            'input_voltage': Gauge('ups_input_voltage', 'Input voltage (V)'),
            'output_voltage': Gauge('ups_output_voltage', 'Output voltage (V)'),
            # 其他metrics定义...
        }
        start_http_server(prom_port)
        
    def update_metrics(self):
        status = self.get_status()
        for name, value in status.items():
            if name in self.metrics:
                self.metrics[name].set(value)

5. 异常处理与优化技巧

在实际部署中，需要考虑以下关键点：

• 通信可靠性 ：

  • 实现命令重试机制（当收到NAK响应时）
  • 添加串口连接心跳检测
  • 记录通信失败日志

• 数据平滑处理 ：

  from collections import deque
  import statistics
  
  class SmoothingFilter:
      def __init__(self, window_size=5):
          self.window = deque(maxlen=window_size)
      
      def add_reading(self, value):
          self.window.append(value)
          return statistics.median(self.window)
  

• 配置管理 ： 使用 .env 文件管理敏感参数：

  # .env
  SERIAL_PORT=/dev/ttyUSB0
  MQTT_BROKER=homeassistant.local
  SHUTDOWN_THRESHOLD=20
  

• 系统服务化 ： 创建systemd服务文件实现开机自启：

  # /etc/systemd/system/ups-monitor.service
  [Unit]
  Description=UPS Monitoring Service
  After=network.target
  
  [Service]
  ExecStart=/usr/bin/python3 /opt/ups-monitor/main.py
  Restart=always
  User=root
  
  [Install]
  WantedBy=multi-user.target