ignace/ups
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

with shutting down plex and raspberry if runtime falls below 300 and 200 secs

Browse Source
This commit is contained in:
ignace
2026-08-10 21:19:06 +02:00
parent e75c3c2358
commit a5b324bf66
3 changed files with 200 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+69
View File
@@ -829,12 +829,81 @@ The collector also sends an ntfy notification when:
- Battery charge crosses below 50%
- Battery charge crosses below 5%
- Estimated battery runtime crosses below 5 minutes
- Local Raspberry Pi shutdown is requested below 200 seconds of runtime
Alert state is stored in `/var/lib/rrd/bx950mi-alert-state.json`, allowing
changes to be detected across cron runs. The first run establishes a baseline
without sending notifications. Threshold notifications can fire again after a
value recovers above the threshold and later drops below it.
### Automatic shutdown of the Plex server
When the UPS status is `OB` or `LB` and estimated runtime is below five
minutes, the collector requests a shutdown of `plex.local` over SSH as user
`ups`. The request is sent once per outage. A failed SSH request is retried by
the next cron run, while a return to `OL` resets the shutdown state.
Generate a dedicated key on the UPS monitor:
```bash
sudo install -d -m 700 /opt/ups/.ssh
sudo ssh-keygen -t ed25519 -f /opt/ups/.ssh/plex_shutdown -N ""
sudo ssh-keyscan -H plex.local | sudo tee /opt/ups/.ssh/known_hosts
sudo chmod 600 /opt/ups/.ssh/plex_shutdown /opt/ups/.ssh/known_hosts
```
Verify the scanned SSH host-key fingerprint against `plex.local` through a
trusted channel before enabling automatic shutdown.
On `plex.local`, create the restricted account and authorize shutdown:
```bash
sudo useradd --create-home --shell /bin/bash ups
sudo visudo -f /etc/sudoers.d/ups-shutdown
```
Add this exact sudoers rule:
```sudoers
ups ALL=(root) NOPASSWD: /usr/sbin/shutdown -h now
```
Add the public key from `/opt/ups/.ssh/plex_shutdown.pub` to
`/home/ups/.ssh/authorized_keys` on `plex.local`, prefixed with a forced command
and SSH restrictions:
```text
restrict,command="sudo /usr/sbin/shutdown -h now" ssh-ed25519 AAAA... ups-shutdown
```
Verify SSH before relying on the automatic action:
```bash
sudo ssh -i /opt/ups/.ssh/plex_shutdown \
-o UserKnownHostsFile=/opt/ups/.ssh/known_hosts ups@plex.local
```
This test will shut down `plex.local`. The hostname, user, key, and known-hosts
file can be overridden for the cron job with:
```text
UPS_SHUTDOWN_HOST
UPS_SHUTDOWN_USER
UPS_SHUTDOWN_KEY
UPS_SHUTDOWN_KNOWN_HOSTS
```
### Automatic shutdown of the Raspberry Pi
When UPS status is `OB` or `LB` and estimated runtime drops below 200 seconds,
the collector sends an ntfy warning and requests shutdown of the Raspberry Pi
running the script. It calls systemd-logind through D-Bus using the `dbus-next`
Python package; it does not invoke a shell command.
Because the collector runs from root's crontab, systemd-logind permits the
power-off request. A failed request is retried by the next cron run, and the
shutdown state resets after the UPS returns to `OL`.
---
# 19. Testing the Python program
requirements.txt
+2
View File
@@ -1,4 +1,6 @@
PyNUTClient>=2.8.5
dbus-next
paramiko
rrdtool
# The NUT server and USB driver are still required to communicate with the UPS.
ups2rrd.py
+129 -1
View File
@@ -33,12 +33,16 @@ NUT UPS name:
BX950MI
"""
import os
import asyncio
import json
import os
import sys
import time
import paramiko
import rrdtool
from dbus_next.aio import MessageBus
from dbus_next.constants import BusType
from PyNUTClient import PyNUT
from ntfy_notify import send_notification
@@ -53,6 +57,20 @@ RRD_DIR = "/var/lib/rrd"
RRD_FILE = os.path.join(RRD_DIR, "bx950mi.rrd")
ALERT_STATE_FILE = os.path.join(RRD_DIR, "bx950mi-alert-state.json")
SHUTDOWN_HOST = os.environ.get("UPS_SHUTDOWN_HOST", "plex.local")
SHUTDOWN_USER = os.environ.get("UPS_SHUTDOWN_USER", "ups")
SHUTDOWN_KEY_FILE = os.environ.get(
"UPS_SHUTDOWN_KEY",
"/opt/ups/.ssh/plex_shutdown",
)
SHUTDOWN_KNOWN_HOSTS = os.environ.get(
"UPS_SHUTDOWN_KNOWN_HOSTS",
"/opt/ups/.ssh/known_hosts",
)
SHUTDOWN_COMMAND = "sudo /usr/sbin/shutdown -h now"
REMOTE_SHUTDOWN_RUNTIME_SECONDS = 5 * 60
LOCAL_SHUTDOWN_RUNTIME_SECONDS = 200
GRAPH_DIR = "/var/www/html/ups"
HTML_FILE = os.path.join(GRAPH_DIR, "index.html")
WEEKLY_HTML_FILE = os.path.join(GRAPH_DIR, "week.html")
@@ -462,6 +480,62 @@ def crossed_below(previous, current, threshold):
)
def shutdown_remote_server(
host=SHUTDOWN_HOST,
user=SHUTDOWN_USER,
key_file=SHUTDOWN_KEY_FILE,
known_hosts=SHUTDOWN_KNOWN_HOSTS,
timeout=10,
):
"""Shut down a remote Linux server over restricted, verified SSH."""
client = paramiko.SSHClient()
client.load_host_keys(known_hosts)
client.set_missing_host_key_policy(paramiko.RejectPolicy())
try:
client.connect(
hostname=host,
username=user,
key_filename=key_file,
timeout=timeout,
banner_timeout=timeout,
auth_timeout=timeout,
look_for_keys=False,
allow_agent=False,
)
client.exec_command(SHUTDOWN_COMMAND, timeout=timeout)
finally:
client.close()
async def request_local_poweroff():
"""Request a local power-off through systemd-logind over D-Bus."""
bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
try:
introspection = await bus.introspect(
"org.freedesktop.login1",
"/org/freedesktop/login1",
)
login = bus.get_proxy_object(
"org.freedesktop.login1",
"/org/freedesktop/login1",
introspection,
)
manager = login.get_interface("org.freedesktop.login1.Manager")
await manager.call_power_off(False)
finally:
bus.disconnect()
def shutdown_local_server():
"""Shut down the Linux server running this collector."""
asyncio.run(request_local_poweroff())
def process_alerts(current):
"""Notify on status changes and configured threshold crossings."""
@@ -472,6 +546,8 @@ def process_alerts(current):
current_charge = current.get("battery_charge")
previous_runtime = previous.get("battery_runtime")
current_runtime = current.get("battery_runtime")
shutdown_sent = previous.get("remote_shutdown_sent", False)
local_shutdown_sent = previous.get("local_shutdown_sent", False)
# An empty state is the initial baseline, not an alert condition.
if previous:
@@ -501,6 +577,56 @@ def process_alerts(current):
f"{current_runtime / 60:.1f} minutes"
)
shutdown_required = (
current_status in {"OB", "LB"}
and current_runtime is not None
and current_runtime < REMOTE_SHUTDOWN_RUNTIME_SECONDS
)
if shutdown_required and not shutdown_sent:
try:
shutdown_remote_server()
shutdown_sent = True
message = (
f"Remote shutdown requested for {SHUTDOWN_HOST}: "
f"UPS runtime is {current_runtime / 60:.1f} minutes"
)
print(message)
notify(message)
except Exception as exc:
print(
f"Remote shutdown of {SHUTDOWN_HOST} failed: {exc}",
file=sys.stderr,
)
local_shutdown_required = (
current_status in {"OB", "LB"}
and current_runtime is not None
and current_runtime < LOCAL_SHUTDOWN_RUNTIME_SECONDS
)
if local_shutdown_required and not local_shutdown_sent:
message = (
"Raspberry Pi shutdown requested: UPS runtime is "
f"{current_runtime:.0f} seconds"
)
notify(message)
try:
shutdown_local_server()
local_shutdown_sent = True
print(message)
except Exception as exc:
print(
f"Local shutdown failed: {exc}",
file=sys.stderr,
)
# A return to utility power starts a new shutdown-alert cycle.
if current_status == "OL":
shutdown_sent = False
local_shutdown_sent = False
# Keep the last valid reading when NUT temporarily omits a value.
state = {
"status": current_status or previous_status,
@@ -510,6 +636,8 @@ def process_alerts(current):
"battery_runtime": (
current_runtime if current_runtime is not None else previous_runtime
),
"remote_shutdown_sent": shutdown_sent,
"local_shutdown_sent": local_shutdown_sent,
}
save_alert_state(state)