From 4ee9bb3f12bd421fdf6b7bf6c99dd0d8702d130c Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 00:48:05 +0200 Subject: [PATCH] add support for all UPower peripherals --- Helpers/Battery.qml | 6 ++ Modules/SysTray/Popouts/UPowerPopout.qml | 4 +- Modules/SysTray/Widgets/UPowerWidget.qml | 8 +- cli/src/zshell/__init__.py | 2 +- cli/src/zshell/subcommands/battery.py | 121 ++++++++++++++--------- 5 files changed, 86 insertions(+), 55 deletions(-) diff --git a/Helpers/Battery.qml b/Helpers/Battery.qml index 21ac558..1f6f8d4 100644 --- a/Helpers/Battery.qml +++ b/Helpers/Battery.qml @@ -10,11 +10,13 @@ import qs.Helpers Singleton { id: root + readonly property list allPeripherals: [...logiDevices, ...upowerDevices.filter(d => !Battery.logiDevices.some(l => l.nativePath === d.nativePath))] readonly property real currentPerc: UPower.displayDevice.percentage readonly property var deviceState: UPower.displayDevice.state readonly property string deviceStateString: UPowerDeviceState.toString(deviceState) readonly property bool isLaptop: UPower.displayDevice.isLaptopBattery readonly property alias logiDevices: adapter.devices + readonly property var lowestPeripheral: allPeripherals.reduce((lowest, current) => current.percentage < lowest.percentage ? current : lowest) readonly property bool onBattery: UPower.onBattery readonly property bool ready: UPower.displayDevice.ready readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty @@ -39,6 +41,10 @@ Singleton { }; } + function startDaemon(): void { + Quickshell.execDetached(["zshell-cli", "battery", "daemon"]); + } + FileView { id: fileView diff --git a/Modules/SysTray/Popouts/UPowerPopout.qml b/Modules/SysTray/Popouts/UPowerPopout.qml index 2ce8187..c09f811 100644 --- a/Modules/SysTray/Popouts/UPowerPopout.qml +++ b/Modules/SysTray/Popouts/UPowerPopout.qml @@ -24,7 +24,7 @@ Column { } Repeater { - model: [...Battery.logiDevices] // ...Battery.upowerDevices] + model: Battery.allPeripherals RowLayout { id: layout @@ -38,7 +38,7 @@ Column { CustomText { Layout.fillWidth: true - text: layout.modelData.name + text: layout.modelData.model } BatteryIcon { diff --git a/Modules/SysTray/Widgets/UPowerWidget.qml b/Modules/SysTray/Widgets/UPowerWidget.qml index 8f45c6c..eccb998 100644 --- a/Modules/SysTray/Widgets/UPowerWidget.qml +++ b/Modules/SysTray/Widgets/UPowerWidget.qml @@ -37,6 +37,8 @@ Item { } } + Component.onCompleted: Battery.startDaemon() + Loader { id: batteryIconLoader @@ -52,12 +54,12 @@ Item { Loader { id: peripheralIconLoader - active: Battery.logiDevices.some(d => d.isPresent) && !batteryIconLoader.active + active: Battery.lowestPeripheral.isPresent && !batteryIconLoader.active anchors.centerIn: parent sourceComponent: BatteryIcon { - devState: Battery.logiDevices[0].state - percentage: Battery.logiDevices[0].percentage + devState: Battery.lowestPeripheral.state + percentage: Battery.lowestPeripheral.percentage } } diff --git a/cli/src/zshell/__init__.py b/cli/src/zshell/__init__.py index 17e2ac5..c7214b5 100644 --- a/cli/src/zshell/__init__.py +++ b/cli/src/zshell/__init__.py @@ -9,12 +9,12 @@ from typer._completion_classes import completion_init from typer._completion_shared import _get_shell_name, install from zshell.subcommands import ( + battery, record, scheme, screenshot, shell, wallpaper, - battery, ) app = typer.Typer(name="zshell-cli", add_completion=False) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index ddde1cb..60d8839 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -1,9 +1,11 @@ from __future__ import annotations import contextlib +import fcntl import json import logging import os +import re import signal import tempfile import time @@ -17,14 +19,31 @@ app = typer.Typer( ) logger = logging.getLogger("zshell.battery") - DEFAULT_OUTPUT = Path.home() / ".cache" / "zshell" / "battery.json" +_RUNTIME_DIR = Path( + os.environ.get("XDG_RUNTIME_DIR") or (Path.home() / ".cache" / "zshell") +) +LOCK = _RUNTIME_DIR / "zshell-battery-daemon.lock" + + +def _acquire_singleton_lock(path: Path) -> int | None: + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o644) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(fd) + return None + os.ftruncate(fd, 0) + os.write(fd, str(os.getpid()).encode()) + return fd @dataclass class DeviceBattery: - name: str + model: str nativePath: str | None + serial: str | None type: str | None wired: bool percentage: float | None @@ -33,6 +52,30 @@ class DeviceBattery: isPresent: bool +_SYSFS_POWER_SUPPLY = Path("/sys/class/power_supply") + + +def _normalize_serial(s: str) -> str: + return re.sub(r"[^0-9A-Za-z]", "", s).upper() + + +def _find_upower_native_path(serial: str | None) -> str | None: + if not serial or not _SYSFS_POWER_SUPPLY.is_dir(): + return None + target = _normalize_serial(serial) + for entry in _SYSFS_POWER_SUPPLY.glob("hidpp_battery_*"): + try: + raw = (entry / "serial_number").read_text().strip() + except OSError: + logger.debug( + "couldn't read serial_number under %s", entry, exc_info=True + ) + continue + if _normalize_serial(raw) == target: + return entry.name + return None + + def _clean_json(path: Path) -> None: _write_json_atomic(path, []) @@ -45,6 +88,7 @@ def _iter_open_devices(): "logitech_receiver isn't importable -- install it with `pip install solaar`." ) from e except ValueError as e: + # gi.require_version() raises ValueError if the GTK3 typelib isn't installed. raise RuntimeError( "logitech_receiver failed to import because a GTK3 typelib is missing " f"({e}). Install the same GTK3 + PyGObject packages Solaar's GUI needs " @@ -59,7 +103,7 @@ def _iter_open_devices(): else: d = receiver.create_receiver(base, dev_info) except OSError as e: - if e.errno == 13: + if e.errno == 13: # EACCES logger.error( "permission denied opening %s -- check the Solaar udev rule " "(rules.d/42-logitech-unify-permissions.rules) is installed " @@ -79,7 +123,7 @@ def _iter_open_devices(): if d.isDevice: yield d else: - yield from d + yield from d # a receiver: walk its currently paired devices def _read_battery(dev) -> DeviceBattery | None: @@ -100,30 +144,23 @@ def _read_battery(dev) -> DeviceBattery | None: return None percentage = battery.level / 100 if isinstance(battery.level, int) else None - status = ( - "charging" - if battery.status.name.lower() == "recharging" - else "fullycharged" - if battery.status.name.lower() == "full" - else battery.status.name.lower() - if battery.status is not None - else None - ) + state = battery.status.name.lower() if battery.status is not None else None + serial = getattr(dev, "serial", None) return DeviceBattery( - name=dev.name or dev.codename or "Unknown device", - nativePath=getattr(dev, "serial", None), + model=dev.name or dev.codename or "Unknown device", + nativePath=_find_upower_native_path(serial), + serial=serial, type=str(dev.kind) if dev.kind is not None else None, wired=dev.receiver is None, percentage=percentage, - state=status, + state=state, charging=battery.charging(), isPresent=battery.ok(), ) def poll_once() -> list[DeviceBattery]: - """Open every attached Logitech device once and read its battery. One-shot, synchronous.""" results = [] for dev in _iter_open_devices(): try: @@ -154,37 +191,6 @@ def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: raise -@app.command() -def show( - as_json: bool = typer.Option( - False, "--json", help="Print raw JSON instead of a table." - ), -): - try: - results = poll_once() - except RuntimeError as e: - typer.echo(f"error: {e}", err=True) - raise typer.Exit(code=1) from None - - if not results: - typer.echo("No Logitech HID++ devices responded.", err=True) - raise typer.Exit(code=1) - - if as_json: - typer.echo(json.dumps([asdict(r) for r in results], indent=2)) - return - - for r in results: - charge_str = ( - f"{r.percentage}%" if r.percentage is not None else "unknown" - ) - flags = [ - f for f, on in (("charging", r.charging), ("wired", r.wired)) if on - ] - flag_str = f" ({', '.join(flags)})" if flags else "" - typer.echo(f"{r.name}: {charge_str}{flag_str}") - - @app.command() def daemon( interval: float = typer.Option( @@ -196,7 +202,22 @@ def daemon( "-o", help="Where to write the JSON status file.", ), + lock: Path = typer.Option( + LOCK, "--lock-file", help="Path to daemon lock file" + ), ): + lock_fd = _acquire_singleton_lock(lock) + if lock_fd is None: + try: + holder_pid = lock.read_text().strip() + except OSError: + holder_pid = "unknown" + typer.echo( + f"error: a battery daemon is already running (pid: {holder_pid}, lock: {lock})", + err=True, + ) + raise typer.Exit(code=1) + running = True def _stop(signum, frame): @@ -205,7 +226,9 @@ def daemon( signal.signal(signal.SIGINT, _stop) signal.signal(signal.SIGTERM, _stop) + typer.echo(f"Polling every {interval}s, writing to {out}") + try: while running: try: @@ -226,7 +249,7 @@ def daemon( with contextlib.suppress(Exception): _clean_json(out) - typer.echo("Stopped.") + typer.echo("Stopped.") if __name__ == "__main__":