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.
636 lines
27 KiB
Python
636 lines
27 KiB
Python
"""Screen capture and encoding.
|
|
|
|
Wayland gives no direct screen access, so the picture comes from the
|
|
xdg-desktop-portal ScreenCast interface: the portal asks which screen to share -
|
|
every time, deliberately, since nothing here is worth sharing by accident - and
|
|
hands back a PipeWire node and a fd for it. GStreamer reads that node, encodes,
|
|
and writes the muxed stream to a pipe the HTTP server fans out to the receiver.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import os
|
|
import re
|
|
import secrets
|
|
import time
|
|
|
|
from dbus_fast import BusType, Message, MessageType, Variant
|
|
from dbus_fast.aio import MessageBus
|
|
|
|
from .util import gst_has, log, runtime_dir, state_dir
|
|
|
|
JPEG_START = b"\xff\xd8\xff"
|
|
JPEG_END = b"\xff\xd9"
|
|
|
|
PORTAL = "org.freedesktop.portal.Desktop"
|
|
PORTAL_PATH = "/org/freedesktop/portal/desktop"
|
|
SCREENCAST = "org.freedesktop.portal.ScreenCast"
|
|
|
|
SOURCE_MONITOR = 1
|
|
SOURCE_WINDOW = 2
|
|
CURSOR_HIDDEN, CURSOR_EMBEDDED = 1, 2
|
|
PERSIST_NONE = 0 # never remember a screen: every cast asks again
|
|
|
|
|
|
class PortalError(Exception):
|
|
pass
|
|
|
|
|
|
class PortalStream:
|
|
def __init__(self, node_id: int, fd: int, props: dict) -> None:
|
|
self.node_id = node_id
|
|
self.fd = fd
|
|
self.props = props
|
|
size = props.get("size")
|
|
self.width, self.height = (int(size[0]), int(size[1])) if size else (0, 0)
|
|
self.source_type = int(props.get("source_type", SOURCE_MONITOR))
|
|
self.name = str(props.get("id", "")) or ("Window" if self.source_type == SOURCE_WINDOW else "Screen")
|
|
|
|
|
|
class Portal:
|
|
"""One ScreenCast session. The bus connection must outlive the cast."""
|
|
|
|
def __init__(self) -> None:
|
|
self.bus: MessageBus | None = None
|
|
self.session: str | None = None
|
|
self._pending: dict[str, asyncio.Future] = {}
|
|
|
|
async def _connect(self) -> MessageBus:
|
|
if self.bus is None or self.bus._disconnected: # noqa: SLF001 - no public flag
|
|
self.bus = await MessageBus(bus_type=BusType.SESSION, negotiate_unix_fd=True).connect()
|
|
await self.bus.call(
|
|
Message(
|
|
destination="org.freedesktop.DBus",
|
|
path="/org/freedesktop/DBus",
|
|
interface="org.freedesktop.DBus",
|
|
member="AddMatch",
|
|
signature="s",
|
|
body=[
|
|
"type='signal',interface='org.freedesktop.portal.Request',member='Response'"
|
|
],
|
|
)
|
|
)
|
|
self.bus.add_message_handler(self._on_signal)
|
|
return self.bus
|
|
|
|
def _on_signal(self, msg: Message):
|
|
if msg.message_type is not MessageType.SIGNAL or msg.member != "Response":
|
|
return
|
|
fut = self._pending.pop(msg.path, None)
|
|
if fut and not fut.done():
|
|
code, results = msg.body[0], msg.body[1]
|
|
fut.set_result((int(code), {k: v.value for k, v in results.items()}))
|
|
|
|
def _request_path(self, token: str) -> str:
|
|
sender = self.bus.unique_name[1:].replace(".", "_")
|
|
return f"{PORTAL_PATH}/request/{sender}/{token}"
|
|
|
|
async def _call(self, member: str, signature: str, body: list, options: dict, timeout: float):
|
|
"""Call a portal method and wait for the Request.Response it answers with."""
|
|
token = "sc" + secrets.token_hex(8)
|
|
options = dict(options)
|
|
options["handle_token"] = Variant("s", token)
|
|
path = self._request_path(token)
|
|
fut: asyncio.Future = asyncio.get_running_loop().create_future()
|
|
self._pending[path] = fut
|
|
reply = await self.bus.call(
|
|
Message(
|
|
destination=PORTAL,
|
|
path=PORTAL_PATH,
|
|
interface=SCREENCAST,
|
|
member=member,
|
|
signature=signature,
|
|
body=body + [options],
|
|
)
|
|
)
|
|
if reply.message_type is MessageType.ERROR:
|
|
self._pending.pop(path, None)
|
|
raise PortalError(f"{member}: {reply.body[0] if reply.body else reply.error_name}")
|
|
try:
|
|
code, results = await asyncio.wait_for(fut, timeout)
|
|
except asyncio.TimeoutError as exc:
|
|
self._pending.pop(path, None)
|
|
raise PortalError("no screen was picked (the share dialog was left unanswered)") from exc
|
|
if code == 1:
|
|
raise PortalError("no screen was picked (the share dialog was dismissed)")
|
|
if code != 0:
|
|
raise PortalError(f"{member}: portal returned {code}")
|
|
return results
|
|
|
|
async def open(self, *, cursor: bool, allow_windows: bool = True) -> PortalStream:
|
|
"""Start a capture session. The portal always asks which screen to share."""
|
|
await self._connect()
|
|
stoken = "sc" + secrets.token_hex(8)
|
|
results = await self._call(
|
|
"CreateSession", "a{sv}", [], {"session_handle_token": Variant("s", stoken)}, 20
|
|
)
|
|
self.session = results.get("session_handle") or f"{PORTAL_PATH}/session/{self.bus.unique_name[1:].replace('.', '_')}/{stoken}"
|
|
|
|
types = SOURCE_MONITOR | (SOURCE_WINDOW if allow_windows else 0)
|
|
opts = {
|
|
"types": Variant("u", types),
|
|
"multiple": Variant("b", False),
|
|
"cursor_mode": Variant("u", CURSOR_EMBEDDED if cursor else CURSOR_HIDDEN),
|
|
"persist_mode": Variant("u", PERSIST_NONE),
|
|
}
|
|
# xdg-desktop-portal-hyprland shows its share picker during
|
|
# SelectSources, not Start, so this is the call that sits waiting for a
|
|
# human. Other portals prompt on Start; both get room.
|
|
await self._call("SelectSources", "oa{sv}", [self.session], opts, 180)
|
|
results = await self._call("Start", "osa{sv}", [self.session, ""], {}, 180)
|
|
streams = results.get("streams") or []
|
|
if not streams:
|
|
raise PortalError("the portal returned no stream")
|
|
node_id, props = streams[0][0], {k: v.value if isinstance(v, Variant) else v
|
|
for k, v in streams[0][1].items()}
|
|
reply = await self.bus.call(
|
|
Message(
|
|
destination=PORTAL,
|
|
path=PORTAL_PATH,
|
|
interface=SCREENCAST,
|
|
member="OpenPipeWireRemote",
|
|
signature="oa{sv}",
|
|
body=[self.session, {}],
|
|
)
|
|
)
|
|
if reply.message_type is MessageType.ERROR:
|
|
raise PortalError(f"OpenPipeWireRemote: {reply.error_name}")
|
|
fd = reply.unix_fds[reply.body[0]]
|
|
return PortalStream(int(node_id), fd, props)
|
|
|
|
async def close(self) -> None:
|
|
if self.bus is not None and self.session:
|
|
try:
|
|
await self.bus.call(
|
|
Message(
|
|
destination=PORTAL,
|
|
path=self.session,
|
|
interface="org.freedesktop.portal.Session",
|
|
member="Close",
|
|
)
|
|
)
|
|
except Exception: # noqa: BLE001 - the session may already be gone
|
|
pass
|
|
self.session = None
|
|
if self.bus is not None:
|
|
self.bus.disconnect()
|
|
self.bus = None
|
|
|
|
|
|
def drop_legacy_token() -> None:
|
|
"""Earlier versions remembered the shared screen; delete what they stored."""
|
|
with contextlib.suppress(OSError):
|
|
(state_dir() / "restore-token.json").unlink()
|
|
|
|
|
|
def pick_encoder(preference: str, container: str) -> str:
|
|
"""Resolve "auto" against what this machine's GStreamer actually has."""
|
|
if container == "webm":
|
|
wanted = ["vp8", "vp9"] if preference in ("auto", "vp8", "nvenc", "vaapi", "x264") else [preference]
|
|
for enc in wanted:
|
|
if gst_has({"vp8": "vp8enc", "vp9": "vp9enc"}[enc]):
|
|
return enc
|
|
raise PortalError("no VP8/VP9 encoder (install gst-plugins-good)")
|
|
order = {
|
|
"auto": ["nvenc", "vaapi", "x264", "openh264"],
|
|
"nvenc": ["nvenc", "vaapi", "x264", "openh264"],
|
|
"vaapi": ["vaapi", "nvenc", "x264", "openh264"],
|
|
"x264": ["x264", "openh264", "nvenc", "vaapi"],
|
|
}.get(preference, ["nvenc", "vaapi", "x264", "openh264"])
|
|
elements = {"nvenc": "nvh264enc", "vaapi": "vah264enc", "x264": "x264enc", "openh264": "openh264enc"}
|
|
for enc in order:
|
|
if gst_has(elements[enc]):
|
|
return enc
|
|
raise PortalError("no H.264 encoder (install gst-plugins-ugly or gst-plugin-va)")
|
|
|
|
|
|
def video_encoder_chain(encoder: str, bitrate_kbps: int, fps: int, *,
|
|
gpu_scale: bool = False) -> list[str]:
|
|
# Half a second between keyframes. It costs a little bitrate and buys two
|
|
# things that matter more here: a receiver can start on the next keyframe
|
|
# instead of waiting a whole second, and the stream can be cut at one when a
|
|
# slow receiver has to be skipped forward (see Fanout).
|
|
gop = max(15, fps // 2)
|
|
if encoder == "nvenc":
|
|
upload = [] if gpu_scale else [
|
|
"cudaupload", "!", "cudaconvertscale", "!", "video/x-raw(memory:CUDAMemory),format=NV12", "!",
|
|
]
|
|
return upload + [
|
|
"nvh264enc", "name=venc", f"bitrate={bitrate_kbps}", f"gop-size={gop}",
|
|
"rc-mode=cbr", "preset=low-latency-hq", "zerolatency=true", "!",
|
|
"h264parse", "config-interval=-1",
|
|
]
|
|
if encoder == "vaapi":
|
|
return [
|
|
"vah264enc", "name=venc", f"bitrate={bitrate_kbps}", f"key-int-max={gop}",
|
|
"rate-control=cbr", "!", "h264parse", "config-interval=-1",
|
|
]
|
|
if encoder == "x264":
|
|
return [
|
|
"x264enc", "name=venc", f"bitrate={bitrate_kbps}", "tune=zerolatency",
|
|
"speed-preset=veryfast", f"key-int-max={gop}", "!", "h264parse", "config-interval=-1",
|
|
]
|
|
if encoder == "openh264":
|
|
return [
|
|
"openh264enc", "name=venc", f"bitrate={bitrate_kbps * 1000}", f"gop-size={gop}", "!",
|
|
"h264parse", "config-interval=-1",
|
|
]
|
|
if encoder == "vp8":
|
|
return [
|
|
"vp8enc", "name=venc", "deadline=1", "cpu-used=6", "threads=8", "end-usage=cbr",
|
|
f"target-bitrate={bitrate_kbps * 1000}", f"keyframe-max-dist={gop}", "error-resilient=1",
|
|
]
|
|
if encoder == "vp9":
|
|
return [
|
|
"vp9enc", "name=venc", "deadline=1", "cpu-used=8", "threads=8", "end-usage=cbr",
|
|
f"target-bitrate={bitrate_kbps * 1000}", f"keyframe-max-dist={gop}",
|
|
]
|
|
raise PortalError(f"unknown encoder {encoder}")
|
|
|
|
|
|
def audio_chain(container: str) -> list[str]:
|
|
if container == "webm":
|
|
return ["opusenc", "bitrate=128000", "!", "queue"]
|
|
if gst_has("avenc_aac"):
|
|
return ["avenc_aac", "bitrate=128000", "!", "aacparse", "!", "queue"]
|
|
if gst_has("fdkaacenc"):
|
|
return ["fdkaacenc", "bitrate=128000", "!", "aacparse", "!", "queue"]
|
|
return ["opusenc", "bitrate=128000", "!", "queue"]
|
|
|
|
|
|
def muxer(container: str) -> list[str]:
|
|
if container == "webm":
|
|
# Just under the keyframe interval, so matroskamux closes each cluster on
|
|
# a keyframe: that is what makes a cluster a safe place to cut.
|
|
return ["webmmux", "name=mux", "streamable=true", "min-cluster-duration=400000000"]
|
|
if container == "mp4":
|
|
return ["mp4mux", "name=mux", "streamable=true", "fragment-duration=200",
|
|
"faststart=false", "presentation-time=0"]
|
|
if container == "mpegts":
|
|
return ["mpegtsmux", "name=mux", "alignment=7"]
|
|
if container == "matroska":
|
|
return ["matroskamux", "name=mux", "streamable=true"]
|
|
raise PortalError(f"unknown container {container}")
|
|
|
|
|
|
def preview_size(width: int, height: int, target: int = 480) -> tuple[int, int]:
|
|
"""Preview dimensions: `target` wide at the source aspect, both even."""
|
|
if width <= 0 or height <= 0:
|
|
return target, (target * 9 // 16 // 2) * 2
|
|
h = max(2, round(height * target / width))
|
|
return (target // 2) * 2, (h // 2) * 2
|
|
|
|
|
|
def build_pipeline(stream: PortalStream, *, container: str, encoder: str, fps: int,
|
|
max_height: int, bitrate_kbps: int, audio_node: str | None,
|
|
hls_dir: str | None = None, gpu_scale: bool = False,
|
|
preview_fd: int | None = None) -> list[str]:
|
|
"""The gst-launch argv for one session (fd 1 carries the muxed stream)."""
|
|
args = ["gst-launch-1.0", "-q"]
|
|
# Scale first, pace second: the pacing element repeats frames, and repeating
|
|
# a 1080p frame is far cheaper than repeating a 4K one.
|
|
args += [
|
|
"pipewiresrc", f"fd={stream.fd}", f"path={stream.node_id}", "do-timestamp=true",
|
|
"keepalive-time=1000", "resend-last=true", "!",
|
|
"queue", "max-size-time=100000000", "leaky=downstream", "!",
|
|
]
|
|
out_w, out_h = output_size(stream.width, stream.height, max_height)
|
|
size_caps = f",width={out_w},height={out_h}" if out_w and out_h else f",height=[16,{max_height}]"
|
|
if gpu_scale:
|
|
# Colour conversion and scaling on the GPU: a 4K screen is otherwise the
|
|
# most expensive thing in the pipeline, ahead of the encoder itself.
|
|
args += [
|
|
"cudaupload", "!", "cudaconvertscale", "!",
|
|
f"video/x-raw(memory:CUDAMemory),format=NV12{size_caps},pixel-aspect-ratio=1/1", "!",
|
|
]
|
|
if encoder not in ("nvenc",):
|
|
args += ["cudadownload", "!", "videoconvert", "!", "video/x-raw,format=I420", "!"]
|
|
else:
|
|
args += [
|
|
"videoconvert", "!", "videoscale", "method=lanczos", "!",
|
|
f"video/x-raw{size_caps},pixel-aspect-ratio=1/1", "!",
|
|
]
|
|
on_gpu = gpu_scale and encoder == "nvenc"
|
|
# A Wayland screen only produces a frame when something on it changes, and
|
|
# videorate can only duplicate once the *next* frame arrives: on a desk left
|
|
# alone the encoder simply stops, the muxer stops, and the receiver sits on a
|
|
# spinner waiting for data that is not coming. A compositor is clock-driven
|
|
# like audiomixer - it emits the last frame again on schedule - so the stream
|
|
# keeps its framerate no matter how still the screen is.
|
|
mixer = "cudacompositor" if (on_gpu and gst_has("cudacompositor")) else "compositor"
|
|
if mixer == "compositor" and on_gpu:
|
|
args += ["cudadownload", "!", "videoconvert", "!", "video/x-raw,format=I420", "!"]
|
|
on_gpu = False
|
|
# Pin the size here, not just the framerate. A compositor places its input
|
|
# at 0,0 and does not scale it, so anything downstream that renegotiates a
|
|
# smaller size (the preview branch's videoscale will happily propose its
|
|
# own 480x270 back through the tee) does not shrink the picture - it crops
|
|
# it to the top-left corner and encodes that.
|
|
size_pin = f",width={out_w},height={out_h}" if out_w and out_h else ""
|
|
rate_caps = (("video/x-raw(memory:CUDAMemory)" if on_gpu else "video/x-raw")
|
|
+ f"{size_pin},framerate={fps}/1")
|
|
args += [
|
|
mixer, "name=vmix", "latency=60000000", "start-time-selection=first",
|
|
# A pad property, not an element one: repeat the last frame for as long
|
|
# as the screen stays still instead of falling back to black.
|
|
"sink_0::max-last-buffer-repeat=-1",
|
|
"!", rate_caps, "!",
|
|
"queue", "max-size-time=100000000", "leaky=downstream", "!",
|
|
]
|
|
if preview_fd is not None:
|
|
# A second, deliberately cheap branch: a few frames a second, small, and
|
|
# leaky, so the panel can show what is actually going out without the
|
|
# preview ever being able to hold the encoder up.
|
|
pw, ph = preview_size(out_w, out_h)
|
|
args += ["tee", "name=vt",
|
|
"vt.", "!", "queue", "max-size-buffers=1", "leaky=downstream", "!"]
|
|
if on_gpu:
|
|
args += ["cudadownload", "!"]
|
|
args += [
|
|
"videorate", "drop-only=true", "!", "video/x-raw,framerate=2/1", "!",
|
|
"videoconvert", "!", "videoscale", "!", f"video/x-raw,width={pw},height={ph}", "!",
|
|
"jpegenc", "quality=65", "!", "fdsink", f"fd={preview_fd}", "sync=false",
|
|
"vt.", "!", "queue", "max-size-time=100000000", "leaky=downstream", "!",
|
|
]
|
|
|
|
args += video_encoder_chain(encoder, bitrate_kbps, fps, gpu_scale=on_gpu)
|
|
# hlssink2 does its own muxing and takes the elementary streams on request
|
|
# pads; everything else goes through one muxer into the pipe.
|
|
hls = hls_dir is not None
|
|
if hls and not gst_safe(hls_dir):
|
|
raise RuntimeError(f"runtime directory {hls_dir!r} cannot go in a pipeline")
|
|
args += ["!", "queue", "!", "hls.video" if hls else "mux."]
|
|
|
|
if audio_node:
|
|
# The monitor of an idle sink can go quiet for minutes at a time (a
|
|
# Bluetooth headset suspends outright), and a muxer with a declared but
|
|
# silent audio track is exactly what leaves a receiver on a spinner:
|
|
# players wait for that track's first packet. Mixing the monitor with a
|
|
# live silence source keeps audio flowing whatever the speakers do.
|
|
rate_caps = "audio/x-raw,format=S16LE,rate=48000,channels=2,layout=interleaved"
|
|
args += [
|
|
"audiomixer", "name=amix", "latency=60000000", "start-time-selection=first", "!",
|
|
rate_caps, "!", "audioconvert", "!",
|
|
]
|
|
args += audio_chain(container)
|
|
args += ["!", "hls.audio" if hls else "mux."]
|
|
args += [
|
|
"pipewiresrc", f"target-object={audio_node}",
|
|
"stream-properties=p,stream.capture.sink=true", "do-timestamp=true", "!",
|
|
"audio/x-raw", "!", "audioconvert", "!", "audioresample", "!", rate_caps, "!",
|
|
"queue", "max-size-time=100000000", "leaky=downstream", "!", "amix.",
|
|
"audiotestsrc", "is-live=true", "wave=silence", "!", rate_caps, "!",
|
|
"queue", "max-size-time=100000000", "leaky=downstream", "!", "amix.",
|
|
]
|
|
|
|
if hls:
|
|
args += ["hlssink2", "name=hls", f"location={hls_dir}/segment%05d.ts",
|
|
f"playlist-location={hls_dir}/index.m3u8", "target-duration=2", "max-files=6",
|
|
"playlist-length=4"]
|
|
else:
|
|
args += muxer(container) + ["!", "fdsink", "fd=1", "sync=false"]
|
|
return args
|
|
|
|
|
|
def output_size(width: int, height: int, max_height: int) -> tuple[int, int]:
|
|
"""The size to encode at: capped to max_height, aspect kept, both even.
|
|
|
|
Encoders want even dimensions, and the scalers want a fixed size — an
|
|
open-ended caps range makes cudaconvertscale fail to negotiate at all.
|
|
"""
|
|
if width <= 0 or height <= 0:
|
|
return 0, 0
|
|
if height > max_height:
|
|
width = round(width * max_height / height)
|
|
height = max_height
|
|
return (width // 2) * 2, (height // 2) * 2
|
|
|
|
|
|
class Capture:
|
|
"""Portal session + gst process, feeding a Fanout."""
|
|
|
|
def __init__(self, on_stopped=None) -> None:
|
|
self.portal: Portal | None = None
|
|
self.proc: asyncio.subprocess.Process | None = None
|
|
self.stream: PortalStream | None = None
|
|
self.encoder = ""
|
|
self.container = ""
|
|
self.error = ""
|
|
self.on_stopped = on_stopped
|
|
# A small JPEG of what is actually going out, refreshed a couple of
|
|
# times a second, so the panel can show the screen it is sharing.
|
|
self.preview_path = runtime_dir() / "preview.jpg"
|
|
self.preview_at = 0.0
|
|
self._tasks: list[asyncio.Task] = []
|
|
self._stderr = ""
|
|
|
|
@property
|
|
def running(self) -> bool:
|
|
return self.proc is not None and self.proc.returncode is None
|
|
|
|
@property
|
|
def size(self) -> tuple[int, int]:
|
|
return (self.stream.width, self.stream.height) if self.stream else (0, 0)
|
|
|
|
@property
|
|
def source_name(self) -> str:
|
|
return self.stream.name if self.stream else ""
|
|
|
|
async def start(self, *, container: str, cfg, fanout,
|
|
hls_dir: str | None = None, on_phase=None) -> None:
|
|
await self.stop()
|
|
self.error = ""
|
|
self.container = container
|
|
self.encoder = pick_encoder(str(cfg["encoder"]), container)
|
|
self.portal = Portal()
|
|
self.stream = await self.portal.open(cursor=bool(cfg["cursor"]))
|
|
if on_phase:
|
|
on_phase("encoder")
|
|
|
|
audio_node = default_sink_node() if cfg["audio"] else None
|
|
preview_r, preview_w = os.pipe()
|
|
argv = build_pipeline(
|
|
self.stream, container=container, encoder=self.encoder, fps=int(cfg["fps"]),
|
|
max_height=int(cfg["maxHeight"]), bitrate_kbps=int(cfg["bitrate"]),
|
|
audio_node=audio_node, hls_dir=hls_dir, gpu_scale=gst_has("cudaconvertscale"),
|
|
preview_fd=preview_w,
|
|
)
|
|
log.info("capture: %dx%d %s/%s @%sfps%s", self.stream.width, self.stream.height,
|
|
self.encoder, container, cfg["fps"], " +audio" if audio_node else "")
|
|
log.debug("gst: %s", " ".join(argv))
|
|
self.proc = await asyncio.create_subprocess_exec(
|
|
*argv,
|
|
stdout=asyncio.subprocess.PIPE if not hls_dir else asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
pass_fds=(self.stream.fd, preview_w),
|
|
)
|
|
os.close(self.stream.fd) # the child owns it now
|
|
self.stream.fd = -1
|
|
os.close(preview_w) # ...and the write end of the preview pipe
|
|
self._tasks.append(asyncio.create_task(self._pump_preview(preview_r)))
|
|
if not hls_dir:
|
|
self._tasks.append(asyncio.create_task(self._pump(fanout)))
|
|
self._tasks.append(asyncio.create_task(self._watch_stderr()))
|
|
self._tasks.append(asyncio.create_task(self._wait()))
|
|
|
|
async def _pump(self, fanout) -> None:
|
|
assert self.proc and self.proc.stdout
|
|
try:
|
|
while True:
|
|
chunk = await self.proc.stdout.read(65536)
|
|
if not chunk:
|
|
break
|
|
fanout.feed(chunk)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
log.debug("capture pump ended: %s", exc)
|
|
|
|
async def _pump_preview(self, fd: int) -> None:
|
|
"""JPEGs off the preview branch, published one whole frame at a time.
|
|
|
|
The panel reads the file while gst keeps writing, so each frame is
|
|
written beside it and renamed over the top - never half a picture.
|
|
"""
|
|
loop = asyncio.get_running_loop()
|
|
reader = asyncio.StreamReader()
|
|
try:
|
|
await loop.connect_read_pipe(
|
|
lambda: asyncio.StreamReaderProtocol(reader), os.fdopen(fd, "rb", 0))
|
|
except Exception as exc: # noqa: BLE001
|
|
log.debug("preview pipe: %s", exc)
|
|
return
|
|
buf = b""
|
|
try:
|
|
while True:
|
|
chunk = await reader.read(65536)
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
while True:
|
|
end = buf.find(JPEG_END)
|
|
if end < 0:
|
|
break
|
|
frame, buf = buf[: end + 2], buf[end + 2 :]
|
|
start = frame.rfind(JPEG_START)
|
|
if start > 0:
|
|
frame = frame[start:]
|
|
if len(frame) > 1024:
|
|
self._publish_preview(frame)
|
|
if len(buf) > 1 << 20: # never a whole frame: give up on it
|
|
buf = b""
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
log.debug("preview ended: %s", exc)
|
|
|
|
def _publish_preview(self, frame: bytes) -> None:
|
|
tmp = self.preview_path.with_suffix(".tmp")
|
|
try:
|
|
tmp.write_bytes(frame)
|
|
os.replace(tmp, self.preview_path)
|
|
self.preview_at = time.time()
|
|
except OSError as exc:
|
|
log.debug("preview write: %s", exc)
|
|
|
|
def clear_preview(self) -> None:
|
|
self.preview_at = 0.0
|
|
with contextlib.suppress(OSError):
|
|
self.preview_path.unlink()
|
|
|
|
async def _watch_stderr(self) -> None:
|
|
assert self.proc and self.proc.stderr
|
|
while True:
|
|
line = await self.proc.stderr.readline()
|
|
if not line:
|
|
break
|
|
text = line.decode(errors="replace").rstrip()
|
|
if not text:
|
|
continue
|
|
self._stderr = (self._stderr + "\n" + text)[-2000:]
|
|
log.warning("gst: %s", text)
|
|
if "ERROR" in text and not self.error:
|
|
# gst-launch keeps running after a negotiation failure; stop it
|
|
# so the session reports the problem instead of hanging.
|
|
self.error = _clean_error(text)
|
|
proc = self.proc
|
|
if proc is not None and proc.returncode is None:
|
|
proc.terminate()
|
|
|
|
async def _wait(self) -> None:
|
|
assert self.proc
|
|
code = await self.proc.wait()
|
|
if code not in (0, -15) and not self.error:
|
|
self.error = _first_error(self._stderr) or f"the encoder exited ({code})"
|
|
log.info("capture stopped (%s)", code)
|
|
if self.on_stopped:
|
|
self.on_stopped(code, self.error)
|
|
|
|
async def stop(self) -> None:
|
|
proc, self.proc = self.proc, None
|
|
for t in self._tasks:
|
|
t.cancel()
|
|
self._tasks = []
|
|
if proc is not None and proc.returncode is None:
|
|
proc.terminate()
|
|
try:
|
|
await asyncio.wait_for(proc.wait(), 3)
|
|
except asyncio.TimeoutError:
|
|
proc.kill()
|
|
if self.stream is not None and self.stream.fd >= 0:
|
|
try:
|
|
os.close(self.stream.fd)
|
|
except OSError:
|
|
pass
|
|
self.stream = None
|
|
self.clear_preview()
|
|
if self.portal is not None:
|
|
await self.portal.close()
|
|
self.portal = None
|
|
|
|
|
|
def _first_error(text: str) -> str:
|
|
for line in text.splitlines():
|
|
if "ERROR" in line:
|
|
return _clean_error(line)
|
|
return ""
|
|
|
|
|
|
def _clean_error(line: str) -> str:
|
|
""""ERROR: from element /GstPipeline:…/GstFoo:foo0: msg" → "foo0: msg"."""
|
|
msg = line.split("ERROR", 1)[1].strip(" :")
|
|
if msg.startswith("from element"):
|
|
msg = msg[len("from element"):].strip()
|
|
msg = msg.rsplit("/", 1)[-1]
|
|
return msg[:200]
|
|
|
|
|
|
# gst-launch-1.0 joins its argv back into one string and re-parses it, so a
|
|
# value containing a space, '!' or ',' does not stay a value - it becomes more
|
|
# pipeline. Everything else we pass is a literal or a clamped int; these two are
|
|
# not, so they get checked.
|
|
GST_SAFE = re.compile(r"^[A-Za-z0-9._:@/+-]+$")
|
|
|
|
|
|
def gst_safe(value: str) -> bool:
|
|
return bool(GST_SAFE.match(value))
|
|
|
|
|
|
def default_sink_node() -> str | None:
|
|
"""node.name of the default audio output, captured in monitor mode."""
|
|
import subprocess
|
|
|
|
for cmd in (["pactl", "get-default-sink"],):
|
|
try:
|
|
out = subprocess.run(cmd, capture_output=True, text=True, timeout=3)
|
|
name = out.stdout.strip()
|
|
if out.returncode == 0 and name and name != "@DEFAULT_SINK@":
|
|
if not gst_safe(name):
|
|
log.warning("audio: sink name %r cannot go in a pipeline", name)
|
|
return None
|
|
return name
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
return None
|