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.
308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""One cast: capture → local HTTP stream → receiver, and the watching of it."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import shutil
|
|
import time
|
|
|
|
from .backends.base import BackendError
|
|
from .capture import Capture, PortalError, pick_encoder
|
|
from .devices import Device
|
|
from .stream import Fanout
|
|
from .util import firewall_hint, local_ip_for, log, runtime_dir, url_host
|
|
|
|
IDLE, STARTING, STREAMING, STOPPING, ERROR = "idle", "starting", "streaming", "stopping", "error"
|
|
|
|
|
|
def choose_container(backend, cfg) -> str:
|
|
"""The best container this receiver and this machine can agree on."""
|
|
preference = str(cfg["encoder"])
|
|
for container in backend.containers:
|
|
if container == "hls":
|
|
return "hls"
|
|
if preference in ("vp8", "vp9") and container != "webm":
|
|
continue
|
|
try:
|
|
pick_encoder(preference, container)
|
|
return container
|
|
except PortalError:
|
|
continue
|
|
# Nothing matched the preference; fall back to whatever the machine has.
|
|
for container in backend.containers:
|
|
try:
|
|
pick_encoder("auto", container)
|
|
return container
|
|
except PortalError:
|
|
continue
|
|
raise PortalError("no usable encoder for this receiver")
|
|
|
|
|
|
class Session:
|
|
def __init__(self, daemon) -> None:
|
|
self.daemon = daemon
|
|
self.state = IDLE
|
|
self.phase = "" # what "starting" is waiting on, for the panel
|
|
self.device: Device | None = None
|
|
self.backend = None
|
|
self.container = ""
|
|
self.encoder = ""
|
|
self.url = ""
|
|
self.error = ""
|
|
self.since = 0.0
|
|
self.player_state = ""
|
|
self.clients = 0
|
|
self.last_client = 0.0
|
|
self.capture = Capture(on_stopped=self._capture_stopped)
|
|
self.fanout: Fanout | None = None
|
|
self.hls_dir = None
|
|
self._watchdog: asyncio.Task | None = None
|
|
self._reconnects = 0
|
|
|
|
# ---- reporting ----
|
|
@property
|
|
def active(self) -> bool:
|
|
return self.state in (STARTING, STREAMING)
|
|
|
|
def to_json(self) -> dict:
|
|
width, height = self.capture.size
|
|
return {
|
|
"state": self.state,
|
|
"phase": self.phase,
|
|
"deviceId": self.device.id if self.device else "",
|
|
"deviceName": self.device.name if self.device else "",
|
|
"deviceKind": self.device.kind if self.device else "",
|
|
"since": self.since,
|
|
"url": self.url,
|
|
"container": self.container,
|
|
"encoder": self.encoder,
|
|
"source": self.capture.source_name,
|
|
"preview": str(self.capture.preview_path) if self.capture.preview_at else "",
|
|
"previewAt": self.capture.preview_at,
|
|
"width": width,
|
|
"height": height,
|
|
"clients": self.clients,
|
|
"playerState": self.player_state,
|
|
"bytes": self.fanout.bytes_out if self.fanout else 0,
|
|
# How far behind the receiver is letting itself fall, in bytes.
|
|
"backlog": self.fanout.backlog if self.fanout else 0,
|
|
"dropped": self.fanout.dropped if self.fanout else 0,
|
|
"error": self.error,
|
|
}
|
|
|
|
# ---- start / stop ----
|
|
async def start(self, device: Device, backend) -> None:
|
|
await self.stop()
|
|
cfg = self.daemon.cfg
|
|
self.device, self.backend = device, backend
|
|
self.state, self.error, self.since = STARTING, "", time.time()
|
|
self.phase = "screen" # the portal is asking which screen to share
|
|
self.player_state, self.clients, self._reconnects = "", 0, 0
|
|
self.daemon.push_state(now=True)
|
|
|
|
try:
|
|
self.container = choose_container(backend, cfg)
|
|
hls_dir = None
|
|
if self.container == "hls":
|
|
hls_dir = runtime_dir() / "hls"
|
|
shutil.rmtree(hls_dir, ignore_errors=True)
|
|
hls_dir.mkdir(parents=True, exist_ok=True)
|
|
self.hls_dir = hls_dir
|
|
self.daemon.stream.hls_dir = hls_dir
|
|
self.fanout = None
|
|
else:
|
|
self.fanout = Fanout(self.container, int(cfg["bitrate"]))
|
|
self.daemon.stream.fanout = self.fanout
|
|
|
|
await self.daemon.stream.start(int(cfg["port"]))
|
|
await self.capture.start(
|
|
container="mpegts" if self.container == "hls" else self.container,
|
|
cfg=cfg, fanout=self.fanout,
|
|
hls_dir=str(hls_dir) if hls_dir else None,
|
|
on_phase=self._set_phase,
|
|
)
|
|
self.encoder = self.capture.encoder
|
|
|
|
ip = local_ip_for(device.host)
|
|
# A fresh token per session is both the access control and the reason
|
|
# a receiver cannot carry its old buffer - and the delay in it - into
|
|
# the new stream: to it, this is a URL it has never seen.
|
|
self.daemon.stream.new_token()
|
|
path = (self.daemon.stream.hls_path() if self.container == "hls"
|
|
else self.daemon.stream.url_path(self.container))
|
|
self.url = f"http://{url_host(ip)}:{self.daemon.stream.port}{path}"
|
|
if self.container == "hls":
|
|
await self._await_playlist()
|
|
|
|
self._set_phase("connect")
|
|
await backend.play(device, self.url, container=self.container, title="Screencast")
|
|
except asyncio.CancelledError:
|
|
await self._teardown()
|
|
self.state, self.device, self.backend = IDLE, None, None
|
|
self.daemon.push_state(now=True)
|
|
raise
|
|
except (PortalError, BackendError) as exc:
|
|
await self.fail(str(exc))
|
|
return
|
|
except Exception as exc: # noqa: BLE001 - never leave a half-started cast behind
|
|
log.exception("cast failed")
|
|
await self.fail(str(exc))
|
|
return
|
|
|
|
self.state = STREAMING
|
|
# Handed over, but nothing is on screen until the receiver fetches it.
|
|
self.phase = "live" if self.clients else "waiting"
|
|
self.daemon.push_state(now=True)
|
|
self._watchdog = asyncio.create_task(self._watch())
|
|
|
|
def _set_phase(self, phase: str) -> None:
|
|
self.phase = phase
|
|
self.daemon.push_state(now=True)
|
|
|
|
async def _await_playlist(self) -> None:
|
|
"""AirPlay fetches the playlist immediately; give hlssink2 time to write one."""
|
|
assert self.hls_dir is not None
|
|
for _ in range(100):
|
|
if (self.hls_dir / "index.m3u8").exists():
|
|
await asyncio.sleep(1.0) # one full segment, so the receiver has something to play
|
|
return
|
|
await asyncio.sleep(0.1)
|
|
raise PortalError("the encoder produced no HLS playlist")
|
|
|
|
async def fail(self, message: str) -> None:
|
|
log.warning("cast failed: %s", message)
|
|
device, backend = self.device, self.backend
|
|
await self._teardown()
|
|
self.device, self.backend = device, backend
|
|
self.state, self.error, self.phase = ERROR, message, ""
|
|
self.daemon.push_state(now=True)
|
|
|
|
async def stop(self, *, quiet: bool = False) -> None:
|
|
if self.state == IDLE and self.device is None:
|
|
return
|
|
self.state = STOPPING
|
|
if not quiet:
|
|
self.daemon.push_state(now=True)
|
|
device, backend = self.device, self.backend
|
|
if backend is not None and device is not None:
|
|
try:
|
|
await backend.stop(device)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.debug("backend stop: %s", exc)
|
|
await self._teardown()
|
|
self.state, self.error, self.phase = IDLE, "", ""
|
|
self.device, self.backend = None, None
|
|
if not quiet:
|
|
self.daemon.push_state(now=True)
|
|
|
|
async def _teardown(self) -> None:
|
|
# The watchdog is usually the one calling this (through fail()), and a
|
|
# task that cancels itself never gets to finish the job: the session
|
|
# would stay "streaming" with a dead encoder behind it, which is exactly
|
|
# what a receiver stuck on a spinner looks like from here.
|
|
watchdog, self._watchdog = self._watchdog, None
|
|
if watchdog is not None and watchdog is not asyncio.current_task():
|
|
watchdog.cancel()
|
|
await self.capture.stop()
|
|
if self.fanout is not None:
|
|
self.fanout.close()
|
|
self.fanout = None
|
|
self.daemon.stream.fanout = None
|
|
self.daemon.stream.hls_dir = None
|
|
if self.hls_dir is not None:
|
|
shutil.rmtree(self.hls_dir, ignore_errors=True)
|
|
self.hls_dir = None
|
|
self.url = ""
|
|
self.clients = 0
|
|
|
|
# ---- live ----
|
|
def note_client(self, peer: str, connected: bool) -> None:
|
|
if self.fanout is None and self.hls_dir is None:
|
|
return
|
|
self.clients = self.fanout.clients if self.fanout else self.clients + (1 if connected else -1)
|
|
self.clients = max(0, self.clients)
|
|
if connected:
|
|
self.last_client = time.time()
|
|
self._reconnects = 0
|
|
if self.state == STREAMING:
|
|
self.phase = "live" if self.clients else "waiting"
|
|
self.daemon.push_state(now=True)
|
|
|
|
def _capture_stopped(self, code: int, error: str) -> None:
|
|
if not self.active:
|
|
return
|
|
asyncio.create_task(self.fail(error or f"screen capture stopped ({code})"))
|
|
|
|
async def _watch(self) -> None:
|
|
"""Poll the receiver, and notice when the picture stops arriving."""
|
|
last_bytes, last_progress = -1, time.time()
|
|
ticks = 0
|
|
while self.active:
|
|
await asyncio.sleep(2)
|
|
ticks += 1
|
|
if self.device is None or self.backend is None:
|
|
return
|
|
# A client that connected and then went quiet (a half-open socket, a
|
|
# receiver that gave up without closing) used to keep the session
|
|
# looking healthy forever, so check the encoder itself rather than
|
|
# trusting the client count.
|
|
if not self.capture.running:
|
|
await self.fail(self.capture.error or "the screen capture stopped")
|
|
return
|
|
if self.fanout is not None:
|
|
if self.fanout.bytes_out != last_bytes:
|
|
last_bytes, last_progress = self.fanout.bytes_out, time.time()
|
|
elif time.time() - last_progress > 15:
|
|
await self.fail("the encoder stopped producing video")
|
|
return
|
|
try:
|
|
status = await self.backend.poll(self.device)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.debug("poll: %s", exc)
|
|
status = {}
|
|
if status:
|
|
self.player_state = status.get("playerState", self.player_state)
|
|
if "volume" in status and status["volume"] is not None:
|
|
self.device.volume = float(status["volume"])
|
|
if "muted" in status:
|
|
self.device.muted = bool(status["muted"])
|
|
self.device.app = status.get("app", self.device.app)
|
|
if status.get("connected") is False:
|
|
await self.fail(f"{self.device.name} dropped the connection")
|
|
return
|
|
# A receiver that never fetched the stream, or that let go of it, is
|
|
# not casting — retry the play once before giving up on it.
|
|
stale = self.fanout is not None and self.clients == 0 and time.time() - self.since > 12
|
|
if stale and time.time() - self.last_client > 12:
|
|
if bool(self.daemon.cfg["autoReconnect"]) and self._reconnects < 2:
|
|
self._reconnects += 1
|
|
log.info("no receiver on the stream; retrying play (%d)", self._reconnects)
|
|
try:
|
|
await self.backend.play(self.device, self.url,
|
|
container=self.container, title="Screencast")
|
|
self.last_client = time.time()
|
|
except BackendError as exc:
|
|
await self.fail(str(exc))
|
|
return
|
|
else:
|
|
hint = firewall_hint(self.daemon.stream.port, self.device.host)
|
|
reason = f"{self.device.name} never picked up the stream"
|
|
if hint:
|
|
# The panel turns this into an "Open port" button.
|
|
self.daemon.firewall_blocked = True
|
|
await self.fail(f"{reason} — {hint}" if hint else reason)
|
|
return
|
|
# How much of the delay is ours: bytes the receiver has not taken
|
|
# yet, and bytes we skipped past because it never would have.
|
|
if self.fanout is not None and ticks % 8 == 0 and self.clients:
|
|
rate = self.fanout.bytes_out / max(1e-3, time.time() - self.fanout.started)
|
|
log.info("stream: %.1f KiB queued (%.1fs behind), %.0f KiB skipped, %.0f KiB/s",
|
|
self.fanout.backlog / 1024, self.fanout.backlog / max(1.0, rate),
|
|
self.fanout.dropped / 1024, rate / 1024)
|
|
self.daemon.push_state(now=True)
|
|
|
|
# ---- source ----
|
|
async def repick_source(self) -> None:
|
|
"""Restart the cast so the portal asks which screen to share again."""
|
|
if self.active and self.device is not None and self.backend is not None:
|
|
await self.start(self.device, self.backend)
|