From 8789d2d32279dd09c59d0e6311a19985fe3083aa Mon Sep 17 00:00:00 2001 From: zach Date: Sun, 12 Jul 2026 17:47:25 +0200 Subject: [PATCH 01/14] initial addition of battery display --- Helpers/Battery.qml | 47 +++-- Modules/SysTray/Common/BatteryIcon.qml | 84 ++++++++ Modules/SysTray/Popouts/UPowerPopout.qml | 27 +++ Modules/SysTray/Widgets/UPowerWidget.qml | 93 ++------- cli/src/zshell/__init__.py | 10 +- cli/src/zshell/subcommands/battery.py | 233 +++++++++++++++++++++++ 6 files changed, 405 insertions(+), 89 deletions(-) create mode 100644 Modules/SysTray/Common/BatteryIcon.qml create mode 100644 cli/src/zshell/subcommands/battery.py diff --git a/Helpers/Battery.qml b/Helpers/Battery.qml index 1ffb61f..21ac558 100644 --- a/Helpers/Battery.qml +++ b/Helpers/Battery.qml @@ -1,34 +1,57 @@ pragma Singleton import Quickshell +import Quickshell.Io import Quickshell.Services.UPower import qs.Config +import qs.Paths +import qs.Helpers Singleton { id: root - readonly property var colors: { - if (deviceState === UPowerDeviceState.Charging || deviceState === UPowerDeviceState.FullyCharged) + 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 bool onBattery: UPower.onBattery + readonly property bool ready: UPower.displayDevice.ready + readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty + readonly property real timeToFull: UPower.displayDevice.timeToFull + readonly property list upowerDevices: UPower.devices.values + + function getColors(percentage: real, state: string): var { + if (state === "charging" || state === "full" || state === "recharging") return { fg: DynamicColors.swapRG(DynamicColors.palette.m3error), bg: DynamicColors.swapRG(DynamicColors.palette.m3onError) }; - else if (currentPerc <= 0.2) + else if (percentage <= 0.2) return { fg: DynamicColors.palette.m3error, bg: DynamicColors.palette.m3onError }; else return { - fg: DynamicColors.palette.m3onSurface, - bg: DynamicColors.palette.m3surface + fg: DynamicColors.palette.m3tertiary, + bg: DynamicColors.palette.m3onTertiary }; } - readonly property real currentPerc: UPower.displayDevice.percentage - readonly property var deviceState: UPower.displayDevice.state - readonly property bool isLaptop: UPower.displayDevice.isLaptopBattery - readonly property bool onBattery: UPower.onBattery - readonly property bool ready: UPower.displayDevice.ready - readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty - readonly property real timeToFull: UPower.displayDevice.timeToFull + + FileView { + id: fileView + + path: `${Paths.cache}/battery.json` + watchChanges: true + + onFileChanged: reload() + + JsonAdapter { + id: adapter + + property list devices: [] + property real updated: 0.0 + } + } } diff --git a/Modules/SysTray/Common/BatteryIcon.qml b/Modules/SysTray/Common/BatteryIcon.qml new file mode 100644 index 0000000..24c53b7 --- /dev/null +++ b/Modules/SysTray/Common/BatteryIcon.qml @@ -0,0 +1,84 @@ +import QtQuick +import qs.Config +import qs.Components +import qs.Helpers + +Row { + id: root + + property real batHeight: Appearance.padding.smaller * 2 + property real batWidth: Appearance.padding.larger * 2 + required property string devState + property real nubHeight: Appearance.padding.small + property real nubWidth: 2 + required property real percentage + property real radius: Appearance.rounding.smallest / 2 + + spacing: 1 + + CustomRect { + id: track + + anchors.verticalCenter: parent.verticalCenter + color: Battery.getColors(root.percentage, root.devState).bg + height: root.batHeight + radius: root.radius + width: root.batWidth + + CustomText { + color: Battery.getColors(root.percentage, root.devState).fg + font.pointSize: Appearance.font.size.larger / 1.5 + font.weight: 800 + height: track.height + horizontalAlignment: Text.AlignHCenter + text: Math.round(root.percentage * 100) + verticalAlignment: Text.AlignVCenter + width: track.width + } + + Item { + clip: true + width: parent.width * root.percentage + + anchors { + bottom: parent.bottom + left: parent.left + top: parent.top + } + + CustomRect { + id: fill + + color: Battery.getColors(root.percentage, root.devState).fg + height: track.height + radius: track.radius + width: track.width + + CustomText { + id: batteryLabel + + clip: true + color: Battery.getColors(root.percentage, root.devState).bg + font.pointSize: Appearance.font.size.larger / 1.5 + font.weight: 800 + height: track.height + horizontalAlignment: Text.AlignHCenter + text: Math.round(root.percentage * 100) + verticalAlignment: Text.AlignVCenter + width: track.width + } + } + } + } + + CustomRect { + id: nub + + anchors.verticalCenter: parent.verticalCenter + bottomRightRadius: 20 + color: root.percentage < 0.99 ? track.color : fill.color + height: root.nubHeight + topRightRadius: 20 + width: root.nubWidth + } +} diff --git a/Modules/SysTray/Popouts/UPowerPopout.qml b/Modules/SysTray/Popouts/UPowerPopout.qml index 4ed3c7c..2ce8187 100644 --- a/Modules/SysTray/Popouts/UPowerPopout.qml +++ b/Modules/SysTray/Popouts/UPowerPopout.qml @@ -1,7 +1,9 @@ pragma ComponentBehavior: Bound import QtQuick +import QtQuick.Layouts import Quickshell.Services.UPower +import qs.Modules.SysTray.Common import qs.Config import qs.Helpers import qs.Components @@ -21,6 +23,31 @@ Column { } } + Repeater { + model: [...Battery.logiDevices] // ...Battery.upowerDevices] + + RowLayout { + id: layout + + required property var modelData + + anchors.left: parent.left + anchors.leftMargin: Appearance.padding.normal + anchors.right: parent.right + anchors.rightMargin: Appearance.padding.normal + + CustomText { + Layout.fillWidth: true + text: layout.modelData.name + } + + BatteryIcon { + devState: layout.modelData.state + percentage: layout.modelData.percentage + } + } + } + CustomText { function formatSeconds(s: int, fallback: string): string { const day = Math.floor(s / 86400); diff --git a/Modules/SysTray/Widgets/UPowerWidget.qml b/Modules/SysTray/Widgets/UPowerWidget.qml index 0c4b4c8..8f45c6c 100644 --- a/Modules/SysTray/Widgets/UPowerWidget.qml +++ b/Modules/SysTray/Widgets/UPowerWidget.qml @@ -1,6 +1,7 @@ import Quickshell.Services.UPower import QtQuick import QtQuick.Layouts +import qs.Modules.SysTray.Common import qs.Components import qs.Config import qs.Helpers @@ -8,12 +9,13 @@ import qs.Helpers Item { id: root + readonly property Item currentItem: batteryIconLoader?.item ?? peripheralIconLoader?.item ?? upowerIconLoader.item readonly property bool shouldBeActive: Config.bar.tray.showPower Layout.preferredHeight: shouldBeActive ? implicitHeight : 0 Layout.preferredWidth: shouldBeActive ? implicitWidth : 0 - implicitHeight: Battery.isLaptop ? batteryIconLoader.item.implicitHeight : upowerIconLoader.item.implicitHeight - implicitWidth: Battery.isLaptop ? batteryIconLoader.item.implicitWidth : upowerIconLoader.item.implicitWidth + implicitHeight: currentItem?.implicitHeight ?? 0 + implicitWidth: currentItem?.implicitWidth ?? 0 opacity: shouldBeActive ? 1 : 0 scale: shouldBeActive ? 1 : 0 visible: opacity > 0 @@ -41,89 +43,28 @@ Item { active: Battery.isLaptop anchors.centerIn: parent - sourceComponent: Row { - id: batteryIcon + sourceComponent: BatteryIcon { + devState: Battery.deviceStateString + percentage: Battery.currentPerc + } + } - property real batHeight: 16 - property real batWidth: 30 - property real nubHeight: 6 - property real nubWidth: 2 - property real radius: Appearance.rounding.smallest / 2 + Loader { + id: peripheralIconLoader - spacing: 1 + active: Battery.logiDevices.some(d => d.isPresent) && !batteryIconLoader.active + anchors.centerIn: parent - CustomRect { - id: track - - anchors.verticalCenter: parent.verticalCenter - color: Battery.colors.bg - height: batteryIcon.batHeight - radius: batteryIcon.radius - width: batteryIcon.batWidth - - CustomText { - color: Battery.colors.fg - font.pointSize: Appearance.font.size.larger / 1.5 - font.weight: 800 - height: track.height - horizontalAlignment: Text.AlignHCenter - text: Math.round(Battery.currentPerc * 100) - verticalAlignment: Text.AlignVCenter - width: track.width - } - - Item { - clip: true - width: parent.width * Battery.currentPerc - - anchors { - bottom: parent.bottom - left: parent.left - top: parent.top - } - - CustomRect { - id: fill - - color: Battery.colors.fg - height: track.height - radius: track.radius - width: track.width - - CustomText { - id: batteryLabel - - clip: true - color: Battery.colors.bg - font.pointSize: 7.5 - font.weight: 800 - height: track.height - horizontalAlignment: Text.AlignHCenter - text: Math.round(Battery.currentPerc * 100) - verticalAlignment: Text.AlignVCenter - width: track.width - } - } - } - } - - CustomRect { - id: nub - - anchors.verticalCenter: parent.verticalCenter - bottomRightRadius: 20 - color: Battery.currentPerc < 0.99 ? track.color : fill.color - height: batteryIcon.nubHeight - topRightRadius: 20 - width: batteryIcon.nubWidth - } + sourceComponent: BatteryIcon { + devState: Battery.logiDevices[0].state + percentage: Battery.logiDevices[0].percentage } } Loader { id: upowerIconLoader - active: !Battery.isLaptop + active: !batteryIconLoader.active && !peripheralIconLoader.active anchors.centerIn: parent sourceComponent: MaterialIcon { diff --git a/cli/src/zshell/__init__.py b/cli/src/zshell/__init__.py index 613ee3e..17e2ac5 100644 --- a/cli/src/zshell/__init__.py +++ b/cli/src/zshell/__init__.py @@ -8,7 +8,14 @@ import typer from typer._completion_classes import completion_init from typer._completion_shared import _get_shell_name, install -from zshell.subcommands import record, scheme, screenshot, shell, wallpaper +from zshell.subcommands import ( + record, + scheme, + screenshot, + shell, + wallpaper, + battery, +) app = typer.Typer(name="zshell-cli", add_completion=False) @@ -17,6 +24,7 @@ app.add_typer(scheme.app, name="scheme") app.add_typer(screenshot.app, name="screenshot") app.add_typer(wallpaper.app, name="wallpaper") app.add_typer(record.app, name="record") +app.add_typer(battery.app, name="battery") def _completion_installed() -> bool: diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py new file mode 100644 index 0000000..ddde1cb --- /dev/null +++ b/cli/src/zshell/subcommands/battery.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import contextlib +import json +import logging +import os +import signal +import tempfile +import time +from dataclasses import asdict, dataclass +from pathlib import Path + +import typer + +app = typer.Typer( + help="Read live battery status from Logitech HID++ peripherals." +) + +logger = logging.getLogger("zshell.battery") + +DEFAULT_OUTPUT = Path.home() / ".cache" / "zshell" / "battery.json" + + +@dataclass +class DeviceBattery: + name: str + nativePath: str | None + type: str | None + wired: bool + percentage: float | None + state: str | None + charging: bool + isPresent: bool + + +def _clean_json(path: Path) -> None: + _write_json_atomic(path, []) + + +def _iter_open_devices(): + try: + from logitech_receiver import base, device, receiver + except ImportError as e: + raise RuntimeError( + "logitech_receiver isn't importable -- install it with `pip install solaar`." + ) from e + except ValueError as e: + 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 " + "(e.g. python3-gi + gir1.2-gtk-3.0 on Debian/Ubuntu, or the gtk3/" + "python3-gobject equivalents on your distro)." + ) from e + + for dev_info in base.receivers_and_devices(): + try: + if dev_info.isDevice: + d = device.create_device(base, dev_info) + else: + d = receiver.create_receiver(base, dev_info) + except OSError as e: + if e.errno == 13: + logger.error( + "permission denied opening %s -- check the Solaar udev rule " + "(rules.d/42-logitech-unify-permissions.rules) is installed " + "and you're in the right group", + dev_info.path, + ) + else: + logger.exception("failed opening %s", dev_info) + continue + except Exception: + logger.exception("failed opening %s", dev_info) + continue + + if d is None: + continue + + if d.isDevice: + yield d + else: + yield from d + + +def _read_battery(dev) -> DeviceBattery | None: + try: + if not dev.ping(): + return None + except Exception: + logger.debug("ping failed for %s", dev, exc_info=True) + return None + + try: + battery = dev.battery() + except Exception: + logger.debug("battery() failed for %s", dev, exc_info=True) + return None + + if battery is 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 + ) + + return DeviceBattery( + name=dev.name or dev.codename or "Unknown device", + nativePath=getattr(dev, "serial", None), + type=str(dev.kind) if dev.kind is not None else None, + wired=dev.receiver is None, + percentage=percentage, + state=status, + 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: + info = _read_battery(dev) + if info is not None: + results.append(info) + finally: + with contextlib.suppress(Exception): + dev.close() + return results + + +def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "updated": time.time(), + "devices": [asdict(r) for r in results], + } + fd, tmp_path = tempfile.mkstemp( + dir=path.parent, prefix=".battery-", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w") as f: + json.dump(payload, f, indent=2) + os.replace(tmp_path, path) # atomic rename on the same filesystem + except Exception: + Path(tmp_path).unlink(missing_ok=True) + 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( + 15.0, "--interval", "-i", help="Seconds between polls." + ), + out: Path = typer.Option( + DEFAULT_OUTPUT, + "--out", + "-o", + help="Where to write the JSON status file.", + ), +): + running = True + + def _stop(signum, frame): + nonlocal running + running = False + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + typer.echo(f"Polling every {interval}s, writing to {out}") + try: + while running: + try: + results = poll_once() + _write_json_atomic(out, results) + except RuntimeError as e: + typer.echo(f"error: {e}", err=True) + raise typer.Exit(code=1) from None + except Exception: + logger.exception("poll cycle failed") + + remaining = interval + while running and remaining > 0: + step = min(0.2, remaining) + time.sleep(step) + remaining -= step + finally: + with contextlib.suppress(Exception): + _clean_json(out) + + typer.echo("Stopped.") + + +if __name__ == "__main__": + app() -- 2.47.3 From 4ee9bb3f12bd421fdf6b7bf6c99dd0d8702d130c Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 00:48:05 +0200 Subject: [PATCH 02/14] 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__": -- 2.47.3 From d6230bef34eaa2d6643599d89e950eb23135e7a8 Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 01:12:55 +0200 Subject: [PATCH 03/14] Fix cmake nuitka build command --- CMakeLists.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 88f0e3c..6baef6c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -93,6 +93,7 @@ if("shell" IN_LIST ENABLE_MODULES) ${NUITKA_EXECUTABLE} --standalone --include-data-dir=${CMAKE_SOURCE_DIR}/cli/src/zshell/assets=zshell/assets + --include-package-data=solaar --output-dir=${ZSHELL_CLI_BUILD_DIR} --output-filename=zshell-cli ${CMAKE_SOURCE_DIR}/cli/src/zshell/ -- 2.47.3 From 259a4dde7a0635000253481b085b698155658a97 Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 01:13:57 +0200 Subject: [PATCH 04/14] gitignore --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index c18220d..de6958c 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,5 @@ dist/ **/test-plugins/ **/Charts/ network-dev/ +**/zshell.build/ +**/zshell.dist/ -- 2.47.3 From 85320e9ea1a8c4d41b9d7505d3e10c67335077dc Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 01:33:55 +0200 Subject: [PATCH 05/14] prevent writing battery json when there has been no change, filter out devices with unknown state --- Helpers/Battery.qml | 2 +- cli/src/zshell/subcommands/battery.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/Helpers/Battery.qml b/Helpers/Battery.qml index 1f6f8d4..a154297 100644 --- a/Helpers/Battery.qml +++ b/Helpers/Battery.qml @@ -21,7 +21,7 @@ Singleton { readonly property bool ready: UPower.displayDevice.ready readonly property real timeToEmpty: UPower.displayDevice.timeToEmpty readonly property real timeToFull: UPower.displayDevice.timeToFull - readonly property list upowerDevices: UPower.devices.values + readonly property list upowerDevices: UPower.devices.values.filter(d => d.state !== UPowerDeviceState.Unknown) function getColors(percentage: real, state: string): var { if (state === "charging" || state === "full" || state === "recharging") diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index 60d8839..0450369 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -229,11 +229,15 @@ def daemon( typer.echo(f"Polling every {interval}s, writing to {out}") + last_payload = None try: while running: try: results = poll_once() - _write_json_atomic(out, results) + payload = [asdict(r) for r in results] + if payload != last_payload: + _write_json_atomic(out, results) + last_payload = payload except RuntimeError as e: typer.echo(f"error: {e}", err=True) raise typer.Exit(code=1) from None -- 2.47.3 From 620fe001a46a59bde3c5685aeb818020313921b1 Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 14:56:57 +0200 Subject: [PATCH 06/14] backend now more robust, threaded writes use locks --- cli/src/zshell/subcommands/battery.py | 527 +++++++++++++++++++++++--- 1 file changed, 479 insertions(+), 48 deletions(-) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index 0450369..dd189b7 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -7,23 +7,32 @@ import logging import os import re import signal +import struct import tempfile +import threading import time +from collections.abc import Callable from dataclasses import asdict, dataclass from pathlib import Path import typer +from logitech_receiver import hidpp10, hidpp20 +from logitech_receiver.common import Battery, BatteryStatus, Notification +from logitech_receiver.hidpp10_constants import Registers +from logitech_receiver.hidpp20_constants import SupportedFeature app = typer.Typer( help="Read live battery status from Logitech HID++ peripherals." ) 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" +DEFAULT_LOCK = _RUNTIME_DIR / "zshell-battery-daemon.lock" def _acquire_singleton_lock(path: Path) -> int | None: @@ -76,6 +85,350 @@ def _find_upower_native_path(serial: str | None) -> str | None: return None +def _process_battery_notification(dev, n) -> bool: + + if not getattr(dev, "isDevice", False): + return False + + if int(n.sub_id) & 0x80: + return False + if n.sub_id == Notification.NO_OPERATION: + return False + + if getattr(dev, "protocol", None) is not None and dev.protocol < 2.0: + if n.sub_id in (Registers.BATTERY_STATUS, Registers.BATTERY_CHARGE): + if n.data[-1:] != b"\x00": + return False + data = bytes([n.address]) + n.data + dev.set_battery_info(hidpp10.parse_battery_status(n.sub_id, data)) + return True + return False + + features = getattr(dev, "features", None) + if not features: + return False + + try: + feature = features.get_feature(n.sub_id) + except Exception: + return False + + if feature == SupportedFeature.BATTERY_STATUS: + if n.address == 0x00: + dev.set_battery_info(hidpp20.decipher_battery_status(n.data)[1]) + return True + return False + + if feature == SupportedFeature.BATTERY_VOLTAGE: + if n.address == 0x00: + dev.set_battery_info(hidpp20.decipher_battery_voltage(n.data)[1]) + return True + return False + + if feature == SupportedFeature.UNIFIED_BATTERY: + if n.address == 0x00: + dev.set_battery_info(hidpp20.decipher_battery_unified(n.data)[1]) + return True + return False + + if feature == SupportedFeature.ADC_MEASUREMENT: + if n.address == 0x00: + result = hidpp20.decipher_adc_measurement(n.data) + if result: + dev.set_battery_info(result[1]) + return True + return False + + if feature == SupportedFeature.CENTURION_BATTERY_SOC: + dev.set_battery_info(hidpp20.decipher_battery_centurion(n.data)[1]) + return True + + if feature == SupportedFeature.SOLAR_DASHBOARD: + if n.data[5:9] == b"GOOD": + charge, lux, adc = struct.unpack("!BHH", n.data[:5]) + status = BatteryStatus.DISCHARGING + if n.address == 0x10 and lux > 200: + status = BatteryStatus.RECHARGING + dev.set_battery_info(Battery(charge, None, status, None, lux)) + return True + return False + + return False + + +def _import_listener_deps(): + try: + from logitech_receiver import base, device, receiver + from logitech_receiver.listener import EventsListener + except ImportError as e: + raise RuntimeError( + "logitech_receiver isn't importable -- install it with `pip install solaar`." + ) from e + except ValueError as e: + 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 " + "(e.g. python3-gi + gir1.2-gtk-3.0 on Debian/Ubuntu, or the gtk3/" + "python3-gobject equivalents on your distro)." + ) from e + + class _BatteryEventsListener(EventsListener): + def __init__(self, receiver_or_device, on_change): + super().__init__(receiver_or_device, self._handle_notification) + self._on_change = on_change + + def has_started(self): + + if not self.receiver.isDevice: + try: + self.receiver.notification_flags = ( + self.receiver.enable_connection_notifications() + ) + self.receiver.notify_devices() + except Exception: + logger.exception( + "failed enabling notifications for %s", self.receiver + ) + + def _handle_notification(self, n): + if self.receiver.isDevice: + try: + _process_battery_notification(self.receiver, n) + except Exception: + logger.exception("processing %s for %s", n, self.receiver) + self._on_change(self.receiver) + return + + if n.devnumber == 0xFF: + try: + _process_battery_notification(self.receiver, n) + except Exception: + logger.exception("processing receiver notification %s", n) + self._on_change(self.receiver) + return + + if not (0 < n.devnumber <= 16): + logger.warning( + "unexpected device number %s in %s", n.devnumber, n + ) + return + + try: + dev = self.receiver[n.devnumber] + except Exception: + logger.exception( + "resolving device number %s on %s", + n.devnumber, + self.receiver, + ) + return + + if not dev: + logger.warning( + "%s: received %s for invalid device %d", + self.receiver, + n, + n.devnumber, + ) + return + + try: + _process_battery_notification(dev, n) + except Exception: + logger.exception("processing %s for %s", n, dev) + + self._on_change(dev) + + return base, device, receiver, _BatteryEventsListener + + +def _start_hotplug_watcher(on_hotplug: Callable[[], None]): + try: + import pyudev + except ImportError: + logger.warning( + "pyudev isn't importable -- hotplugged devices will only be " + "noticed on the next --rescan-interval tick, not instantly. " + "Install it with `pip install pyudev` (it's a core Solaar " + "dependency, so it's normally already present)." + ) + return None + + debounce_lock = threading.Lock() + debounce_state = {"timer": None} + DEBOUNCE_SECONDS = 0.5 + + def _fire(): + with debounce_lock: + debounce_state["timer"] = None + on_hotplug() + + def _handle_event(_device): + if _device.action not in ("add", "remove"): + return + with debounce_lock: + existing = debounce_state["timer"] + if existing is not None: + existing.cancel() + timer = threading.Timer(DEBOUNCE_SECONDS, _fire) + timer.daemon = True + debounce_state["timer"] = timer + timer.start() + + context = pyudev.Context() + monitor = pyudev.Monitor.from_netlink(context) + monitor.filter_by(subsystem="hidraw") + observer = pyudev.MonitorObserver(monitor, callback=_handle_event) + observer.start() + return observer + + +class _ListenerRegistry: + def __init__(self, on_change): + self._on_change = on_change + self._entries = {} + self._last_change: dict[int, float] = {} + self._lock = threading.Lock() + self._rescan_lock = threading.Lock() + self._min_rescan_interval = 1.0 + self._last_rescan_time = 0.0 + self._hidpp_io_lock = threading.RLock() + + def _touch_and_notify(self, dev_or_receiver, *args, **kwargs): + with self._lock: + self._last_change[id(dev_or_receiver)] = time.time() + self._on_change(dev_or_receiver, *args, **kwargs) + + def last_change(self, dev) -> float: + with self._lock: + return self._last_change.get(id(dev), 0.0) + + def rescan(self): + now = time.monotonic() + if now - self._last_rescan_time < self._min_rescan_interval: + return + if not self._rescan_lock.acquire(blocking=False): + return + try: + self._last_rescan_time = now + with self._hidpp_io_lock: + base, device, receiver, ListenerClass = _import_listener_deps() + + with self._lock: + dead = [ + p + for p, (_obj, listener) in self._entries.items() + if not listener._active + ] + for p in dead: + del self._entries[p] + known_paths = set(self._entries.keys()) + + for dev_info in base.receivers_and_devices(): + if dev_info.path in known_paths: + continue + try: + if dev_info.isDevice: + obj = device.create_device(base, dev_info) + else: + obj = receiver.create_receiver(base, dev_info) + except OSError as e: + if e.errno == 13: + logger.error( + "permission denied opening %s -- check the Solaar udev rule " + "(rules.d/42-logitech-unify-permissions.rules) is installed", + dev_info.path, + ) + else: + logger.exception("failed opening %s", dev_info) + continue + except Exception: + logger.exception("failed opening %s", dev_info) + continue + + if obj is None: + continue + + listener = ListenerClass(obj, self._touch_and_notify) + listener.start() + + with self._lock: + self._entries[dev_info.path] = (obj, listener) + self._last_change[id(obj)] = time.time() + + logger.info( + "listening on %s (%s)", + dev_info.path, + "device" if dev_info.isDevice else "receiver", + ) + finally: + self._rescan_lock.release() + + def known_devices(self): + with self._lock: + entries = list(self._entries.values()) + + result = [] + for obj, _listener in entries: + if obj.isDevice: + result.append(obj) + else: + result.extend(list(obj)) + return result + + def stop(self): + with self._lock: + entries = list(self._entries.values()) + for _obj, listener in entries: + try: + listener.stop() + except Exception: + logger.exception("error stopping listener") + for _obj, listener in entries: + listener.join(timeout=2.0) + + +def _dedupe_by_identity( + pairs: list[tuple[object, DeviceBattery]], + freshness: Callable[[object], float], +) -> list[DeviceBattery]: + passthrough: list[DeviceBattery] = [] + best: dict[str, tuple[float, DeviceBattery]] = {} + + for dev, entry in pairs: + key = entry.serial + if not key: + passthrough.append(entry) + continue + ts = freshness(dev) + current = best.get(key) + if current is None or ts >= current[0]: + best[key] = (ts, entry) + + return passthrough + [entry for _ts, entry in best.values()] + + +def _snapshot(registry: _ListenerRegistry) -> list[DeviceBattery]: + pairs: list[tuple[object, DeviceBattery]] = [] + + with registry._hidpp_io_lock: + for dev in registry.known_devices(): + info = dev.battery_info + if info is None: + try: + if dev.ping(): + info = dev.battery() + except Exception: + logger.debug( + "initial battery() failed for %s", dev, exc_info=True + ) + info = None + entry = _to_device_battery(dev, info) + if entry is not None: + pairs.append((dev, entry)) + return _dedupe_by_identity(pairs, registry.last_change) + + def _clean_json(path: Path) -> None: _write_json_atomic(path, []) @@ -88,7 +441,6 @@ 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 " @@ -103,7 +455,7 @@ def _iter_open_devices(): else: d = receiver.create_receiver(base, dev_info) except OSError as e: - if e.errno == 13: # EACCES + if e.errno == 13: logger.error( "permission denied opening %s -- check the Solaar udev rule " "(rules.d/42-logitech-unify-permissions.rules) is installed " @@ -123,7 +475,31 @@ def _iter_open_devices(): if d.isDevice: yield d else: - yield from d # a receiver: walk its currently paired devices + yield from d + + +def _to_device_battery(dev, battery) -> DeviceBattery | None: + if battery is None: + return None + + percentage = battery.level / 100 if isinstance(battery.level, int) else None + state = battery.status.name.lower() if battery.status is not None else None + + serial = ( + getattr(dev, "serial", None) or getattr(dev, "unitId", None) or None + ) + + return DeviceBattery( + 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=state, + charging=battery.charging(), + isPresent=battery.ok(), + ) def _read_battery(dev) -> DeviceBattery | None: @@ -140,37 +516,23 @@ def _read_battery(dev) -> DeviceBattery | None: logger.debug("battery() failed for %s", dev, exc_info=True) return None - if battery is None: - return None - - percentage = battery.level / 100 if isinstance(battery.level, int) else None - state = battery.status.name.lower() if battery.status is not None else None - serial = getattr(dev, "serial", None) - - return DeviceBattery( - 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=state, - charging=battery.charging(), - isPresent=battery.ok(), - ) + return _to_device_battery(dev, battery) def poll_once() -> list[DeviceBattery]: - results = [] + pairs: list[tuple[object, DeviceBattery]] = [] for dev in _iter_open_devices(): try: info = _read_battery(dev) if info is not None: - results.append(info) + pairs.append((dev, info)) finally: with contextlib.suppress(Exception): dev.close() - return results + + return _dedupe_by_identity( + pairs, lambda dev: 1.0 if dev.receiver is None else 0.0 + ) def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: @@ -185,7 +547,7 @@ def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: try: with os.fdopen(fd, "w") as f: json.dump(payload, f, indent=2) - os.replace(tmp_path, path) # atomic rename on the same filesystem + os.replace(tmp_path, path) except Exception: Path(tmp_path).unlink(missing_ok=True) raise @@ -193,8 +555,18 @@ def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: @app.command() def daemon( - interval: float = typer.Option( - 15.0, "--interval", "-i", help="Seconds between polls." + rescan_interval: float = typer.Option( + 60.0, + "--rescan-interval", + "-i", + help=( + "Seconds between fallback rescans. This is NOT a battery poll " + "interval -- battery updates are event-driven and written as " + "soon as a device reports a change. New/removed devices " + "(e.g. plugging a keyboard in to charge) are normally noticed " + "within milliseconds via a udev hotplug watcher; this interval " + "is only a safety net in case a udev event is ever missed." + ), ), out: Path = typer.Option( DEFAULT_OUTPUT, @@ -202,22 +574,86 @@ 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_file: Path = typer.Option( + DEFAULT_LOCK, + "--lock-file", + help="Path used to ensure only one daemon runs at a time.", + ), + verbose: bool = typer.Option( + False, + "--verbose", + "-v", + help="Log every listener start and battery-change write to stderr.", ), ): - lock_fd = _acquire_singleton_lock(lock) + lock_fd = _acquire_singleton_lock(lock_file) if lock_fd is None: try: - holder_pid = lock.read_text().strip() + holder_pid = lock_file.read_text().strip() except OSError: holder_pid = "unknown" typer.echo( - f"error: a battery daemon is already running (pid: {holder_pid}, lock: {lock})", + f"error: a battery daemon is already running (pid {holder_pid}, lock: {lock_file})", err=True, ) raise typer.Exit(code=1) + if verbose: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(name)s: %(message)s" + ) + + write_lock = threading.Lock() + last_payload = None + + pending_write_timer: list[threading.Timer | None] = [None] + pending_write_lock = threading.Lock() + WRITE_SETTLE_SECONDS = 0.3 + + def write_if_changed(): + nonlocal last_payload + with write_lock: + results = _snapshot(registry) + payload = [asdict(r) for r in results] + if payload != last_payload: + _write_json_atomic(out, results) + last_payload = payload + logger.info( + "battery status changed, wrote %d device(s)", len(results) + ) + + def write_if_changed_coalesced(): + with pending_write_lock: + existing = pending_write_timer[0] + if existing is not None: + existing.cancel() + timer = threading.Timer(WRITE_SETTLE_SECONDS, write_if_changed) + timer.daemon = True + pending_write_timer[0] = timer + timer.start() + + def on_change(_device_or_receiver, alert=None, reason=None): + write_if_changed_coalesced() + + registry = _ListenerRegistry(on_change) + + def rescan_and_write(): + try: + registry.rescan() + except Exception: + logger.exception("rescan failed") + write_if_changed_coalesced() + + try: + registry.rescan() + except RuntimeError as e: + typer.echo(f"error: {e}", err=True) + raise typer.Exit(code=1) from None + + write_if_changed() + + hotplug_observer = _start_hotplug_watcher(rescan_and_write) + running = True def _stop(signum, frame): @@ -227,32 +663,27 @@ def daemon( signal.signal(signal.SIGINT, _stop) signal.signal(signal.SIGTERM, _stop) - typer.echo(f"Polling every {interval}s, writing to {out}") + typer.echo( + f"Listening for battery events, writing to {out} (lock: {lock_file})" + ) - last_payload = None try: while running: - try: - results = poll_once() - payload = [asdict(r) for r in results] - if payload != last_payload: - _write_json_atomic(out, results) - last_payload = payload - except RuntimeError as e: - typer.echo(f"error: {e}", err=True) - raise typer.Exit(code=1) from None - except Exception: - logger.exception("poll cycle failed") - - remaining = interval + remaining = rescan_interval while running and remaining > 0: step = min(0.2, remaining) time.sleep(step) remaining -= step + if running: + rescan_and_write() finally: + if hotplug_observer is not None: + with contextlib.suppress(Exception): + hotplug_observer.stop() with contextlib.suppress(Exception): _clean_json(out) + registry.stop() typer.echo("Stopped.") -- 2.47.3 From 57147dc7c14ad68d9dd83b33a70633b81183e2cf Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 14:59:52 +0200 Subject: [PATCH 07/14] fix: cmake nuitka command --- CMakeLists.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 6baef6c..933c6e5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -93,7 +93,7 @@ if("shell" IN_LIST ENABLE_MODULES) ${NUITKA_EXECUTABLE} --standalone --include-data-dir=${CMAKE_SOURCE_DIR}/cli/src/zshell/assets=zshell/assets - --include-package-data=solaar + --include-package=gi.overrides --output-dir=${ZSHELL_CLI_BUILD_DIR} --output-filename=zshell-cli ${CMAKE_SOURCE_DIR}/cli/src/zshell/ -- 2.47.3 From bb50cf756d1e109f90f5cf3fab82a13fc9850fc4 Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 15:03:16 +0200 Subject: [PATCH 08/14] fix: cmake nuitka command again --- CMakeLists.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/CMakeLists.txt b/CMakeLists.txt index 933c6e5..23bbaec 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,7 @@ if("shell" IN_LIST ENABLE_MODULES) --standalone --include-data-dir=${CMAKE_SOURCE_DIR}/cli/src/zshell/assets=zshell/assets --include-package=gi.overrides + --include-package-data=solaar --output-dir=${ZSHELL_CLI_BUILD_DIR} --output-filename=zshell-cli ${CMAKE_SOURCE_DIR}/cli/src/zshell/ -- 2.47.3 From a9bbb21f54bb909c58879b23cb5447430b0a1ae5 Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 19:31:40 +0200 Subject: [PATCH 09/14] improve device detection when plugging/unplugging --- cli/src/zshell/subcommands/battery.py | 144 +++++++++++++++++++++++++- 1 file changed, 139 insertions(+), 5 deletions(-) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index dd189b7..e5f0568 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -292,6 +292,7 @@ class _ListenerRegistry: self._rescan_lock = threading.Lock() self._min_rescan_interval = 1.0 self._last_rescan_time = 0.0 + self._hidpp_io_lock = threading.RLock() def _touch_and_notify(self, dev_or_receiver, *args, **kwargs): @@ -309,6 +310,7 @@ class _ListenerRegistry: return if not self._rescan_lock.acquire(blocking=False): return + try: self._last_rescan_time = now with self._hidpp_io_lock: @@ -321,9 +323,38 @@ class _ListenerRegistry: if not listener._active ] for p in dead: + dying_obj, _dying_listener = self._entries[p] + + if dying_obj.isDevice: + ident = getattr( + dying_obj, "unitId", None + ) or getattr(dying_obj, "serial", None) + info = getattr(dying_obj, "battery_info", None) + if ident and info is not None: + for ( + other_path, + (other_obj, _other_listener), + ) in self._entries.items(): + if other_path == p or other_obj.isDevice: + continue + for child in other_obj: + child_ident = getattr( + child, "unitId", None + ) or getattr(child, "serial", None) + if child_ident == ident: + child.set_battery_info(info) + break del self._entries[p] known_paths = set(self._entries.keys()) + with self._lock: + id_to_entry = {} + for path, (obj, listener) in self._entries.items(): + if obj.isDevice: + ident = obj.unitId or obj.serial + if ident: + id_to_entry[ident] = (path, obj, listener) + for dev_info in base.receivers_and_devices(): if dev_info.path in known_paths: continue @@ -350,17 +381,108 @@ class _ListenerRegistry: continue listener = ListenerClass(obj, self._touch_and_notify) - listener.start() + + if obj.isDevice: + ident = None + try: + if obj.protocol >= 2.0: + obj.get_ids() + ident = obj.unitId or obj.serial + except Exception: + pass + + if ident: + with self._lock: + existing = id_to_entry.get(ident) + if existing: + old_path, old_obj, old_listener = existing + + if old_path != dev_info.path: + logger.info( + "replacing device %s (old path %s) with new path %s", + ident, + old_path, + dev_info.path, + ) + + if ( + getattr(obj, "battery_info", None) + is None + and getattr( + old_obj, "battery_info", None + ) + is not None + ): + obj.set_battery_info( + old_obj.battery_info + ) + + old_listener.stop() + + del self._entries[old_path] + + self._entries[dev_info.path] = ( + obj, + listener, + ) + self._last_change[id(obj)] = time.time() + + to_join = old_listener + break + else: + pass + else: + if ( + getattr(obj, "battery_info", None) + is None + ): + for ( + other_path, + (other_obj, _other_listener), + ) in self._entries.items(): + if other_obj.isDevice: + continue + for child in other_obj: + child_ident = getattr( + child, "unitId", None + ) or getattr( + child, "serial", None + ) + if child_ident == ident: + obj.set_battery_info( + child.battery_info + ) + break + if ( + getattr( + obj, + "battery_info", + None, + ) + is not None + ): + break with self._lock: - self._entries[dev_info.path] = (obj, listener) - self._last_change[id(obj)] = time.time() + if dev_info.path not in self._entries: + self._entries[dev_info.path] = (obj, listener) + self._last_change[id(obj)] = time.time() + + listener.start() logger.info( "listening on %s (%s)", dev_info.path, "device" if dev_info.isDevice else "receiver", ) + + if "to_join" in locals(): + try: + to_join.join(timeout=1.0) + except Exception: + logger.exception("error joining replaced listener") + del to_join + finally: self._rescan_lock.release() @@ -610,10 +732,22 @@ def daemon( pending_write_lock = threading.Lock() WRITE_SETTLE_SECONDS = 0.3 + write_seq_lock = threading.Lock() + write_seq = 0 + last_written_seq = 0 + def write_if_changed(): - nonlocal last_payload + nonlocal last_payload, write_seq, last_written_seq + with write_seq_lock: + write_seq += 1 + my_seq = write_seq + + results = _snapshot(registry) with write_lock: - results = _snapshot(registry) + with write_seq_lock: + if my_seq <= last_written_seq: + return + last_written_seq = my_seq payload = [asdict(r) for r in results] if payload != last_payload: _write_json_atomic(out, results) -- 2.47.3 From bf5a8b80494ed5a9416c9a5f2e1868384ad1e04a Mon Sep 17 00:00:00 2001 From: zach Date: Mon, 13 Jul 2026 22:10:41 +0200 Subject: [PATCH 10/14] simplify dedupe --- cli/src/zshell/subcommands/battery.py | 57 +++++++++++++-------------- 1 file changed, 27 insertions(+), 30 deletions(-) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index e5f0568..075ae59 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -491,11 +491,23 @@ class _ListenerRegistry: entries = list(self._entries.values()) result = [] + seen_idents = set() for obj, _listener in entries: if obj.isDevice: + ident = obj.unitId or obj.serial + if ident and ident in seen_idents: + continue + if ident: + seen_idents.add(ident) result.append(obj) else: - result.extend(list(obj)) + for child in obj: + ident = child.unitId or child.serial + if ident and ident in seen_idents: + continue + if ident: + seen_idents.add(ident) + result.append(child) return result def stop(self): @@ -510,28 +522,8 @@ class _ListenerRegistry: listener.join(timeout=2.0) -def _dedupe_by_identity( - pairs: list[tuple[object, DeviceBattery]], - freshness: Callable[[object], float], -) -> list[DeviceBattery]: - passthrough: list[DeviceBattery] = [] - best: dict[str, tuple[float, DeviceBattery]] = {} - - for dev, entry in pairs: - key = entry.serial - if not key: - passthrough.append(entry) - continue - ts = freshness(dev) - current = best.get(key) - if current is None or ts >= current[0]: - best[key] = (ts, entry) - - return passthrough + [entry for _ts, entry in best.values()] - - def _snapshot(registry: _ListenerRegistry) -> list[DeviceBattery]: - pairs: list[tuple[object, DeviceBattery]] = [] + results: list[DeviceBattery] = [] with registry._hidpp_io_lock: for dev in registry.known_devices(): @@ -547,8 +539,8 @@ def _snapshot(registry: _ListenerRegistry) -> list[DeviceBattery]: info = None entry = _to_device_battery(dev, info) if entry is not None: - pairs.append((dev, entry)) - return _dedupe_by_identity(pairs, registry.last_change) + results.append(entry) + return results def _clean_json(path: Path) -> None: @@ -642,19 +634,24 @@ def _read_battery(dev) -> DeviceBattery | None: def poll_once() -> list[DeviceBattery]: - pairs: list[tuple[object, DeviceBattery]] = [] + results: list[DeviceBattery] = [] + seen_idents = set() for dev in _iter_open_devices(): try: info = _read_battery(dev) if info is not None: - pairs.append((dev, info)) + entry = _to_device_battery(dev, info) + if entry: + ident = entry.serial or (getattr(dev, "unitId", None)) + if ident and ident in seen_idents: + continue + if ident: + seen_idents.add(ident) + results.append(entry) finally: with contextlib.suppress(Exception): dev.close() - - return _dedupe_by_identity( - pairs, lambda dev: 1.0 if dev.receiver is None else 0.0 - ) + return results def _write_json_atomic(path: Path, results: list[DeviceBattery]) -> None: -- 2.47.3 From cf907089dafc31d264724050eb39f84009b1285a Mon Sep 17 00:00:00 2001 From: AramJonghu Date: Sat, 15 Aug 2026 21:13:26 +0200 Subject: [PATCH 11/14] ci(python): rework of jobs + fix import issue --- .gitea/workflows/python.yml | 109 ++++++++---------------------------- 1 file changed, 22 insertions(+), 87 deletions(-) diff --git a/.gitea/workflows/python.yml b/.gitea/workflows/python.yml index 011ba51..06377a8 100644 --- a/.gitea/workflows/python.yml +++ b/.gitea/workflows/python.yml @@ -1,12 +1,15 @@ name: Python +env: + APT_DEPS: git python3 python3-pip python3-venv python3-gi python3-cairo python3-dbus python3-pyudev python3-psutil python3-evdev python3-yaml python3-xlib python3-pillow gir1.2-gtk-3.0 + on: pull_request: jobs: - fmt: - runs-on: alpine - container: node:26-alpine + static: + runs-on: debian + container: node:26-trixie-slim steps: - name: Checkout @@ -14,10 +17,8 @@ jobs: - name: Install tools run: | - apk add --no-cache \ - git \ - python3 \ - py3-pip + apt-get update + apt-get install -y --no-install-recommends git python3 python3-pip python3-venv python3 -m venv .venv . .venv/bin/activate pip install --no-cache-dir ruff @@ -27,32 +28,15 @@ jobs: . .venv/bin/activate ruff format --check . - lint: - runs-on: alpine - container: node:26-alpine - - steps: - - name: Checkout - uses: actions/checkout@v4 - - - name: Install tools - run: | - apk add --no-cache \ - git \ - python3 \ - py3-pip - python3 -m venv .venv - . .venv/bin/activate - pip install --no-cache-dir ruff - - name: Lint run: | . .venv/bin/activate ruff check . - test: - runs-on: alpine - container: node:26-alpine + verify: + if: always() + runs-on: debian + container: node:26-trixie-slim steps: - name: Checkout @@ -60,20 +44,16 @@ jobs: - name: Install tools run: | - apk add --no-cache \ - git \ - python3 \ - py3-pip \ - py3-pillow \ - build-base - python3 -m venv .venv + apt-get update + apt-get install -y --no-install-recommends $APT_DEPS build-essential python3-dev + python3 -m venv --system-site-packages .venv . .venv/bin/activate - pip install --no-cache-dir \ - typer \ - pillow \ - materialyoucolor \ - jinja2 \ - pytest + pip install --no-cache-dir basedpyright nuitka . + + - name: Type check + run: | + . .venv/bin/activate + basedpyright - name: Test run: | @@ -81,52 +61,7 @@ jobs: cd cli python -m pytest tests/ -v - typecheck: - runs-on: alpine - container: node:26-alpine - - steps: - - name: Checkout - uses: actions/checkout@v4 - - - name: Install tools - run: | - apk add --no-cache \ - git \ - python3 \ - py3-pip - python3 -m venv .venv - . .venv/bin/activate - pip install --no-cache-dir basedpyright typer pillow materialyoucolor jinja2 - - - name: Type check - run: | - . .venv/bin/activate - basedpyright - - buildcheck: - runs-on: alpine - container: node:26-alpine - - steps: - - name: Checkout - uses: actions/checkout@v4 - - - name: Install tools - run: | - apk add --no-cache \ - git \ - python3 \ - py3-pip \ - build-base \ - python3-dev \ - gcc \ - g++ - python3 -m venv .venv - . .venv/bin/activate - pip install --no-cache-dir nuitka - - name: Nuitka module check run: | . .venv/bin/activate - nuitka --module --include-package=zshell cli/src/zshell/ + nuitka --module --include-package=zshell cli/src/zshell/ \ No newline at end of file -- 2.47.3 From 8883d98b007e63efb883452a2b0ff854314d5bc3 Mon Sep 17 00:00:00 2001 From: AramJonghu Date: Sat, 15 Aug 2026 21:13:26 +0200 Subject: [PATCH 12/14] chore: added missing dependency to pyproject --- pyproject.toml | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9d0589d..d72ea6c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,12 +10,11 @@ dependencies = [ "typer", "pillow", "jinja2", - "materialyoucolor" + "materialyoucolor", + "solaar", + "pytest" ] -[project.optional-dependencies] -dev = ["pytest"] - [project.scripts] zshell-cli = "zshell:main" -- 2.47.3 From ba567efc22df0d074b46c2639f9949a7bad45898 Mon Sep 17 00:00:00 2001 From: zach Date: Sun, 16 Aug 2026 11:23:57 +0200 Subject: [PATCH 13/14] fix unused vars --- cli/src/zshell/subcommands/battery.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index 075ae59..15494ff 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -145,7 +145,7 @@ def _process_battery_notification(dev, n) -> bool: if feature == SupportedFeature.SOLAR_DASHBOARD: if n.data[5:9] == b"GOOD": - charge, lux, adc = struct.unpack("!BHH", n.data[:5]) + charge, lux, _ = struct.unpack("!BHH", n.data[:5]) status = BatteryStatus.DISCHARGING if n.address == 0x10 and lux > 200: status = BatteryStatus.RECHARGING @@ -437,7 +437,7 @@ class _ListenerRegistry: is None ): for ( - other_path, + _, (other_obj, _other_listener), ) in self._entries.items(): if other_obj.isDevice: -- 2.47.3 From 33c94323900c290141e048be33b2e8a921eb9503 Mon Sep 17 00:00:00 2001 From: zach Date: Sun, 16 Aug 2026 13:11:17 +0200 Subject: [PATCH 14/14] fix python errors --- cli/src/zshell/subcommands/battery.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/cli/src/zshell/subcommands/battery.py b/cli/src/zshell/subcommands/battery.py index 15494ff..676c008 100644 --- a/cli/src/zshell/subcommands/battery.py +++ b/cli/src/zshell/subcommands/battery.py @@ -255,7 +255,7 @@ def _start_hotplug_watcher(on_hotplug: Callable[[], None]): return None debounce_lock = threading.Lock() - debounce_state = {"timer": None} + debounce_state: dict[str, threading.Timer | None] = {"timer": None} DEBOUNCE_SECONDS = 0.5 def _fire(): @@ -381,6 +381,7 @@ class _ListenerRegistry: continue listener = ListenerClass(obj, self._touch_and_notify) + to_join = None if obj.isDevice: ident = None @@ -476,7 +477,7 @@ class _ListenerRegistry: "device" if dev_info.isDevice else "receiver", ) - if "to_join" in locals(): + if to_join is not None: try: to_join.join(timeout=1.0) except Exception: @@ -565,9 +566,13 @@ def _iter_open_devices(): for dev_info in base.receivers_and_devices(): try: if dev_info.isDevice: - d = device.create_device(base, dev_info) + d = device.create_device(base, dev_info) # pyright: ignore[reportArgumentType] + if d is not None: + yield d else: - d = receiver.create_receiver(base, dev_info) + d = receiver.create_receiver(base, dev_info) # pyright: ignore[reportArgumentType] + if d is not None: + yield from d except OSError as e: if e.errno == 13: logger.error( @@ -583,14 +588,6 @@ def _iter_open_devices(): logger.exception("failed opening %s", dev_info) continue - if d is None: - continue - - if d.isDevice: - yield d - else: - yield from d - def _to_device_battery(dev, battery) -> DeviceBattery | None: if battery is None: -- 2.47.3