Something went wrong. Try again.
Global — KDE Connect–compatible mobile companion (Gleam + egui/vidya dist-ui)
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675#!/usr/bin/env python3"""Host peer operations for Global (Valent / GSConnect / kdeconnect-cli).
Usage: peer-ops list peer-ops pair <device_id> peer-ops unpair <device_id> peer-ops ring <device_id> peer-ops clipboard-push <device_id> [text] peer-ops clipboard-pull <device_id> peer-ops media-play-pause <device_id> [session_id]
Always prints one JSON object to stdout: {"ok": true|false, "action": "...", "device_id": "...", "message": "...", "error": null|string, "devices": [...], "discovered": [...], "detail": {...}}
Never reports ok=true for a silent no-op when the peer is offline / missing."""from __future__ import annotations
import jsonimport osimport reimport shutilimport subprocessimport sysfrom pathlib import Path
HOME = Path.home()SCRIPT_DIR = Path(__file__).resolve().parentDISCOVER = SCRIPT_DIR / "discover-peers"VALENT_DEST = "ca.andyholmes.Valent"VALENT_PATH = "/ca/andyholmes/Valent"
def _run(cmd: list[str], timeout: float = 8.0) -> tuple[int, str, str]: try: p = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, env={**os.environ, "LC_ALL": "C"}, ) return p.returncode, p.stdout or "", p.stderr or "" except FileNotFoundError: return 127, "", "not found" except subprocess.TimeoutExpired: return 124, "", "timeout"
def out(obj: dict) -> int: print(json.dumps(obj, ensure_ascii=False)) return 0 if obj.get("ok") else 1
def fail(action: str, device_id: str, error: str, **extra) -> int: payload = { "ok": False, "action": action, "device_id": device_id, "message": "", "error": error, "devices": [], "discovered": [], "detail": {}, } payload.update(extra) return out(payload)
def discover() -> dict: if not DISCOVER.is_file(): return {"ok": False, "error": "discover-peers missing", "devices": [], "sources": []} code, stdout, stderr = _run([str(DISCOVER)], timeout=20) if code != 0 and not stdout.strip(): return { "ok": False, "error": stderr.strip() or f"discover exit {code}", "devices": [], "sources": [], } try: data = json.loads(stdout) except json.JSONDecodeError as e: return {"ok": False, "error": f"bad discover JSON: {e}", "devices": [], "sources": []} return data
def split_lists(disc: dict) -> tuple[list, list]: devices = [] discovered = [] for d in disc.get("devices") or []: if d.get("paired"): devices.append(d) else: discovered.append(d) return devices, discovered
def find_peer(disc: dict, device_id: str) -> dict | None: for d in disc.get("devices") or []: if d.get("id") == device_id or d.get("id", "").replace("-", "_") == device_id.replace( "-", "_" ): return d return None
def is_online(peer: dict | None) -> bool: if not peer: return False return peer.get("link") in ("online", "idle")
def local_device_id() -> str | None: cert = HOME / ".config" / "valent" / "certificate.pem" if not cert.is_file(): return None code, o, _ = _run( ["openssl", "x509", "-in", str(cert), "-noout", "-subject"], timeout=2 ) if code != 0: return None m = re.search(r"CN\s*=\s*([^/\n,]+)", o) return m.group(1).strip() if m else None
def valent_device_action(device_id: str, action: str, extra: list | None = None) -> tuple[bool, str]: """Invoke Valent application action `device` with (ssav).
Returns (called_ok, note). Valent often returns success even when the plugin is inactive; callers MUST verify post-conditions. """ # gdbus: Activate device "[<('id', 'action', @av [])>]" "{}" param = f"[<('{device_id}', '{action}', @av [])>]" code, stdout, stderr = _run( [ "gdbus", "call", "--session", "--dest", VALENT_DEST, "--object-path", VALENT_PATH, "--method", "org.gtk.Actions.Activate", "device", param, "{}", ], timeout=6, ) if code == 0: return True, stdout.strip() or "activated" # Also try object path if device is exported path = f"/ca/andyholmes/Valent/Device/{device_id}" code2, out2, err2 = _run( [ "gdbus", "call", "--session", "--dest", VALENT_DEST, "--object-path", path, "--method", "org.gtk.Actions.Activate", action, "[]", "{}", ], timeout=6, ) if code2 == 0: return True, out2.strip() or "device-path activated" return False, (stderr or err2 or "valent activate failed").strip()
def kdeconnect_cli(*args: str) -> tuple[bool, str]: if shutil.which("kdeconnect-cli") is None: return False, "kdeconnect-cli not installed" code, stdout, stderr = _run(["kdeconnect-cli", *args], timeout=10) text = (stdout + "\n" + stderr).strip() if code == 0: return True, text or "ok" return False, text or f"kdeconnect-cli exit {code}"
def ensure_not_local(device_id: str) -> str | None: local = local_device_id() if local and ( local == device_id or local.replace("-", "_") == device_id.replace("-", "_") ): return "refusing to operate on the local host device id" return None
def cmd_list() -> int: disc = discover() if not disc.get("ok", True) and disc.get("error"): # discover-peers always sets ok:true when it runs; treat devices present as success if not disc.get("devices") and disc.get("error"): return fail("list", "", disc["error"]) devices, discovered = split_lists(disc) return out( { "ok": True, "action": "list", "device_id": "", "message": f"{len(devices)} paired · {len(discovered)} nearby", "error": None, "devices": devices, "discovered": discovered, "detail": {"sources": disc.get("sources") or [], "errors": disc.get("errors") or []}, } )
def cmd_pair(device_id: str) -> int: if err := ensure_not_local(device_id): return fail("pair", device_id, err) before = discover() peer = find_peer(before, device_id) notes: list[str] = [] host_ok = False
ok_v, note_v = valent_device_action(device_id, "pair") notes.append(f"valent:{note_v}") host_ok = host_ok or ok_v ok_v2, note_v2 = valent_device_action(device_id, "state.pair") notes.append(f"valent-state:{note_v2}") host_ok = host_ok or ok_v2
ok_k, note_k = kdeconnect_cli("--pair", "-d", device_id) if ok_k or "not installed" not in note_k: notes.append(f"kdeconnect-cli:{note_k}") host_ok = host_ok or ok_k
after = discover() peer_after = find_peer(after, device_id) devices, discovered = split_lists(after)
# Success only if paired flag true after if peer_after and peer_after.get("paired"): return out( { "ok": True, "action": "pair", "device_id": device_id, "message": f"Paired {peer_after.get('name') or device_id}", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "link": peer_after.get("link")}, } )
# If host tools unavailable entirely if not host_ok and shutil.which("gdbus") is None and shutil.which("kdeconnect-cli") is None: return fail( "pair", device_id, "no host pairing backend (gdbus/Valent or kdeconnect-cli)", devices=devices, discovered=discovered, detail={"notes": notes}, )
# Peer unknown / offline — honest failure if peer is None and peer_after is None: return fail( "pair", device_id, "device not visible to host discovery; cannot confirm pair", devices=devices, discovered=discovered, detail={"notes": notes}, )
return fail( "pair", device_id, "pair requested but host still reports device unpaired (peer offline or rejected)", devices=devices, discovered=discovered, detail={"notes": notes, "peer_before": peer, "peer_after": peer_after}, )
def cmd_unpair(device_id: str) -> int: if err := ensure_not_local(device_id): return fail("unpair", device_id, err) before = discover() peer = find_peer(before, device_id) if peer is None: devices, discovered = split_lists(before) return fail( "unpair", device_id, "device not in host peer list", devices=devices, discovered=discovered, )
notes: list[str] = [] host_ok = False for action in ("unpair", "state.unpair"): ok_v, note_v = valent_device_action(device_id, action) notes.append(f"valent:{action}:{note_v}") host_ok = host_ok or ok_v
ok_k, note_k = kdeconnect_cli("--unpair", "-d", device_id) notes.append(f"kdeconnect-cli:{note_k}") host_ok = host_ok or ok_k
# If still paired offline-only in config, remove Valent device dir after host attempt # only when tools accepted the call OR config is the only source. config_dir = HOME / ".config" / "valent" / "device" / device_id.replace("-", "_") # also try raw id alt = HOME / ".config" / "valent" / "device" / device_id removed = False after = discover() peer_after = find_peer(after, device_id) if peer_after and peer_after.get("paired") and peer_after.get("source") == "valent-config": for d in (config_dir, alt): if d.is_dir(): cert = d / "certificate.pem" # remove only the pairing material we know about try: if cert.is_file(): cert.unlink() # remove empty dir for child in d.iterdir(): if child.is_file(): child.unlink() d.rmdir() removed = True notes.append(f"removed-config:{d}") except OSError as e: notes.append(f"config-rm-error:{e}") after = discover() peer_after = find_peer(after, device_id)
devices, discovered = split_lists(after) still_paired = bool(peer_after and peer_after.get("paired")) if not still_paired: return out( { "ok": True, "action": "unpair", "device_id": device_id, "message": f"Unpaired {device_id}" + (" (config cleared)" if removed else ""), "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "config_removed": removed}, } )
return fail( "unpair", device_id, "unpair requested but host still lists device as paired", devices=devices, discovered=discovered, detail={"notes": notes, "host_ok": host_ok}, )
def cmd_ring(device_id: str) -> int: if err := ensure_not_local(device_id): return fail("ring", device_id, err) disc = discover() peer = find_peer(disc, device_id) devices, discovered = split_lists(disc) if peer is None: return fail( "ring", device_id, "device not found in host discovery", devices=devices, discovered=discovered, ) if not is_online(peer): return fail( "ring", device_id, f"peer offline ({peer.get('link')}); cannot ring", devices=devices, discovered=discovered, detail={"peer": peer}, )
notes: list[str] = [] # Prefer kdeconnect-cli when available — clear exit codes ok_k, note_k = kdeconnect_cli("--ring", "-d", device_id) notes.append(f"kdeconnect-cli:{note_k}") if ok_k: return out( { "ok": True, "action": "ring", "device_id": device_id, "message": f"Ring sent to {peer.get('name') or device_id}", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "kdeconnect-cli"}, } )
# Valent plugin action names any_call = False for action in ( "findmyphone.ring", "findmyphone", "ring", ): ok_v, note_v = valent_device_action(device_id, action) notes.append(f"valent:{action}:{note_v}") any_call = any_call or ok_v
if any_call: # Peer was online and host accepted the call — best confirmation we have without live feedback return out( { "ok": True, "action": "ring", "device_id": device_id, "message": f"Ring requested for online peer {peer.get('name') or device_id} via Valent", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "valent", "peer_link": peer.get("link")}, } )
return fail( "ring", device_id, "no host ring backend succeeded (install kdeconnect-cli or ensure Valent findmyphone plugin)", devices=devices, discovered=discovered, detail={"notes": notes}, )
def cmd_clipboard_push(device_id: str, text: str) -> int: if err := ensure_not_local(device_id): return fail("clipboard-push", device_id, err) disc = discover() peer = find_peer(disc, device_id) devices, discovered = split_lists(disc) if peer is None: return fail( "clipboard-push", device_id, "device not found", devices=devices, discovered=discovered, ) if not is_online(peer): return fail( "clipboard-push", device_id, f"peer offline ({peer.get('link')}); cannot push clipboard", devices=devices, discovered=discovered, )
notes: list[str] = [] # kdeconnect-cli share-text / send-clipboard variants if text: ok_k, note_k = kdeconnect_cli("--share-text", text, "-d", device_id) notes.append(f"share-text:{note_k}") if ok_k: return out( { "ok": True, "action": "clipboard-push", "device_id": device_id, "message": "Shared text to peer", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "kdeconnect-cli"}, } ) ok_k2, note_k2 = kdeconnect_cli("--send-clipboard", "-d", device_id) notes.append(f"send-clipboard:{note_k2}") if ok_k2: return out( { "ok": True, "action": "clipboard-push", "device_id": device_id, "message": "Clipboard pushed to peer", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "kdeconnect-cli"}, } )
any_call = False for action in ("share.text", "clipboard.push", "share", "clipboard"): # pass text as string variant when provided if text and action == "share.text": # Activate with extra params is best-effort; gdbus av empty for now pass ok_v, note_v = valent_device_action(device_id, action) notes.append(f"valent:{action}:{note_v}") any_call = any_call or ok_v
if any_call: return out( { "ok": True, "action": "clipboard-push", "device_id": device_id, "message": f"Clipboard/share requested for online peer {peer.get('name') or device_id}", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "valent"}, } )
return fail( "clipboard-push", device_id, "no host clipboard/share backend succeeded", devices=devices, discovered=discovered, detail={"notes": notes}, )
def cmd_clipboard_pull(device_id: str) -> int: if err := ensure_not_local(device_id): return fail("clipboard-pull", device_id, err) disc = discover() peer = find_peer(disc, device_id) devices, discovered = split_lists(disc) if peer is None: return fail( "clipboard-pull", device_id, "device not found", devices=devices, discovered=discovered, ) if not is_online(peer): return fail( "clipboard-pull", device_id, f"peer offline ({peer.get('link')}); cannot pull clipboard", devices=devices, discovered=discovered, )
notes: list[str] = [] # No reliable kdeconnect-cli pull on all versions — try Valent any_call = False for action in ("clipboard.pull", "clipboard"): ok_v, note_v = valent_device_action(device_id, action) notes.append(f"valent:{action}:{note_v}") any_call = any_call or ok_v
if any_call: return out( { "ok": True, "action": "clipboard-pull", "device_id": device_id, "message": f"Clipboard pull requested for online peer {peer.get('name') or device_id}", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "backend": "valent"}, } )
return fail( "clipboard-pull", device_id, "no host clipboard-pull backend succeeded", devices=devices, discovered=discovered, detail={"notes": notes}, )
def cmd_media_play_pause(device_id: str, session_id: str) -> int: if err := ensure_not_local(device_id): return fail("media-play-pause", device_id, err) disc = discover() peer = find_peer(disc, device_id) devices, discovered = split_lists(disc) if peer is None: return fail( "media-play-pause", device_id, "device not found", devices=devices, discovered=discovered, ) if not is_online(peer): return fail( "media-play-pause", device_id, f"peer offline ({peer.get('link')}); cannot control media", devices=devices, discovered=discovered, )
notes: list[str] = [] any_call = False for action in ("mpris.playpause", "media.playpause", "mpris"): ok_v, note_v = valent_device_action(device_id, action) notes.append(f"valent:{action}:{note_v}") any_call = any_call or ok_v
if any_call: return out( { "ok": True, "action": "media-play-pause", "device_id": device_id, "message": f"Media play/pause requested for {peer.get('name') or device_id}", "error": None, "devices": devices, "discovered": discovered, "detail": {"notes": notes, "session_id": session_id}, } )
return fail( "media-play-pause", device_id, "no host media backend succeeded", devices=devices, discovered=discovered, detail={"notes": notes, "session_id": session_id}, )
def main(argv: list[str]) -> int: if len(argv) < 2: return fail("", "", "usage: peer-ops <list|pair|unpair|ring|clipboard-push|clipboard-pull|media-play-pause> ...") cmd = argv[1] if cmd == "list": return cmd_list() if cmd == "pair" and len(argv) >= 3: return cmd_pair(argv[2]) if cmd == "unpair" and len(argv) >= 3: return cmd_unpair(argv[2]) if cmd == "ring" and len(argv) >= 3: return cmd_ring(argv[2]) if cmd == "clipboard-push" and len(argv) >= 3: text = argv[3] if len(argv) >= 4 else "" return cmd_clipboard_push(argv[2], text) if cmd == "clipboard-pull" and len(argv) >= 3: return cmd_clipboard_pull(argv[2]) if cmd == "media-play-pause" and len(argv) >= 3: session = argv[3] if len(argv) >= 4 else "" return cmd_media_play_pause(argv[2], session) return fail(cmd, "", f"unknown or incomplete command: {cmd}")
if __name__ == "__main__": sys.exit(main(sys.argv))