add support for all UPower peripherals

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)

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__":