MrBlakeandClaude Opus 5.5
ea7ce49452
PredatorSense replacement for the Acer Predator Helios 16 (PH16-71) on Linux: - predatord: root service for profiles, fans (incl. temperature curves), keyboard RGB, lightbar and battery health; restores settings at boot and after resume, switches profiles on AC/battery changes - Qt desktop app with dashboard, fan curve editor and per-key RGB studio - Plasma 6 panel widget with quick controls - predatorctl command line tool - predator_wmi kernel module (DKMS) as a minimal WMI method bridge Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""Client for the predatord socket."""
|
|
|
|
import json
|
|
import socket
|
|
|
|
from . import SOCKET_PATH, STATE_PATH
|
|
|
|
|
|
class DaemonError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class Client:
|
|
def __init__(self, path=SOCKET_PATH, timeout=10):
|
|
self.path = path
|
|
self.timeout = timeout
|
|
self._sock = None
|
|
self._file = None
|
|
|
|
def _connect(self):
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.settimeout(self.timeout)
|
|
try:
|
|
s.connect(self.path)
|
|
except OSError as e:
|
|
s.close()
|
|
raise DaemonError("cannot reach the predatord service (%s). "
|
|
"Check 'systemctl status predatord'." % e) from e
|
|
self._sock = s
|
|
self._file = s.makefile("rwb")
|
|
|
|
def close(self):
|
|
if self._sock:
|
|
self._file.close()
|
|
self._sock.close()
|
|
self._sock = self._file = None
|
|
|
|
def call(self, cmd, **args):
|
|
for attempt in (0, 1):
|
|
if self._sock is None:
|
|
self._connect()
|
|
try:
|
|
self._file.write((json.dumps({"cmd": cmd, "args": args}) + "\n").encode())
|
|
self._file.flush()
|
|
line = self._file.readline()
|
|
if not line:
|
|
raise OSError("connection closed")
|
|
break
|
|
except OSError:
|
|
self.close()
|
|
if attempt:
|
|
raise DaemonError("lost connection to the service")
|
|
res = json.loads(line)
|
|
if not res.get("ok"):
|
|
raise DaemonError(res.get("error", "unknown error"))
|
|
return res.get("result")
|
|
|
|
|
|
def read_state():
|
|
"""Read the latest state snapshot (no socket needed)."""
|
|
try:
|
|
with open(STATE_PATH) as f:
|
|
return json.load(f)
|
|
except (OSError, ValueError):
|
|
return None
|