initial addition of battery display

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:

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()