alan
1ed57291f9
An Omarchy shell plugin: a Python/asyncio daemon that discovers network displays and serves an encoded screen capture for them to pull, plus a Quickshell bar widget and panel to drive it. The screen comes from the xdg-desktop-portal ScreenCast interface (asked for every time, never remembered), goes through GStreamer, and is served from a local HTTP port. Desktop audio is mixed in from the default output's monitor. The container follows the receiver: WebM/VP8 for Cast, MPEG-TS/H.264 for DLNA, HLS for AirPlay video. The stream port has to be reachable from the LAN, and none of these protocols can carry a credential, so the capability is the URL: a fresh random path per session, refused to anything off the LAN, capped at four concurrent readers.
231 lines
8.7 KiB
Python
231 lines
8.7 KiB
Python
"""Command line entry point.
|
|
|
|
screencast-server run the daemon the shell plugin starts
|
|
screencast-server devices list what is on the network
|
|
screencast-server cast "<name>" start casting the screen
|
|
screencast-server stop | status
|
|
screencast-server pair "<name>" AirPlay PIN pairing
|
|
screencast-server set fps=30 bitrate=8000
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
from . import VERSION
|
|
from .server import Daemon
|
|
from .util import log, runtime_dir, setup_logging
|
|
|
|
|
|
async def _call(msg: dict, *, wait_state: bool = True, timeout: float = 200.0) -> dict:
|
|
"""Send one command to the running daemon and return its reply."""
|
|
path = runtime_dir() / "ctl.sock"
|
|
try:
|
|
reader, writer = await asyncio.open_unix_connection(str(path))
|
|
except (FileNotFoundError, ConnectionRefusedError):
|
|
print("screencast-server is not running (start it with: screencast-server run)", file=sys.stderr)
|
|
raise SystemExit(4)
|
|
state: dict = {}
|
|
writer.write((json.dumps(msg) + "\n").encode())
|
|
await writer.drain()
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
try:
|
|
while True:
|
|
remaining = deadline - asyncio.get_running_loop().time()
|
|
if remaining <= 0:
|
|
break
|
|
line = await asyncio.wait_for(reader.readline(), remaining)
|
|
if not line:
|
|
break
|
|
data = json.loads(line)
|
|
if data.get("type") == "state":
|
|
state = data
|
|
if msg.get("cmd") == "get" and wait_state:
|
|
break
|
|
elif data.get("type") == "reply":
|
|
data["state"] = state
|
|
return data
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
finally:
|
|
writer.close()
|
|
return {"ok": True, "state": state}
|
|
|
|
|
|
def _print_devices(state: dict) -> None:
|
|
devices = state.get("devices") or []
|
|
if not devices:
|
|
print("no receivers found yet")
|
|
return
|
|
width = max(len(d["name"]) for d in devices)
|
|
for dev in devices:
|
|
extra = f" · {dev['app']}" if dev.get("app") else ""
|
|
print(f"{dev['name']:<{width}} {dev['kindLabel']:<12} {dev['host']:<15} {dev['id']}{extra}")
|
|
|
|
|
|
def _print_status(state: dict) -> None:
|
|
session = state.get("session") or {}
|
|
if session.get("state") in (None, "", "idle"):
|
|
print("idle")
|
|
else:
|
|
print(
|
|
f"{session['state']}: {session.get('deviceName', '')} "
|
|
f"({session.get('width')}x{session.get('height')} {session.get('encoder')}/"
|
|
f"{session.get('container')}, {session.get('clients')} client(s))"
|
|
)
|
|
if session.get("url"):
|
|
print(f" {session['url']}")
|
|
if session.get("error"):
|
|
print(f"error: {session['error']}")
|
|
elif state.get("error"):
|
|
print(f"error: {state['error']}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="screencast-server", description=__doc__)
|
|
parser.add_argument("-v", "--verbose", action="store_true")
|
|
parser.add_argument("--version", action="version", version=VERSION)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
sub.add_parser("run", help="run the daemon")
|
|
sub.add_parser("devices", help="list receivers")
|
|
sub.add_parser("status", help="what is being cast")
|
|
sub.add_parser("stop", help="stop casting")
|
|
sub.add_parser("rescan", help="scan the network again")
|
|
sub.add_parser("open-firewall", help="let receivers reach the stream port (asks for a password)")
|
|
p_cast = sub.add_parser("cast", help="cast the screen to a receiver")
|
|
p_cast.add_argument("device", help="device id, or part of its name")
|
|
p_vol = sub.add_parser("volume", help="set the receiver volume (0-100)")
|
|
p_vol.add_argument("percent", type=float)
|
|
p_pair = sub.add_parser("pair", help="pair with an AirPlay receiver")
|
|
p_pair.add_argument("device")
|
|
p_set = sub.add_parser("set", help="change settings, e.g. fps=30 audio=false")
|
|
p_set.add_argument("assignments", nargs="+")
|
|
args = parser.parse_args(argv)
|
|
|
|
setup_logging(args.verbose)
|
|
if args.command == "run":
|
|
try:
|
|
code = asyncio.run(Daemon().run())
|
|
except KeyboardInterrupt:
|
|
code = 0
|
|
# Same reason as the timer in Daemon.shutdown: library threads would
|
|
# otherwise hold the process open after everything is closed.
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
os._exit(code)
|
|
|
|
if args.command == "devices":
|
|
state = asyncio.run(_call({"cmd": "rescan"}, timeout=8)).get("state") or {}
|
|
_print_devices(state)
|
|
return 0
|
|
if args.command == "status":
|
|
_print_status(asyncio.run(_call({"cmd": "get"}, timeout=8)).get("state") or {})
|
|
return 0
|
|
if args.command == "stop":
|
|
asyncio.run(_call({"cmd": "stop"}, timeout=20))
|
|
return 0
|
|
if args.command == "rescan":
|
|
_print_devices(asyncio.run(_call({"cmd": "rescan"}, timeout=10)).get("state") or {})
|
|
return 0
|
|
if args.command == "open-firewall":
|
|
reply = asyncio.run(_call({"cmd": "openFirewall"}, timeout=200))
|
|
if reply.get("ok"):
|
|
print(reply.get("note") or "the stream port is open")
|
|
return 0
|
|
print(reply.get("error") or "could not open the port", file=sys.stderr)
|
|
return 1
|
|
if args.command == "cast":
|
|
state = asyncio.run(_cast(args.device))
|
|
_print_status(state)
|
|
return 0 if (state.get("session") or {}).get("state") == "streaming" else 1
|
|
if args.command == "volume":
|
|
reply = asyncio.run(_call({"cmd": "volume", "value": args.percent / 100.0}, timeout=15))
|
|
return 0 if reply.get("ok") else 1
|
|
if args.command == "pair":
|
|
return _pair(args.device)
|
|
if args.command == "set":
|
|
patch: dict = {}
|
|
for item in args.assignments:
|
|
if "=" not in item:
|
|
print(f"expected key=value, got “{item}”", file=sys.stderr)
|
|
return 2
|
|
key, value = item.split("=", 1)
|
|
if value.lower() in ("true", "false"):
|
|
patch[key] = value.lower() == "true"
|
|
else:
|
|
try:
|
|
patch[key] = int(value)
|
|
except ValueError:
|
|
patch[key] = value
|
|
reply = asyncio.run(_call({"cmd": "set", "settings": patch}, timeout=15))
|
|
print(json.dumps((reply.get("state") or {}).get("settings", {}), indent=2))
|
|
return 0
|
|
return 2
|
|
|
|
|
|
async def _cast(device: str) -> dict:
|
|
"""Start the cast, then follow the state pushes until it settles."""
|
|
path = runtime_dir() / "ctl.sock"
|
|
try:
|
|
reader, writer = await asyncio.open_unix_connection(str(path))
|
|
except (FileNotFoundError, ConnectionRefusedError):
|
|
print("screencast-server is not running", file=sys.stderr)
|
|
raise SystemExit(4)
|
|
writer.write((json.dumps({"cmd": "cast", "device": device}) + "\n").encode())
|
|
await writer.drain()
|
|
state: dict = {}
|
|
# The daemon pushes its state on connect, before it has seen the command:
|
|
# that first one says "idle" and must not end the wait.
|
|
first = True
|
|
deadline = asyncio.get_running_loop().time() + 300
|
|
try:
|
|
while True:
|
|
remaining = deadline - asyncio.get_running_loop().time()
|
|
if remaining <= 0:
|
|
break
|
|
line = await asyncio.wait_for(reader.readline(), remaining)
|
|
if not line:
|
|
break
|
|
data = json.loads(line)
|
|
if data.get("type") == "reply" and not data.get("ok"):
|
|
print(f"error: {data.get('error')}", file=sys.stderr)
|
|
break
|
|
if data.get("type") != "state":
|
|
continue
|
|
state = data
|
|
if first:
|
|
first = False
|
|
continue
|
|
session_state = (data.get("session") or {}).get("state", "")
|
|
if session_state in ("streaming", "error", "idle") and state.get("busy", "") == "":
|
|
break
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
finally:
|
|
writer.close()
|
|
return state
|
|
|
|
|
|
def _pair(device: str) -> int:
|
|
reply = asyncio.run(_call({"cmd": "pair", "device": device}, timeout=30))
|
|
if not reply.get("ok"):
|
|
print(f"error: {reply.get('error')}", file=sys.stderr)
|
|
return 1
|
|
pin = input("PIN shown on the receiver: ").strip()
|
|
reply = asyncio.run(_call({"cmd": "pairPin", "pin": pin}, timeout=30))
|
|
if not reply.get("ok"):
|
|
print(f"error: {reply.get('error') or 'pairing failed'}", file=sys.stderr)
|
|
return 1
|
|
print("paired")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except BrokenPipeError:
|
|
log.debug("stdout closed")
|