Как подключиться к конвейеру GStreamer ⇐ Python

[Anonymous]

Я пытаюсь сделать снимок экрана в KDE и Wayland. Как я могу это сделать? Я делаю это неправильно?
Я попробовал код ниже, но получаю

gi.repository.GLib.GError: g-io-error-quark: GDBus.Error:org.freedesktop.portal.Error.NotAllowed: службы определения местоположения отключены (36)

онлайн

response = screen_cast.CreateSession(options)

Мой код:

Код: Выделить всё

#!/usr/bin/env python3
import gi

gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib

import cv2
import numpy as np
import sys
from pydbus import SessionBus

# ---------------------------
# 1️⃣ Initialize GStreamer
# ---------------------------
Gst.init(None)

# ---------------------------
# 2️⃣ Request screencast session via KDE portal
# ---------------------------
bus = SessionBus()
screen_cast = bus.get("org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop")

# Set options for the portal
# Here, capture monitor (1) with cursor
options = {
"session_handle_token": GLib.Variant("s", "kde_screencast_example"),
"types": GLib.Variant("as", ["monitor"]),  # 'as' = array of strings
"multiple": GLib.Variant("b", False),     # 'b' = boolean
"cursor_mode": GLib.Variant("s", "metadata")
}

# This call pops up the KDE screen selection dialog
# It returns a PipeWire node ID in the response
print("📌 Requesting screencast session...")
response = screen_cast.CreateSession(options)
session_path = response  # D-Bus object path for the session

# Start the session and select sources
screen_cast.Start(session_path, {})

# ---------------------------
# 3️⃣ Connect to PipeWire node
# ---------------------------
# The portal will emit the PipeWire node ID via D-Bus property
# Wait for it to appear
import time

node_id = None
for _ in range(30):  # wait max 15 seconds
props = screen_cast.GetSession(session_path)
if "streams" in props and props["streams"]:
node_id = props["streams"][0]["pipewire_node_id"]
break
time.sleep(0.5)

if node_id is None:
print("❌ Failed to get PipeWire node from portal")
sys.exit(1)

print(f"✅ Got PipeWire node: {node_id}")

# ---------------------------
# 4️⃣ Build GStreamer pipeline
# ---------------------------
# Connect pipewiresrc to appsink for OpenCV
pipeline_desc = (
f"pipewiresrc node={node_id} ! "
"video/x-raw,format=RGBA ! videoconvert ! "
"appsink name=sink max-buffers=1 drop=true emit-signals=true"
)
pipeline = Gst.parse_launch(pipeline_desc)
appsink = pipeline.get_by_name("sink")
appsink.set_property("emit-signals", True)

# ---------------------------
# 5️⃣ Callback to process frames
# ---------------------------
def on_new_sample(sink):
(omited)

appsink.connect("new-sample", on_new_sample)

# ---------------------------
# 6️⃣ Start pipeline
# ---------------------------
pipeline.set_state(Gst.State.PLAYING)

# ---------------------------
# 7️⃣ Run main loop
# ---------------------------
loop = GLib.MainLoop()
try:
loop.run()
except KeyboardInterrupt:
pipeline.set_state(Gst.State.NULL)
cv2.destroyAllWindows()

Я знаю о переходе на X11. Но я бы хотел сделать это чисто.

Подробнее здесь: https://stackoverflow.com/questions/798 ... r-pipeline