#!/usr/bin/env python3 import os import shutil 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, QVBoxLayout, QWidget, ) NET_GROUPS = { "local": ["lo"], "wan": ["eth0"], "lan": ["eth1"], "wifi24": ["wlan0"], "wifi5": ["wlan1"], } PC_IP = "192.168.50.34" PC_MAC = "00:00:00:00:00:d7" PC_DHCP_LEASE_FILE = "/var/lib/misc/dnsmasq.leases" SYNOLOGY_PING_HOST = "chaegeon.com" SAMPLE_INTERVAL_MS = 100 FAN_INTERVAL_MS = 500 WAN_GRAPH_SECONDS = 20 TEMP_GRAPH_SECONDS = 20 WAN_GRAPH_POINTS = WAN_GRAPH_SECONDS * 1000 // SAMPLE_INTERVAL_MS TEMP_GRAPH_POINTS = TEMP_GRAPH_SECONDS * 1000 // SAMPLE_INTERVAL_MS 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 SYSTEM_INTERVAL_MS = 1000 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 load_average(): try: return float(read_text("/proc/loadavg").split()[0]) except (IndexError, ValueError): return 0.0 def memory_used_percent(): 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]) total = values.get("MemTotal", 0) available = values.get("MemAvailable", 0) return None 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 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 def kernel_activity(): values = {} for line in read_text("/proc/stat").splitlines(): parts = line.split() if not parts: continue key = parts[0] if key == "cpu": ticks = [int(value) for value in parts[1:] if value.isdigit()] if len(ticks) >= 7: values["CPU_TOTAL"] = sum(ticks) values["CPU_IDLE"] = ticks[3] + (ticks[4] if len(ticks) > 4 else 0) values["CPU_USER"] = ticks[0] values["CPU_NICE"] = ticks[1] values["CPU_SYSTEM"] = ticks[2] values["CPU_IOWAIT"] = ticks[4] if len(ticks) > 4 else 0 values["CPU_IRQ"] = ticks[5] if len(ticks) > 5 else 0 values["CPU_SOFTIRQ"] = ticks[6] if len(ticks) > 6 else 0 elif key in {"intr", "ctxt", "processes", "procs_running", "procs_blocked", "softirq"} and len(parts) > 1: values[key.upper()] = int(parts[1]) if parts[1].isdigit() else 0 snmp_headers = {} for line in read_text("/proc/net/snmp").splitlines(): parts = line.split() if not parts: continue family = parts[0].rstrip(":") if family not in {"Tcp", "Udp"}: continue if family not in snmp_headers: snmp_headers[family] = parts[1:] continue row = dict(zip(snmp_headers[family], parts[1:])) if family == "Tcp": values["TCP_IN"] = int(row.get("InSegs", "0") or 0) values["TCP_OUT"] = int(row.get("OutSegs", "0") or 0) values["TCP_RETRANS"] = int(row.get("RetransSegs", "0") or 0) elif family == "Udp": values["UDP_IN"] = int(row.get("InDatagrams", "0") or 0) values["UDP_OUT"] = int(row.get("OutDatagrams", "0") or 0) values["UDP_ERRORS"] = ( int(row.get("InErrors", "0") or 0) + int(row.get("RcvbufErrors", "0") or 0) + int(row.get("SndbufErrors", "0") or 0) ) del snmp_headers[family] for line in read_text("/proc/vmstat").splitlines(): parts = line.split() if len(parts) >= 2 and parts[0] in {"pgfault", "pgmajfault", "pgpgin", "pgpgout"}: values[parts[0].upper()] = int(parts[1]) if parts[1].isdigit() else 0 return values def compact_count(value): value = int(max(0, value)) if value >= 1_000_000: return f"{value / 1_000_000:.1f}M" if value >= 1_000: return f"{value / 1_000:.1f}k" return str(value) def normalize_mac(mac): clean = mac.replace(":", "").replace("-", "") if len(clean) != 12: return "" try: bytes.fromhex(clean) except ValueError: return "" return ":".join(clean[i:i + 2] for i in range(0, 12, 2)).lower() def dhcp_ip_for_mac(mac): mac = normalize_mac(mac) if not mac: return "" for line in read_text(PC_DHCP_LEASE_FILE).splitlines(): parts = line.split() if len(parts) >= 3 and normalize_mac(parts[1]) == mac: ip = parts[2] octets = ip.split(".") if len(octets) == 4 and all(part.isdigit() and 0 <= int(part) <= 255 for part in octets): return ip return "" def broadcast_targets_for_ip(ip): parts = ip.split(".") targets = [] if len(parts) == 4 and all(part.isdigit() and 0 <= int(part) <= 255 for part in parts): targets.append(".".join(parts[:3] + ["255"])) targets.append("255.255.255.255") return list(dict.fromkeys(targets)) def send_magic_packet(mac, ip=PC_IP): mac = normalize_mac(mac) if not mac: return False clean = mac.replace(":", "") 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 broadcast_targets_for_ip(ip): 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.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 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("깨우는 중") pc_ip = dhcp_ip_for_mac(PC_MAC) or PC_IP if not send_magic_packet(PC_MAC, pc_ip): self.wait_until = 0.0 self.error_until = now + 5 self.error_dot.set_steady(True) self.write_dot.set_steady(False) self.detail.setText("실패") super().mousePressEvent(event) def tick(self): self.error_dot.tick() 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_rpi_activity = {} 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.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.net_history = { "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.fan_timer = QTimer(self) self.fan_timer.timeout.connect(self.update_fan_status) self.fan_timer.start(FAN_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_timer = QTimer(self) self.disk_temp_timer.timeout.connect(self.update_disk_temp) self.disk_temp_timer.start(DISK_TEMP_INTERVAL_MS) self.system_timer = QTimer(self) self.system_timer.timeout.connect(self.update_system_status) self.system_timer.start(SYSTEM_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.ping_pc() self.update_disk_temp() self.update_system_status() self.ping_synology() self.update_rpi_throttle() def _build_ui(self): root = QWidget() layout = QHBoxLayout(root) layout.setContentsMargins(4, 8, 10, 8) layout.setSpacing(8) status_rows = [ (("io", "SSD I/O"), ("synology_ping", "PING"), ("rpi_fan", "FAN")), (("rpi_throttle", "THROT"), ("rpi_cpu_activity", "CPU"), ("rpi_irq", "IRQ")), (("rpi_ctx", "CTX"), ("rpi_fork", "FORK"), ("local", "LOCAL")), (("wan", "UPLINK"), ("lan", "ETH1"), ("wifi5", "5G")), (("wifi24", "2.4G"), ("rpi_softirq", "SOFTIRQ"), ("rpi_tcp", "TCP")), (("rpi_udp", "UDP"), ("rpi_vm", "VM"), ("rpi_runq", "RUNQ")), (("rpi_load", "LOAD"), ("rpi_mem", "MEM"), ("rpi_root", "ROOT")), ] self.pc_button = PcWakeButton() self.pc_button.setFixedHeight(48) left = self.build_status_panel(None, status_rows, fixed_width=398, footer_widget=self.pc_button) right = QWidget() right_layout = QVBoxLayout(right) right_layout.setContentsMargins(0, 0, 0, 0) right_layout.setSpacing(8) for key, chart in [ ( "rpi_temp", Sparkline( "온도 / RPM", "#f97316", "#38bdf8", "C", "rpm", max_points=TEMP_GRAPH_POINTS, ), ), ( "rpi_wan", Sparkline( "WAN 트래픽", "#f97316", "#22c55e", auto_byte_unit=True, max_points=WAN_GRAPH_POINTS, ), ), ]: self.charts[key] = chart right_layout.addWidget(chart, 1) layout.addWidget(left) layout.addWidget(right, 1) self.setCentralWidget(root) def build_status_panel(self, title_text, rows, fixed_width=None, footer_widget=None): 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) if title_text: 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 footer_widget is not None: panel_layout.addWidget(footer_widget) 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_graphs() self.render_status(groups, rates, activity, io_read_activity, io_write_activity, disk_error) self.render_rpi_activity() 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 render_rpi_activity(self): self.render_activity_tiles("rpi", kernel_activity(), "last_rpi_activity") def render_activity_tiles(self, prefix, values, previous_attr): row_keys = [ "cpu_activity", "irq", "ctx", "fork", "softirq", "tcp", "udp", "vm", "runq", ] if not values: for suffix in row_keys: self.rows[f"{prefix}_{suffix}"].set_state("-", read_steady=False, write_steady=False, error_steady=True) return current = {key: self.int_value(value) for key, value in values.items()} previous = getattr(self, previous_attr) has_previous = bool(previous) def delta(name): if not has_previous: return 0 return max(0, current.get(name, 0) - previous.get(name, current.get(name, 0))) def changed(name): return has_previous and current.get(name, 0) != previous.get(name, 0) total_delta = delta("CPU_TOTAL") idle_delta = delta("CPU_IDLE") cpu_used = 0.0 if total_delta <= 0 else max(0.0, (total_delta - idle_delta) * 100 / total_delta) self.rows[f"{prefix}_cpu_activity"].set_state( f"{cpu_used:.0f}%", read_activity=delta("CPU_USER") + delta("CPU_NICE") > 0, write_activity=delta("CPU_SYSTEM") + delta("CPU_IRQ") + delta("CPU_SOFTIRQ") > 0, error=delta("CPU_IOWAIT") > 0, ) irq_delta = delta("INTR") self.rows[f"{prefix}_irq"].set_state(f"+{compact_count(irq_delta)}", read_activity=irq_delta > 0) ctx_delta = delta("CTXT") self.rows[f"{prefix}_ctx"].set_state(f"+{compact_count(ctx_delta)}", read_activity=ctx_delta > 0) fork_delta = delta("PROCESSES") self.rows[f"{prefix}_fork"].set_state(f"+{compact_count(fork_delta)}", write_activity=fork_delta > 0) softirq_delta = delta("SOFTIRQ") self.rows[f"{prefix}_softirq"].set_state(f"+{compact_count(softirq_delta)}", write_activity=softirq_delta > 0) tcp_in = delta("TCP_IN") tcp_out = delta("TCP_OUT") tcp_retrans = delta("TCP_RETRANS") self.rows[f"{prefix}_tcp"].set_state( f"I{compact_count(tcp_in)}/O{compact_count(tcp_out)}", error=tcp_retrans > 0, read_activity=tcp_in > 0, write_activity=tcp_out > 0, ) udp_in = delta("UDP_IN") udp_out = delta("UDP_OUT") udp_errors = delta("UDP_ERRORS") self.rows[f"{prefix}_udp"].set_state( f"I{compact_count(udp_in)}/O{compact_count(udp_out)}", error=udp_errors > 0, read_activity=udp_in > 0, write_activity=udp_out > 0, ) page_faults = delta("PGFAULT") major_faults = delta("PGMAJFAULT") paging = delta("PGPGIN") + delta("PGPGOUT") self.rows[f"{prefix}_vm"].set_state( f"F{compact_count(page_faults)}", error=major_faults > 0, read_activity=page_faults > 0, write_activity=paging > 0, ) running = current.get("PROCS_RUNNING", 0) blocked = current.get("PROCS_BLOCKED", 0) self.rows[f"{prefix}_runq"].set_state( f"R{running}/B{blocked}", read_activity=changed("PROCS_RUNNING"), write_activity=changed("PROCS_BLOCKED"), error_steady=blocked > 0, ) setattr(self, previous_attr, current) def ping_pc(self): if self.pc_ping.state() != QProcess.NotRunning: return pc_ip = dhcp_ip_for_mac(PC_MAC) or PC_IP 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 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() def update_system_status(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) self.set_percent_tile("rpi_root", disk_used_percent("/"), warn=80, critical=92) def render_rpi_fan(self, rpm): self.render_fan_tile("rpi_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 update_temp_graphs(self, rpi_temp, rpi_rpm): if "rpi_temp" not in self.charts: return 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, ) def update_graphs(self): hottest = max(float(self.latest_temp), self.disk_temp or 0.0) self.update_temp_graphs(hottest, self.latest_rpm) self.update_wan_graph() 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 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 @staticmethod def int_value(value): try: return int(float(value)) except (TypeError, ValueError): return 0 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; } """ ) win = TouchPanel() win.showFullScreen() sys.exit(app.exec_()) if __name__ == "__main__": main()