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: