diff --git a/config/dot/quickshell/scripts/panama-kdeconnect b/config/dot/quickshell/scripts/panama-kdeconnect index fbbf5f5..5e670df 100755 --- a/config/dot/quickshell/scripts/panama-kdeconnect +++ b/config/dot/quickshell/scripts/panama-kdeconnect @@ -25,6 +25,9 @@ PLUGIN_ACTIONS = { "kdeconnect_share": "share", } DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices" +DEVICE_OBJECT_LINE = re.compile( + rf"(?P{re.escape(DEVICE_OBJECT_PREFIX)}/(?P[A-Fa-f0-9]{{32,64}}))$" +) Runner = Callable[..., subprocess.CompletedProcess[str]] @@ -107,6 +110,13 @@ def parse_string_property(output: str) -> str: return parts[1] if len(parts) == 2 and parts[0] == "s" else "" +def parse_bool_property(output: str) -> bool | None: + parts = output.split() + if len(parts) != 2 or parts[0] != "b" or parts[1] not in {"true", "false"}: + return None + return parts[1] == "true" + + def run_command( command: list[str], *, @@ -179,6 +189,82 @@ def reported_type(device_id: str, runner: Runner = subprocess.run) -> str: return parse_string_property(result.stdout) if result.returncode == 0 else "" +def device_property( + device_id: str, + member: str, + runner: Runner = subprocess.run, +) -> subprocess.CompletedProcess[str]: + return run_command( + [ + "busctl", + "--user", + "get-property", + "org.kde.kdeconnect", + device_object(device_id), + "org.kde.kdeconnect.device", + member, + ], + runner=runner, + ) + + +def dbus_device_ids(runner: Runner = subprocess.run) -> list[str]: + try: + result = run_command( + ["busctl", "--user", "tree", "org.kde.kdeconnect"], + runner=runner, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return [] + if result.returncode != 0: + return [] + return [ + match.group("id") + for line in result.stdout.splitlines() + if (match := DEVICE_OBJECT_LINE.search(line.strip())) is not None + ] + + +def dbus_devices(runner: Runner = subprocess.run) -> list[dict[str, object]]: + devices: list[dict[str, object]] = [] + for device_id in dbus_device_ids(runner): + try: + name_result = device_property(device_id, "name", runner) + type_result = device_property(device_id, "type", runner) + paired_result = device_property(device_id, "isPaired", runner) + reachable_result = device_property(device_id, "isReachable", runner) + except (FileNotFoundError, subprocess.TimeoutExpired): + continue + name = parse_string_property(name_result.stdout) if name_result.returncode == 0 else "" + device_type = parse_string_property(type_result.stdout) if type_result.returncode == 0 else "" + paired = parse_bool_property(paired_result.stdout) if paired_result.returncode == 0 else None + reachable = parse_bool_property(reachable_result.stdout) if reachable_result.returncode == 0 else None + if not name or paired is not True or reachable is None: + continue + try: + plugins = device_plugins(device_id, runner) + except (FileNotFoundError, subprocess.TimeoutExpired): + plugins = [] + actions = sorted( + { + action + for plugin, action in PLUGIN_ACTIONS.items() + if plugin in plugins + } + ) + devices.append( + { + "id": device_id, + "name": name, + "type": device_type or inferred_type(name), + "paired": paired, + "reachable": reachable, + "actions": actions, + } + ) + return devices + + def collect_status(runner: Runner = subprocess.run) -> dict[str, object]: try: listing = run_command( @@ -200,15 +286,24 @@ def collect_status(runner: Runner = subprocess.run) -> dict[str, object]: continue device_id = match.group("id") try: - device = normalize_device_line( - line, - device_plugins(device_id, runner), - reported_type(device_id, runner), - ) + plugins = device_plugins(device_id, runner) + device_type = reported_type(device_id, runner) + except (FileNotFoundError, subprocess.TimeoutExpired): + plugins = [] + device_type = "" + try: + device = normalize_device_line(line, plugins, device_type) except ValueError: continue devices.append(device) + known_ids = {str(device["id"]) for device in devices} + devices.extend( + device + for device in dbus_devices(runner) + if str(device["id"]) not in known_ids + ) + devices.sort( key=lambda device: ( not bool(device["reachable"]), diff --git a/config/dot/quickshell/services/Connectivity.qml b/config/dot/quickshell/services/Connectivity.qml index b14fd8a..fc63542 100644 --- a/config/dot/quickshell/services/Connectivity.qml +++ b/config/dot/quickshell/services/Connectivity.qml @@ -33,11 +33,16 @@ Singleton { } readonly property var wiredDevice: { + let fallback = null; for (const device of Networking.devices.values) { - if (device.type === DeviceType.Wired) + if (device.type !== DeviceType.Wired) + continue; + if (device.connected) return device; + if (!fallback) + fallback = device; } - return null; + return fallback; } readonly property var adapter: Bluetooth.defaultAdapter diff --git a/tests/quickshell/kdeconnect_bridge_test.py b/tests/quickshell/kdeconnect_bridge_test.py index 2981450..41d0d73 100644 --- a/tests/quickshell/kdeconnect_bridge_test.py +++ b/tests/quickshell/kdeconnect_bridge_test.py @@ -130,6 +130,134 @@ class KdeConnectBridgeTest(unittest.TestCase): ], ) + def test_status_falls_back_to_paired_dbus_device_when_cli_is_empty(self) -> None: + device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + device_path = f"/modules/kdeconnect/devices/{device_id}" + + def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command == ["kdeconnect-cli", "--list-devices"]: + return subprocess.CompletedProcess(command, 0, "0 devices found\n", "") + if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: + return subprocess.CompletedProcess(command, 0, f"└─ {device_path}\n", "") + if command[:3] == ["busctl", "--user", "call"]: + return subprocess.CompletedProcess(command, 0, "as 0\n", "") + if command[:3] == ["busctl", "--user", "get-property"]: + values = { + "name": 's "Fixture iPhone"\n', + "type": 's "phone"\n', + "isPaired": "b true\n", + "isReachable": "b false\n", + "supportedPlugins": ( + 'as 3 "kdeconnect_share" "kdeconnect_clipboard" ' + '"kdeconnect_findmyphone"\n' + ), + } + return subprocess.CompletedProcess(command, 0, values[command[-1]], "") + raise AssertionError(command) + + self.assertEqual( + bridge.collect_status(runner), + { + "available": True, + "devices": [ + { + "id": device_id, + "name": "Fixture iPhone", + "type": "phone", + "paired": True, + "reachable": False, + "actions": ["clipboard", "ring", "share"], + } + ], + "error": "", + }, + ) + + def test_dbus_plugin_timeout_keeps_device_with_no_actions(self) -> None: + device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + + def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: + path = f"/modules/kdeconnect/devices/{device_id}" + return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") + if command[:3] == ["busctl", "--user", "call"]: + return subprocess.CompletedProcess(command, 0, "as 0\n", "") + if command[:3] == ["busctl", "--user", "get-property"]: + values = { + "name": 's "Fixture iPhone"\n', + "type": 's "phone"\n', + "isPaired": "b true\n", + "isReachable": "b false\n", + } + if command[-1] == "supportedPlugins": + raise subprocess.TimeoutExpired(command, 8) + return subprocess.CompletedProcess(command, 0, values[command[-1]], "") + raise AssertionError(command) + + self.assertEqual(bridge.dbus_devices(runner)[0]["actions"], []) + + def test_cli_device_plugin_timeout_fails_closed(self) -> None: + device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + listing = f"- Fixture iPhone: {device_id} (paired and reachable)\n" + + def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command == ["kdeconnect-cli", "--list-devices"]: + return subprocess.CompletedProcess(command, 0, listing, "") + raise subprocess.TimeoutExpired(command, 8) + + status = bridge.collect_status(runner) + + self.assertEqual(status["devices"][0]["type"], "phone") + self.assertEqual(status["devices"][0]["actions"], []) + + def test_dbus_inventory_excludes_unpaired_peers(self) -> None: + device_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + + def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: + path = f"/modules/kdeconnect/devices/{device_id}" + return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") + if command[:3] == ["busctl", "--user", "get-property"]: + values = { + "name": 's "Nearby Stranger"\n', + "type": 's "phone"\n', + "isPaired": "b false\n", + "isReachable": "b true\n", + } + return subprocess.CompletedProcess(command, 0, values[command[-1]], "") + raise AssertionError(command) + + self.assertEqual(bridge.dbus_devices(runner), []) + + def test_status_merges_paired_dbus_device_missing_from_cli(self) -> None: + cli_id = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + dbus_id = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" + listing = f"- Fixture Laptop: {cli_id} (paired and reachable)\n" + + def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command == ["kdeconnect-cli", "--list-devices"]: + return subprocess.CompletedProcess(command, 0, listing, "") + if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: + path = f"/modules/kdeconnect/devices/{dbus_id}" + return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") + if command[:3] == ["busctl", "--user", "call"]: + return subprocess.CompletedProcess(command, 0, "as 0\n", "") + if command[:3] == ["busctl", "--user", "get-property"]: + is_dbus_device = dbus_id in command[4] + values = { + "name": 's "Fixture iPhone"\n', + "type": 's "phone"\n' if is_dbus_device else 's "desktop"\n', + "isPaired": "b true\n", + "isReachable": "b false\n", + "supportedPlugins": "as 0\n", + } + return subprocess.CompletedProcess(command, 0, values[command[-1]], "") + raise AssertionError(command) + + status = bridge.collect_status(runner) + + self.assertEqual({device["id"] for device in status["devices"]}, {cli_id, dbus_id}) + if __name__ == "__main__": unittest.main()