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: