Panama/config/dot/quickshell/scripts/panama-kdeconnect

#!/usr/bin/env python3

"""Capability-aware KDE Connect boundary for the Panama shell."""

from __future__ import annotations

import json
import pathlib
import re
import subprocess
import sys
from collections.abc import Callable


DEVICE_ID = re.compile(r"^[A-Fa-f0-9]{32,64}$")
DEVICE_LINE = re.compile(
    r"^-\s+(?P<name>.+):\s+(?P<id>[A-Fa-f0-9]{32,64})"
    r"(?:\s+on\s+.+?)?\s+\((?P<state>[^)]+)\)\s*$"
)
PLUGIN_ACTIONS = {
    "kdeconnect_clipboard": "clipboard",
    "kdeconnect_findmyphone": "ring",
    "kdeconnect_ping": "ping",
    "kdeconnect_share": "share",
}
DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices"
DEVICE_OBJECT_LINE = re.compile(
    rf"(?P<path>{re.escape(DEVICE_OBJECT_PREFIX)}/(?P<id>[A-Fa-f0-9]{{32,64}}))$"
)
BATTERY_INTERFACE = "org.kde.kdeconnect.device.battery"
CONNECTIVITY_INTERFACE = "org.kde.kdeconnect.device.connectivity_report"
# busctl spells every integer width with its own type code; a property that is
# an int32 today may be read back as a uint32 by another kdeconnectd build.
INT_PROPERTY_TYPES = frozenset({"y", "n", "q", "i", "u", "x", "t"})
# Vitals are enrichment, not the answer: a phone that has just gone out of range
# can leave a plugin object that blocks, and the status read must not stall
# behind it. Shorter than the eight seconds the identity reads are allowed.
VITALS_TIMEOUT = 3
# Every device carries both keys so the shell never has to tell "not read" from
# "no such field"; absent vitals are null, not missing.
EMPTY_VITALS: dict[str, object] = {"battery": None, "signal": None}
Runner = Callable[..., subprocess.CompletedProcess[str]]


def compact_json(value: dict[str, object]) -> str:
    return json.dumps(value, separators=(",", ":"))


def validate_device_id(value: str) -> str:
    if not DEVICE_ID.fullmatch(value):
        raise ValueError("invalid-device")
    return value


def validate_file(value: pathlib.Path) -> pathlib.Path:
    try:
        resolved = value.expanduser().resolve(strict=True)
    except OSError as error:
        raise ValueError("invalid-file") from error
    if not resolved.is_file():
        raise ValueError("invalid-file")
    return resolved


def inferred_type(name: str) -> str:
    normalized = name.casefold()
    if "iphone" in normalized or "phone" in normalized:
        return "phone"
    if "tablet" in normalized or "ipad" in normalized:
        return "tablet"
    return "device"


def normalize_device_line(
    line: str,
    plugins: list[str],
    device_type: str = "",
) -> dict[str, object]:
    match = DEVICE_LINE.fullmatch(line.strip())
    if match is None:
        raise ValueError("invalid-device-line")

    state = match.group("state").casefold()
    actions = sorted(
        {
            action
            for plugin, action in PLUGIN_ACTIONS.items()
            if plugin in plugins
        }
    )
    name = match.group("name").strip()
    return {
        "id": match.group("id"),
        "name": name,
        "type": device_type or inferred_type(name),
        "paired": "paired" in state,
        "reachable": "reachable" in state,
        "actions": actions,
    }


# busctl properties are read with --json=short, not in its default text form.
#
# The text form escapes every non-ASCII byte in octal, and it escapes them into
# the OUTPUT rather than into a quoted string a shell-style parser can undo -- so
# a phone named "Gib's iPhone" with a typographic apostrophe arrived as the
# literal characters "Gib\342\200\231s iPhone" and was displayed that way.
# That is not specific to apostrophes: any name with an accent, an emoji, a
# quote, or a backslash was affected the same way.
#
# The JSON form returns real UTF-8 and needs no unescaping, which is why these
# parse a document rather than splitting words.
def property_payload(output: str | None) -> dict[str, object] | None:
    """The decoded document of a busctl --json=short read, or None when the
    output is missing or is not a property document at all."""
    try:
        payload = json.loads(output)
    except (json.JSONDecodeError, TypeError):
        return None
    return payload if isinstance(payload, dict) else None


def parse_property(output: str | None, expected: str) -> object | None:
    """The value of a busctl --json=short property read, or None if it is not
    the type asked for."""
    payload = property_payload(output)
    if payload is None or payload.get("type") != expected:
        return None
    return payload.get("data")


def parse_loaded_plugins(output: str) -> list[str]:
    data = parse_property(output, "as")
    if not isinstance(data, list):
        return []
    return [str(item) for item in data]


def parse_string_property(output: str | None) -> str:
    data = parse_property(output, "s")
    return data if isinstance(data, str) else ""


def parse_bool_property(output: str | None) -> bool | None:
    data = parse_property(output, "b")
    return data if isinstance(data, bool) else None


def parse_int_property(output: str | None) -> int | None:
    """The integer value of a busctl --json=short property read, or None.

    JSON has one number type, so booleans are rejected explicitly: `true` is an
    int to isinstance and would otherwise read back as a charge of 1.
    """
    payload = property_payload(output)
    if payload is None or payload.get("type") not in INT_PROPERTY_TYPES:
        return None
    data = payload.get("data")
    if isinstance(data, bool) or not isinstance(data, int):
        return None
    return data


def run_command(
    command: list[str],
    *,
    runner: Runner = subprocess.run,
    timeout: float = 8,
) -> subprocess.CompletedProcess[str]:
    return runner(
        command,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )


def device_object(device_id: str) -> str:
    return f"{DEVICE_OBJECT_PREFIX}/{validate_device_id(device_id)}"


def loaded_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "call",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "loadedPlugins",
        ],
        runner=runner,
    )
    return parse_loaded_plugins(result.stdout) if result.returncode == 0 else []


def supported_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "supportedPlugins",
        ],
        runner=runner,
    )
    return parse_loaded_plugins(result.stdout) if result.returncode == 0 else []


def device_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    active = loaded_plugins(device_id, runner)
    return active if active else supported_plugins(device_id, runner)


def reported_type(device_id: str, runner: Runner = subprocess.run) -> str:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "type",
        ],
        runner=runner,
    )
    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",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            member,
        ],
        runner=runner,
    )


# Each KDE Connect plugin hangs its own object off the device path, and those
# objects exist only while the device is paired, reachable AND the plugin is
# loaded. A phone with the battery plugin turned off, or one that just walked
# out of range, makes busctl exit non-zero -- that is an absence of data, never
# a fault to report, so every failure mode here collapses to None.
def plugin_property(
    device_id: str,
    plugin: str,
    interface: str,
    member: str,
    runner: Runner = subprocess.run,
) -> str | None:
    """A plugin object's property read as raw busctl output, or None when the
    object, the plugin, or the daemon is not there."""
    try:
        result = run_command(
            [
                "busctl",
                "--user",
                "--json=short",
                "get-property",
                "org.kde.kdeconnect",
                f"{device_object(device_id)}/{plugin}",
                interface,
                member,
            ],
            runner=runner,
            timeout=VITALS_TIMEOUT,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return None
    return result.stdout if result.returncode == 0 else None


def device_battery(
    device_id: str,
    runner: Runner = subprocess.run,
) -> dict[str, object] | None:
    """Charge and charging state, or None when the phone has no battery to
    report.

    Charge is read first because it is the cheapest proof the plugin object
    exists at all: when it is missing the other two reads are skipped, which is
    the common case for a device that is merely paired. kdeconnectd reports -1
    for "not known yet", and anything outside a percentage is not a charge.
    """
    charge = parse_int_property(
        plugin_property(device_id, "battery", BATTERY_INTERFACE, "charge", runner)
    )
    if charge is None or not 0 <= charge <= 100:
        return None
    has_battery = parse_bool_property(
        plugin_property(device_id, "battery", BATTERY_INTERFACE, "hasBattery", runner)
    )
    if has_battery is False:
        return None
    charging = parse_bool_property(
        plugin_property(device_id, "battery", BATTERY_INTERFACE, "isCharging", runner)
    )
    return {"charge": charge, "charging": charging is True}


def device_signal(
    device_id: str,
    runner: Runner = subprocess.run,
) -> dict[str, object] | None:
    """Cell network type and bar count, or None when there is no cellular
    report -- strength is -1 on a phone with no modem or no service."""
    strength = parse_int_property(
        plugin_property(
            device_id,
            "connectivity_report",
            CONNECTIVITY_INTERFACE,
            "cellularNetworkStrength",
            runner,
        )
    )
    if strength is None or strength < 0:
        return None
    network_type = parse_string_property(
        plugin_property(
            device_id,
            "connectivity_report",
            CONNECTIVITY_INTERFACE,
            "cellularNetworkType",
            runner,
        )
    )
    return {"networkType": network_type, "strength": strength}


def device_vitals(
    device: dict[str, object],
    runner: Runner = subprocess.run,
) -> dict[str, object]:
    """The battery and signal fields for a device entry.

    Only a paired, reachable device is queried; for anything else the plugin
    objects cannot exist, so asking would spend busctl calls to learn nothing.
    """
    if not (device.get("paired") and device.get("reachable")):
        return dict(EMPTY_VITALS)
    device_id = str(device["id"])
    return {
        "battery": device_battery(device_id, runner),
        "signal": device_signal(device_id, 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,
    *,
    vitals: bool = True,
) -> 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
            }
        )
        device = {
            "id": device_id,
            "name": name,
            "type": device_type or inferred_type(name),
            "paired": paired,
            "reachable": reachable,
            "actions": actions,
        }
        device.update(device_vitals(device, runner) if vitals else EMPTY_VITALS)
        devices.append(device)
    return devices


# `vitals` is off for the read that guards an action: ringing a phone needs its
# identity and its action list, not its charge, and the extra busctl round trips
# would sit between the tap and the ring.
def collect_status(
    runner: Runner = subprocess.run,
    *,
    vitals: bool = True,
) -> dict[str, object]:
    try:
        listing = run_command(
            ["kdeconnect-cli", "--list-devices"],
            runner=runner,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"available": False, "devices": [], "error": "unavailable"}

    if listing.returncode != 0:
        return {"available": False, "devices": [], "error": "unavailable"}

    devices: list[dict[str, object]] = []
    for line in listing.stdout.splitlines():
        if not line.lstrip().startswith("-"):
            continue
        match = DEVICE_LINE.fullmatch(line.strip())
        if match is None:
            continue
        device_id = match.group("id")
        try:
            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
        device.update(device_vitals(device, runner) if vitals else EMPTY_VITALS)
        devices.append(device)

    known_ids = {str(device["id"]) for device in devices}
    devices.extend(
        device
        for device in dbus_devices(runner, vitals=vitals)
        if str(device["id"]) not in known_ids
    )

    devices.sort(
        key=lambda device: (
            not bool(device["reachable"]),
            not bool(device["paired"]),
            str(device["name"]).casefold(),
        )
    )
    return {"available": True, "devices": devices, "error": ""}


def action_command(
    action: str,
    device_id: str,
    file_path: pathlib.Path | None = None,
) -> list[str]:
    device_id = validate_device_id(device_id)
    options = {
        "ring": ["--ring"],
        "clipboard": ["--send-clipboard"],
    }
    if action == "share" and file_path is not None:
        return [
            "kdeconnect-cli",
            "-d",
            device_id,
            "--share",
            str(validate_file(file_path)),
        ]
    if action not in options:
        raise ValueError("unsupported-action")
    return ["kdeconnect-cli", "-d", device_id, *options[action]]


def invoke_action(
    action: str,
    device_id: str,
    file_path: pathlib.Path | None = None,
    runner: Runner = subprocess.run,
) -> dict[str, object]:
    device_id = validate_device_id(device_id)
    status = collect_status(runner, vitals=False)
    device = next(
        (
            item
            for item in status["devices"]
            if isinstance(item, dict) and item.get("id") == device_id
        ),
        None,
    )
    if not device or not device.get("paired") or not device.get("reachable"):
        return {"ok": False, "action": action, "error": "device-offline"}
    if action not in device.get("actions", []):
        return {"ok": False, "action": action, "error": "unsupported-action"}

    command = action_command(action, device_id, file_path)
    try:
        result = run_command(command, runner=runner, timeout=3600)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"ok": False, "action": action, "error": "action-failed"}
    if result.returncode != 0:
        return {"ok": False, "action": action, "error": "action-failed"}

    response: dict[str, object] = {"ok": True, "action": action, "error": ""}
    if action == "share" and file_path is not None:
        response["fileName"] = validate_file(file_path).name
    return response


def print_result(value: dict[str, object]) -> None:
    sys.stdout.write(compact_json(value) + "\n")


def main(argv: list[str]) -> int:
    if not argv or argv[0] == "status":
        print_result(collect_status())
        return 0

    command = argv[0]
    try:
        if command == "ring" and len(argv) == 2:
            result = invoke_action("ring", argv[1])
        elif command == "send-clipboard" and len(argv) == 2:
            result = invoke_action("clipboard", argv[1])
        elif command == "send-file" and len(argv) == 3:
            result = invoke_action("share", argv[1], pathlib.Path(argv[2]))
        else:
            result = {"ok": False, "action": command, "error": "invalid-command"}
    except ValueError as error:
        result = {"ok": False, "action": command, "error": str(error)}
    print_result(result)
    return 0 if bool(result.get("ok")) else 2


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))