GAMEMAX IceBurg LCD Linux bridge: sender + tray app
Reverse-engineered Linux driver for the LCD panel of the GAMEMAX IceBurg Digital AIO cooler (HID 5131:2007). Includes: - src/gamemax_report.py pure-stdlib sender: reads hwmon/proc sensors and streams the 65-byte HID report the firmware understands (no deps) - src/gamemax_gui.py PySide6 tray app: animated monitor, fan/temperature channel pickers, autostart, blank mode, single instance, crash log - packaging/ AppImage build script, udev rule, systemd user unit - tools/ IL dump helpers used to read the vendor MSIL - docs/ full reverse-engineering analysis - DISCLAIMER.md Linux-compatibility project, not affiliated with vendor Co-authored-by: REA (managed-code analysis)
This commit is contained in:
oxpoeandREA (managed-code analysis) committed
1 parent
b083919692
commit
f58a011f4d
15 files changed
+1893
No files matched your search
Executable
+804
@@ -0,0 +1,804 @@
|
||||
#!/usr/bin/env python3
|
||||
"""GAMEMAX IceBurg LCD — GUI tray: status, nyala/mati, pemantauan beranimasi, pengaturan.
|
||||
|
||||
Streaming tetap dijalankan oleh layanan systemd (gamemax-report.service); GUI ini
|
||||
mengendalikannya dan memakai gamemax_report.py sebagai pengirim.
|
||||
|
||||
gamemax_gui.py # jendela + tray (default)
|
||||
gamemax_gui.py --tray-only # tanpa jendela
|
||||
gamemax_gui.py --selftest # uji muat tanpa tampilan (untuk build AppImage)
|
||||
"""
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
|
||||
from PySide6.QtCore import QPointF, QRectF, Qt, QThread, QTimer, Signal
|
||||
from PySide6.QtGui import (QAction, QColor, QConicalGradient, QFont, QIcon,
|
||||
QPainter, QPen, QPixmap)
|
||||
from PySide6.QtWidgets import (
|
||||
QApplication, QCheckBox, QComboBox, QFormLayout, QGroupBox, QHBoxLayout, QLabel,
|
||||
QMenu, QPlainTextEdit, QProgressBar, QPushButton, QSystemTrayIcon, QVBoxLayout, QWidget,
|
||||
)
|
||||
from PySide6.QtNetwork import QLocalServer, QLocalSocket
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
import gamemax_report as gr # noqa: E402
|
||||
|
||||
SERVICE = "gamemax-report.service"
|
||||
CONFIG_PATH = gr.CONFIG_PATH
|
||||
INSTALL_DIR = os.path.expanduser("~/.local/share/gamemax-report")
|
||||
UNIT_PATH = os.path.expanduser(f"~/.config/systemd/user/{SERVICE}")
|
||||
RULE_PATH = "/etc/udev/rules.d/70-gamemax.rules"
|
||||
|
||||
UNIT_TEMPLATE = """[Unit]
|
||||
Description=GAMEMAX HID sensor display bridge (PcInfoMonitorPro replacement)
|
||||
After=graphical-session.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python3 {script} --wait --interval {interval} --sample {sample}
|
||||
Restart=always
|
||||
RestartSec=3
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
|
||||
RULE_TEMPLATE = ('SUBSYSTEM=="hidraw", ATTRS{idVendor}=="5131", ATTRS{idProduct}=="2007", '
|
||||
'TAG+="uaccess"\n')
|
||||
|
||||
CRASH_LOG = os.path.join(os.path.dirname(CONFIG_PATH), "gui-crash.log")
|
||||
ACCENT = QColor(222, 247, 80) # warna aksen GAMEMAX dari appConfig.ini
|
||||
DIM = QColor(120, 128, 120)
|
||||
|
||||
|
||||
def crash_log(text):
|
||||
"""Catat galat ke berkas supaya bisa diperiksa setelah aplikasi berhenti."""
|
||||
try:
|
||||
os.makedirs(os.path.dirname(CRASH_LOG), exist_ok=True)
|
||||
with open(CRASH_LOG, "a") as fh:
|
||||
fh.write(time.strftime("[%Y-%m-%d %H:%M:%S] ") + text.rstrip() + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def systemctl(*args):
|
||||
return subprocess.run(["systemctl", "--user", *args], capture_output=True, text=True)
|
||||
|
||||
|
||||
def service_active():
|
||||
return systemctl("is-active", SERVICE).stdout.strip() == "active"
|
||||
|
||||
|
||||
def service_enabled():
|
||||
return systemctl("is-enabled", SERVICE).stdout.strip() == "enabled"
|
||||
|
||||
|
||||
def dimmed_icon(path, opacity=0.35):
|
||||
"""Ikon tray versi redup untuk menandai layar tidak menampilkan sensor."""
|
||||
icon = QIcon(path)
|
||||
pixmap = icon.pixmap(64, 64)
|
||||
if pixmap.isNull():
|
||||
return icon
|
||||
dim = QPixmap(pixmap.size())
|
||||
dim.fill(Qt.transparent)
|
||||
painter = QPainter(dim)
|
||||
painter.setOpacity(opacity)
|
||||
painter.drawPixmap(0, 0, pixmap)
|
||||
painter.end()
|
||||
return QIcon(dim)
|
||||
|
||||
|
||||
def icon_path():
|
||||
for name in ("icon.png", "icon.ico", "gamemax.png"):
|
||||
path = os.path.join(HERE, name)
|
||||
if os.path.exists(path):
|
||||
return path
|
||||
return ""
|
||||
|
||||
|
||||
class Task(QThread):
|
||||
"""Menjalankan aksi lambat (systemctl/pkexec/salin berkas) di luar thread UI.
|
||||
|
||||
PENTING: method yang dijalankan di sini tidak boleh menyentuh widget Qt
|
||||
(membaca combo/checkbox atau memanggil refresh()); ambil nilainya di thread UI
|
||||
dan kirim sebagai argumen.
|
||||
"""
|
||||
|
||||
done = Signal(str, bool, str)
|
||||
|
||||
def __init__(self, label, fn):
|
||||
super().__init__()
|
||||
self.label, self.fn = label, fn
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
self.done.emit(self.label, True, self.fn() or "")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.done.emit(self.label, False, str(exc))
|
||||
|
||||
|
||||
class Animated:
|
||||
"""Nilai yang bergerak halus menuju target (digerakkan QTimer 33 ms)."""
|
||||
|
||||
def __init__(self, value=0.0):
|
||||
self.shown = value
|
||||
self.target = value
|
||||
|
||||
def tick(self, ease=0.18):
|
||||
if abs(self.target - self.shown) < 0.5:
|
||||
self.shown = self.target
|
||||
else:
|
||||
self.shown += (self.target - self.shown) * ease
|
||||
return self.shown
|
||||
|
||||
|
||||
class Gauge(QWidget):
|
||||
"""Busur 240 derajat untuk suhu (0..99 °C), dengan jarum dan angka besar."""
|
||||
|
||||
def __init__(self, title, vmax=99, unit="°C"):
|
||||
super().__init__()
|
||||
self.title, self.vmax, self.unit = title, vmax, unit
|
||||
self.value = Animated()
|
||||
self.setMinimumSize(220, 220)
|
||||
|
||||
def set(self, value):
|
||||
self.value.target = 0.0 if value is None else min(self.vmax, value)
|
||||
|
||||
def tick(self):
|
||||
self.value.tick(0.25)
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, _):
|
||||
p = QPainter(self)
|
||||
p.setRenderHint(QPainter.Antialiasing)
|
||||
w, h = self.width(), self.height()
|
||||
side = min(w, h) - 12
|
||||
rect = QRectF((w - side) / 2, (h - side) / 2 + 6, side, side)
|
||||
start, span = 210, -240 # derajat: kiri-bawah -> kanan-bawah
|
||||
frac = max(0.0, min(1.0, self.value.shown / self.vmax))
|
||||
|
||||
p.setPen(QPen(QColor(58, 62, 58), 14, Qt.SolidLine, Qt.RoundCap))
|
||||
p.drawArc(rect, start * 16, span * 16)
|
||||
if frac > 0:
|
||||
grad = QConicalGradient(rect.center(), -60)
|
||||
grad.setColorAt(0.0, ACCENT)
|
||||
grad.setColorAt(0.5, QColor(255, 170, 60))
|
||||
grad.setColorAt(1.0, QColor(255, 90, 60))
|
||||
p.setPen(QPen(grad, 14, Qt.SolidLine, Qt.RoundCap))
|
||||
p.drawArc(rect, start * 16, int(span * frac * 16))
|
||||
|
||||
angle = math.radians(start + span * frac)
|
||||
r = side / 2 - 8
|
||||
cx, cy = rect.center().x(), rect.center().y()
|
||||
p.setPen(QPen(ACCENT, 5, Qt.SolidLine, Qt.RoundCap))
|
||||
p.drawLine(QPointF(cx, cy), QPointF(cx + r * math.cos(angle), cy - r * math.sin(angle)))
|
||||
|
||||
p.setPen(ACCENT)
|
||||
f = QFont(); f.setPointSize(max(22, side // 5)); f.setBold(True)
|
||||
p.setFont(f)
|
||||
p.drawText(QRectF(rect.x(), cy - side / 6, side, side / 3), Qt.AlignCenter,
|
||||
f"{int(round(self.value.shown))}")
|
||||
p.setPen(QColor(210, 214, 210))
|
||||
f.setPointSize(max(9, side // 14)); f.setBold(False)
|
||||
p.setFont(f)
|
||||
p.drawText(QRectF(rect.x(), cy + side / 8, side, side / 5), Qt.AlignCenter,
|
||||
f"{self.unit} · {self.title}")
|
||||
|
||||
|
||||
class Bar(QWidget):
|
||||
"""Bar horizontal beranimasi dengan label dan nilai."""
|
||||
|
||||
def __init__(self, title, vmax, unit=""):
|
||||
super().__init__()
|
||||
self.title, self.vmax, self.unit = title, vmax, unit
|
||||
self.value = Animated()
|
||||
self.setMinimumHeight(46)
|
||||
|
||||
def set(self, value):
|
||||
self.value.target = 0.0 if value is None else max(0.0, min(self.vmax, value))
|
||||
|
||||
def tick(self):
|
||||
self.value.tick()
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, _):
|
||||
p = QPainter(self)
|
||||
p.setRenderHint(QPainter.Antialiasing)
|
||||
w, h = self.width(), self.height()
|
||||
f = QFont(); f.setPointSize(10)
|
||||
p.setFont(f)
|
||||
p.setPen(QColor(210, 214, 210))
|
||||
frac = self.value.shown / self.vmax
|
||||
shown = int(round(self.value.shown))
|
||||
p.drawText(QRectF(0, 0, w, 20), Qt.AlignLeft, self.title)
|
||||
p.setPen(ACCENT)
|
||||
p.drawText(QRectF(0, 0, w, 20), Qt.AlignRight, f"{shown}{self.unit}")
|
||||
track = QRectF(0, h - 18, w, 12)
|
||||
p.setPen(Qt.NoPen)
|
||||
p.setBrush(QColor(58, 62, 58))
|
||||
p.drawRoundedRect(track, 6, 6)
|
||||
if frac > 0.001:
|
||||
fill = QRectF(0, h - 18, max(6.0, w * frac), 12)
|
||||
p.setBrush(ACCENT if frac < 0.8 else QColor(255, 120, 70))
|
||||
p.drawRoundedRect(fill, 6, 6)
|
||||
|
||||
|
||||
class Window(QWidget):
|
||||
def __init__(self, tray):
|
||||
super().__init__()
|
||||
self.tray = tray
|
||||
self.prev_stat = gr.stat_snapshot()
|
||||
self.setWindowTitle("GAMEMAX IceBurg LCD")
|
||||
self.setWindowIcon(QIcon(icon_path()))
|
||||
self.setMinimumWidth(640)
|
||||
root = QVBoxLayout(self)
|
||||
|
||||
hint = QLabel("Kontrol layar LCD AIO GAMEMAX. Pilih slot kipas/sumber suhu di "
|
||||
"<b>Pengaturan</b> lalu klik <b>Simpan & terapkan</b>.")
|
||||
hint.setWordWrap(True)
|
||||
hint.setStyleSheet("color:#c8ccc8;")
|
||||
root.addWidget(hint)
|
||||
|
||||
# ---- banner mode layar (jelas: normal / kosong / berhenti)
|
||||
self.mode_banner = QLabel()
|
||||
self.mode_banner.setAlignment(Qt.AlignCenter)
|
||||
self.mode_banner.setMinimumHeight(34)
|
||||
root.addWidget(self.mode_banner)
|
||||
|
||||
# ---- header status
|
||||
self.status = QLabel()
|
||||
self.status.setStyleSheet("font-size: 13px;")
|
||||
root.addWidget(self.status)
|
||||
|
||||
# ---- kontrol utama
|
||||
controls = QHBoxLayout()
|
||||
self.toggle = QPushButton("Aktifkan layar")
|
||||
self.toggle.setMinimumHeight(38)
|
||||
self.toggle.setStyleSheet("font-weight: bold;")
|
||||
self.toggle.clicked.connect(lambda: self.start_task("mengubah status layanan", self.on_toggle))
|
||||
self.toggle.setToolTip("Mulai/hentikan pengiriman data sensor ke layar LCD")
|
||||
controls.addWidget(self.toggle, 2)
|
||||
self.blank = QCheckBox("Layar kosong")
|
||||
self.blank.toggled.connect(self.on_blank_clicked)
|
||||
self.blank.setToolTip("Kirim nilai nol ke layar (panel tetap beranimasi)")
|
||||
controls.addWidget(self.blank)
|
||||
test = QPushButton("Kirim uji")
|
||||
test.setToolTip("Kirim satu report contoh: 55 °C, beban 50 %, kipas 1500 RPM")
|
||||
test.clicked.connect(self.send_test)
|
||||
controls.addWidget(test)
|
||||
root.addLayout(controls)
|
||||
|
||||
# indikator proses berjalan (aksi dijalankan di thread terpisah)
|
||||
self.busy = QProgressBar()
|
||||
self.busy.setRange(0, 0) # mode tak tentu -> animasi berjalan
|
||||
self.busy.setTextVisible(False)
|
||||
self.busy.setFixedHeight(10)
|
||||
self.busy.hide()
|
||||
root.addWidget(self.busy)
|
||||
self.busy_label = QLabel()
|
||||
self.busy_label.setStyleSheet("color:#deF750;")
|
||||
self.busy_label.hide()
|
||||
root.addWidget(self.busy_label)
|
||||
self.buttons = [self.toggle, test]
|
||||
|
||||
# ---- papan monitor beranimasi
|
||||
board = QGroupBox("Pemantauan langsung")
|
||||
self.board = board
|
||||
bl = QHBoxLayout(board)
|
||||
self.gauge = Gauge("CPU", 99, "°C")
|
||||
bl.addWidget(self.gauge, 1)
|
||||
bars = QVBoxLayout()
|
||||
self.bar_load = Bar("Beban CPU", 100, " %")
|
||||
self.bar_fan = Bar("Kipas CPU (slot RPM)", 2000, " RPM")
|
||||
self.bar_clock = Bar("Frekuensi CPU", 6000, " MHz")
|
||||
for b in (self.bar_load, self.bar_fan, self.bar_clock):
|
||||
bars.addWidget(b)
|
||||
self.info = QLabel()
|
||||
self.info.setStyleSheet("color:#c8ccc8;")
|
||||
bars.addWidget(self.info)
|
||||
bars.addStretch(1)
|
||||
bl.addLayout(bars, 2)
|
||||
root.addWidget(board)
|
||||
|
||||
# ---- pengaturan
|
||||
settings = QGroupBox("Pengaturan")
|
||||
form = QFormLayout(settings)
|
||||
self.fan = QComboBox()
|
||||
self.fan.setToolTip("Kanal kipas yang dikirim ke layar; pilih kanal yang berputar "
|
||||
"kalau kipas dipindah ke header lain")
|
||||
self.temp = QComboBox()
|
||||
form.addRow("Slot kipas (RPM)", self.fan)
|
||||
form.addRow("Sumber suhu CPU", self.temp)
|
||||
self.autostart = QCheckBox("Mulai otomatis saat login")
|
||||
self.autostart.toggled.connect(
|
||||
lambda checked: self.start_task("mulai otomatis saat login",
|
||||
lambda: self.on_autostart(checked)))
|
||||
form.addRow("", self.autostart)
|
||||
row = QHBoxLayout()
|
||||
save = QPushButton("Simpan && terapkan")
|
||||
save.clicked.connect(self.on_save_clicked)
|
||||
row.addWidget(save)
|
||||
install = QPushButton("Pasang / perbarui layanan")
|
||||
install.clicked.connect(lambda: self.start_task("memasang layanan", self.install_service))
|
||||
row.addWidget(install)
|
||||
rule = QPushButton("Pasang aturan udev")
|
||||
rule.clicked.connect(lambda: self.start_task("memasang aturan udev", self.install_rule))
|
||||
row.addWidget(rule)
|
||||
self.buttons += [save, install, rule]
|
||||
form.addRow("", row)
|
||||
root.addWidget(settings)
|
||||
|
||||
# ---- detail + log
|
||||
self.detail = QPlainTextEdit()
|
||||
self.detail.setReadOnly(True)
|
||||
self.detail.setFixedHeight(110)
|
||||
root.addWidget(self.detail)
|
||||
self.logview = QPlainTextEdit()
|
||||
self.logview.setReadOnly(True)
|
||||
self.logview.setFixedHeight(64)
|
||||
root.addWidget(self.logview)
|
||||
|
||||
self.reload_choices()
|
||||
self.reload_config()
|
||||
|
||||
# 33 ms -> animasi; 1000 ms -> baca sensor
|
||||
self.anim = QTimer(self, timeout=self.tick, interval=33)
|
||||
self.anim.start()
|
||||
self.poll = QTimer(self, timeout=self.refresh, interval=1000)
|
||||
self.poll.start()
|
||||
self.refresh()
|
||||
if gr.load_config(CONFIG_PATH).get("blank"):
|
||||
self.log("PERHATIAN: mode layar kosong sedang aktif dari konfigurasi "
|
||||
"(layar menampilkan nilai nol). Hilangkan centang 'Layar kosong' "
|
||||
"untuk kembali ke nilai sensor.")
|
||||
|
||||
# -------------------------------------------------------------- pengaturan
|
||||
def reload_choices(self):
|
||||
self.fan.clear()
|
||||
self.fan.addItem("Otomatis (kipas tercepat yang berputar)", "")
|
||||
for c in gr.hwmon_channels("fan*_input"):
|
||||
rpm = "" if c["value"] is None else f" — {int(c['value'])} RPM"
|
||||
self.fan.addItem(f"{c['chip']} {c['file']}{rpm}", f"{c['chip']}/{c['file']}")
|
||||
self.temp.clear()
|
||||
self.temp.addItem("Otomatis (sensor CPU tertinggi)", "")
|
||||
channels = [c for c in gr.hwmon_channels("temp*_input") if c["value"] is not None]
|
||||
channels.sort(key=lambda c: (c["chip"] not in gr.CPU_CHIPS, c["chip"], c["file"]))
|
||||
for c in channels:
|
||||
name = c["label"] or c["file"]
|
||||
tag = "" if c["chip"] in gr.CPU_CHIPS else " (bukan CPU) "
|
||||
self.temp.addItem(f"{c['chip']}{tag} {name} — {c['value'] / 1000:.1f} °C",
|
||||
f"{c['chip']}/{c['file']}")
|
||||
|
||||
def reload_config(self):
|
||||
cfg = gr.load_config(CONFIG_PATH)
|
||||
for combo, key in ((self.fan, "fan_channel"), (self.temp, "temp_channel")):
|
||||
idx = combo.findData(cfg.get(key) or "")
|
||||
combo.setCurrentIndex(idx if idx >= 0 else 0)
|
||||
self.autostart.blockSignals(True)
|
||||
self.autostart.setChecked(service_enabled())
|
||||
self.autostart.blockSignals(False)
|
||||
|
||||
def config(self):
|
||||
return {
|
||||
"fan_channel": self.fan.currentData() or None,
|
||||
"temp_channel": self.temp.currentData() or None,
|
||||
"unit": "c", # panel hanya punya label °C
|
||||
"interval": 0.5,
|
||||
"sample": 0.5,
|
||||
"vendor_layout": False,
|
||||
"blank": bool(self.blank.isChecked()),
|
||||
}
|
||||
|
||||
# -------------------------------------------------------------- tugas latar
|
||||
def start_task(self, label, fn):
|
||||
if getattr(self, "_task", None) is not None and self._task.isRunning():
|
||||
self.log("masih ada proses berjalan — tunggu sampai indikator selesai")
|
||||
return
|
||||
self.busy_label.setText(f"Sedang bekerja: {label}…")
|
||||
self.busy.show()
|
||||
self.busy_label.show()
|
||||
for button in self.buttons:
|
||||
button.setEnabled(False)
|
||||
task = Task(label, fn)
|
||||
task.done.connect(self.on_task_done)
|
||||
self._task = task
|
||||
task.start()
|
||||
|
||||
def on_task_done(self, label, ok, message):
|
||||
self.busy.hide()
|
||||
self.busy_label.hide()
|
||||
for button in self.buttons:
|
||||
button.setEnabled(True)
|
||||
self.log(("selesai: " if ok else "GAGAL: ") + label + (f" — {message}" if message else ""))
|
||||
self.refresh()
|
||||
|
||||
# -------------------------------------------------------------- aksi
|
||||
def on_toggle(self):
|
||||
try:
|
||||
self.ensure_unit()
|
||||
want_active = not service_active()
|
||||
for attempt in (1, 2):
|
||||
systemctl("start" if want_active else "stop", SERVICE)
|
||||
if service_active() == want_active:
|
||||
return (f"layanan {'AKTIF' if want_active else 'nonaktif'}"
|
||||
+ ("" if attempt == 1 else f" (percobaan ke-{attempt})"))
|
||||
else:
|
||||
return "status layanan tidak berubah seperti yang diminta"
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"galat: {exc}"
|
||||
|
||||
def on_autostart(self, checked):
|
||||
try:
|
||||
self.ensure_unit()
|
||||
systemctl("daemon-reload")
|
||||
res = systemctl("enable" if checked else "disable", SERVICE)
|
||||
state = service_enabled()
|
||||
return (f"mulai otomatis saat login: {'AKTIF' if state else 'nonaktif'}"
|
||||
+ ("" if res.returncode == 0 else f" ({res.stderr.strip()})"))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"galat autostart: {exc}"
|
||||
|
||||
def on_blank_clicked(self, checked):
|
||||
"""Dijalankan di thread UI: baca widget di sini, kerjakan sisanya di latar."""
|
||||
cfg = self.config()
|
||||
cfg["blank"] = bool(checked)
|
||||
self.start_task("mode layar kosong", lambda: self.on_blank(checked, cfg))
|
||||
|
||||
def on_blank(self, checked, cfg):
|
||||
"""Mode layar kosong ditangani pengirim (config "blank")."""
|
||||
try:
|
||||
self.write_config(cfg)
|
||||
self.ensure_unit(cfg)
|
||||
if checked and not service_active():
|
||||
systemctl("start", SERVICE)
|
||||
elif service_active():
|
||||
systemctl("restart", SERVICE)
|
||||
return ("layar kosong: AKTIF (report bernilai nol)" if checked
|
||||
else "layar kosong: nonaktif (kembali ke nilai sensor)")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"galat mode layar kosong: {exc}"
|
||||
|
||||
def write_config(self, cfg):
|
||||
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
|
||||
with open(CONFIG_PATH, "w") as fh:
|
||||
json.dump(cfg, fh, indent=2)
|
||||
|
||||
def ensure_unit(self, cfg=None):
|
||||
if not os.path.exists(UNIT_PATH):
|
||||
self.install_service(cfg)
|
||||
|
||||
def send_test(self):
|
||||
node = gr.find_device()
|
||||
if not node:
|
||||
self.log("perangkat tidak terdeteksi")
|
||||
return
|
||||
v = gr.read_sensors(self.prev_stat, {"fan_channel": self.fan.currentData(),
|
||||
"temp_channel": self.temp.currentData()})
|
||||
v.update({"cpu_temp": 55, "gpu_temp": 60, "cpu_load": 50, "cpu_fan": 1500,
|
||||
"cpu_clock": 4000, "cpu_power": 66})
|
||||
try:
|
||||
with open(node, "wb", buffering=0) as fh:
|
||||
fh.write(gr.build_report(v, "c")[1:])
|
||||
self.log(f"report uji terkirim ke {node} (55 °C / 50 % / 1500 RPM)")
|
||||
except OSError as exc:
|
||||
self.log(f"gagal mengirim uji: {exc}")
|
||||
|
||||
def install_service(self, cfg=None):
|
||||
try:
|
||||
os.makedirs(INSTALL_DIR, exist_ok=True)
|
||||
script = os.path.join(INSTALL_DIR, "gamemax_report.py")
|
||||
shutil.copyfile(os.path.join(HERE, "gamemax_report.py"), script)
|
||||
os.makedirs(os.path.dirname(UNIT_PATH), exist_ok=True)
|
||||
cfg = cfg or {}
|
||||
with open(UNIT_PATH, "w") as fh:
|
||||
fh.write(UNIT_TEMPLATE.format(script=script,
|
||||
interval=cfg.get("interval", 0.5),
|
||||
sample=cfg.get("sample", 0.5)))
|
||||
systemctl("daemon-reload")
|
||||
return f"layanan dipasang: {UNIT_PATH}"
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"gagal memasang layanan: {exc}"
|
||||
|
||||
def install_rule(self):
|
||||
script = "/tmp/gamemax-udev-install.sh"
|
||||
try:
|
||||
with open(script, "w") as fh:
|
||||
fh.write("#!/bin/sh\n"
|
||||
f"cat > {RULE_PATH} <<'EOF'\n{RULE_TEMPLATE}EOF\n"
|
||||
"udevadm control --reload\n"
|
||||
"udevadm trigger --subsystem-match=hidraw\n")
|
||||
os.chmod(script, 0o755)
|
||||
res = subprocess.run(["pkexec", "sh", script], capture_output=True, text=True)
|
||||
return ("aturan udev dipasang" if res.returncode == 0
|
||||
else f"gagal memasang udev: {res.stderr.strip() or res.returncode}")
|
||||
except FileNotFoundError:
|
||||
return f"pkexec tidak tersedia; jalankan manual: sudo sh {script}"
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"galat udev: {exc}"
|
||||
|
||||
def fix_fan_choice(self):
|
||||
"""Slot kipas terpilih yang diam sementara ada kipas lain berputar akan dialihkan."""
|
||||
if not self.fan.currentData():
|
||||
return ""
|
||||
spinning = [(c["chip"], c["file"], int(c["value"])) for c in gr.hwmon_channels("fan*_input")
|
||||
if c["value"] and c["chip"] not in gr.GPU_CHIPS]
|
||||
if gr.channel_value(self.fan.currentData()) or not spinning:
|
||||
return ""
|
||||
chip, fname, rpm = max(spinning, key=lambda t: t[2])
|
||||
idx = self.fan.findData(f"{chip}/{fname}")
|
||||
if idx < 0:
|
||||
return ""
|
||||
old = self.fan.currentText()
|
||||
self.fan.setCurrentIndex(idx)
|
||||
return f"slot lama membaca 0 RPM ({old.split(' —')[0]}) -> dialihkan ke {chip} {fname} ({rpm} RPM)"
|
||||
|
||||
def on_save_clicked(self):
|
||||
"""Thread UI: koreksi slot kipas lalu baca pengaturan sebelum dikerjakan."""
|
||||
note = self.fix_fan_choice()
|
||||
self.start_task("menyimpan pengaturan",
|
||||
lambda: self.save_settings(self.config(), note))
|
||||
|
||||
def save_settings(self, cfg, note=""):
|
||||
try:
|
||||
self.write_config(cfg)
|
||||
self.ensure_unit(cfg)
|
||||
if service_active():
|
||||
systemctl("restart", SERVICE)
|
||||
tail = " (layanan dijalankan ulang)"
|
||||
else:
|
||||
systemctl("start", SERVICE)
|
||||
tail = " (layanan dinyalakan)"
|
||||
return "pengaturan disimpan & diterapkan" + tail + (f" | {note}" if note else "")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"gagal menyimpan: {exc}"
|
||||
|
||||
def log(self, text):
|
||||
self.logview.appendPlainText(text)
|
||||
|
||||
def warn_once(self, text):
|
||||
if getattr(self, "_warned", None) != text:
|
||||
self._warned = text
|
||||
self.log(text)
|
||||
|
||||
# -------------------------------------------------------------- refresh
|
||||
def refresh(self):
|
||||
try:
|
||||
active = service_active()
|
||||
node = gr.find_device()
|
||||
enabled = service_enabled()
|
||||
blank_on = bool(gr.load_config(CONFIG_PATH).get("blank"))
|
||||
mode = "normal" if active and not blank_on else ("kosong" if blank_on else "berhenti")
|
||||
banner = {
|
||||
"normal": ("Layar menampilkan data sensor", "#1d3b1d", "#b8f5b8"),
|
||||
"kosong": ("LAYAR KOSONG — panel menampilkan nilai nol (animasi firmware tetap jalan)",
|
||||
"#4a3a10", "#ffd479"),
|
||||
"berhenti": ("PENGIRIMAN DIHENTIKAN — panel menampilkan data terakhir yang diterima",
|
||||
"#3a3a3a", "#d0d0d0"),
|
||||
}[mode]
|
||||
self.mode_banner.setText(banner[0])
|
||||
self.board.setTitle({
|
||||
"normal": "Pemantauan langsung — nilai ini yang dikirim ke layar",
|
||||
"kosong": "Pemantauan langsung — nilai sensor (yang DIKIRIM ke layar: nol)",
|
||||
"berhenti": "Pemantauan langsung — nilai sensor (TIDAK dikirim, layanan mati)",
|
||||
}[mode])
|
||||
self.mode_banner.setStyleSheet(
|
||||
f"background:{banner[1]}; color:{banner[2]}; border-radius:4px; "
|
||||
"font-weight:bold; padding:4px;")
|
||||
if getattr(self, "_mode", None) != mode:
|
||||
self._mode = mode
|
||||
labels = {"normal": "Layar kembali menampilkan data sensor",
|
||||
"kosong": "Layar dikosongkan (nilai nol)",
|
||||
"berhenti": "Pengiriman data dihentikan"}
|
||||
if mode != "normal" or getattr(self, "_mode_seen", False):
|
||||
subprocess.run(["notify-send", "-a", "GAMEMAX LCD", "Status layar",
|
||||
labels[mode]], capture_output=True)
|
||||
self._mode_seen = True
|
||||
dot = "🟢" if active else "⚪"
|
||||
self.status.setText(
|
||||
f"{dot} Perangkat: <b>{node or 'tidak terdeteksi'}</b> · "
|
||||
f"Layanan: <b>{'AKTIF' if active else 'nonaktif'}</b> · "
|
||||
f"Mulai otomatis: <b>{'ya' if enabled else 'tidak'}</b>"
|
||||
+ (" · Layar: <b style='color:#ff9060'>KOSONG (nilai nol)</b>"
|
||||
if blank_on else ""))
|
||||
self.toggle.setText("Matikan layar" if active else "Aktifkan layar")
|
||||
self.autostart.blockSignals(True)
|
||||
self.autostart.setChecked(enabled)
|
||||
self.autostart.blockSignals(False)
|
||||
self.blank.blockSignals(True)
|
||||
self.blank.setChecked(bool(gr.load_config(CONFIG_PATH).get("blank")))
|
||||
self.blank.blockSignals(False)
|
||||
self.tray.sync(active)
|
||||
|
||||
cfg = {"fan_channel": self.fan.currentData(),
|
||||
"temp_channel": self.temp.currentData()}
|
||||
v = gr.read_sensors(self.prev_stat, cfg)
|
||||
self.prev_stat = gr.stat_snapshot()
|
||||
self.gauge.set(v["cpu_temp"])
|
||||
self.bar_load.set(v["cpu_load"])
|
||||
self.bar_fan.set(v["cpu_fan"])
|
||||
spinning = [(c["chip"], c["file"], int(c["value"])) for c in gr.hwmon_channels("fan*_input")
|
||||
if c["value"] and c["chip"] not in gr.GPU_CHIPS]
|
||||
if self.fan.currentData() and not (v["cpu_fan"] or 0) and spinning:
|
||||
best = max(spinning, key=lambda t: t[2])
|
||||
self.warn_once(f"slot kipas terpilih membaca 0 RPM; kipas berputar ada di "
|
||||
f"{best[0]} {best[1]} ({best[2]} RPM) — pilih itu lalu Simpan")
|
||||
self.bar_clock.set(v["cpu_clock"])
|
||||
self.detail.setPlainText(
|
||||
(f"Daya CPU : {v['cpu_power']:.0f} W\n" if v["cpu_power"] else "Daya CPU : —\n") +
|
||||
(f"GPU : {v['gpu_temp']:.0f} °C · beban {v['gpu_load'] or '—'} % · "
|
||||
f"{int(v['gpu_clock'] or 0)} MHz\n" if v["gpu_temp"] else "GPU : —\n") +
|
||||
(f"Memori : {v['ram_load']} % dari {int(v['ram_mb'])} MiB\n"
|
||||
if v["ram_load"] and v["ram_mb"] else "Memori : —\n") +
|
||||
(f"Penyimpanan : {v['disk_free']} % kosong dari {v['disk_gb']:.0f} GB\n"
|
||||
if v["disk_free"] and v["disk_gb"] else "Penyimpanan : —\n") +
|
||||
"Isi report : " + gr.build_report(v, "c").hex(" "))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.log(f"galat pembaruan: {exc}")
|
||||
if os.environ.get("GAMEMAX_DEBUG"):
|
||||
self.log(traceback.format_exc())
|
||||
|
||||
def tick(self):
|
||||
for widget in (self.gauge, self.bar_load, self.bar_fan, self.bar_clock):
|
||||
widget.tick()
|
||||
|
||||
|
||||
class Tray:
|
||||
def __init__(self, app):
|
||||
self.app = app
|
||||
self.icon = QSystemTrayIcon(QIcon(icon_path()), app)
|
||||
self.blank_action = QAction("Layar kosong (nilai nol)", app)
|
||||
self.blank_action.setCheckable(True)
|
||||
menu = QMenu()
|
||||
show = QAction("Buka jendela", menu)
|
||||
show.triggered.connect(self.show_window)
|
||||
toggle = QAction("Aktifkan / matikan", menu)
|
||||
toggle.triggered.connect(lambda: self.window.on_toggle())
|
||||
self.blank_action.triggered.connect(lambda checked: self.window.blank.setChecked(checked))
|
||||
quit_action = QAction("Keluar", menu)
|
||||
quit_action.triggered.connect(app.quit)
|
||||
menu.addAction(show)
|
||||
menu.addAction(toggle)
|
||||
menu.addAction(self.blank_action)
|
||||
menu.addSeparator()
|
||||
menu.addAction(quit_action)
|
||||
self.icon.setContextMenu(menu)
|
||||
self.icon.activated.connect(
|
||||
lambda reason: self.show_window() if reason == QSystemTrayIcon.Trigger else None)
|
||||
self.window = Window(self)
|
||||
self.icon.show()
|
||||
|
||||
def sync(self, active):
|
||||
blank = bool(gr.load_config(CONFIG_PATH).get("blank"))
|
||||
state = ("layar kosong (nilai nol)" if blank and active
|
||||
else ("pengiriman dihentikan" if not active else "normal"))
|
||||
self.icon.setToolTip(f"GAMEMAX LCD: {state}")
|
||||
base = icon_path()
|
||||
self.icon.setIcon(dimmed_icon(base) if state != "normal" else QIcon(base))
|
||||
window = getattr(self, "window", None)
|
||||
if window is not None:
|
||||
self.blank_action.setChecked(window.blank.isChecked())
|
||||
|
||||
def show_window(self):
|
||||
self.window.showNormal() # keluar dari minimize
|
||||
self.window.setWindowState((self.window.windowState() & ~Qt.WindowMinimized)
|
||||
| Qt.WindowActive)
|
||||
self.window.raise_()
|
||||
self.window.activateWindow()
|
||||
if QSystemTrayIcon.isSystemTrayAvailable():
|
||||
self.icon.showMessage("GAMEMAX LCD", "Jendela dibuka dari permintaan kedua.",
|
||||
QSystemTrayIcon.Information, 3000)
|
||||
|
||||
|
||||
SOCKET_NAME = "gamemax-lcd-" + os.environ.get("USER", "user")
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
return _main()
|
||||
except Exception: # noqa: BLE001
|
||||
crash_log("FATAL\n" + traceback.format_exc())
|
||||
raise
|
||||
|
||||
|
||||
def _main():
|
||||
app = QApplication(sys.argv)
|
||||
app.setApplicationName("GAMEMAX LCD")
|
||||
app.setQuitOnLastWindowClosed(False)
|
||||
|
||||
# hanya satu instance: yang kedua memunculkan jendela yang sudah ada lalu keluar
|
||||
probe = QLocalSocket()
|
||||
probe.connectToServer(SOCKET_NAME)
|
||||
if probe.waitForConnected(400):
|
||||
probe.write(b"raise")
|
||||
probe.waitForBytesWritten(400)
|
||||
print("GAMEMAX LCD sudah berjalan — instance pertama diminta membuka jendelanya.")
|
||||
# Beri tahu lewat notifikasi desktop juga, karena keluarnya proses kedua bisa
|
||||
# terlihat seperti crash saat diluncurkan dari menu.
|
||||
subprocess.run(["notify-send", "-a", "GAMEMAX LCD", "Sudah berjalan",
|
||||
"Aplikasi sudah aktif — jendelanya dibuka, cek taskbar/tray."],
|
||||
capture_output=True)
|
||||
return 0
|
||||
QLocalServer.removeServer(SOCKET_NAME)
|
||||
server = QLocalServer()
|
||||
server.listen(SOCKET_NAME)
|
||||
|
||||
tray = Tray(app)
|
||||
server.newConnection.connect(lambda: tray.show_window())
|
||||
app.aboutToQuit.connect(lambda: (server.close(), QLocalServer.removeServer(SOCKET_NAME)))
|
||||
# Jendela selalu dibuka saat diluncurkan; tray tetap ada kalau didukung desktop.
|
||||
tray.show_window()
|
||||
if "--tray-only" in sys.argv:
|
||||
tray.window.hide()
|
||||
if not QSystemTrayIcon.isSystemTrayAvailable():
|
||||
tray.window.log("Tray tidak tersedia di desktop ini — pakai jendela ini untuk kontrol.")
|
||||
def report_exception(*tb):
|
||||
text = "".join(traceback.format_exception(*tb)).strip()
|
||||
crash_log(text)
|
||||
try:
|
||||
tray.window.log("GALAT: " + text.splitlines()[-1])
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
sys.excepthook = report_exception
|
||||
threading.excepthook = lambda args: report_exception(args.exc_type, args.exc_value, args.exc_traceback)
|
||||
if "--selftest-actions" in sys.argv: # uji aksi GUI pada bundel beku
|
||||
w = tray.window
|
||||
steps = (("simpan & terapkan", lambda: w.save_settings(w.config())),
|
||||
("layar kosong ON", lambda: w.on_blank(True, {**w.config(), "blank": True})),
|
||||
("layar kosong OFF", lambda: w.on_blank(False, {**w.config(), "blank": False})),
|
||||
("matikan layar", w.on_toggle),
|
||||
("nyalakan layar", w.on_toggle),
|
||||
("autostart ON", lambda: w.on_autostart(True)),
|
||||
("kirim uji", w.send_test),
|
||||
("pasang layanan", lambda: w.install_service(w.config())))
|
||||
for name, fn in steps:
|
||||
try:
|
||||
fn()
|
||||
print(f"[ OK ] {name}")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"[GAGAL] {name}: {exc}")
|
||||
# jalur asinkron yang dipakai tombol: centang layar kosong + simpan
|
||||
def wait_task(seconds=10):
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < seconds:
|
||||
task = getattr(w, "_task", None)
|
||||
if task is not None and not task.isRunning():
|
||||
break
|
||||
app.processEvents()
|
||||
time.sleep(0.05)
|
||||
for _ in range(10):
|
||||
app.processEvents()
|
||||
time.sleep(0.02)
|
||||
|
||||
for name, trigger in (("centang layar kosong", lambda: w.blank.setChecked(True)),
|
||||
("lepas centang", lambda: w.blank.setChecked(False)),
|
||||
("simpan lewat tombol", w.on_save_clicked)):
|
||||
try:
|
||||
trigger()
|
||||
wait_task()
|
||||
print(f"[ OK ] {name} (asinkron)")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"[GAGAL] {name} (asinkron): {exc}")
|
||||
w.refresh()
|
||||
print("log:", " | ".join(w.logview.toPlainText().splitlines()[-6:]))
|
||||
return 0
|
||||
if "--selftest" in sys.argv:
|
||||
QTimer.singleShot(1200, app.quit)
|
||||
rc = app.exec()
|
||||
if "--selftest" in sys.argv:
|
||||
print(f"selftest OK (kipas={tray.window.fan.count()} suhu={tray.window.temp.count()})")
|
||||
return rc
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--selftest" in sys.argv:
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
sys.exit(main())
|
||||
Executable
+443
@@ -0,0 +1,443 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Push hardware sensor data to a GAMEMAX AIO LCD (HID VID 0x5131, PID 0x2007).
|
||||
|
||||
Reverse-engineered from PcInfoMonitorPro.exe (GAMEMAX V1.0.0.04) — see ../01-recon.md.
|
||||
Report layout, derived from DMainWidget.onGetPcInfoOk:
|
||||
|
||||
byte 0 report ID, always 0
|
||||
byte 1 64 (0x40), payload length
|
||||
byte 2 CPU temp (0..127). The original app ORs 0x80 for Celsius, but this
|
||||
firmware ignores every value >= 128, so we send plain numbers
|
||||
byte 3 CPU load %
|
||||
byte 4..5 RPM slot (big-endian u16) -- panel renders this field next to the fan icon
|
||||
byte 6..7 CPU clock MHz (big-endian u16) -- original app's byte order
|
||||
byte 8..9 CPU power W (big-endian u16)
|
||||
byte 10 GPU temp (0..127), same rule as CPU temp
|
||||
byte 11 GPU load %
|
||||
byte 12..13 GPU clock MHz (big-endian u16)
|
||||
byte 14..15 GPU fan RPM (big-endian u16)
|
||||
byte 16..17 GPU power W (big-endian u16)
|
||||
byte 18 RAM usage %
|
||||
byte 19..20 total RAM MiB (big-endian u16)
|
||||
byte 21 disk free %
|
||||
byte 22..23 total disk GB (big-endian u16)
|
||||
byte 24..64 zero
|
||||
|
||||
The panel reads byte 4..5 for the "RPM" field while the original app puts the CPU
|
||||
clock there, so by default the two values are swapped so the panel shows a real fan
|
||||
speed; --vendor-layout restores the original placement.
|
||||
|
||||
Observed ceiling: the interrupt OUT endpoint (bInterval 160 ms) accepts about
|
||||
8 reports/s and each hidraw write blocks until its URB completes.
|
||||
|
||||
Plug and play: with --wait the process stays alive, waits for the device to appear
|
||||
and keeps streaming through unplug/replug, which is what the systemd unit uses.
|
||||
|
||||
python gamemax_report.py --dry-run --cpu-temp 55
|
||||
python gamemax_report.py --wait # service mode
|
||||
python gamemax_report.py --list-sensors # JSON for the GUI
|
||||
"""
|
||||
import argparse
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import select
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
VID, PID = 0x5131, 0x2007
|
||||
REPORT_SIZE = 65
|
||||
CONFIG_PATH = os.path.expanduser("~/.config/gamemax-report/config.json")
|
||||
VALUE_KEYS = ("cpu_temp", "gpu_temp", "cpu_load", "cpu_clock", "cpu_fan", "cpu_power",
|
||||
"gpu_load", "gpu_clock", "gpu_fan", "gpu_power",
|
||||
"ram_load", "ram_mb", "disk_free", "disk_gb")
|
||||
CONFIG_KEYS = ("fan_channel", "temp_channel", "unit", "interval", "sample",
|
||||
"flag_celsius", "vendor_layout", "disk_path", "blank")
|
||||
|
||||
|
||||
def load_config(path=CONFIG_PATH):
|
||||
try:
|
||||
with open(path) as fh:
|
||||
data = json.load(fh)
|
||||
return data if isinstance(data, dict) else {}
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def u16(value):
|
||||
v = max(0, min(0xFFFF, int(value or 0)))
|
||||
return bytes((v >> 8, v & 0xFF))
|
||||
|
||||
|
||||
def temp_byte(value, unit, flag_celsius=False):
|
||||
"""Wire values are Celsius 0..127.
|
||||
|
||||
The original app ORs 0x80 when the UI unit is Celsius; the IceBurg panel rejects
|
||||
values >= 128 (it keeps showing the last accepted one), so we only do that when
|
||||
--flag-celsius is requested. --unit f converts the supplied Fahrenheit values.
|
||||
"""
|
||||
if value is None:
|
||||
return 0
|
||||
# The panel shows at most two digits and its unit label is fixed to Celsius, so
|
||||
# values above 99 are clamped (verified on hardware); --unit f sends the number
|
||||
# as-is for fidelity with the original app, but the panel still labels it °C.
|
||||
v = max(0, min(99, int(round(value))))
|
||||
return v | 0x80 if flag_celsius else v
|
||||
|
||||
|
||||
def build_report(v, unit, flag_celsius=False, vendor_layout=False):
|
||||
r = bytearray(REPORT_SIZE)
|
||||
clock, fan = v.get("cpu_clock"), v.get("cpu_fan")
|
||||
if not vendor_layout: # this panel renders byte 4..5 in the slot labelled RPM
|
||||
clock, fan = fan, clock
|
||||
r[0] = 0
|
||||
r[1] = 64
|
||||
r[2] = temp_byte(v.get("cpu_temp"), unit, flag_celsius)
|
||||
r[3] = int(v.get("cpu_load") or 0) & 0xFF
|
||||
r[4:6] = u16(clock)
|
||||
r[6:8] = u16(fan)
|
||||
r[8:10] = u16(v.get("cpu_power"))
|
||||
r[10] = temp_byte(v.get("gpu_temp"), unit, flag_celsius)
|
||||
r[11] = int(v.get("gpu_load") or 0) & 0xFF
|
||||
r[12:14] = u16(v.get("gpu_clock"))
|
||||
r[14:16] = u16(v.get("gpu_fan"))
|
||||
r[16:18] = u16(v.get("gpu_power"))
|
||||
r[18] = int(v.get("ram_load") or 0) & 0xFF
|
||||
r[19:21] = u16(v.get("ram_mb"))
|
||||
r[21] = int(v.get("disk_free") or 0) & 0xFF
|
||||
r[22:24] = u16(v.get("disk_gb"))
|
||||
return bytes(r)
|
||||
|
||||
|
||||
def self_check():
|
||||
"""Lock the protocol shape the original enforces."""
|
||||
r = build_report({"cpu_temp": 60, "gpu_temp": 70, "cpu_clock": 5000, "cpu_fan": 1200,
|
||||
"cpu_power": 88, "gpu_power": 250, "ram_load": 42, "ram_mb": 32768,
|
||||
"disk_free": 30, "disk_gb": 1951}, "c")
|
||||
assert len(r) == REPORT_SIZE, len(r)
|
||||
assert r[0] == 0 and r[1] == 64, r[:2].hex()
|
||||
assert r[2] == 60 and r[10] == 70, (r[2], r[10]) # plain Celsius, no flag
|
||||
flagged = build_report({"cpu_temp": 60}, "c", flag_celsius=True)
|
||||
assert flagged[2] == 0xBC, hex(flagged[2]) # original app's byte
|
||||
assert r[4:6] == b"\x04\xb0" and r[6:8] == b"\x13\x88" # fan in RPM slot, BE
|
||||
vendor = build_report({"cpu_clock": 5000, "cpu_fan": 1200}, "c", vendor_layout=True)
|
||||
assert vendor[4:6] == b"\x13\x88" and vendor[6:8] == b"\x04\xb0" # original placement
|
||||
assert r[8:10] == b"\x00\x58" and r[16:18] == b"\x00\xfa"
|
||||
assert r[18] == 42 and r[19:21] == b"\x80\x00" # 32768 MiB
|
||||
assert r[21] == 30 and r[22:24] == b"\x07\x9f" # 1951 GB
|
||||
assert r[24:] == bytes(41)
|
||||
assert build_report({"cpu_temp": 100}, "f")[2] == 99 # clamped to two digits
|
||||
assert build_report({"cpu_temp": 77}, "f")[2] == 77 # Fahrenheit sent as-is
|
||||
return r
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- hwmon access
|
||||
|
||||
def hwmon_read(path, scale=1.0):
|
||||
try:
|
||||
with open(path) as fh:
|
||||
return float(fh.read().strip()) / scale
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def hwmon_channels(pattern="fan*_input"):
|
||||
"""List every hwmon channel as {"chip", "file", "label", "value"}."""
|
||||
out = []
|
||||
for path in sorted(glob.glob(f"/sys/class/hwmon/hwmon*/{pattern}")):
|
||||
base, fname = os.path.dirname(path), os.path.basename(path)
|
||||
try:
|
||||
chip = open(os.path.join(base, "name")).read().strip()
|
||||
except OSError:
|
||||
continue
|
||||
label = None
|
||||
try:
|
||||
label = open(os.path.join(base, fname.replace("_input", "_label"))).read().strip()
|
||||
except OSError:
|
||||
pass
|
||||
out.append({"chip": chip, "file": fname, "label": label,
|
||||
"value": hwmon_read(path)})
|
||||
return out
|
||||
|
||||
|
||||
def channel_value(spec, scale=1.0):
|
||||
"""Read an explicit channel like "nct6798/fan2_input"."""
|
||||
if not spec or "/" not in spec:
|
||||
return None
|
||||
chip, fname = spec.split("/", 1)
|
||||
for path in glob.glob(f"/sys/class/hwmon/hwmon*/{fname}"):
|
||||
try:
|
||||
if open(os.path.join(os.path.dirname(path), "name")).read().strip() != chip:
|
||||
continue
|
||||
except OSError:
|
||||
continue
|
||||
return hwmon_read(path, scale)
|
||||
return None
|
||||
|
||||
|
||||
def hwmon_max(chips, pattern="temp*_input", scale=1.0):
|
||||
"""Highest matching reading; a CPU exposes Tctl plus per-CCD Tccd sensors."""
|
||||
best = None
|
||||
for ch in hwmon_channels(pattern):
|
||||
if not any(c == "*" or c in ch["chip"] for c in chips):
|
||||
continue
|
||||
if ch["value"] is None:
|
||||
continue
|
||||
value = ch["value"] / scale
|
||||
if best is None or value > best:
|
||||
best = value
|
||||
return best
|
||||
|
||||
|
||||
CPU_CHIPS = ["k10temp", "coretemp", "zenpower", "cpu_thermal"]
|
||||
GPU_CHIPS = ["amdgpu", "nouveau", "i915"]
|
||||
|
||||
|
||||
def cpu_temp(cfg):
|
||||
scale = 1000.0
|
||||
explicit = channel_value(cfg.get("temp_channel"), scale)
|
||||
if explicit is not None:
|
||||
return explicit
|
||||
return (hwmon_max(CPU_CHIPS, "temp*_input", scale)
|
||||
or hwmon_max(CPU_CHIPS, "temp1_input", scale))
|
||||
|
||||
|
||||
def cpu_fan(cfg):
|
||||
explicit = channel_value(cfg.get("fan_channel"))
|
||||
if explicit is not None:
|
||||
return explicit
|
||||
fans = [c["value"] for c in hwmon_channels("fan*_input")
|
||||
if c["value"] and c["chip"] not in GPU_CHIPS]
|
||||
return max(fans) if fans else 0
|
||||
|
||||
|
||||
def meminfo(key):
|
||||
try:
|
||||
with open("/proc/meminfo") as fh:
|
||||
for line in fh:
|
||||
if line.startswith(key):
|
||||
return float(line.split()[1]) # kB
|
||||
except OSError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def stat_snapshot():
|
||||
with open("/proc/stat") as fh:
|
||||
p = [int(x) for x in fh.readline().split()[1:]]
|
||||
return p[3] + p[4], sum(p)
|
||||
|
||||
|
||||
def load_pct(prev, now):
|
||||
idle, total = prev
|
||||
idle2, total2 = now
|
||||
return None if total2 == total else round(100 * (1 - (idle2 - idle) / (total2 - total)))
|
||||
|
||||
|
||||
def read_sensors(prev=None, cfg=None):
|
||||
cfg = cfg or {}
|
||||
total_kb, avail_kb = meminfo("MemTotal"), meminfo("MemAvailable")
|
||||
try:
|
||||
st = os.statvfs(cfg.get("disk_path") or "/")
|
||||
disk_free = round(100 * st.f_bavail / st.f_blocks) if st.f_blocks else None
|
||||
disk_gb = st.f_blocks * st.f_frsize / 2**30
|
||||
except OSError:
|
||||
disk_free = disk_gb = None
|
||||
return {
|
||||
"cpu_temp": cpu_temp(cfg), # Celsius; temp_byte handles the unit
|
||||
"gpu_temp": hwmon_max(GPU_CHIPS, "temp*_input", 1000.0),
|
||||
"cpu_load": None if prev is None else load_pct(prev, stat_snapshot()),
|
||||
"cpu_clock": hwmon_read("/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq", 1000.0),
|
||||
"cpu_fan": cpu_fan(cfg),
|
||||
"cpu_power": hwmon_max(CPU_CHIPS, "power*_input", 1e6),
|
||||
"gpu_clock": (hwmon_max(GPU_CHIPS, "freq*_input", 1e6)
|
||||
or hwmon_read("/sys/class/drm/card0/device/pp_dpm_sclk")),
|
||||
"gpu_fan": hwmon_max(GPU_CHIPS, "fan*_input") or 0,
|
||||
"gpu_load": hwmon_max(GPU_CHIPS, "gpu_busy_percent"),
|
||||
"gpu_power": hwmon_max(GPU_CHIPS, "power*_input", 1e6),
|
||||
"ram_load": None if not (total_kb and avail_kb) else round(100 * (1 - avail_kb / total_kb)),
|
||||
"ram_mb": None if not total_kb else total_kb / 1024,
|
||||
"disk_free": disk_free,
|
||||
"disk_gb": disk_gb,
|
||||
}
|
||||
|
||||
|
||||
class Sampler(threading.Thread):
|
||||
"""Refresh sensor values in the background so the send cadence stays independent."""
|
||||
|
||||
def __init__(self, interval, cfg=None):
|
||||
super().__init__(daemon=True)
|
||||
self.interval = interval
|
||||
self.cfg = cfg or {}
|
||||
self.values = {}
|
||||
self.lock = threading.Lock()
|
||||
self.stop = threading.Event()
|
||||
self.prev = stat_snapshot()
|
||||
|
||||
def run(self):
|
||||
while not self.stop.is_set():
|
||||
now = stat_snapshot()
|
||||
v = read_sensors(self.prev, self.cfg)
|
||||
self.prev = now
|
||||
with self.lock:
|
||||
self.values = v
|
||||
self.stop.wait(self.interval)
|
||||
|
||||
def snapshot(self):
|
||||
with self.lock:
|
||||
return dict(self.values)
|
||||
|
||||
|
||||
def find_device():
|
||||
"""Return the hidraw node whose HID_ID reports VID:PID 5131:2007."""
|
||||
want = f"{VID:08X}:{PID:08X}"
|
||||
for uevent in sorted(glob.glob("/sys/class/hidraw/hidraw*/device/uevent")):
|
||||
try:
|
||||
text = open(uevent).read()
|
||||
except OSError:
|
||||
continue
|
||||
if want in text.upper():
|
||||
return "/dev/" + uevent.split("/")[4]
|
||||
return None
|
||||
|
||||
|
||||
def stream(a, overrides):
|
||||
"""Open the device and stream reports. Returns (sent, seconds)."""
|
||||
node = a.device or find_device()
|
||||
if not node:
|
||||
return None
|
||||
fd = os.open(node, os.O_RDWR)
|
||||
print(f"terbuka {node} ({VID:#06x}:{PID:#06x})", flush=True)
|
||||
sampler = Sampler(a.sample, {"fan_channel": a.fan_channel, "temp_channel": a.temp_channel})
|
||||
sampler.start()
|
||||
time.sleep(min(a.sample, 0.5)) # wait for the first sample
|
||||
sent, t0 = 0, time.time()
|
||||
try:
|
||||
while a.count == 0 or sent < a.count:
|
||||
v = {} if a.blank else sampler.snapshot()
|
||||
v.update(overrides)
|
||||
os.write(fd, build_report(v, a.unit, a.flag_celsius, a.vendor_layout)[1:])
|
||||
sent += 1
|
||||
time.sleep(a.interval)
|
||||
finally:
|
||||
sampler.stop.set()
|
||||
os.close(fd)
|
||||
return sent, time.time() - t0
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--dry-run", action="store_true", help="print the report instead of sending it")
|
||||
ap.add_argument("--listen", type=float, metavar="SECONDS",
|
||||
help="passively print device->host reports for this long")
|
||||
ap.add_argument("--device", help="hidraw node; default: auto-detect by VID:PID")
|
||||
ap.add_argument("--unit", choices=["c", "f"], help="unit of the values you supply")
|
||||
ap.add_argument("--interval", type=float, help="seconds between reports (device caps at ~8/s)")
|
||||
ap.add_argument("--sample", type=float, help="sensor refresh interval")
|
||||
ap.add_argument("--count", type=int, default=0, help="stop after N reports (0 = run until Ctrl-C)")
|
||||
ap.add_argument("--flag-celsius", action="store_true", default=None,
|
||||
help="emit the original app's 0x80 Celsius flag (this firmware ignores it)")
|
||||
ap.add_argument("--vendor-layout", action="store_true", default=None,
|
||||
help="original app's byte order (clock lands in the panel's RPM slot)")
|
||||
ap.add_argument("--fan-channel", help='explicit fan channel, e.g. "nct6798/fan2_input"')
|
||||
ap.add_argument("--temp-channel", help='explicit temp channel, e.g. "k10temp/temp3_input"')
|
||||
ap.add_argument("--blank", action="store_true", default=None,
|
||||
help="kirim report bernilai nol (mode 'layar kosong')")
|
||||
ap.add_argument("--wait", action="store_true",
|
||||
help="stay alive and wait for the device (plug and play, service mode)")
|
||||
ap.add_argument("--list-sensors", action="store_true",
|
||||
help="print available fan/temp channels as JSON and exit")
|
||||
ap.add_argument("--config", default=CONFIG_PATH, help="JSON config path")
|
||||
for key in VALUE_KEYS:
|
||||
ap.add_argument(f"--{key.replace('_', '-')}", type=float, default=None)
|
||||
a = ap.parse_args()
|
||||
|
||||
if a.list_sensors:
|
||||
print(json.dumps({
|
||||
"fans": hwmon_channels("fan*_input"),
|
||||
"temps": hwmon_channels("temp*_input"),
|
||||
"cpu_chips": CPU_CHIPS,
|
||||
"gpu_chips": GPU_CHIPS,
|
||||
"device": find_device(),
|
||||
"config": load_config(a.config),
|
||||
}, indent=2))
|
||||
return 0
|
||||
|
||||
print("pemeriksaan format:", self_check().hex(" "), flush=True)
|
||||
|
||||
# config provides defaults; explicit flags win
|
||||
cfg = load_config(a.config)
|
||||
for key in CONFIG_KEYS:
|
||||
if getattr(a, key, None) is None and cfg.get(key) is not None:
|
||||
setattr(a, key, cfg[key])
|
||||
a.unit = a.unit or "c"
|
||||
a.interval = 0.5 if a.interval is None else a.interval
|
||||
a.sample = 0.5 if a.sample is None else a.sample
|
||||
a.flag_celsius = bool(a.flag_celsius)
|
||||
a.vendor_layout = bool(a.vendor_layout)
|
||||
a.blank = bool(a.blank)
|
||||
|
||||
overrides = {k: getattr(a, k) for k in VALUE_KEYS if getattr(a, k) is not None}
|
||||
|
||||
if a.dry_run:
|
||||
v = {} if a.blank else read_sensors(stat_snapshot(), {"fan_channel": a.fan_channel,
|
||||
"temp_channel": a.temp_channel})
|
||||
v.update(overrides)
|
||||
print("nilai:", {k: (round(x, 1) if isinstance(x, float) else x)
|
||||
for k, x in sorted(v.items())})
|
||||
print("isi report:", build_report(v, a.unit, a.flag_celsius, a.vendor_layout).hex(" "))
|
||||
return 0
|
||||
|
||||
if a.listen:
|
||||
node = a.device or find_device()
|
||||
if not node:
|
||||
print(f"node hidraw untuk {VID:#06x}:{PID:#06x} tidak ditemukan", file=sys.stderr)
|
||||
return 1
|
||||
fd = os.open(node, os.O_RDONLY | os.O_NONBLOCK)
|
||||
print(f"mendengarkan di {node} selama {a.listen} detik")
|
||||
deadline, seen = time.time() + a.listen, 0
|
||||
try:
|
||||
while time.time() < deadline:
|
||||
if select.select([fd], [], [], 0.3)[0]:
|
||||
seen += 1
|
||||
print(f" report {seen}: {os.read(fd, 64).hex(' ')}")
|
||||
finally:
|
||||
os.close(fd)
|
||||
print(f"report diterima: {seen}")
|
||||
return 0
|
||||
|
||||
if not a.wait:
|
||||
result = stream(a, overrides)
|
||||
if result is None:
|
||||
print(f"node hidraw untuk {VID:#06x}:{PID:#06x} tidak ditemukan", file=sys.stderr)
|
||||
return 1
|
||||
sent, dt = result
|
||||
print(f"terkirim {sent} report ({sent / dt:.1f}/detik)")
|
||||
return 0
|
||||
|
||||
# plug and play: wait for the device, stream, and go back to waiting on unplug
|
||||
stop = threading.Event()
|
||||
signal.signal(signal.SIGTERM, lambda *_: stop.set())
|
||||
signal.signal(signal.SIGINT, lambda *_: stop.set())
|
||||
print(f"menunggu {VID:#06x}:{PID:#06x} (Ctrl-C atau SIGTERM untuk berhenti)", flush=True)
|
||||
while not stop.is_set():
|
||||
if not find_device():
|
||||
stop.wait(2)
|
||||
continue
|
||||
try:
|
||||
result = stream(a, overrides)
|
||||
if result:
|
||||
sent, dt = result
|
||||
print(f"perangkat terlepas setelah {sent} report ({sent / dt:.1f}/detik), menunggu lagi", flush=True)
|
||||
except OSError as exc:
|
||||
print(f"galat perangkat ({exc}), menunggu lagi", flush=True)
|
||||
stop.wait(2)
|
||||
print("berhenti")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 14 KiB |
Reference in new issue
Block a user