1246 lines
48 KiB
Python
Executable File
1246 lines
48 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import sys
|
|
import time
|
|
import os
|
|
import shutil
|
|
from collections import deque
|
|
from pathlib import Path
|
|
|
|
from PyQt5.QtCore import QPointF, QProcess, Qt, QTimer
|
|
from PyQt5.QtGui import QColor, QPainter, QPainterPath, QPen
|
|
from PyQt5.QtWidgets import (
|
|
QApplication,
|
|
QFrame,
|
|
QGridLayout,
|
|
QHBoxLayout,
|
|
QLabel,
|
|
QMainWindow,
|
|
QVBoxLayout,
|
|
QWidget,
|
|
)
|
|
|
|
|
|
NET_GROUPS = {
|
|
"local": ["lo"],
|
|
"wan": ["eth0"],
|
|
"lan": ["eth1"],
|
|
"wifi24": ["wlan0"],
|
|
"wifi5": ["wlan1"],
|
|
}
|
|
|
|
RPI_SERVICE_UNITS = [
|
|
("rpi_docker", "docker.service"),
|
|
("rpi_nginx", "nginx.service"),
|
|
("rpi_mariadb", "mariadb.service"),
|
|
("rpi_control", "control-websocket.service"),
|
|
("rpi_fanpolicy", "fanpanel-apply.service"),
|
|
]
|
|
SYNOLOGY_PING_HOST = "chaegeon.com"
|
|
SYNOLOGY_SSH = [
|
|
"/usr/bin/ssh",
|
|
"-i",
|
|
"/home/seo/.ssh/id_ed25519",
|
|
"-o",
|
|
"BatchMode=yes",
|
|
"-o",
|
|
"IdentitiesOnly=yes",
|
|
"-o",
|
|
"ConnectTimeout=5",
|
|
"-p",
|
|
"1026",
|
|
"seo@chaegeon.com",
|
|
]
|
|
SYNOLOGY_STATUS_SCRIPT = r"""
|
|
while :; do
|
|
wan="$(awk '$2=="00000000"{print $1; exit}' /proc/net/route)"
|
|
[ -n "$wan" ] || wan=ovs_eth0
|
|
echo "WAN_IFACE=$wan"
|
|
awk -v wan="$wan" '
|
|
NR > 2 {
|
|
split($0, row, ":")
|
|
iface = row[1]
|
|
gsub(/^[ \t]+|[ \t]+$/, "", iface)
|
|
data = row[2]
|
|
gsub(/^[ \t]+/, "", data)
|
|
split(data, f, /[ \t]+/)
|
|
}
|
|
iface == wan {printf "WAN_RX=%s\nWAN_TX=%s\nWAN_RX_ERR=%s\nWAN_TX_ERR=%s\nWAN_RX_DROP=%s\nWAN_TX_DROP=%s\n", f[1], f[9], f[3], f[11], f[4], f[12]}
|
|
iface == "lo" {printf "LOCAL_RX=%s\nLOCAL_TX=%s\nLOCAL_RX_ERR=%s\nLOCAL_TX_ERR=%s\nLOCAL_RX_DROP=%s\nLOCAL_TX_DROP=%s\n", f[1], f[9], f[3], f[11], f[4], f[12]}
|
|
' /proc/net/dev
|
|
awk '
|
|
$3 == "sata1" {printf "SATA1_READS=%s\nSATA1_WRITES=%s\n", $6, $10}
|
|
$3 == "sata2" {printf "SATA2_READS=%s\nSATA2_WRITES=%s\n", $6, $10}
|
|
$3 == "sata4" {printf "SATA4_READS=%s\nSATA4_WRITES=%s\n", $6, $10}
|
|
' /proc/diskstats
|
|
awk '{printf "LOAD1=%s\nLOAD5=%s\nLOAD15=%s\n", $1, $2, $3}' /proc/loadavg
|
|
awk '
|
|
/MemTotal:/ {mem_total=$2}
|
|
/MemAvailable:/ {mem_avail=$2}
|
|
/SwapTotal:/ {swap_total=$2}
|
|
/SwapFree:/ {swap_free=$2}
|
|
END {
|
|
if (mem_total > 0) printf "MEM_USED_PCT=%.0f\n", (mem_total - mem_avail) * 100 / mem_total
|
|
if (swap_total > 0) printf "SWAP_USED_PCT=%.0f\n", (swap_total - swap_free) * 100 / swap_total
|
|
else printf "SWAP_USED_PCT=0\n"
|
|
}
|
|
' /proc/meminfo
|
|
awk '{printf "UPTIME=%d\n", $1}' /proc/uptime
|
|
df -P / /volume1 /volume2 2>/dev/null | awk '
|
|
NR > 1 {
|
|
key = ""
|
|
if ($6 == "/") key = "ROOT"
|
|
if ($6 == "/volume1") key = "VOL1"
|
|
if ($6 == "/volume2") key = "VOL2"
|
|
if (key != "") {
|
|
pct = $5
|
|
gsub("%", "", pct)
|
|
printf "%s_FS_PCT=%s\n", key, pct
|
|
}
|
|
}
|
|
'
|
|
echo END=1
|
|
sleep 0.1
|
|
done
|
|
"""
|
|
SYNOLOGY_HEALTH_SCRIPT = r"""
|
|
cpu="$(cat /sys/class/hwmon/hwmon0/temp1_input 2>/dev/null)"
|
|
fan="$(cat /sys/class/hwmon/hwmon1/fan2_input 2>/dev/null)"
|
|
pwm="$(cat /sys/class/hwmon/hwmon1/pwm2 2>/dev/null)"
|
|
enable="$(cat /sys/class/hwmon/hwmon1/pwm2_enable 2>/dev/null)"
|
|
guard="$(systemctl is-active synology-fan-guard.service 2>/dev/null || printf unknown)"
|
|
case "$cpu" in ''|*[!0-9]*) cpu=0 ;; esac
|
|
case "$fan" in ''|*[!0-9]*) fan=0 ;; esac
|
|
case "$pwm" in ''|*[!0-9]*) pwm=0 ;; esac
|
|
case "$enable" in ''|*[!0-9]*) enable=-1 ;; esac
|
|
awk -v cpu="$cpu" -v fan="$fan" -v pwm="$pwm" -v enable="$enable" -v guard="$guard" 'BEGIN {printf "SYNO_CPU=%.1f\nSYNO_FAN=%d\nSYNO_GUARD=%s\nSYNO_PWM=%d\nSYNO_PWM_ENABLE=%d\n", cpu / 1000, fan, guard, pwm, enable}'
|
|
"""
|
|
SAMPLE_INTERVAL_MS = 100
|
|
DASHBOARD_INTERVAL_MS = 500
|
|
FAN_INTERVAL_MS = 500
|
|
WAN_GRAPH_SECONDS = 10
|
|
TEMP_GRAPH_SECONDS = 60
|
|
WAN_GRAPH_POINT_DENSITY = 2
|
|
WAN_GRAPH_POINTS = WAN_GRAPH_SECONDS * 1000 // SAMPLE_INTERVAL_MS * WAN_GRAPH_POINT_DENSITY
|
|
TEMP_GRAPH_POINTS = TEMP_GRAPH_SECONDS * 1000 // DASHBOARD_INTERVAL_MS
|
|
SYNOLOGY_RESTART_INTERVAL_MS = 1000
|
|
TEMP_INTERVAL_MS = 500
|
|
DISK_TEMP_INTERVAL_MS = TEMP_INTERVAL_MS
|
|
DISK_TEMP_CACHE_FILE = "/run/control-rpi-disk-temp"
|
|
DISK_TEMP_CACHE_MAX_AGE_SECONDS = 30
|
|
RPI_THROTTLE_INTERVAL_MS = SAMPLE_INTERVAL_MS
|
|
SYNOLOGY_HEALTH_INTERVAL_MS = FAN_INTERVAL_MS
|
|
RPI_SERVICE_INTERVAL_MS = 5000
|
|
|
|
|
|
def byte_rate_unit(values):
|
|
max_value = max([abs(float(value)) for value in values] + [0.0])
|
|
units = [
|
|
(1024 ** 3, "GB/s"),
|
|
(1024 ** 2, "MB/s"),
|
|
(1024, "KB/s"),
|
|
]
|
|
for factor, label in units:
|
|
if max_value >= factor:
|
|
return factor, label
|
|
return 1, "B/s"
|
|
|
|
|
|
def format_byte_rate(value, factor=None, unit=None, fixed=False):
|
|
if factor is None or unit is None:
|
|
factor, unit = byte_rate_unit([value])
|
|
scaled = float(value) / factor
|
|
if fixed:
|
|
text = f"{scaled:6.2f}"
|
|
elif abs(scaled) >= 100:
|
|
text = f"{scaled:.0f}"
|
|
elif abs(scaled) >= 10:
|
|
text = f"{scaled:.1f}"
|
|
else:
|
|
text = f"{scaled:.2f}"
|
|
return f"{text} {unit}"
|
|
|
|
|
|
def format_wan_caption(rx_rate, tx_rate):
|
|
factor, unit = byte_rate_unit([rx_rate, tx_rate])
|
|
return f"{format_byte_rate(rx_rate, factor, unit, True)} / {format_byte_rate(tx_rate, factor, unit, True)}"
|
|
|
|
|
|
def format_temp_rpm_caption(temp, rpm):
|
|
return f"{temp:5.1f} C / {int(rpm):5d} rpm"
|
|
|
|
|
|
def format_uptime(seconds):
|
|
seconds = max(0, int(float(seconds or 0)))
|
|
days = seconds // 86400
|
|
hours = (seconds % 86400) // 3600
|
|
minutes = (seconds % 3600) // 60
|
|
if days > 0:
|
|
return f"{days}d {hours}h"
|
|
if hours > 0:
|
|
return f"{hours}h {minutes}m"
|
|
return f"{minutes}m"
|
|
|
|
|
|
def load_average():
|
|
parts = read_text("/proc/loadavg").split()
|
|
try:
|
|
return float(parts[0])
|
|
except (IndexError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def memory_used_percent(swap=False):
|
|
values = {}
|
|
for line in read_text("/proc/meminfo").splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
key, rest = line.split(":", 1)
|
|
parts = rest.strip().split()
|
|
if parts and parts[0].isdigit():
|
|
values[key] = int(parts[0])
|
|
|
|
if swap:
|
|
total = values.get("SwapTotal", 0)
|
|
free = values.get("SwapFree", 0)
|
|
return 0.0 if total <= 0 else (total - free) * 100 / total
|
|
|
|
total = values.get("MemTotal", 0)
|
|
available = values.get("MemAvailable", 0)
|
|
return 0.0 if total <= 0 else (total - available) * 100 / total
|
|
|
|
|
|
def disk_used_percent(path):
|
|
try:
|
|
usage = shutil.disk_usage(path)
|
|
except OSError:
|
|
return None
|
|
return usage.used * 100 / usage.total if usage.total > 0 else None
|
|
|
|
|
|
def uptime_seconds():
|
|
try:
|
|
return float(read_text("/proc/uptime").split()[0])
|
|
except (IndexError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def read_text(path, default=""):
|
|
try:
|
|
return Path(path).read_text(encoding="utf-8").strip()
|
|
except OSError:
|
|
return default
|
|
|
|
|
|
def read_int(path, default=0):
|
|
raw = read_text(path)
|
|
return int(raw) if raw.isdigit() else default
|
|
|
|
|
|
def cpu_temp():
|
|
raw = read_int("/sys/class/thermal/thermal_zone0/temp")
|
|
return round(raw / 1000, 1) if raw > 0 else 0.0
|
|
|
|
|
|
def cached_disk_temp():
|
|
values = {}
|
|
for line in read_text(DISK_TEMP_CACHE_FILE).splitlines():
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip()
|
|
|
|
try:
|
|
updated = float(values.get("UPDATED", "0") or 0)
|
|
temp = float(values.get("TEMP", ""))
|
|
except ValueError:
|
|
return None
|
|
|
|
if updated <= 0 or (time.time() - updated) > DISK_TEMP_CACHE_MAX_AGE_SECONDS:
|
|
return None
|
|
return temp
|
|
|
|
|
|
def fan_paths():
|
|
for pwm in Path("/sys/devices/platform/cooling_fan/hwmon").glob("hwmon*/pwm1"):
|
|
base = pwm.parent
|
|
return {
|
|
"pwm": str(pwm),
|
|
"enable": str(base / "pwm1_enable"),
|
|
"rpm": str(base / "fan1_input"),
|
|
}
|
|
for pwm in Path("/sys/class/hwmon").glob("hwmon*/pwm1"):
|
|
base = pwm.parent
|
|
return {
|
|
"pwm": str(pwm),
|
|
"enable": str(base / "pwm1_enable"),
|
|
"rpm": str(base / "fan1_input"),
|
|
}
|
|
return {"pwm": "", "enable": "", "rpm": ""}
|
|
|
|
|
|
def default_iface():
|
|
for line in read_text("/proc/net/route").splitlines()[1:]:
|
|
parts = line.split()
|
|
if len(parts) >= 2 and parts[1] == "00000000":
|
|
return parts[0]
|
|
return "eth0"
|
|
|
|
|
|
def iface_stats(name):
|
|
base = Path("/sys/class/net") / name
|
|
stats = base / "statistics"
|
|
return {
|
|
"name": name,
|
|
"state": read_text(base / "operstate", "unknown"),
|
|
"carrier": read_text(base / "carrier", "0") == "1",
|
|
"rx_bytes": read_int(stats / "rx_bytes"),
|
|
"tx_bytes": read_int(stats / "tx_bytes"),
|
|
"rx_packets": read_int(stats / "rx_packets"),
|
|
"tx_packets": read_int(stats / "tx_packets"),
|
|
"rx_errors": read_int(stats / "rx_errors"),
|
|
"tx_errors": read_int(stats / "tx_errors"),
|
|
"rx_dropped": read_int(stats / "rx_dropped"),
|
|
"tx_dropped": read_int(stats / "tx_dropped"),
|
|
}
|
|
|
|
|
|
def group_stats(names):
|
|
rows = [iface_stats(name) for name in names if (Path("/sys/class/net") / name).exists()]
|
|
if not rows:
|
|
return iface_stats(names[0])
|
|
return {
|
|
"name": "+".join(row["name"] for row in rows),
|
|
"state": "up" if any(row["state"] == "up" for row in rows) else rows[0]["state"],
|
|
"carrier": any(row["carrier"] for row in rows),
|
|
"rx_bytes": sum(row["rx_bytes"] for row in rows),
|
|
"tx_bytes": sum(row["tx_bytes"] for row in rows),
|
|
"rx_packets": sum(row["rx_packets"] for row in rows),
|
|
"tx_packets": sum(row["tx_packets"] for row in rows),
|
|
"rx_errors": sum(row["rx_errors"] for row in rows),
|
|
"tx_errors": sum(row["tx_errors"] for row in rows),
|
|
"rx_dropped": sum(row["rx_dropped"] for row in rows),
|
|
"tx_dropped": sum(row["tx_dropped"] for row in rows),
|
|
}
|
|
|
|
|
|
def disk_io():
|
|
reads = 0
|
|
writes = 0
|
|
found = False
|
|
for line in read_text("/proc/diskstats").splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 14:
|
|
continue
|
|
name = parts[2]
|
|
if name.startswith(("loop", "ram", "zram", "dm-")):
|
|
continue
|
|
found = True
|
|
reads += int(parts[3])
|
|
writes += int(parts[7])
|
|
return reads, writes, not found
|
|
|
|
|
|
class StatusDot(QWidget):
|
|
def __init__(self, active_color):
|
|
super().__init__()
|
|
self.active_color = QColor(active_color)
|
|
self.steady = False
|
|
self.pulse = 0
|
|
self.setFixedSize(15, 15)
|
|
|
|
def set_steady(self, steady=False):
|
|
self.steady = steady
|
|
self.update()
|
|
|
|
def blink(self):
|
|
self.pulse = 1
|
|
self.update()
|
|
|
|
def tick(self):
|
|
if self.pulse > 0:
|
|
self.pulse -= 1
|
|
self.update()
|
|
|
|
def paintEvent(self, event):
|
|
del event
|
|
painter = QPainter(self)
|
|
painter.setRenderHint(QPainter.Antialiasing)
|
|
base = QColor("#8fa0aa")
|
|
if self.steady or self.pulse > 0:
|
|
base = self.active_color
|
|
painter.setBrush(base)
|
|
painter.setPen(Qt.NoPen)
|
|
painter.drawEllipse(2, 2, 11, 11)
|
|
painter.setBrush(QColor(255, 255, 255, 55))
|
|
painter.drawEllipse(6, 4, 4, 4)
|
|
|
|
|
|
class StatusTile(QFrame):
|
|
ACTIVITY_HOLD_SECONDS = 0.12
|
|
ERROR_HOLD_SECONDS = 2.0
|
|
ERROR_BLINK_SECONDS = 0.5
|
|
|
|
def __init__(self, title):
|
|
super().__init__()
|
|
self.read_dot = StatusDot("#16a34a")
|
|
self.write_dot = StatusDot("#f97316")
|
|
self.error_dot = StatusDot("#dc2626")
|
|
self.last_read_at = 0.0
|
|
self.last_write_at = 0.0
|
|
self.last_error_at = 0.0
|
|
self.read_steady = None
|
|
self.write_steady = None
|
|
self.error_steady = None
|
|
self.error_blink = False
|
|
self.error_blink_seconds = self.ERROR_BLINK_SECONDS
|
|
self.title = QLabel(title)
|
|
self.title.setObjectName("statusTitle")
|
|
self.detail = QLabel("-")
|
|
self.detail.setObjectName("statusDetail")
|
|
self.detail.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
|
|
self.setFixedHeight(48)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(6, 4, 6, 4)
|
|
layout.setSpacing(1)
|
|
texts = QVBoxLayout()
|
|
texts.setContentsMargins(0, 0, 0, 0)
|
|
texts.setSpacing(0)
|
|
texts.addWidget(self.title)
|
|
texts.addWidget(self.detail)
|
|
layout.addLayout(texts)
|
|
|
|
dots = QHBoxLayout()
|
|
dots.setContentsMargins(0, 0, 0, 0)
|
|
dots.setSpacing(4)
|
|
dots.addWidget(self.read_dot, 0, Qt.AlignLeft)
|
|
dots.addWidget(self.write_dot, 0, Qt.AlignLeft)
|
|
dots.addWidget(self.error_dot, 0, Qt.AlignLeft)
|
|
dots.addStretch(1)
|
|
layout.addLayout(dots)
|
|
self.setObjectName("statusTile")
|
|
|
|
def set_state(
|
|
self,
|
|
detail,
|
|
error=False,
|
|
read_activity=False,
|
|
write_activity=False,
|
|
read_steady=None,
|
|
write_steady=None,
|
|
error_steady=None,
|
|
error_blink=False,
|
|
error_blink_seconds=None,
|
|
):
|
|
now = time.monotonic()
|
|
self.read_steady = read_steady
|
|
self.write_steady = write_steady
|
|
self.error_steady = error_steady
|
|
self.error_blink = error_blink
|
|
self.error_blink_seconds = error_blink_seconds or self.ERROR_BLINK_SECONDS
|
|
if read_activity:
|
|
self.last_read_at = now
|
|
if write_activity:
|
|
self.last_write_at = now
|
|
if error:
|
|
self.last_error_at = now
|
|
self.update_dots(now)
|
|
self.detail.setText(detail)
|
|
|
|
def tick(self):
|
|
self.update_dots(time.monotonic())
|
|
|
|
def update_dots(self, now):
|
|
read_on = self.read_steady if self.read_steady is not None else now - self.last_read_at <= self.ACTIVITY_HOLD_SECONDS
|
|
write_on = self.write_steady if self.write_steady is not None else now - self.last_write_at <= self.ACTIVITY_HOLD_SECONDS
|
|
if self.error_blink:
|
|
error_on = int(now / self.error_blink_seconds) % 2 == 0
|
|
else:
|
|
error_on = self.error_steady if self.error_steady is not None else now - self.last_error_at <= self.ERROR_HOLD_SECONDS
|
|
self.read_dot.set_steady(read_on)
|
|
self.write_dot.set_steady(write_on)
|
|
self.error_dot.set_steady(error_on)
|
|
|
|
|
|
class Sparkline(QFrame):
|
|
def __init__(self, title, color="#38bdf8", second_color=None, y_unit="", right_unit="", auto_byte_unit=False, max_points=120):
|
|
super().__init__()
|
|
self.setObjectName("chartCard")
|
|
self.color = QColor(color)
|
|
self.second_color = QColor(second_color) if second_color else None
|
|
self.y_unit = y_unit
|
|
self.right_unit = right_unit
|
|
self.auto_byte_unit = auto_byte_unit
|
|
self.max_points = max_points
|
|
self.values = []
|
|
self.second_values = []
|
|
self.title = QLabel(title)
|
|
self.title.setObjectName("chartTitle")
|
|
self.caption = QLabel("-")
|
|
self.caption.setObjectName("chartCaption")
|
|
self.canvas = QWidget()
|
|
self.canvas.paintEvent = self.paint_chart
|
|
self.canvas.setMinimumHeight(58)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(10, 8, 10, 8)
|
|
layout.setSpacing(4)
|
|
top = QHBoxLayout()
|
|
top.addWidget(self.title)
|
|
top.addStretch(1)
|
|
top.addWidget(self.caption)
|
|
layout.addLayout(top)
|
|
layout.addWidget(self.canvas, 1)
|
|
|
|
def set_values(self, values, caption, second_values=None):
|
|
self.values = list(values)[-self.max_points:]
|
|
self.second_values = list(second_values or [])[-self.max_points:]
|
|
self.caption.setText(caption)
|
|
self.canvas.update()
|
|
|
|
def paint_chart(self, event):
|
|
del event
|
|
painter = QPainter(self.canvas)
|
|
painter.setRenderHint(QPainter.Antialiasing)
|
|
right_margin = 34 if self.second_values and self.right_unit else 2
|
|
rect = self.canvas.rect().adjusted(34, 4, -right_margin, -13)
|
|
painter.fillRect(rect, QColor("#0f172a"))
|
|
painter.setPen(QPen(QColor("#263650"), 1))
|
|
painter.drawLine(rect.left(), rect.bottom(), rect.right(), rect.bottom())
|
|
if len(self.values) < 2:
|
|
return
|
|
values = list(self.values)
|
|
second_values = list(self.second_values)
|
|
unit_factor = 1
|
|
unit_label = self.y_unit
|
|
if self.auto_byte_unit:
|
|
unit_factor, unit_label = byte_rate_unit(values + second_values)
|
|
values = [value / unit_factor for value in values]
|
|
second_values = [value / unit_factor for value in second_values]
|
|
scale_values = values + second_values if self.auto_byte_unit and second_values else values
|
|
low = min(scale_values)
|
|
high = max(scale_values)
|
|
if high <= low:
|
|
high = low + 1
|
|
painter.setPen(QColor("#93a4bd"))
|
|
painter.drawText(1, rect.top() + 10, self.axis_text(high, unit_label))
|
|
painter.drawText(1, rect.bottom(), self.axis_text(low, unit_label))
|
|
self.draw_series(painter, rect, values, low, high, self.color)
|
|
if self.second_color and second_values:
|
|
second_low = min(second_values)
|
|
second_high = max(second_values)
|
|
if second_high <= second_low:
|
|
second_high = second_low + 1
|
|
if self.right_unit and not self.auto_byte_unit:
|
|
painter.setPen(QColor("#93a4bd"))
|
|
right_x = self.canvas.rect().right() - 31
|
|
painter.drawText(right_x, rect.top() + 10, self.axis_text(second_high, self.right_unit))
|
|
painter.drawText(right_x, rect.bottom(), self.axis_text(second_low, self.right_unit))
|
|
self.draw_series(painter, rect, second_values, second_low, second_high, self.second_color)
|
|
else:
|
|
self.draw_series(painter, rect, second_values, low, high, self.second_color)
|
|
|
|
def axis_text(self, value, unit=None):
|
|
if abs(value) >= 100:
|
|
text = f"{value:.0f}"
|
|
elif abs(value) >= 10:
|
|
text = f"{value:.1f}"
|
|
else:
|
|
text = f"{value:.2f}"
|
|
return f"{text}{self.y_unit if unit is None else unit}"
|
|
|
|
@staticmethod
|
|
def draw_series(painter, rect, values, low, high, color):
|
|
path = QPainterPath()
|
|
for index, value in enumerate(values):
|
|
x = rect.left() + rect.width() * index / max(1, len(values) - 1)
|
|
y = rect.bottom() - rect.height() * ((float(value) - low) / (high - low))
|
|
point = QPointF(x, y)
|
|
path.moveTo(point) if index == 0 else path.lineTo(point)
|
|
painter.setPen(QPen(color, 2))
|
|
painter.drawPath(path)
|
|
|
|
|
|
class TouchPanel(QMainWindow):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.rows = {}
|
|
self.charts = {}
|
|
self.fan = fan_paths()
|
|
self.wan_iface = default_iface()
|
|
NET_GROUPS["wan"] = [self.wan_iface]
|
|
self.last_net = {}
|
|
self.last_synology = {}
|
|
self.last_synology_net_rate = {}
|
|
self.synology_buffer = ""
|
|
self.synology_values = {}
|
|
self.synology_cpu_temp = 0.0
|
|
self.synology_fan_rpm = 0
|
|
self.last_io = None
|
|
self.last_sample = None
|
|
self.last_graph = None
|
|
self.current_rates = {"wan": {"rx": 0.0, "tx": 0.0}}
|
|
self.current_synology_rates = {"wan": {"rx": 0.0, "tx": 0.0}}
|
|
self.disk_temp = None
|
|
self.latest_temp = 0.0
|
|
self.latest_rpm = 0
|
|
self.rpi_temp_history = deque(maxlen=TEMP_GRAPH_POINTS)
|
|
self.rpi_rpm_history = deque(maxlen=TEMP_GRAPH_POINTS)
|
|
self.synology_temp_history = deque(maxlen=TEMP_GRAPH_POINTS)
|
|
self.synology_rpm_history = deque(maxlen=TEMP_GRAPH_POINTS)
|
|
self.net_history = {
|
|
"wan": {"rx": deque(maxlen=WAN_GRAPH_POINTS), "tx": deque(maxlen=WAN_GRAPH_POINTS)},
|
|
"synology_wan": {"rx": deque(maxlen=WAN_GRAPH_POINTS), "tx": deque(maxlen=WAN_GRAPH_POINTS)},
|
|
}
|
|
|
|
self.setWindowTitle("Standalone Touch Records")
|
|
self.setCursor(Qt.BlankCursor)
|
|
self.setMinimumSize(800, 480)
|
|
self._build_ui()
|
|
|
|
self.sample_timer = QTimer(self)
|
|
self.sample_timer.timeout.connect(self.sample)
|
|
self.sample_timer.start(SAMPLE_INTERVAL_MS)
|
|
|
|
self.dashboard_timer = QTimer(self)
|
|
self.dashboard_timer.timeout.connect(self.update_dashboard)
|
|
self.dashboard_timer.start(DASHBOARD_INTERVAL_MS)
|
|
|
|
self.fan_timer = QTimer(self)
|
|
self.fan_timer.timeout.connect(self.update_fan_status)
|
|
self.fan_timer.start(FAN_INTERVAL_MS)
|
|
|
|
self.disk_temp_timer = QTimer(self)
|
|
self.disk_temp_timer.timeout.connect(self.update_disk_temp)
|
|
self.disk_temp_timer.start(DISK_TEMP_INTERVAL_MS)
|
|
|
|
self.rpi_services_proc = QProcess(self)
|
|
self.rpi_services_proc.finished.connect(self.rpi_services_finished)
|
|
self.rpi_services_timer = QTimer(self)
|
|
self.rpi_services_timer.timeout.connect(self.update_rpi_services)
|
|
self.rpi_services_timer.start(RPI_SERVICE_INTERVAL_MS)
|
|
|
|
self.synology_proc = QProcess(self)
|
|
self.synology_proc.readyReadStandardOutput.connect(self.synology_status_ready)
|
|
self.synology_proc.finished.connect(self.synology_status_finished)
|
|
self.synology_timer = QTimer(self)
|
|
self.synology_timer.timeout.connect(self.update_synology_status)
|
|
self.synology_timer.start(SYNOLOGY_RESTART_INTERVAL_MS)
|
|
|
|
self.synology_health_proc = QProcess(self)
|
|
self.synology_health_proc.finished.connect(self.synology_health_finished)
|
|
self.synology_health_timer = QTimer(self)
|
|
self.synology_health_timer.timeout.connect(self.update_synology_health)
|
|
self.synology_health_timer.start(SYNOLOGY_HEALTH_INTERVAL_MS)
|
|
|
|
self.synology_ping_proc = QProcess(self)
|
|
self.synology_ping_buffer = ""
|
|
self.synology_ping_proc.readyReadStandardOutput.connect(self.synology_ping_ready)
|
|
self.synology_ping_proc.readyReadStandardError.connect(self.synology_ping_error)
|
|
self.synology_ping_proc.finished.connect(self.synology_ping_finished)
|
|
self.synology_ping_timer = QTimer(self)
|
|
self.synology_ping_timer.timeout.connect(self.ping_synology)
|
|
self.synology_ping_timer.start(1000)
|
|
|
|
self.rpi_throttle_proc = QProcess(self)
|
|
self.rpi_throttle_proc.finished.connect(self.rpi_throttle_finished)
|
|
self.rpi_throttle_timer = QTimer(self)
|
|
self.rpi_throttle_timer.timeout.connect(self.update_rpi_throttle)
|
|
self.rpi_throttle_timer.start(RPI_THROTTLE_INTERVAL_MS)
|
|
|
|
self.pulse_timer = QTimer(self)
|
|
self.pulse_timer.timeout.connect(self.tick_pulses)
|
|
self.pulse_timer.start(25)
|
|
self.sample()
|
|
self.update_fan_status()
|
|
self.update_disk_temp()
|
|
self.update_rpi_services()
|
|
self.update_synology_status()
|
|
self.update_synology_health()
|
|
self.ping_synology()
|
|
self.update_rpi_throttle()
|
|
self.update_dashboard()
|
|
|
|
def _build_ui(self):
|
|
root = QWidget()
|
|
layout = QHBoxLayout(root)
|
|
layout.setContentsMargins(4, 8, 10, 8)
|
|
layout.setSpacing(8)
|
|
|
|
rpi_rows = [
|
|
(("io", "SSD I/O"), ("rpi_temp", "TEMP"), ("rpi_fan", "FAN")),
|
|
(("rpi_throttle", "THROT"), ("rpi_load", "LOAD"), ("rpi_mem", "MEM")),
|
|
(("rpi_root", "ROOT"), ("rpi_uptime", "UPTIME"), ("local", "LOCAL")),
|
|
(("wan", "UPLINK"), ("lan", "ETH1"), ("wifi5", "5G")),
|
|
(("wifi24", "2.4G"), ("rpi_docker", "DOCKER"), ("rpi_nginx", "NGINX")),
|
|
(("rpi_mariadb", "MARIADB"), ("rpi_control", "CONTROL"), ("rpi_fanpolicy", "FANPOL")),
|
|
]
|
|
synology_rows = [
|
|
(("synology_ping", "PING"), ("synology_temp", "TEMP"), ("synology_fan", "FAN")),
|
|
(("synology_guard", "GUARD"), ("synology_pwm", "PWM"), ("synology_load", "LOAD")),
|
|
(("synology_mem", "MEM"), ("synology_swap", "SWAP"), ("synology_uptime", "UPTIME")),
|
|
(("synology_root", "ROOT"), ("synology_vol1", "VOL1"), ("synology_vol2", "VOL2")),
|
|
(("synology_local", "LOCAL"), ("synology_wan", "UPLINK"), ("synology_ssh", "SSH")),
|
|
(("synology_sata1", "SATA1 I/O"), ("synology_sata2", "SATA2 I/O"), ("synology_sata4", "SATA4 I/O")),
|
|
]
|
|
|
|
left = self.build_status_panel("RASPBERRY", rpi_rows, fixed_width=398, bottom_blank=True)
|
|
right = self.build_status_panel("SYNOLOGY", synology_rows)
|
|
|
|
header = QHBoxLayout()
|
|
self.clock = QLabel("--:--:--")
|
|
self.clock.setObjectName("clock")
|
|
self.status_text = QLabel("")
|
|
self.status_text.setObjectName("statusText")
|
|
header.addStretch(1)
|
|
header.addWidget(self.status_text)
|
|
header.addWidget(self.clock)
|
|
right.layout().insertLayout(1, header)
|
|
|
|
layout.addWidget(left)
|
|
layout.addWidget(right, 1)
|
|
self.setCentralWidget(root)
|
|
|
|
def build_status_panel(self, title_text, rows, fixed_width=None, bottom_blank=False):
|
|
panel = QFrame()
|
|
panel.setObjectName("statusPanel")
|
|
if fixed_width:
|
|
panel.setFixedWidth(fixed_width)
|
|
panel_layout = QVBoxLayout(panel)
|
|
panel_layout.setContentsMargins(6, 8, 6, 8)
|
|
panel_layout.setSpacing(6)
|
|
|
|
title = QLabel(title_text)
|
|
title.setObjectName("sectionTitle")
|
|
panel_layout.addWidget(title)
|
|
|
|
grid = QGridLayout()
|
|
grid.setContentsMargins(0, 0, 0, 0)
|
|
grid.setHorizontalSpacing(6)
|
|
grid.setVerticalSpacing(6)
|
|
for row_index, row in enumerate(rows):
|
|
for column_index, spec in enumerate(row):
|
|
self.add_status_grid_cell(grid, row_index, column_index, spec)
|
|
panel_layout.addLayout(grid)
|
|
panel_layout.addStretch(1)
|
|
|
|
if bottom_blank:
|
|
blank = QWidget()
|
|
blank.setFixedHeight(48)
|
|
blank.setObjectName("statusEmpty")
|
|
panel_layout.addWidget(blank)
|
|
|
|
return panel
|
|
|
|
def add_status_grid_cell(self, grid, row, column, spec):
|
|
if spec is None:
|
|
spacer = QWidget()
|
|
spacer.setFixedHeight(48)
|
|
spacer.setObjectName("statusEmpty")
|
|
grid.addWidget(spacer, row, column)
|
|
return
|
|
key, label = spec
|
|
tile = StatusTile(label)
|
|
self.rows[key] = tile
|
|
grid.addWidget(tile, row, column)
|
|
|
|
def sample(self):
|
|
now = time.monotonic()
|
|
dt = max(0.001, now - self.last_sample) if self.last_sample else 0.001
|
|
self.last_sample = now
|
|
|
|
reads, writes, disk_error = disk_io()
|
|
io_read_activity = self.last_io is not None and reads != self.last_io[0]
|
|
io_write_activity = self.last_io is not None and writes != self.last_io[1]
|
|
self.last_io = (reads, writes)
|
|
|
|
groups = {key: group_stats(names) for key, names in NET_GROUPS.items()}
|
|
rates = {}
|
|
activity = {}
|
|
for key, row in groups.items():
|
|
previous = self.last_net.get(key)
|
|
current = (
|
|
row["rx_bytes"],
|
|
row["tx_bytes"],
|
|
row["rx_packets"],
|
|
row["tx_packets"],
|
|
row["rx_errors"],
|
|
row["tx_errors"],
|
|
row["rx_dropped"],
|
|
row["tx_dropped"],
|
|
)
|
|
activity[key] = {
|
|
"read": previous is not None and (current[1] > previous[1] or current[3] > previous[3]),
|
|
"write": previous is not None and (current[0] > previous[0] or current[2] > previous[2]),
|
|
"error": previous is not None and any(current[index] > previous[index] for index in [4, 5, 6, 7]),
|
|
}
|
|
if previous:
|
|
rates[key] = {
|
|
"rx": max(0.0, (current[0] - previous[0]) / dt),
|
|
"tx": max(0.0, (current[1] - previous[1]) / dt),
|
|
}
|
|
else:
|
|
rates[key] = {"rx": 0.0, "tx": 0.0}
|
|
self.last_net[key] = current
|
|
|
|
self.current_rates = rates
|
|
self.update_wan_graph()
|
|
self.render_status(groups, rates, activity, io_read_activity, io_write_activity, disk_error)
|
|
|
|
def render_status(self, groups, rates, activity, io_read_activity, io_write_activity, disk_error):
|
|
disk_temp_text = f"{self.disk_temp:.1f} C" if self.disk_temp is not None else ""
|
|
self.rows["io"].set_state(disk_temp_text, disk_error, io_read_activity, io_write_activity)
|
|
self.rows["local"].set_state(self.iface_detail(groups["local"], "lo"), activity["local"]["error"], activity["local"]["read"], activity["local"]["write"])
|
|
self.rows["wan"].set_state(self.iface_detail(groups["wan"], self.wan_iface), activity["wan"]["error"], activity["wan"]["read"], activity["wan"]["write"])
|
|
self.rows["lan"].set_state(self.iface_detail(groups["lan"], "eth1"), activity["lan"]["error"], activity["lan"]["read"], activity["lan"]["write"])
|
|
self.rows["wifi24"].set_state(self.iface_detail(groups["wifi24"], "wlan0"), activity["wifi24"]["error"], activity["wifi24"]["read"], activity["wifi24"]["write"])
|
|
self.rows["wifi5"].set_state(self.iface_detail(groups["wifi5"], "wlan1"), activity["wifi5"]["error"], activity["wifi5"]["read"], activity["wifi5"]["write"])
|
|
|
|
def update_fan_status(self):
|
|
rpm = read_int(self.fan["rpm"])
|
|
self.render_rpi_fan(rpm)
|
|
|
|
self.latest_rpm = rpm
|
|
|
|
def update_dashboard(self):
|
|
self.clock.setText(time.strftime("%H:%M:%S"))
|
|
self.render_rpi_system()
|
|
|
|
def render_rpi_system(self):
|
|
cores = max(1, os.cpu_count() or 1)
|
|
load = load_average()
|
|
self.rows["rpi_load"].set_state(
|
|
f"{load:.2f}",
|
|
read_steady=load < cores * 0.7,
|
|
write_steady=cores * 0.7 <= load < cores * 1.2,
|
|
error_steady=load >= cores * 1.2,
|
|
)
|
|
self.set_percent_tile("rpi_mem", memory_used_percent(), warn=80, critical=92)
|
|
root_used = disk_used_percent("/")
|
|
self.set_percent_tile("rpi_root", root_used, warn=80, critical=92)
|
|
self.rows["rpi_uptime"].set_state(format_uptime(uptime_seconds()), read_steady=True, write_steady=False, error_steady=False)
|
|
|
|
def update_rpi_services(self):
|
|
if self.rpi_services_proc.state() != QProcess.NotRunning:
|
|
return
|
|
units = [unit for _key, unit in RPI_SERVICE_UNITS]
|
|
self.rpi_services_proc.start("/usr/bin/systemctl", ["is-active"] + units)
|
|
|
|
def rpi_services_finished(self, exit_code, exit_status):
|
|
del exit_code, exit_status
|
|
output = bytes(self.rpi_services_proc.readAllStandardOutput()).decode("utf-8", "ignore")
|
|
states = output.splitlines()
|
|
for index, (key, _unit) in enumerate(RPI_SERVICE_UNITS):
|
|
state = states[index].strip() if index < len(states) else "unknown"
|
|
active = state == "active"
|
|
detail = "OK" if active else state.upper()[:6]
|
|
self.rows[key].set_state(detail, read_steady=active, write_steady=False, error_steady=not active)
|
|
|
|
def ping_synology(self):
|
|
if self.synology_ping_proc.state() != QProcess.NotRunning:
|
|
return
|
|
self.synology_ping_buffer = ""
|
|
self.synology_ping_proc.start("/usr/bin/ping", ["-n", "-i", "0.5", SYNOLOGY_PING_HOST])
|
|
|
|
def synology_ping_ready(self):
|
|
output = bytes(self.synology_ping_proc.readAllStandardOutput()).decode("utf-8", "ignore")
|
|
self.synology_ping_buffer += output
|
|
while "\n" in self.synology_ping_buffer:
|
|
line, self.synology_ping_buffer = self.synology_ping_buffer.split("\n", 1)
|
|
latency_ms = self.parse_ping_latency(line)
|
|
if latency_ms is not None:
|
|
self.render_synology_ping(latency_ms)
|
|
|
|
def synology_ping_error(self):
|
|
output = bytes(self.synology_ping_proc.readAllStandardError()).decode("utf-8", "ignore").strip()
|
|
if output:
|
|
self.rows["synology_ping"].set_state("fail", read_steady=False, write_steady=False, error_steady=False, error_blink=True)
|
|
|
|
def synology_ping_finished(self, exit_code, exit_status):
|
|
del exit_code, exit_status
|
|
self.rows["synology_ping"].set_state("fail", read_steady=False, write_steady=False, error_steady=False, error_blink=True)
|
|
QTimer.singleShot(1000, self.ping_synology)
|
|
|
|
@staticmethod
|
|
def parse_ping_latency(line):
|
|
for token in line.split():
|
|
if not token.startswith("time="):
|
|
continue
|
|
try:
|
|
return float(token.split("=", 1)[1])
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
def render_synology_ping(self, latency_ms):
|
|
detail = f"{latency_ms:.1f}ms"
|
|
if latency_ms <= 15:
|
|
self.rows["synology_ping"].set_state(detail, read_steady=True, write_steady=False, error_steady=False)
|
|
elif latency_ms <= 30:
|
|
self.rows["synology_ping"].set_state(detail, read_steady=False, write_steady=True, error_steady=False)
|
|
elif latency_ms <= 50:
|
|
self.rows["synology_ping"].set_state(detail, read_steady=False, write_steady=False, error_steady=True)
|
|
else:
|
|
self.rows["synology_ping"].set_state(detail, read_steady=False, write_steady=False, error_steady=False, error_blink=True)
|
|
|
|
def update_rpi_throttle(self):
|
|
if self.rpi_throttle_proc.state() != QProcess.NotRunning:
|
|
return
|
|
self.rpi_throttle_proc.start("/usr/bin/vcgencmd", ["get_throttled"])
|
|
|
|
def rpi_throttle_finished(self, exit_code, exit_status):
|
|
del exit_status
|
|
output = bytes(self.rpi_throttle_proc.readAllStandardOutput()).decode("utf-8", "ignore").strip()
|
|
if exit_code != 0 or "throttled=" not in output:
|
|
self.rows["rpi_throttle"].set_state("N/A", read_steady=False, write_steady=False, error_steady=True)
|
|
return
|
|
try:
|
|
flags = int(output.split("=", 1)[1], 16)
|
|
except ValueError:
|
|
self.rows["rpi_throttle"].set_state("N/A", read_steady=False, write_steady=False, error_steady=True)
|
|
return
|
|
undervoltage_now = bool(flags & 0x1)
|
|
throttled_now = bool(flags & 0x4)
|
|
history = flags & 0xF0000
|
|
if undervoltage_now and throttled_now:
|
|
self.rows["rpi_throttle"].set_state("UV+TH", read_steady=False, write_steady=False, error_steady=False, error_blink=True, error_blink_seconds=0.1)
|
|
elif throttled_now:
|
|
self.rows["rpi_throttle"].set_state("THROT", read_steady=False, write_steady=False, error_steady=False, error_blink=True, error_blink_seconds=0.5)
|
|
elif undervoltage_now:
|
|
self.rows["rpi_throttle"].set_state("UV", read_steady=False, write_steady=False, error_steady=True)
|
|
elif history:
|
|
self.rows["rpi_throttle"].set_state("HIST", read_steady=False, write_steady=True, error_steady=False)
|
|
else:
|
|
self.rows["rpi_throttle"].set_state("OK", read_steady=True, write_steady=False, error_steady=False)
|
|
|
|
def update_disk_temp(self):
|
|
self.latest_temp = cpu_temp()
|
|
self.disk_temp = cached_disk_temp()
|
|
self.render_rpi_temp(self.latest_temp)
|
|
|
|
def update_synology_status(self):
|
|
if self.synology_proc.state() != QProcess.NotRunning:
|
|
return
|
|
self.synology_buffer = ""
|
|
self.synology_values = {}
|
|
self.synology_proc.start(SYNOLOGY_SSH[0], SYNOLOGY_SSH[1:] + [SYNOLOGY_STATUS_SCRIPT])
|
|
|
|
def update_synology_health(self):
|
|
if self.synology_health_proc.state() != QProcess.NotRunning:
|
|
return
|
|
self.synology_health_proc.start(SYNOLOGY_SSH[0], SYNOLOGY_SSH[1:] + [SYNOLOGY_HEALTH_SCRIPT])
|
|
|
|
def synology_health_finished(self, exit_code, exit_status):
|
|
del exit_status
|
|
if exit_code != 0:
|
|
return
|
|
output = bytes(self.synology_health_proc.readAllStandardOutput()).decode("utf-8", "ignore")
|
|
values = {}
|
|
for line in output.splitlines():
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip()
|
|
try:
|
|
self.synology_cpu_temp = float(values.get("SYNO_CPU", "0") or 0)
|
|
except ValueError:
|
|
self.synology_cpu_temp = 0.0
|
|
self.synology_fan_rpm = int(values.get("SYNO_FAN", "0") or 0)
|
|
self.render_synology_temp()
|
|
self.render_synology_fan()
|
|
self.render_synology_guard(values)
|
|
|
|
def synology_status_ready(self):
|
|
output = bytes(self.synology_proc.readAllStandardOutput()).decode("utf-8", "ignore")
|
|
self.synology_buffer += output
|
|
while "\n" in self.synology_buffer:
|
|
line, self.synology_buffer = self.synology_buffer.split("\n", 1)
|
|
if line == "END=1":
|
|
self.render_synology_status(self.synology_values)
|
|
self.synology_values = {}
|
|
continue
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
self.synology_values[key.strip()] = value.strip()
|
|
|
|
def synology_status_finished(self, exit_code, exit_status):
|
|
del exit_code, exit_status
|
|
self.set_synology_offline()
|
|
|
|
def set_synology_offline(self):
|
|
for key in [
|
|
"synology_sata1",
|
|
"synology_sata2",
|
|
"synology_sata4",
|
|
"synology_wan",
|
|
"synology_local",
|
|
"synology_guard",
|
|
"synology_pwm",
|
|
"synology_load",
|
|
"synology_mem",
|
|
"synology_swap",
|
|
"synology_uptime",
|
|
"synology_root",
|
|
"synology_vol1",
|
|
"synology_vol2",
|
|
"synology_ssh",
|
|
]:
|
|
self.rows[key].set_state("-", read_steady=False, write_steady=False, error_steady=True)
|
|
|
|
def render_synology_status(self, values):
|
|
if not values:
|
|
self.set_synology_offline()
|
|
return
|
|
for disk in ["SATA1", "SATA2", "SATA4"]:
|
|
self.set_synology_io_tile(
|
|
f"synology_{disk.lower()}",
|
|
int(values.get(f"{disk}_READS", "0") or 0),
|
|
int(values.get(f"{disk}_WRITES", "0") or 0),
|
|
self.synology_disk_temp_detail(disk),
|
|
)
|
|
self.set_synology_net_tile(
|
|
"synology_wan",
|
|
int(values.get("WAN_RX", "0") or 0),
|
|
int(values.get("WAN_TX", "0") or 0),
|
|
int(values.get("WAN_RX_ERR", "0") or 0) + int(values.get("WAN_TX_ERR", "0") or 0),
|
|
int(values.get("WAN_RX_DROP", "0") or 0) + int(values.get("WAN_TX_DROP", "0") or 0),
|
|
values.get("WAN_IFACE", "wan"),
|
|
)
|
|
self.set_synology_net_tile(
|
|
"synology_local",
|
|
int(values.get("LOCAL_RX", "0") or 0),
|
|
int(values.get("LOCAL_TX", "0") or 0),
|
|
int(values.get("LOCAL_RX_ERR", "0") or 0) + int(values.get("LOCAL_TX_ERR", "0") or 0),
|
|
int(values.get("LOCAL_RX_DROP", "0") or 0) + int(values.get("LOCAL_TX_DROP", "0") or 0),
|
|
"lo",
|
|
)
|
|
self.render_synology_system(values)
|
|
self.rows["synology_ssh"].set_state("OK", read_steady=True, write_steady=False, error_steady=False)
|
|
|
|
def set_synology_io_tile(self, key, reads, writes, detail):
|
|
previous = self.last_synology.get(key)
|
|
self.last_synology[key] = (reads, writes, 0, 0)
|
|
read_activity = previous is not None and reads > previous[0]
|
|
write_activity = previous is not None and writes > previous[1]
|
|
self.rows[key].set_state(detail, read_activity=read_activity, write_activity=write_activity, error_steady=False)
|
|
|
|
def render_rpi_temp(self, pi_temp):
|
|
pi_disk = self.disk_temp or 0.0
|
|
hottest = max(float(pi_temp), pi_disk)
|
|
detail = f"{hottest:.0f} C" if hottest > 0 else "- C"
|
|
self.rows["rpi_temp"].set_state(
|
|
detail,
|
|
read_steady=0 < hottest < 60,
|
|
write_steady=60 <= hottest < 80,
|
|
error_steady=False,
|
|
error_blink=hottest >= 80,
|
|
)
|
|
|
|
def render_rpi_fan(self, rpm):
|
|
self.render_fan_tile("rpi_fan", rpm)
|
|
|
|
def render_synology_temp(self):
|
|
temp = self.synology_cpu_temp
|
|
detail = f"{temp:.0f} C" if temp > 0 else "- C"
|
|
self.rows["synology_temp"].set_state(
|
|
detail,
|
|
read_steady=0 < temp < 60,
|
|
write_steady=60 <= temp < 80,
|
|
error_steady=False,
|
|
error_blink=temp >= 80,
|
|
)
|
|
|
|
def render_synology_fan(self):
|
|
self.render_fan_tile("synology_fan", self.synology_fan_rpm)
|
|
|
|
def render_synology_guard(self, values):
|
|
guard = values.get("SYNO_GUARD", "unknown")
|
|
try:
|
|
pwm = int(values.get("SYNO_PWM", "0") or 0)
|
|
enable = int(values.get("SYNO_PWM_ENABLE", "-1") or -1)
|
|
except ValueError:
|
|
pwm = 0
|
|
enable = -1
|
|
if guard != "active" or self.synology_fan_rpm <= 0:
|
|
self.rows["synology_guard"].set_state("FAIL", read_steady=False, write_steady=False, error_steady=True)
|
|
elif pwm >= 255 or enable != 5:
|
|
self.rows["synology_guard"].set_state("FULL", read_steady=False, write_steady=True, error_steady=False)
|
|
else:
|
|
self.rows["synology_guard"].set_state("AUTO", read_steady=True, write_steady=False, error_steady=False)
|
|
self.rows["synology_pwm"].set_state(
|
|
f"{round(pwm * 100 / 255):.0f}%",
|
|
read_steady=0 < pwm < 170,
|
|
write_steady=170 <= pwm <= 255,
|
|
error_steady=pwm <= 0 or enable != 5,
|
|
)
|
|
|
|
def render_synology_system(self, values):
|
|
try:
|
|
load = float(values.get("LOAD1", "0") or 0)
|
|
except ValueError:
|
|
load = 0.0
|
|
self.rows["synology_load"].set_state(
|
|
f"{load:.2f}",
|
|
read_steady=load < 2.8,
|
|
write_steady=2.8 <= load < 4.8,
|
|
error_steady=load >= 4.8,
|
|
)
|
|
self.set_percent_tile("synology_mem", self.float_value(values.get("MEM_USED_PCT")), warn=80, critical=92)
|
|
self.set_percent_tile("synology_swap", self.float_value(values.get("SWAP_USED_PCT")), warn=50, critical=80)
|
|
self.set_percent_tile("synology_root", self.float_value(values.get("ROOT_FS_PCT")), warn=80, critical=92)
|
|
self.set_percent_tile("synology_vol1", self.float_value(values.get("VOL1_FS_PCT")), warn=80, critical=92)
|
|
self.set_percent_tile("synology_vol2", self.float_value(values.get("VOL2_FS_PCT")), warn=80, critical=92)
|
|
self.rows["synology_uptime"].set_state(format_uptime(self.float_value(values.get("UPTIME"))), read_steady=True, write_steady=False, error_steady=False)
|
|
|
|
def render_fan_tile(self, key, rpm):
|
|
detail = f"{int(rpm):,} rpm" if rpm > 0 else "0 rpm"
|
|
self.rows[key].set_state(
|
|
detail,
|
|
read_steady=500 <= rpm <= 4500,
|
|
write_steady=4500 < rpm <= 7000,
|
|
error_steady=rpm == 0 or rpm > 7000,
|
|
)
|
|
|
|
def synology_disk_temp_detail(self, disk):
|
|
del disk
|
|
return "조회 중지"
|
|
|
|
def set_synology_net_tile(self, key, rx, tx, errors, drops, detail):
|
|
previous = self.last_synology.get(key)
|
|
self.last_synology[key] = (rx, tx, errors, drops)
|
|
read_activity = previous is not None and tx > previous[1]
|
|
write_activity = previous is not None and rx > previous[0]
|
|
error_activity = previous is not None and (errors > previous[2] or drops > previous[3])
|
|
self.rows[key].set_state(self.drop_detail(drops, detail), error_activity, read_activity, write_activity)
|
|
if key == "synology_wan":
|
|
self.update_synology_wan_rate(rx, tx)
|
|
|
|
def update_synology_wan_rate(self, rx, tx):
|
|
now = time.monotonic()
|
|
previous = self.last_synology_net_rate.get("wan")
|
|
if previous is not None:
|
|
prev_rx, prev_tx, prev_time = previous
|
|
dt = max(0.001, now - prev_time)
|
|
self.current_synology_rates["wan"] = {
|
|
"rx": max(0.0, (rx - prev_rx) / dt),
|
|
"tx": max(0.0, (tx - prev_tx) / dt),
|
|
}
|
|
self.last_synology_net_rate["wan"] = (rx, tx, now)
|
|
self.update_synology_wan_graph()
|
|
|
|
def update_temp_graphs(self, rpi_temp, rpi_rpm):
|
|
if "rpi_temp" not in self.charts or "synology_temp" not in self.charts:
|
|
return
|
|
synology_temp = getattr(self, "synology_cpu_temp", 0.0)
|
|
synology_rpm = getattr(self, "synology_fan_rpm", 0)
|
|
|
|
self.rpi_temp_history.append(rpi_temp)
|
|
self.rpi_rpm_history.append(rpi_rpm)
|
|
self.charts["rpi_temp"].set_values(
|
|
self.rpi_temp_history,
|
|
format_temp_rpm_caption(rpi_temp, rpi_rpm),
|
|
self.rpi_rpm_history,
|
|
)
|
|
|
|
self.synology_temp_history.append(synology_temp)
|
|
self.synology_rpm_history.append(synology_rpm)
|
|
self.charts["synology_temp"].set_values(
|
|
self.synology_temp_history,
|
|
format_temp_rpm_caption(synology_temp, synology_rpm),
|
|
self.synology_rpm_history,
|
|
)
|
|
|
|
def update_wan_graph(self):
|
|
if "rpi_wan" not in self.charts:
|
|
return
|
|
wan = self.current_rates.get("wan", {"rx": 0.0, "tx": 0.0})
|
|
self.net_history["wan"]["rx"].append(wan["rx"])
|
|
self.net_history["wan"]["tx"].append(wan["tx"])
|
|
self.charts["rpi_wan"].set_values(
|
|
self.net_history["wan"]["rx"],
|
|
format_wan_caption(wan["rx"], wan["tx"]),
|
|
self.net_history["wan"]["tx"],
|
|
)
|
|
|
|
def update_synology_wan_graph(self):
|
|
if "synology_wan" not in self.charts:
|
|
return
|
|
wan = self.current_synology_rates.get("wan", {"rx": 0.0, "tx": 0.0})
|
|
self.net_history["synology_wan"]["rx"].append(wan["rx"])
|
|
self.net_history["synology_wan"]["tx"].append(wan["tx"])
|
|
self.charts["synology_wan"].set_values(
|
|
self.net_history["synology_wan"]["rx"],
|
|
format_wan_caption(wan["rx"], wan["tx"]),
|
|
self.net_history["synology_wan"]["tx"],
|
|
)
|
|
|
|
def tick_pulses(self):
|
|
for row in self.rows.values():
|
|
row.tick()
|
|
|
|
@staticmethod
|
|
def iface_detail(row, normal_label="정상"):
|
|
if not row["carrier"]:
|
|
return "꺼짐"
|
|
drops = row["rx_dropped"] + row["tx_dropped"]
|
|
return TouchPanel.drop_detail(drops, normal_label)
|
|
|
|
@staticmethod
|
|
def drop_detail(drops, normal_label):
|
|
del normal_label
|
|
return f"D {drops}"
|
|
|
|
@staticmethod
|
|
def float_value(value):
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def set_percent_tile(self, key, percent, warn=80, critical=92):
|
|
if percent is None:
|
|
self.rows[key].set_state("-", read_steady=False, write_steady=False, error_steady=True)
|
|
return
|
|
detail = f"{percent:.0f}%"
|
|
self.rows[key].set_state(
|
|
detail,
|
|
read_steady=0 <= percent < warn,
|
|
write_steady=warn <= percent < critical,
|
|
error_steady=percent >= critical,
|
|
)
|
|
|
|
@staticmethod
|
|
def iface_warn(row):
|
|
return row["rx_errors"] > 0 or row["tx_errors"] > 0
|
|
|
|
|
|
def main():
|
|
app = QApplication(sys.argv)
|
|
app.setOverrideCursor(Qt.BlankCursor)
|
|
app.setStyleSheet(
|
|
"""
|
|
QWidget { background: #0b1220; color: #e5edf7; font-family: "Noto Sans CJK KR", "Noto Sans CJK KR Regular", "Noto Sans KR", sans-serif; }
|
|
#statusPanel { background: #101827; border-right: 1px solid #263650; }
|
|
#sectionTitle { font-size: 20px; font-weight: 700; }
|
|
#statusGroupTitle { color: #8fa7c5; font-size: 10px; font-weight: 700; background: transparent; }
|
|
#statusTile { background: #d7d9dc; border: 2px groove #f4f4f4; border-radius: 2px; }
|
|
#statusTitle { font-size: 13px; font-weight: 700; color: #111827; background: transparent; }
|
|
#statusDetail { font-size: 9px; color: #475569; background: transparent; }
|
|
#pcWake { background: transparent; border: 0; }
|
|
#pcLabel { color: #dce8f8; font-size: 13px; font-weight: 700; background: transparent; }
|
|
#chartCard { background: #111a2e; border: 1px solid #27364f; border-radius: 12px; }
|
|
#chartTitle { color: #dce8f8; font-size: 14px; font-weight: 700; background: transparent; }
|
|
#chartCaption { color: #93a4bd; font-size: 11px; background: transparent; font-family: "DejaVu Sans Mono", monospace; }
|
|
#clock { font-size: 18px; color: #8fb3ff; font-weight: 700; }
|
|
#statusText { color: #93a4bd; font-size: 13px; background: transparent; }
|
|
"""
|
|
)
|
|
win = TouchPanel()
|
|
win.showFullScreen()
|
|
sys.exit(app.exec_())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|