diff --git a/README.md b/README.md index 2229ce3..1c51974 100644 --- a/README.md +++ b/README.md @@ -68,8 +68,8 @@ Control은 라즈베리파이/리눅스 호스트의 팬과 시스템 상태를 - `apply_policy.php`: CLI/cron 팬 정책 적용 - `bin/control_ws.php`: WebSocket 서버. 핵심 PHP 파일 변경 감지 시 종료되고 systemd가 재시작해 새 코드를 로드 - `bin/wifi_observe.php`: 5G WiFi 연결 시간 보정을 위한 시스템단 관측 CLI -- `touch_panel/panel.py`: 라즈베리파이 정품 터치 디스플레이용 Qt 관리자 패널 프로토타입 -- `touch_panel/control_touch_bridge.php`: Qt 패널이 기존 Control PHP 함수를 호출하는 로컬 JSON 브리지 +- `touch_panel/panel.py`: 라즈베리파이 정품 터치 디스플레이용 standalone Qt 대시보드 +- `touch_panel/control_touch_bridge.php`: 이전 Qt 패널용 로컬 JSON 브리지. 현재 standalone 패널은 사용하지 않습니다. - `touch_panel/control-touch-panel.desktop`: GUI 로그인 후 Qt 패널을 띄우는 데스크톱 autostart 파일 - `systemd/control-wifi-observe.*`: WiFi 관측 CLI를 주기적으로 실행하는 systemd unit 예시 - `systemd/control-touch-panel.service`: GUI 세션에서 Qt 관리자 패널을 띄우는 systemd unit 예시 @@ -128,25 +128,27 @@ API 응답의 `battery.enabled`는 false, `battery.removed`는 true로 내려가 설정 모달이 열려 있는 동안 1초 상태 갱신은 폼 DOM을 다시 만들지 않습니다. 입력 중인 값은 저장, 기본값 초기화, 닫기 전까지 유지되며, 백그라운드 snapshot은 내부 설정 payload만 갱신합니다. 단순 boolean 설정은 `켜기`/`끄기` 세그먼트로 표시합니다. -## 터치 패널 프로토타입 +## 터치 패널 대시보드 -라즈베리파이 정품 7인치 DSI 터치 디스플레이에 띄울 로컬 Qt 관리자 패널 초안을 둡니다. 현재 라즈베리파이는 `multi-user.target` 콘솔 부팅이며 PyQt5는 설치되어 있습니다. GUI로 원복하려면 X/데스크톱 패키지를 다시 설치한 뒤 graphical target으로 바꾸고 재부팅해야 합니다. +라즈베리파이 정품 7인치 DSI 터치 디스플레이에 띄울 로컬 Qt 대시보드를 둡니다. 현재 GUI 패키지와 `lightdm`이 설치되어 있고 기본 부팅 target은 `graphical.target`입니다. `seo` GUI 세션의 데스크톱 autostart로 패널을 띄웁니다. ```bash sudo raspi-config sudo systemctl set-default graphical.target ``` -패키지가 제거된 상태면 Raspberry Pi OS 데스크톱 패키지와 X 서버를 먼저 복구해야 합니다. 운영 패키지 설치는 사용자가 확인한 뒤 진행합니다. +패키지가 제거된 상태면 Raspberry Pi OS 데스크톱 패키지와 X 서버를 먼저 복구해야 합니다. -프로토타입 수동 실행은 GUI 세션에서 다음처럼 합니다. +수동 실행은 GUI 세션에서 다음처럼 합니다. ```bash cd /var/www/control /usr/bin/python3 touch_panel/panel.py ``` -패널은 `touch_panel/control_touch_bridge.php`를 통해 기존 `collect_snapshot()`, `set_control_state()`를 재사용합니다. 왼쪽에는 `IO`, `WAN`, `LAN`, `2.4G`, `5G` 상태 타일을 두고, 원은 평소 회색으로 두었다가 해당 상태가 살아 있는 갱신 순간에만 초록으로 짧게 깜빡입니다. 오른쪽에는 팬 RPM, 판단 온도, PWM, `auto`/`manual`/`off`, PWM 증감 버튼을 둡니다. 팬 버튼은 우선 Control DB의 팬 상태를 저장하고, 실제 적용은 기존 Control 팬 정책 루프가 이어받는 구조입니다. +패널은 Control PHP/API/DB snapshot을 호출하지 않는 standalone 방식입니다. Python이 `/sys/class/net`, `/proc/diskstats`, `/sys/class/thermal`, 팬 hwmon 파일을 직접 읽어 120ms 간격으로 상태를 갱신합니다. 왼쪽 `IO`, `WAN`, `LAN`, `2.4G`, `5G` 원은 장식용 heartbeat가 아니라 실제 커널 카운터 증가에 반응합니다. `IO`는 디스크 read/write 카운터가 증가할 때, 네트워크는 RX/TX byte 또는 packet 카운터가 증가할 때 짧게 점멸합니다. + +오른쪽에는 팬 RPM, CPU 온도, PWM, WAN 실시간 속도와 온도/RPM, WAN/LAN/WiFi 트래픽 그래프를 표시합니다. 팬 버튼과 슬라이더는 Control DB를 거치지 않고 팬 hwmon sysfs에 직접 쓰기를 시도합니다. GUI 사용자의 sysfs 쓰기 권한이 없으면 대시보드는 계속 동작하되 `팬 쓰기 권한 없음`을 표시합니다. GUI 자동 실행은 데스크톱 autostart가 1차 기준입니다. diff --git a/touch_panel/panel.py b/touch_panel/panel.py old mode 100755 new mode 100644 index 771a120..d25f4e4 --- a/touch_panel/panel.py +++ b/touch_panel/panel.py @@ -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; }