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 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/ diff --git a/CMakeLists.txt b/CMakeLists.txt index bcf8318..95c3714 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -93,6 +93,8 @@ if("shell" IN_LIST ENABLE_MODULES) ${NUITKA_EXECUTABLE} --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/ diff --git a/Helpers/Battery.qml b/Helpers/Battery.qml index 2df72d9..44129b1 100644 --- a/Helpers/Battery.qml +++ b/Helpers/Battery.qml @@ -1,35 +1,63 @@ pragma Singleton import Quickshell +import Quickshell.Io import Quickshell.Services.UPower import ZShell.Config import qs.Services +import qs.Paths Singleton { id: root - readonly property var colors: { - if (deviceState === UPowerDeviceState.Charging || deviceState === UPowerDeviceState.FullyCharged) + 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 + readonly property real timeToFull: UPower.displayDevice.timeToFull + 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") return { fg: Colors.swapRG(Colors.palette.m3error), bg: Colors.swapRG(Colors.palette.m3onError) }; - else if (currentPerc <= 0.2) + else if (percentage <= 0.2) return { fg: Colors.palette.m3error, bg: Colors.palette.m3onError }; else return { - fg: Colors.palette.m3onSurface, - bg: Colors.palette.m3surface + fg: Colors.palette.m3tertiary, + bg: Colors.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 + + function startDaemon(): void { + Quickshell.execDetached(["zshell-cli", "battery", "daemon"]); + } + + 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/Bar/Components/Common/BatteryIcon.qml b/Modules/Bar/Components/Common/BatteryIcon.qml new file mode 100644 index 0000000..31f4a1e --- /dev/null +++ b/Modules/Bar/Components/Common/BatteryIcon.qml @@ -0,0 +1,85 @@ +import QtQuick +import ZShell.Config +import qs.Services +import qs.Components +import qs.Helpers + +Row { + id: root + + property real batHeight: Tokens.padding.smaller * 2 + property real batWidth: Tokens.padding.larger * 2 + required property string devState + property real nubHeight: Tokens.padding.small + property real nubWidth: 2 + required property real percentage + property real radius: Tokens.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: Tokens.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: Tokens.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/Bar/Components/StatusIcons.qml b/Modules/Bar/Components/StatusIcons.qml index 66f6158..bb520e6 100644 --- a/Modules/Bar/Components/StatusIcons.qml +++ b/Modules/Bar/Components/StatusIcons.qml @@ -112,7 +112,6 @@ WidgetBase { name: "upower" UPowerWidget { - horizontal: root.horizontal } } } diff --git a/Modules/Bar/Components/StatusIcons/UPowerWidget.qml b/Modules/Bar/Components/StatusIcons/UPowerWidget.qml index 380b912..0b1ab6d 100644 --- a/Modules/Bar/Components/StatusIcons/UPowerWidget.qml +++ b/Modules/Bar/Components/StatusIcons/UPowerWidget.qml @@ -2,16 +2,17 @@ import Quickshell.Services.UPower import QtQuick import QtQuick.Layouts import ZShell.Config +import qs.Modules.Bar.Components.Common import qs.Components import qs.Helpers Item { id: root - required property bool horizontal + readonly property Item currentItem: batteryIconLoader?.item ?? peripheralIconLoader?.item ?? upowerIconLoader.item - 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 Behavior on Layout.preferredHeight { Anim { @@ -30,95 +31,36 @@ Item { } } + Component.onCompleted: Battery.startDaemon() + Loader { id: batteryIconLoader 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: Tokens.rounding.smallest / 2 + Loader { + id: peripheralIconLoader - spacing: 1 + active: (Battery.lowestPeripheral?.isPresent ?? false) && !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: Tokens.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.lowestPeripheral?.state ?? "" + percentage: Battery.lowestPeripheral?.percentage ?? 0 } } Loader { id: upowerIconLoader - active: !Battery.isLaptop + active: !batteryIconLoader.active && !peripheralIconLoader.active anchors.centerIn: parent sourceComponent: MaterialIcon { diff --git a/Modules/Bar/Popouts/UPowerPopout.qml b/Modules/Bar/Popouts/UPowerPopout.qml index 1ceec24..4aee034 100644 --- a/Modules/Bar/Popouts/UPowerPopout.qml +++ b/Modules/Bar/Popouts/UPowerPopout.qml @@ -1,8 +1,10 @@ pragma ComponentBehavior: Bound import QtQuick +import QtQuick.Layouts import Quickshell.Services.UPower import ZShell.Config +import qs.Modules.Bar.Components.Common import qs.Helpers import qs.Components import qs.Services @@ -22,6 +24,31 @@ Column { } } + Repeater { + model: Battery.allPeripherals + + RowLayout { + id: layout + + required property var modelData + + anchors.left: parent.left + anchors.leftMargin: Tokens.padding.normal + anchors.right: parent.right + anchors.rightMargin: Tokens.padding.normal + + CustomText { + Layout.fillWidth: true + text: layout.modelData.model + } + + BatteryIcon { + devState: layout.modelData.state + percentage: layout.modelData.percentage + } + } + } + CustomText { function formatSeconds(s: int, fallback: string): string { const day = Math.floor(s / 86400); @@ -75,7 +102,7 @@ Column { CustomText { anchors.verticalCenter: parent.verticalCenter color: Colors.palette.m3onError - font.family: Appearance.font.family.mono + font.family: Config.appearance.font.family.mono text: qsTr("Performance Degraded") } diff --git a/cli/src/zshell/__init__.py b/cli/src/zshell/__init__.py index 613ee3e..c7214b5 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 ( + battery, + record, + scheme, + screenshot, + shell, + wallpaper, +) 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..676c008 --- /dev/null +++ b/cli/src/zshell/subcommands/battery.py @@ -0,0 +1,819 @@ +from __future__ import annotations + +import contextlib +import fcntl +import json +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") +) +DEFAULT_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: + model: str + nativePath: str | None + serial: str | None + type: str | None + wired: bool + percentage: float | None + state: str | None + charging: bool + 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 _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, _ = 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: dict[str, threading.Timer | None] = {"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: + 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 + 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) + to_join = None + + 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_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: + 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 is not None: + try: + to_join.join(timeout=1.0) + except Exception: + logger.exception("error joining replaced listener") + del to_join + + finally: + self._rescan_lock.release() + + def known_devices(self): + with self._lock: + 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: + 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): + 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 _snapshot(registry: _ListenerRegistry) -> list[DeviceBattery]: + results: list[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: + results.append(entry) + return results + + +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) # pyright: ignore[reportArgumentType] + if d is not None: + yield d + else: + 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( + "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 + + +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: + 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 + + return _to_device_battery(dev, battery) + + +def poll_once() -> list[DeviceBattery]: + results: list[DeviceBattery] = [] + seen_idents = set() + for dev in _iter_open_devices(): + try: + info = _read_battery(dev) + if info is not None: + 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 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) + except Exception: + Path(tmp_path).unlink(missing_ok=True) + raise + + +@app.command() +def daemon( + 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, + "--out", + "-o", + help="Where to write the JSON status 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_file) + if lock_fd is None: + try: + 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_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 + + write_seq_lock = threading.Lock() + write_seq = 0 + last_written_seq = 0 + + def write_if_changed(): + 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: + 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) + 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): + nonlocal running + running = False + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + + typer.echo( + f"Listening for battery events, writing to {out} (lock: {lock_file})" + ) + + try: + while running: + 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.") + + +if __name__ == "__main__": + app() 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"