터치 패널 단독 실시간 대시보드 전환

This commit is contained in:
seo committed 2026-07-30 04:57:25 +09:00
1 parent ed76621201
commit 5061a03135
2 files changed
+274 -300

No files matched your search

Executable → Regular
+265 -293
View File
@@ -1,10 +1,10 @@
#!/usr/bin/env python3
import json
import subprocess
import sys
import time
from collections import deque
from pathlib import Path
from PyQt5.QtCore import QObject, QPointF, Qt, QThread, QTimer, pyqtSignal
from PyQt5.QtCore import QPointF, Qt, QTimer
from PyQt5.QtGui import QColor, QPainter, QPainterPath, QPen
from PyQt5.QtWidgets import (
QApplication,
@@ -20,51 +20,139 @@ from PyQt5.QtWidgets import (
)
ROOT = Path(__file__).resolve().parents[1]
BRIDGE = ROOT / "touch_panel" / "control_touch_bridge.php"
PHP = "/usr/bin/php"
NET_GROUPS = {
"wan": ["eth0"],
"lan": ["br0", "eth1"],
"wifi24": ["wlan0"],
"wifi5": ["wlan1"],
}
def run_bridge(*args):
proc = subprocess.run(
[PHP, str(BRIDGE), *map(str, args)],
cwd=str(ROOT),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=8,
check=False,
)
if proc.returncode != 0:
raise RuntimeError((proc.stderr or proc.stdout or "bridge failed").strip())
payload = json.loads(proc.stdout)
if not payload.get("ok"):
raise RuntimeError(payload.get("message") or payload.get("error") or "bridge error")
return payload["data"]
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 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
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
reads += int(parts[3])
writes += int(parts[7])
return reads, writes
def write_sys(path, value):
if not path:
return False
try:
Path(path).write_text(str(value) + "\n", encoding="utf-8")
return True
except OSError:
return False
class StatusDot(QWidget):
def __init__(self):
super().__init__()
self.active = False
self.warning = False
self.pulse = 0
self.phase = False
self.setFixedSize(22, 22)
def set_warning(self, warning=False):
self.warning = warning
self.update()
def start_pulse(self):
self.active = True
self.pulse = 4
def blink(self):
self.pulse = 3
self.phase = not self.phase
self.update()
def tick(self):
if self.pulse > 0:
self.pulse -= 1
if self.pulse <= 0:
self.active = False
self.update()
def paintEvent(self, event):
@@ -74,8 +162,8 @@ class StatusDot(QWidget):
base = QColor("#8fa0aa")
if self.warning:
base = QColor("#f59e0b")
if self.pulse > 0 and self.active:
base = QColor("#86efac") if self.pulse % 2 == 0 else QColor("#16a34a")
if self.pulse > 0:
base = QColor("#86efac") if self.phase else QColor("#16a34a")
painter.setBrush(base)
painter.setPen(Qt.NoPen)
painter.drawEllipse(3, 3, 16, 16)
@@ -84,16 +172,13 @@ class StatusDot(QWidget):
class StatusTile(QFrame):
def __init__(self, title, pulse_delay_ms=0):
def __init__(self, title):
super().__init__()
self.pulse_delay_ms = pulse_delay_ms
self.dot = StatusDot()
self.title = QLabel(title)
self.title.setObjectName("statusTitle")
self.detail = QLabel("-")
self.detail.setObjectName("statusDetail")
self.detail.setWordWrap(True)
self.detail.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
self.setFixedHeight(58)
@@ -102,47 +187,24 @@ class StatusTile(QFrame):
layout.setSpacing(6)
layout.addWidget(self.dot, 0, Qt.AlignLeft | Qt.AlignVCenter)
text_layout = QVBoxLayout()
text_layout.setContentsMargins(0, 0, 0, 0)
text_layout.setSpacing(1)
text_layout.addWidget(self.title)
text_layout.addWidget(self.detail)
layout.addLayout(text_layout, 1)
texts = QVBoxLayout()
texts.setContentsMargins(0, 0, 0, 0)
texts.setSpacing(1)
texts.addWidget(self.title)
texts.addWidget(self.detail)
layout.addLayout(texts, 1)
self.setObjectName("statusTile")
def update_state(self, active, detail, warning=False):
def set_state(self, detail, warning=False, activity=False):
self.dot.set_warning(warning)
if active:
QTimer.singleShot(self.pulse_delay_ms, self.dot.start_pulse)
if activity:
self.dot.blink()
self.detail.setText(detail)
def tick(self):
self.dot.tick()
class FanButton(QPushButton):
def __init__(self, text):
super().__init__(text)
self.setCheckable(True)
self.setMinimumHeight(44)
class BridgeWorker(QObject):
finished = pyqtSignal(str, object)
failed = pyqtSignal(str, str)
def __init__(self, action, args):
super().__init__()
self.action = action
self.args = args
def run(self):
try:
self.finished.emit(self.action, run_bridge(*self.args))
except Exception as exc:
self.failed.emit(self.action, str(exc))
class MetricCard(QFrame):
def __init__(self, title):
super().__init__()
@@ -151,7 +213,6 @@ class MetricCard(QFrame):
self.title.setObjectName("metricTitle")
self.value = QLabel("-")
self.value.setObjectName("metricValue")
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 8, 10, 8)
layout.setSpacing(2)
@@ -189,8 +250,8 @@ class Sparkline(QFrame):
layout.addWidget(self.canvas, 1)
def set_values(self, values, caption, second_values=None):
self.values = [float(v or 0) for v in values][-80:]
self.second_values = [float(v or 0) for v in (second_values or [])][-80:]
self.values = list(values)[-80:]
self.second_values = list(second_values or [])[-80:]
self.caption.setText(caption)
self.canvas.update()
@@ -202,7 +263,6 @@ class Sparkline(QFrame):
painter.fillRect(rect, QColor("#0f172a"))
painter.setPen(QPen(QColor("#263650"), 1))
painter.drawLine(rect.left(), rect.bottom(), rect.right(), rect.bottom())
all_values = self.values + self.second_values
if len(all_values) < 2:
return
@@ -210,57 +270,53 @@ class Sparkline(QFrame):
high = max(all_values)
if high <= low:
high = low + 1
self.draw_series(painter, rect, self.values, low, high, self.color, 2)
self.draw_series(painter, rect, self.values, low, high, self.color)
if self.second_color and self.second_values:
self.draw_series(painter, rect, self.second_values, low, high, self.second_color, 2)
self.draw_series(painter, rect, self.second_values, low, high, self.second_color)
@staticmethod
def draw_series(painter, rect, values, low, high, color, width):
if len(values) < 2:
return
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))
ratio = (float(value) - low) / (high - low)
y = rect.bottom() - (rect.height() * ratio)
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)
if index == 0:
path.moveTo(point)
else:
path.lineTo(point)
painter.setPen(QPen(color, width))
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.snapshot = {}
self.rows = {}
self.mode_buttons = {}
self.metrics = {}
self.charts = {}
self.last_counters = {}
self.bridge_busy = False
self.worker_thread = None
self.worker = None
self.current_mode = "auto"
self.current_pwm = 120
self.mode_buttons = {}
self.fan = fan_paths()
self.wan_iface = default_iface()
NET_GROUPS["wan"] = [self.wan_iface]
self.last_net = {}
self.last_io = None
self.last_sample = None
self.current_pwm = read_int(self.fan["pwm"])
self.temp_history = deque(maxlen=80)
self.rpm_history = deque(maxlen=80)
self.net_history = {key: {"rx": deque(maxlen=80), "tx": deque(maxlen=80)} for key in ["wan", "lan", "wifi"]}
self.setWindowTitle("Control Touch Panel")
self.setWindowTitle("Standalone Touch Dashboard")
self.setCursor(Qt.BlankCursor)
self.setMinimumSize(800, 480)
self._build_ui()
self.refresh_timer = QTimer(self)
self.refresh_timer.timeout.connect(self.refresh)
self.refresh_timer.start(1500)
self.sample_timer = QTimer(self)
self.sample_timer.timeout.connect(self.sample)
self.sample_timer.start(120)
self.pulse_timer = QTimer(self)
self.pulse_timer.timeout.connect(self.tick_pulses)
self.pulse_timer.start(180)
self.refresh()
self.pulse_timer.start(70)
self.sample()
def _build_ui(self):
root = QWidget()
@@ -274,21 +330,14 @@ class TouchPanel(QMainWindow):
left_layout = QVBoxLayout(left)
left_layout.setContentsMargins(6, 8, 6, 8)
left_layout.setSpacing(6)
title = QLabel("상태")
title.setObjectName("sectionTitle")
left_layout.addWidget(title)
for index, (key, label) in enumerate([
("io", "IO"),
("wan", "WAN"),
("lan", "LAN"),
("wifi24", "2.4G"),
("wifi5", "5G"),
]):
row = StatusTile(label, index * 70)
self.rows[key] = row
left_layout.addWidget(row)
for key, label in [("io", "IO"), ("wan", "WAN"), ("lan", "LAN"), ("wifi24", "2.4G"), ("wifi5", "5G")]:
tile = StatusTile(label)
self.rows[key] = tile
left_layout.addWidget(tile)
left_layout.addStretch(1)
right = QWidget()
@@ -297,13 +346,13 @@ class TouchPanel(QMainWindow):
right_layout.setSpacing(8)
header = QHBoxLayout()
fan_title = QLabel("Control")
fan_title.setObjectName("sectionTitle")
heading = QLabel("Dashboard")
heading.setObjectName("sectionTitle")
self.clock = QLabel("--:--:--")
self.clock.setObjectName("clock")
self.status_text = QLabel("")
self.status_text.setObjectName("statusText")
header.addWidget(fan_title)
header.addWidget(heading)
header.addStretch(1)
header.addWidget(self.status_text)
header.addWidget(self.clock)
@@ -312,21 +361,17 @@ class TouchPanel(QMainWindow):
metric_grid = QGridLayout()
metric_grid.setHorizontalSpacing(8)
metric_grid.setVerticalSpacing(8)
for index, (key, title) in enumerate([
("rpm", "팬 RPM"),
("temp", "판단 온도"),
("pwm", "PWM"),
("wan", "WAN"),
]):
for index, (key, title) in enumerate([("rpm", "팬 RPM"), ("temp", "CPU 온도"), ("pwm", "PWM"), ("wan", "WAN")]):
card = MetricCard(title)
self.metrics[key] = card
metric_grid.addWidget(card, index // 4, index % 4)
metric_grid.addWidget(card, 0, index)
right_layout.addLayout(metric_grid)
mode_layout = QHBoxLayout()
for mode in ["auto", "manual", "off"]:
btn = FanButton(mode)
btn.clicked.connect(lambda checked=False, m=mode: self.set_fan(m, self.current_pwm))
for mode, label in [("auto", "auto"), ("manual", "manual"), ("off", "off")]:
btn = QPushButton(label)
btn.setCheckable(True)
btn.clicked.connect(lambda checked=False, m=mode: self.set_fan(m))
self.mode_buttons[mode] = btn
mode_layout.addWidget(btn)
right_layout.addLayout(mode_layout)
@@ -340,9 +385,9 @@ class TouchPanel(QMainWindow):
plus.clicked.connect(lambda: self.step_pwm(10))
self.slider = QSlider(Qt.Horizontal)
self.slider.setRange(0, 255)
self.slider.setValue(120)
self.slider.sliderReleased.connect(lambda: self.set_fan("manual", self.slider.value()))
self.slider.setValue(self.current_pwm)
self.slider.valueChanged.connect(self.slider_changed)
self.slider.sliderReleased.connect(lambda: self.set_fan("manual"))
slider_wrap.addWidget(minus)
slider_wrap.addWidget(self.slider, 1)
slider_wrap.addWidget(plus)
@@ -361,186 +406,114 @@ class TouchPanel(QMainWindow):
chart_grid.addWidget(chart, index // 2, index % 2)
right_layout.addLayout(chart_grid, 1)
right_layout.addStretch(1)
layout.addWidget(left)
layout.addWidget(right, 1)
self.setCentralWidget(root)
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_io()
io_activity = self.last_io is not None and (reads, writes) != self.last_io
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"])
activity[key] = previous is not None and current != previous
if previous:
rates[key] = {
"rx": max(0.0, (current[0] - previous[0]) * 8 / dt / 1_000_000),
"tx": max(0.0, (current[1] - previous[1]) * 8 / dt / 1_000_000),
}
else:
rates[key] = {"rx": 0.0, "tx": 0.0}
self.last_net[key] = current
self.render(temp, rpm, pwm, percent, enable, groups, rates, activity, io_activity)
def render(self, temp, rpm, pwm, percent, enable, groups, rates, activity, io_activity):
self.rows["io"].set_state(f"{temp:.1f} C", temp >= 75, io_activity)
self.rows["wan"].set_state(self.iface_detail(groups["wan"]), self.iface_warn(groups["wan"]), activity["wan"])
self.rows["lan"].set_state(self.iface_detail(groups["lan"]), self.iface_warn(groups["lan"]), activity["lan"])
self.rows["wifi24"].set_state(self.iface_detail(groups["wifi24"]), self.iface_warn(groups["wifi24"]), activity["wifi24"])
self.rows["wifi5"].set_state(self.iface_detail(groups["wifi5"]), self.iface_warn(groups["wifi5"]), activity["wifi5"])
self.metrics["rpm"].set_value(f"{rpm:,}")
self.metrics["temp"].set_value(f"{temp:.1f} C")
self.metrics["pwm"].set_value(f"{pwm} / {percent:.1f}%")
self.metrics["wan"].set_value(f"{rates['wan']['rx']:.2f}/{rates['wan']['tx']:.2f} Mbps")
self.slider.blockSignals(True)
self.slider.setValue(pwm)
self.slider.blockSignals(False)
self.mode_buttons["off"].setChecked(enable == 0 or pwm == 0)
self.mode_buttons["manual"].setChecked(enable == 1 and pwm > 0)
self.mode_buttons["auto"].setChecked(enable not in [0, 1])
self.clock.setText(time.strftime("%H:%M:%S"))
self.temp_history.append(temp)
self.rpm_history.append(rpm / 100)
self.charts["temp"].set_values(self.temp_history, "C / RPMx100", self.rpm_history)
for key, chart_key in [("wan", "wan"), ("lan", "lan")]:
self.net_history[key]["rx"].append(rates[key]["rx"])
self.net_history[key]["tx"].append(rates[key]["tx"])
self.charts[chart_key].set_values(
self.net_history[key]["rx"],
f"{rates[key]['rx']:.2f}/{rates[key]['tx']:.2f} Mbps",
self.net_history[key]["tx"],
)
wifi_rx = rates["wifi24"]["rx"] + rates["wifi5"]["rx"]
wifi_tx = rates["wifi24"]["tx"] + rates["wifi5"]["tx"]
self.net_history["wifi"]["rx"].append(wifi_rx)
self.net_history["wifi"]["tx"].append(wifi_tx)
self.charts["wifi"].set_values(self.net_history["wifi"]["rx"], f"{wifi_rx:.2f}/{wifi_tx:.2f} Mbps", self.net_history["wifi"]["tx"])
def set_fan(self, mode):
pwm = self.slider.value()
ok = True
if mode == "off":
ok = write_sys(self.fan["enable"], 0) and write_sys(self.fan["pwm"], 0)
elif mode == "manual":
ok = write_sys(self.fan["enable"], 1) and write_sys(self.fan["pwm"], pwm)
elif mode == "auto":
ok = write_sys(self.fan["enable"], 2)
self.status_text.setText("" if ok else "팬 쓰기 권한 없음")
def step_pwm(self, delta):
self.slider.setValue(max(0, min(255, self.slider.value() + delta)))
self.set_fan("manual")
def slider_changed(self, value):
self.current_pwm = value
self.metrics["pwm"].set_value(f"{value} / {value / 255 * 100:.1f}%")
def tick_pulses(self):
for row in self.rows.values():
row.tick()
def slider_changed(self, value):
self.current_pwm = value
if "pwm" in self.metrics:
self.metrics["pwm"].set_value(f"{value} / {value / 255 * 100:.1f}%")
def set_fan(self, mode, pwm):
if self.start_bridge("fan", "fan", mode, pwm):
self.status_text.setText("팬 변경 중")
def step_pwm(self, delta):
value = max(0, min(255, self.slider.value() + delta))
self.slider.setValue(value)
self.set_fan("manual", value)
def refresh(self):
self.start_bridge("status", "status")
def start_bridge(self, action, *args):
if self.bridge_busy:
return False
self.bridge_busy = True
self.worker_thread = QThread(self)
self.worker = BridgeWorker(action, args)
self.worker.moveToThread(self.worker_thread)
self.worker_thread.started.connect(self.worker.run)
self.worker.finished.connect(self.bridge_done)
self.worker.failed.connect(self.bridge_failed)
self.worker.finished.connect(self.worker_thread.quit)
self.worker.failed.connect(self.worker_thread.quit)
self.worker.finished.connect(self.worker.deleteLater)
self.worker.failed.connect(self.worker.deleteLater)
self.worker_thread.finished.connect(self.worker_thread.deleteLater)
self.worker_thread.finished.connect(self.bridge_finished)
self.worker_thread.start()
return True
def bridge_done(self, action, data):
self.snapshot = data
self.status_text.setText("팬 변경 완료" if action == "fan" else "")
self.render()
def bridge_failed(self, action, message):
del action
self.status_text.setText("통신 오류")
self.rows["io"].update_state(False, "snapshot 실패", True)
print(message, file=sys.stderr)
def bridge_finished(self):
self.bridge_busy = False
self.worker_thread = None
self.worker = None
def render(self):
data = self.snapshot
fan = data.get("fan", {})
system = data.get("system", {})
network = data.get("network", {})
wifi = data.get("wifi", {})
touch = data.get("touch_display", {})
io = data.get("io", {})
history = data.get("history", [])
network_history = data.get("network_history", [])
interfaces = {row.get("name"): row for row in network.get("interfaces", [])}
clients = wifi.get("clients", [])
wifi24 = [c for c in clients if c.get("band") == "2.4G"]
wifi5 = [c for c in clients if c.get("band") == "5G"]
lan_clients = [c for c in clients if c.get("band") == "LAN"]
temp = float(system.get("temp_c") or 0)
rpm = int(fan.get("rpm") or 0)
pwm = int(fan.get("pwm") or 0)
mode = str(fan.get("mode") or "auto")
wan = interfaces.get(network.get("default_interface") or "eth0", {})
lan = interfaces.get("br0") or interfaces.get("eth1") or {}
w24 = interfaces.get("wlan0") or {}
w5 = interfaces.get("wlan1") or {}
io_active = bool(touch.get("connected")) and bool(fan.get("policy_ok", True)) and self.counter_activity(
"io",
int(io.get("reads") or 0),
int(io.get("writes") or 0),
)
io_detail = f"{temp:.1f} C"
self.rows["io"].update_state(io_active, io_detail, temp >= 75)
self.rows["wan"].update_state(self.iface_activity("wan", wan), self.iface_detail(wan), self.iface_warn(wan))
self.rows["lan"].update_state(self.iface_activity("lan", lan), f"{len(lan_clients)}대 {self.iface_detail(lan)}", self.iface_warn(lan))
self.rows["wifi24"].update_state(self.iface_activity("wifi24", w24), f"{len(wifi24)}대 {self.iface_detail(w24)}", self.iface_warn(w24))
self.rows["wifi5"].update_state(self.iface_activity("wifi5", w5), f"{len(wifi5)}대 {self.iface_detail(w5)}", self.iface_warn(w5))
self.current_mode = mode
self.current_pwm = pwm
control_temp = float(fan.get("control_temp_c") or temp)
self.metrics["rpm"].set_value(f"{rpm:,}")
self.metrics["temp"].set_value(f"{control_temp:.1f} C")
self.metrics["pwm"].set_value(f"{pwm} / {float(fan.get('percent') or 0):.1f}%")
self.metrics["wan"].set_value(self.traffic_caption(wan))
self.slider.blockSignals(True)
self.slider.setValue(pwm)
self.slider.blockSignals(False)
for key, btn in self.mode_buttons.items():
btn.setChecked(key == mode)
self.clock.setText(str(data.get("generated_at", "--:--:--")).split(" ")[-1])
self.update_charts(history, network_history, network.get("default_interface") or "eth0")
def iface_activity(self, key, row):
if not self.iface_up(row):
self.last_counters[key] = None
return False
return self.counter_activity(
key,
int(row.get("rx_bytes") or 0),
int(row.get("tx_bytes") or 0),
int(row.get("rx_packets") or 0),
int(row.get("tx_packets") or 0),
)
def counter_activity(self, key, *values):
current = tuple(values)
previous = self.last_counters.get(key)
self.last_counters[key] = current
return previous is not None and current != previous
def update_charts(self, history, network_history, wan_iface):
temps = [row.get("temp_c") for row in history if isinstance(row, dict)]
rpms = [float(row.get("fan_rpm") or 0) / 100 for row in history if isinstance(row, dict)]
self.charts["temp"].set_values(temps, "C / RPMx100", rpms)
wan_rows = self.iface_history(network_history, [wan_iface])
lan_rows = self.iface_history(network_history, ["br0", "eth1"])
wifi_rows = self.iface_history(network_history, ["wlan0", "wlan1"])
self.set_traffic_chart("wan", wan_rows)
self.set_traffic_chart("lan", lan_rows)
self.set_traffic_chart("wifi", wifi_rows)
def set_traffic_chart(self, key, rows):
rx = [float(row.get("rx_mbps") or 0) for row in rows]
tx = [float(row.get("tx_mbps") or 0) for row in rows]
latest_rx = rx[-1] if rx else 0
latest_tx = tx[-1] if tx else 0
self.charts[key].set_values(rx, f"{latest_rx:.2f}/{latest_tx:.2f} Mbps", tx)
@staticmethod
def iface_history(rows, names):
names = set(names)
return [row for row in rows if isinstance(row, dict) and row.get("iface") in names]
@staticmethod
def iface_up(row):
return row.get("state") == "up" and row.get("carrier") == "up"
def iface_detail(row):
if not row["carrier"]:
return "down"
drops = row["rx_dropped"] + row["tx_dropped"]
return f"up D{drops}"
@staticmethod
def iface_warn(row):
return int(row.get("rx_errors") or 0) > 0 or int(row.get("tx_errors") or 0) > 0
@staticmethod
def iface_detail(row):
if not row:
return "없음"
rx_drop = int(row.get("rx_dropped") or 0)
tx_drop = int(row.get("tx_dropped") or 0)
return f"{row.get('state', '-')} D{rx_drop + tx_drop}"
@staticmethod
def traffic_caption(row):
if not row:
return "-"
rx = int(row.get("rx_bytes") or 0)
tx = int(row.get("tx_bytes") or 0)
return f"{rx / 1024 / 1024:.0f}/{tx / 1024 / 1024:.0f}MB"
return row["rx_errors"] > 0 or row["tx_errors"] > 0
def main():
@@ -549,7 +522,6 @@ def main():
app.setStyleSheet(
"""
QWidget { background: #0b1220; color: #e5edf7; font-family: "Noto Sans CJK KR", "Noto Sans CJK KR Regular", "Noto Sans KR", sans-serif; }
#panel { background: #111a2e; border: 1px solid #27364f; border-radius: 16px; }
#statusPanel { background: #101827; border-right: 1px solid #263650; }
#sectionTitle { font-size: 20px; font-weight: 700; }
#statusTile { background: #d7d9dc; border: 2px groove #f4f4f4; border-radius: 2px; }