Files
control/touch_panel/panel.py
T

1105 lines
42 KiB
Python
Executable File

#!/usr/bin/env python3
import sys
import time
import socket
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,
QSlider,
QVBoxLayout,
QWidget,
)
NET_GROUPS = {
"local": ["lo"],
"wan": ["eth0"],
"lan": ["eth1"],
"wifi24": ["wlan0"],
"wifi5": ["wlan1"],
}
PC_IP = "192.168.50.73"
PC_MAC = "74:56:3C:CD:CD:79"
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
echo END=1
sleep 0.05
done
"""
SYNOLOGY_DISK_TEMP_SCRIPT = r"""
for disk in /dev/sata1 /dev/sata2 /dev/sata4; do
temp="-"
if [ -b "$disk" ]; then
temp="$(
/usr/bin/timeout 3 sudo -n /usr/bin/smartctl -A "$disk" 2>/dev/null |
awk '/Temperature_Celsius|Airflow_Temperature_Cel/ {print $10; exit}'
)"
case "$temp" in
''|*[!0-9]*) temp="-" ;;
esac
fi
printf '%s=%s\n' "${disk##/dev/}" "$temp"
done |
awk -F= '
$1 == "sata1" {printf "SATA1_TEMP=%s\n", $2 ? $2 : "-"}
$1 == "sata2" {printf "SATA2_TEMP=%s\n", $2 ? $2 : "-"}
$1 == "sata4" {printf "SATA4_TEMP=%s\n", $2 ? $2 : "-"}
'
"""
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)"
case "$cpu" in ''|*[!0-9]*) cpu=0 ;; esac
case "$fan" in ''|*[!0-9]*) fan=0 ;; esac
awk -v cpu="$cpu" -v fan="$fan" 'BEGIN {printf "SYNO_CPU=%.1f\nSYNO_FAN=%d\n", cpu / 1000, fan}'
"""
SAMPLE_INTERVAL_MS = 50
GRAPH_SECONDS = 10
GRAPH_POINTS = GRAPH_SECONDS * 1000 // SAMPLE_INTERVAL_MS
SYNOLOGY_RESTART_INTERVAL_MS = 1000
DISK_TEMP_INTERVAL_MS = SAMPLE_INTERVAL_MS
SYNOLOGY_DISK_TEMP_INTERVAL_MS = SAMPLE_INTERVAL_MS
SYNOLOGY_HEALTH_INTERVAL_MS = SAMPLE_INTERVAL_MS
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 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 memory_percent():
rows = {}
for line in read_text("/proc/meminfo").splitlines():
if ":" not in line:
continue
key, value = line.split(":", 1)
parts = value.strip().split()
if parts and parts[0].isdigit():
rows[key] = int(parts[0])
total = rows.get("MemTotal", 0)
available = rows.get("MemAvailable", 0)
if total <= 0:
return 0.0
return round((total - available) / total * 100, 1)
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
def send_magic_packet(mac):
clean = mac.replace(":", "").replace("-", "")
if len(clean) != 12:
return False
packet = bytes.fromhex("ff" * 6 + clean * 16)
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
for target in ["192.168.50.255", "255.255.255.255"]:
sock.sendto(packet, (target, 9))
return True
except OSError:
return False
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.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,
):
now = time.monotonic()
self.read_steady = read_steady
self.write_steady = write_steady
self.error_steady = error_steady
self.error_blink = error_blink
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 PcWakeButton(StatusTile):
def __init__(self):
super().__init__("PC")
self.online = False
self.last_wake_at = 0.0
self.wait_until = 0.0
self.error_until = 0.0
self.wait_phase = False
self.blink_timer = QTimer(self)
self.blink_timer.timeout.connect(self.tick_waiting)
self.blink_timer.start(250)
def set_online(self, online):
self.online = online
self.read_dot.set_steady(online)
now = time.monotonic()
self.error_dot.set_steady(now < self.error_until)
if online:
self.wait_until = 0.0
self.wait_phase = False
self.write_dot.set_steady(False)
self.detail.setText("켜짐")
elif now < self.wait_until:
remaining = int(self.wait_until - now)
self.detail.setText(f"대기 {remaining}s")
else:
self.write_dot.set_steady(False)
self.detail.setText("꺼짐")
def tick_waiting(self):
now = time.monotonic()
if not self.online and now < self.wait_until:
self.wait_phase = not self.wait_phase
self.write_dot.set_steady(self.wait_phase)
if not self.online and self.wait_until > 0 and now >= self.wait_until:
self.wait_until = 0.0
self.wait_phase = False
self.write_dot.set_steady(False)
self.error_until = now + 60
self.error_dot.set_steady(True)
self.detail.setText("실패")
if self.error_until > 0 and now >= self.error_until:
self.error_until = 0.0
self.error_dot.set_steady(False)
def mousePressEvent(self, event):
if event.button() == Qt.LeftButton:
now = time.monotonic()
if not self.online and now >= self.wait_until:
self.last_wake_at = now
self.wait_until = now + 60
self.wait_phase = True
self.write_dot.set_steady(True)
self.error_until = 0.0
self.error_dot.set_steady(False)
self.detail.setText("깨우는 중")
if not send_magic_packet(PC_MAC):
self.wait_until = 0.0
self.error_until = now + 5
self.error_dot.set_steady(True)
super().mousePressEvent(event)
def tick(self):
self.error_dot.tick()
class MetricCard(QFrame):
def __init__(self, title):
super().__init__()
self.setObjectName("metricCard")
self.title = QLabel(title)
self.title.setObjectName("metricTitle")
self.value = QLabel("-")
self.value.setObjectName("metricValue")
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 8, 10, 8)
layout.setSpacing(2)
layout.addWidget(self.title)
layout.addWidget(self.value)
def set_value(self, text):
self.value.setText(text)
class Sparkline(QFrame):
def __init__(self, title, color="#38bdf8", second_color=None, y_unit="", right_unit="", auto_byte_unit=False):
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.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)[-GRAPH_POINTS:]
self.second_values = list(second_values or [])[-GRAPH_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]
low = min(values)
high = max(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:
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.metrics = {}
self.charts = {}
self.fan = fan_paths()
self.wan_iface = default_iface()
NET_GROUPS["wan"] = [self.wan_iface]
self.last_net = {}
self.last_synology = {}
self.synology_buffer = ""
self.synology_values = {}
self.synology_disk_temps = {}
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.disk_temp = None
self.current_pwm = read_int(self.fan["pwm"])
self.latest_temp = 0.0
self.latest_rpm = 0
self.latest_pwm = self.current_pwm
self.latest_percent = 0.0
self.temp_history = deque(maxlen=GRAPH_POINTS)
self.rpm_history = deque(maxlen=GRAPH_POINTS)
self.net_history = {"wan": {"rx": deque(maxlen=GRAPH_POINTS), "tx": deque(maxlen=GRAPH_POINTS)}}
self.setWindowTitle("Standalone Touch Dashboard")
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(SAMPLE_INTERVAL_MS)
self.pc_ping = QProcess(self)
self.pc_ping.finished.connect(self.pc_ping_finished)
self.pc_timer = QTimer(self)
self.pc_timer.timeout.connect(self.ping_pc)
self.pc_timer.start(1000)
self.disk_temp_proc = QProcess(self)
self.disk_temp_proc.finished.connect(self.disk_temp_finished)
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.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_disk_temp_proc = QProcess(self)
self.synology_disk_temp_proc.finished.connect(self.synology_disk_temp_finished)
self.synology_disk_temp_timer = QTimer(self)
self.synology_disk_temp_timer.timeout.connect(self.update_synology_disk_temp)
self.synology_disk_temp_timer.start(SYNOLOGY_DISK_TEMP_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.pulse_timer = QTimer(self)
self.pulse_timer.timeout.connect(self.tick_pulses)
self.pulse_timer.start(25)
self.sample()
self.ping_pc()
self.update_disk_temp()
self.update_synology_status()
self.update_synology_disk_temp()
self.update_synology_health()
self.update_dashboard()
def _build_ui(self):
root = QWidget()
layout = QHBoxLayout(root)
layout.setContentsMargins(4, 8, 10, 8)
layout.setSpacing(8)
left = QFrame()
left.setObjectName("statusPanel")
left.setFixedWidth(398)
left_layout = QVBoxLayout(left)
left_layout.setContentsMargins(6, 8, 6, 8)
left_layout.setSpacing(6)
title = QLabel("상태")
title.setObjectName("sectionTitle")
left_layout.addWidget(title)
status_grid = QGridLayout()
status_grid.setContentsMargins(0, 0, 0, 0)
status_grid.setHorizontalSpacing(6)
status_grid.setVerticalSpacing(6)
pi_title = QLabel("RASPBERRY")
pi_title.setObjectName("statusGroupTitle")
synology_title = QLabel("SYNOLOGY")
synology_title.setObjectName("statusGroupTitle")
extra_title = QLabel("CHECK")
extra_title.setObjectName("statusGroupTitle")
status_grid.addWidget(pi_title, 0, 0)
status_grid.addWidget(synology_title, 0, 1)
status_grid.addWidget(extra_title, 0, 2)
status_rows = [
(("io", "SSD I/O"), ("synology_sata1", "SATA1 I/O"), ("rpi_temp", "RPI TEMP")),
(("local", "LOCAL"), ("synology_sata2", "SATA2 I/O"), ("rpi_fan", "RPI FAN")),
(("wan", "UPLINK"), ("synology_sata4", "SATA4 I/O"), ("synology_temp", "SYNO TEMP")),
(("lan", "ETH1"), ("synology_local", "LOCAL"), ("synology_fan", "SYNO FAN")),
(("wifi5", "5G"), ("synology_wan", "UPLINK"), None),
(("wifi24", "2.4G"), None, None),
]
for index, (pi_row, synology_row, extra_row) in enumerate(status_rows, 1):
key, label = pi_row
tile = StatusTile(label)
self.rows[key] = tile
status_grid.addWidget(tile, index, 0)
self.add_status_grid_cell(status_grid, index, 1, synology_row)
self.add_status_grid_cell(status_grid, index, 2, extra_row)
left_layout.addLayout(status_grid)
left_layout.addStretch(1)
self.pc_button = PcWakeButton()
left_layout.addWidget(self.pc_button)
right = QWidget()
right_layout = QVBoxLayout(right)
right_layout.setContentsMargins(0, 0, 0, 0)
right_layout.setSpacing(8)
header = QHBoxLayout()
heading = QLabel("Dashboard")
heading.setObjectName("sectionTitle")
self.clock = QLabel("--:--:--")
self.clock.setObjectName("clock")
self.status_text = QLabel("")
self.status_text.setObjectName("statusText")
header.addWidget(heading)
header.addStretch(1)
header.addWidget(self.status_text)
header.addWidget(self.clock)
right_layout.addLayout(header)
metric_grid = QGridLayout()
metric_grid.setHorizontalSpacing(8)
metric_grid.setVerticalSpacing(8)
for index, (key, title) in enumerate([("rpm", "팬 RPM"), ("temp", "CPU 온도"), ("pwm", "PWM"), ("mem", "메모리")]):
card = MetricCard(title)
self.metrics[key] = card
metric_grid.addWidget(card, 0, index)
right_layout.addLayout(metric_grid)
slider_wrap = QHBoxLayout()
self.slider = QSlider(Qt.Horizontal)
self.slider.setRange(0, 255)
self.slider.setValue(self.current_pwm)
self.slider.setFocusPolicy(Qt.NoFocus)
self.slider.setAttribute(Qt.WA_TransparentForMouseEvents, True)
slider_wrap.addWidget(self.slider, 1)
right_layout.addLayout(slider_wrap)
chart_grid = QGridLayout()
chart_grid.setHorizontalSpacing(8)
chart_grid.setVerticalSpacing(8)
for index, (key, chart) in enumerate([
("temp", Sparkline("온도 / RPM", "#f97316", "#38bdf8", "C", "rpm")),
("wan", Sparkline("WAN 트래픽", "#38bdf8", "#22c55e", auto_byte_unit=True)),
]):
self.charts[key] = chart
chart_grid.addWidget(chart, index, 0)
right_layout.addLayout(chart_grid, 1)
layout.addWidget(left)
layout.addWidget(right, 1)
self.setCentralWidget(root)
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
temp = cpu_temp()
rpm = read_int(self.fan["rpm"])
pwm = read_int(self.fan["pwm"])
enable = read_int(self.fan["enable"], 2)
percent = pwm / 255 * 100 if pwm > 0 else 0
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.render_status(temp, rpm, pwm, percent, enable, groups, rates, activity, io_read_activity, io_write_activity, disk_error)
def render_status(self, temp, rpm, pwm, percent, enable, 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 "- C"
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"])
self.render_rpi_temp(temp)
self.render_rpi_fan(rpm)
self.latest_temp = temp
self.latest_rpm = rpm
self.latest_pwm = pwm
self.latest_percent = percent
del enable
def update_dashboard(self):
rpm = getattr(self, "latest_rpm", read_int(self.fan["rpm"]))
temp = getattr(self, "latest_temp", cpu_temp())
pwm = getattr(self, "latest_pwm", read_int(self.fan["pwm"]))
percent = getattr(self, "latest_percent", pwm / 255 * 100 if pwm > 0 else 0)
mem_percent = memory_percent()
self.metrics["rpm"].set_value(f"{rpm:,}")
self.metrics["temp"].set_value(f"{temp:.1f} C")
self.metrics["pwm"].set_value(f"{percent:.1f}%")
self.metrics["mem"].set_value(f"{mem_percent:.1f}%")
self.slider.blockSignals(True)
self.slider.setValue(pwm)
self.slider.blockSignals(False)
self.clock.setText(time.strftime("%H:%M:%S"))
self.update_graphs()
def ping_pc(self):
if self.pc_ping.state() != QProcess.NotRunning:
return
self.pc_ping.start("/usr/bin/ping", ["-c", "1", "-W", "1", PC_IP])
def pc_ping_finished(self, exit_code, exit_status):
del exit_status
self.pc_button.set_online(exit_code == 0)
def update_disk_temp(self):
if self.disk_temp_proc.state() != QProcess.NotRunning:
return
self.disk_temp_proc.start("/usr/bin/sudo", ["-n", "/usr/sbin/smartctl", "-A", "-d", "sntasmedia", "/dev/sda"])
def disk_temp_finished(self, exit_code, exit_status):
del exit_code, exit_status
output = bytes(self.disk_temp_proc.readAllStandardOutput()).decode("utf-8", "ignore")
for line in output.splitlines():
if line.strip().startswith("Temperature:"):
parts = line.split()
for part in parts:
if part.isdigit():
self.disk_temp = float(part)
self.render_rpi_temp(getattr(self, "latest_temp", cpu_temp()))
return
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_disk_temp(self):
if self.synology_disk_temp_proc.state() != QProcess.NotRunning:
return
self.synology_disk_temp_proc.start(SYNOLOGY_SSH[0], SYNOLOGY_SSH[1:] + [SYNOLOGY_DISK_TEMP_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_disk_temp_finished(self, exit_code, exit_status):
del exit_status
if exit_code != 0:
return
output = bytes(self.synology_disk_temp_proc.readAllStandardOutput()).decode("utf-8", "ignore").strip()
temps = {}
for line in output.splitlines():
if "=" not in line:
continue
key, value = line.split("=", 1)
temps[key.strip()] = value.strip()
self.synology_disk_temps = temps
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()
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"]:
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",
)
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_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):
value = self.synology_disk_temps.get(f"{disk}_TEMP", "")
if not value:
return "- C"
return f"{value} C"
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(detail, error_activity, read_activity, write_activity)
def update_graphs(self):
temp = getattr(self, "latest_temp", cpu_temp())
rpm = getattr(self, "latest_rpm", read_int(self.fan["rpm"]))
self.temp_history.append(temp)
self.rpm_history.append(rpm)
self.charts["temp"].set_values(self.temp_history, format_temp_rpm_caption(temp, rpm), self.rpm_history)
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["wan"].set_values(
self.net_history["wan"]["rx"],
format_wan_caption(wan["rx"], wan["tx"]),
self.net_history["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 normal_label if drops <= 0 else f"드롭 {drops}"
@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; }
#metricCard, #chartCard { background: #111a2e; border: 1px solid #27364f; border-radius: 12px; }
#metricTitle { color: #8fa7c5; font-size: 13px; background: transparent; }
#metricValue { color: #f8fafc; font-size: 24px; font-weight: 800; background: transparent; }
#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; }
QSlider::groove:horizontal { height: 10px; background: #263650; border-radius: 5px; }
QSlider::sub-page:horizontal { background: #22c55e; border-radius: 5px; }
QSlider::handle:horizontal { width: 28px; margin: -10px 0; border-radius: 14px; background: #f8fafc; }
"""
)
win = TouchPanel()
win.showFullScreen()
sys.exit(app.exec_())
if __name__ == "__main__":
main()