网络结构：光猫->PVE主机->PVE内部虚拟路由器系统->群晖
原来正常的工作流程：
- 市电断开->UPS供电->群晖识别低电量自动关机并发送邮件通知->PVE nut-monitor->PVE 自动关机
- 市电恢复->群晖、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%)
软件依赖
apt update
apt install nut nut-server nut-client snmpd snmpNUT部分
/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
#!/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