PVE虚拟机系统接入USB UPS并通过SNMP伪装为群晖UPS

网络结构：光猫->PVE主机->PVE内部虚拟路由器系统->群晖

原来正常的工作流程：

1. 市电断开->UPS供电->群晖识别低电量自动关机并发送邮件通知->PVE nut-monitor->PVE 自动关机
2. 市电恢复->群晖、PVE来电自启->PVE恢复网络通信->群晖发邮件通知UPS恢复正常

群晖通过USB接入UPS后，所有信息都来源于UPS通告，购入的 `APC BK650M2_CH`默认低电量通报为93%，目前无法通过DSM界面进行任何修改

PVE主机设置nut-client slave，监控群晖的UPS状态，在电量低于93%并且在耗尽前恢复市电的情况下，电池状态会被错误标记为 `FSD OL CHRG`(Forced Shutdown + Online + Charging)，导致PVE主机在来电自启后就会马上识别到 `FSD`信号并迅速关机，只能通过控制面板手动开关UPS功能保存应用刷新为正常的 `OL CHRG`

逼不得已求助AI，将UPS转向接入PVE主机作主设备master，群晖降为slave，这样就能强行覆写不合适的低电量阈值(下述配置为15%)

###### 软件依赖

```bash
apt update
apt install nut nut-server nut-client snmpd snmp
```

##### NUT部分

###### `/etc/nut/nut.conf`

```
MODE=netserver
```

###### `/etc/nut/ups.conf`

```
pollinterval = 5
[ups]
    driver = usbhid-ups
    port = auto
# APC BK650M2_CH默认报告93%为低电量, 不然会让PVE和群晖过早触发关机, 覆写为15%
    override.battery.charge.low = 15
```

###### `/etc/nut/upsd.conf`

```
LISTEN 0.0.0.0 3493
```

###### `/etc/nut/upsd.users`

```
# 群晖用, 这里使用monuser/secret作为用户名/密码, slave模式
[monuser]
password = secret
upsmon slave

# PVE宿主机自身用, 拥有高权限
[pvemon]
password = pvemon_password
upsmon primary
actions = SET
instcmds = ALL
```

###### `/etc/nut/upsmon.conf`

```
MONITOR ups@localhost 1 pvemon pvemon_password primary

# 关机指令PVE识别不了默认的now参数，改为+0
SHUTDOWNCMD "/sbin/shutdown -h +0"
```

##### SNMP部分

###### `/etc/snmp/snmpd.conf`

```
# 添加
view    upsview    included   .1.3.6.1.2.1.33

# 给community组ups_ro分配权限
rocommunity ups_ro 127.0.0.1 -V upsview
rocommunity ups_ro 192.168.9.11 -V upsview

# 将NUT相关信息传递给后续的nut-ups-snmp python脚本
pass_persist .1.3.6.1.2.1.33 /usr/local/bin/nut-ups-snmp
```

###### `/usr/local/bin/nut-ups-snmp`

AI生成，基于 `RFC1628`传递相对应的NUT信息到SNMP

```python
#!/usr/bin/env python3

import sys
import subprocess

BASE = "1.3.6.1.2.1.33"
NUT = "ups@localhost"

def get_nut():
    try:
        p = subprocess.run(
            ["upsc", NUT],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            timeout=3,
        )

data = {}

for line in p.stdout.splitlines():
            if ":" not in line:
                continue

k, v = line.split(":", 1)
            data[k.strip()] = v.strip()

return data

except Exception:
        return {}

def integer(value, default=0):
    try:
        return int(float(value))
    except Exception:
        return default

def build_objects():

d = get_nut()

manufacturer = d.get(
        "device.mfr",
        d.get("ups.mfr", "Unknown")
    )

model = d.get(
        "device.model",
        d.get("ups.model", "Unknown")
    )

serial = d.get(
        "device.serial",
        d.get("ups.serial", "")
    )

software = d.get(
        "driver.version.data",
        d.get("driver.version", "NUT")
    )

charge = integer(d.get("battery.charge"), 0)
    runtime = integer(d.get("battery.runtime"), 0)

input_voltage = integer(
        d.get("input.voltage"),
        0
    )

output_voltage = integer(
        d.get("output.voltage"),
        0
    )

load = integer(
        d.get("ups.load"),
        0
    )

status = d.get("ups.status", "")

objects = {

# ============================================================
        # upsIdent
        # ============================================================

f"{BASE}.1.1.1.0":
            ("string", manufacturer),

f"{BASE}.1.1.2.0":
            ("string", model),

f"{BASE}.1.1.3.0":
            ("string", software),

f"{BASE}.1.1.4.0":
            ("string", "NUT 2.8.1"),

f"{BASE}.1.1.5.0":
            ("string", serial),

# ============================================================
        # upsBattery
        # ============================================================

# upsBatteryStatus
        # 1 unknown
        # 2 battery normal
        # 3 battery low
        # 4 battery depleted
        f"{BASE}.1.2.1.0":
            (
                "integer",
                3 if "LB" in status
                else 2 if "OB" in status
                else 2
            ),

# upsSecondsOnBattery
        f"{BASE}.1.2.2.0":
            (
                "integer",
                integer(d.get("battery.runtime"), 0)
                if "OB" in status else 0
            ),

# upsEstimatedMinutesRemaining
        f"{BASE}.1.2.3.0":
            (
                "integer",
                max(0, runtime // 60)
            ),

# upsEstimatedChargeRemaining
        f"{BASE}.1.2.4.0":
            (
                "integer",
                charge
            ),

# upsBatteryVoltage
        # RFC1628 uses 0.1V
        f"{BASE}.1.2.5.0":
            (
                "integer",
                integer(float(d.get("battery.voltage", 0)) * 10)
            ),

# upsBatteryCurrent
        f"{BASE}.1.2.6.0":
            ("integer", 0),

# upsBatteryTemperature
        f"{BASE}.1.2.7.0":
            ("integer", 0),

# upsBatteryCurrent
        f"{BASE}.1.2.8.0":
            ("integer", 0),

# ============================================================
        # upsInput
        # ============================================================

# upsInputLineBads
        f"{BASE}.3.1.1.0":
            ("counter", 0),

# upsInputNumLines
        f"{BASE}.3.1.2.0":
            ("integer", 1),

# upsInputTable index 1
        #
        # upsInputLineIndex
        f"{BASE}.3.3.1.2.1":
            ("integer", 1),

# upsInputFrequency
        # 0.1 Hz
        f"{BASE}.3.3.1.3.1":
            (
                "integer",
                integer(float(d.get("input.frequency", 0)) * 10)
            ),

# upsInputVoltage
        f"{BASE}.3.3.1.4.1":
            (
                "integer",
                input_voltage
            ),

# upsInputCurrent
        f"{BASE}.3.3.1.5.1":
            ("integer", 0),

# ============================================================
        # upsOutput
        # ============================================================

# upsOutputSource
        #
        # 1 other
        # 2 none
        # 3 normal
        # 4 bypass
        # 5 battery
        # 6 booster
        # 7 reducer
        f"{BASE}.4.1.1.0":
            (
                "integer",
                5 if "OB" in status else 3
            ),

# upsOutputFrequency
        f"{BASE}.4.1.2.0":
            (
                "integer",
                integer(float(d.get("output.frequency", 0)) * 10)
            ),

# upsOutputNumLines
        f"{BASE}.4.1.3.0":
            ("integer", 1),

# upsOutputTable index 1

# upsOutputLineIndex
        f"{BASE}.4.4.1.2.1":
            ("integer", 1),

# upsOutputVoltage
        f"{BASE}.4.4.1.3.1":
            (
                "integer",
                output_voltage
            ),

# upsOutputCurrent
        f"{BASE}.4.4.1.4.1":
            ("integer", 0),

# upsOutputPower
        f"{BASE}.4.4.1.5.1":
            (
                "integer",
                integer(d.get("ups.realpower"), 0)
            ),

# ============================================================
        # upsAlarms
        # ============================================================

# upsAlarmsPresent
        f"{BASE}.6.1.0":
            (
                "gauge",
                1 if "OB" in status else 0
            ),

# ============================================================
        # upsTest
        # ============================================================

# upsTestBatteryStatus
        f"{BASE}.7.1.0":
            ("integer", 1),

# ============================================================
        # upsControl
        # ============================================================

# upsShutdownType
        f"{BASE}.8.1.0":
            ("integer", 1),

# upsShutdownAfterDelay
        f"{BASE}.8.2.0":
            (
                "integer",
                integer(d.get("ups.delay.shutdown"), 0)
            ),

# upsStartupAfterDelay
        f"{BASE}.8.3.0":
            ("integer", 0),
    }

return objects

def find_next(objects, oid):

try:
        requested = tuple(int(x) for x in oid.strip(".").split("."))
    except Exception:
        return None

candidates = []

for key in objects:
        try:
            current = tuple(int(x) for x in key.split("."))
        except Exception:
            continue

if current > requested:
            candidates.append((current, key))

if not candidates:
        return None

candidates.sort()
    return candidates[0][1]

def handle():

objects = build_objects()

while True:

line = sys.stdin.readline()

if not line:
            break

command = line.strip()

# ------------------------------------------------------------
        # PING
        # ------------------------------------------------------------

if command == "PING":
            print("PONG", flush=True)
            continue

# ------------------------------------------------------------
        # GET / GETNEXT
        # ------------------------------------------------------------

if command in ("get", "getnext"):

oid_line = sys.stdin.readline()

if not oid_line:
                break

oid = oid_line.strip().lstrip(".")

objects = build_objects()

if command == "get":
                target = oid
            else:
                target = find_next(objects, oid)

if target is None or target not in objects:
                print("NONE", flush=True)
                continue

typ, value = objects[target]

print(target, flush=True)
            print(typ, flush=True)
            print(value, flush=True)

continue

# ------------------------------------------------------------
        # SET
        # ------------------------------------------------------------

if command == "set":

oid_line = sys.stdin.readline()

if not oid_line:
                break

oid = oid_line.strip().lstrip(".")

value_line = sys.stdin.readline()

if not value_line:
                break

# 当前只提供只读 UPS 信息
            print("not-writable", flush=True)
            continue

# ------------------------------------------------------------
        # Unknown command
        # ------------------------------------------------------------

print("NONE", flush=True)

if __name__ == "__main__":
    handle()
```

###### 相关服务

```
systemctl restart nut-driver@ups
systemctl restart nut-server
systemctl restart nut-monitor
```