vbridge.py

#!/usr/bin/env python3
"""Bridge Vesktop's voice control socket to the Omarchy Discord panel.

Speaks the exact stdin/stdout line-JSON contract of rpc.py, so Rpc.qml can run
either script unchanged: commands arrive on stdin, state lines leave on
stdout. The far end is the VesktopVoiceControl Vencord plugin, which needs no
OAuth, no client id and no secret -- the code runs inside the user's own
client.
"""

import json
import os
import select
import socket
import sys
import time

PROTOCOL_HEADER = "VESKTOP_VOICE_CONTROL/1.0"
RECONNECT_DELAY_SEC = 2.0


def socket_path():
    runtime_dir = os.environ.get("XDG_RUNTIME_DIR") or "/run/user/%d" % os.getuid()
    return os.path.join(runtime_dir, "vesktop-voice-control.sock")


def emit(payload):
    sys.stdout.write(json.dumps(payload) + "\n")
    sys.stdout.flush()


def state_from_control(message):
    def number(value, fallback):
        try:
            return round(float(value))
        except (TypeError, ValueError):
            return fallback

    speaking = message.get("speaking")
    return {
        "ok": True,
        "channel": str(message.get("channel") or ""),
        "guild": str(message.get("guild") or ""),
        "mute": message.get("mute") is True,
        "deaf": message.get("deaf") is True,
        "inputVolume": number(message.get("inputVolume"), 100),
        "speaking": speaking if isinstance(speaking, list) else [],
        "error": "",
        "ping": number(message.get("ping"), 0),
        "voiceState": str(message.get("voiceState") or ""),
    }


def read_lines(fd, buffer):
    """Drain one file descriptor, return (new buffer, complete lines)."""
    try:
        chunk = os.read(fd, 4096)
    except (BlockingIOError, InterruptedError):
        return buffer, []
    except OSError:
        chunk = b""
    if not chunk:
        return buffer, None
    buffer += chunk.decode("utf-8", "replace")
    lines = []
    while "\n" in buffer:
        line, buffer = buffer.split("\n", 1)
        line = line.strip()
        if line:
            lines.append(line)
    return buffer, lines


def probe():
    path = socket_path()
    try:
        conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        conn.connect(path)
    except OSError as error:
        print("control socket: %s" % error)
        return 1
    try:
        conn.settimeout(5)
        header = conn.makefile().readline().strip()
        if header != PROTOCOL_HEADER:
            print("unexpected header: %r" % header)
            return 1
        conn.sendall(b'{"cmd":"refresh"}\n')
        line = conn.makefile().readline().strip()
        print("state: %s" % line)
        return 0
    finally:
        conn.close()


def main():
    if "--probe" in sys.argv:
        return probe()

    path = socket_path()
    conn = None
    stdin_buffer = ""
    socket_buffer = ""

    while True:
        if conn is None:
            try:
                conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                conn.connect(path)
                conn.setblocking(False)
                socket_buffer = ""
                # The header is a fixed greeting; consume it and move on.
                try:
                    first = conn.recv(256).decode("utf-8", "replace")
                    if PROTOCOL_HEADER not in first.split("\n")[0]:
                        raise OSError("unexpected protocol header")
                except (BlockingIOError, InterruptedError):
                    pass
            except OSError:
                if conn is not None:
                    try:
                        conn.close()
                    except OSError:
                        pass
                conn = None
                # Stay alive and retry: the plugin may not be enabled yet, and
                # exiting here would burn Rpc.qml's restart budget.
                time.sleep(RECONNECT_DELAY_SEC)
                continue

        watched = [conn] + [sys.stdin]
        try:
            readable, _, _ = select.select(watched, [], [], 1.0)
        except InterruptedError:
            continue

        if conn in readable:
            socket_buffer, lines = read_lines(conn.fileno(), socket_buffer)
            if lines is None:
                # Vesktop closed the socket; go back to reconnecting.
                try:
                    conn.close()
                except OSError:
                    pass
                conn = None
                continue
            for line in lines:
                try:
                    message = json.loads(line)
                except ValueError:
                    continue
                if message.get("type") == "state":
                    emit(state_from_control(message))

        if sys.stdin in readable:
            stdin_buffer, lines = read_lines(sys.stdin.fileno(), stdin_buffer)
            if lines is None:
                # The panel closed stdin; the bridge's working life is over.
                return 0
            for line in lines:
                try:
                    payload = json.loads(line)
                except ValueError:
                    continue
                if conn is not None:
                    try:
                        conn.sendall((json.dumps(payload) + "\n").encode())
                    except OSError:
                        try:
                            conn.close()
                        except OSError:
                            pass
                        conn = None


if __name__ == "__main__":
    sys.exit(main() or 0)