#3750 freedesktop portal screencast support

xpra/codecs/gstreamer/capture.py

@@ -7,6 +7,7 @@
 from gi.repository import GObject  # @UnresolvedImport
 
 from xpra.gst_common import  import_gst
+from xpra.gtk_common.gobject_util import one_arg_signal
 from xpra.gst_pipeline import Pipeline, GST_FLOW_OK
 from xpra.codecs.gstreamer.codec_common import (
     get_version, get_type, get_info,
@@ -27,6 +28,7 @@ class Capture(Pipeline):
     Uses a GStreamer pipeline to capture the screen
     """
     __gsignals__ = Pipeline.__generic_signals__.copy()
+    __gsignals__["new-image"] = one_arg_signal
 
     def __init__(self, element : str="ximagesrc", pixel_format : str="BGRX",
                  width : int=0, height : int=0):
@@ -38,14 +40,15 @@ def __init__(self, element : str="ximagesrc", pixel_format : str="BGRX",
         self.framerate : int  = 10
         self.image = Queue(maxsize=1)
         self.create_pipeline(element)
+        assert width>0 and height>0
 
     def create_pipeline(self, capture_element:str="ximagesrc"):
         #CAPS = f"video/x-raw,width={self.width},height={self.height},format=(string){self.pixel_format},framerate={self.framerate}/1,interlace=progressive"
         elements = [
             capture_element,   #ie: ximagesrc
-            f"video/x-raw,framerate={self.framerate}/1",
+            #f"video/x-raw,framerate={self.framerate}/1",
             "videoconvert",
-            "videorate",
+            #"videorate",
             #"videoscale ! video/x-raw,width=800,height=600 ! autovideosink
             "appsink name=sink emit-signals=true max-buffers=10 drop=true sync=false async=false qos=false",
             ]
@@ -77,6 +80,8 @@ def on_new_sample(self, _bus):
                 self.image.put_nowait(image)
             except Full:
                 log("image queue is already full")
+            else:
+                self.emit("new-image", self.frames)
         return GST_FLOW_OK
 
     def on_new_preroll(self, _appsink):
@@ -85,6 +90,8 @@ def on_new_preroll(self, _appsink):
 
     def get_image(self, x:int=0, y:int=0, width:int=0, height:int=0):
         log("get_image%s", (x, y, width, height))
+        if self.state=="stopped":
+            return None
         try:
             return self.image.get(timeout=5)
         except Empty:

xpra/gst_pipeline.py

@@ -12,7 +12,7 @@
 Gst = import_gst()
 if not Gst:
     raise ImportError("GStreamer bindings not found")
-from gi.repository import GLib, GObject
+from gi.repository import GLib, GObject  # @UnresolvedImport
 
 from xpra.util import AtomicInteger, noerr, first_time
 from xpra.gtk_common.gobject_util import one_arg_signal
@@ -273,6 +273,9 @@ def on_message(self, _bus, message):
                             self.gstlogwarn("                  %s", l.strip("\n\r"))
         elif t in (Gst.MessageType.NEED_CONTEXT, Gst.MessageType.HAVE_CONTEXT):
             log("context message: %s", message)
+        elif t == Gst.MessageType.QOS:
+            qos = message.parse_qos()
+            log.warn(f"qos={qos}")
         else:
             self.gstlogwarn("unhandled bus message type %s: %s", t, message)
         self.emit_info()

xpra/platform/xposix/screencast.py

@@ -9,9 +9,11 @@
 from enum import IntEnum
 
 from xpra.exit_codes import ExitCode
+from xpra.util import typedict
 from xpra.dbus.common import loop_init, init_session_bus
+from xpra.dbus.helper import dbus_to_native
 from xpra.codecs.gstreamer.capture import Capture
-#from xpra.server.shadow.root_window_model import RootWindowModel
+from xpra.server.shadow.root_window_model import RootWindowModel
 from xpra.server.shadow.gtk_shadow_server_base import GTKShadowServerBase
 from xpra.log import Logger
 
@@ -21,7 +23,7 @@
 PORTAL_REQUEST = "org.freedesktop.portal.Request"
 PORTAL_DESKTOP_INTERFACE = "org.freedesktop.portal.Desktop"
 PORTAL_DESKTOP_PATH = "/org/freedesktop/portal/desktop"
-SCREENCAST_IFACE = 'org.freedesktop.portal.ScreenCast'
+SCREENCAST_IFACE = "org.freedesktop.portal.ScreenCast"
 
 loop_init()
 bus = init_session_bus()
@@ -43,16 +45,39 @@ def __init__(self, multi_window=True):
         self.session_handler = 0
         self.portal_interface = None
         self.dbus_sender_name : str = re.sub(r'\.', r'_', bus.get_unique_name()[1:])
+        self.portal_interface = bus.get_object(PORTAL_DESKTOP_INTERFACE, PORTAL_DESKTOP_PATH)
+        log(f"setup_capture() self.portal_interface={self.portal_interface}")
 
     #def init(self, opts):
     #    GTKShadowServerBase.init(self, opts)
 
+    def notify_new_user(self, ss):
+        log("notify_new_user() start capture")
+        super().notify_new_user(ss)
+        if not self._window_to_id:
+            self.dbus_request_screenscast()
+
+    def last_client_exited(self):
+        super().last_client_exited()
+        c = self.capture
+        if c:
+            self.capture = None
+            c.stop()
+
+
+    def makeRootWindowModels(self):
+        log("makeRootWindowModels()")
+        return []
+
+    def makeDynamicWindowModels(self):
+        log("makeDynamicWindowModels()")
+        return []
+
     def set_keymap(self, server_source, force=False):
         log.info("keymap support not implemented in pipewire screencast shadow server")
 
     def setup_capture(self):
-        self.portal_interface = bus.get_object(PORTAL_DESKTOP_INTERFACE, PORTAL_DESKTOP_PATH)
-        log(f"setup_capture() self.portal_interface={self.portal_interface}")
+        pass
 
     def cleanup(self):
         GTKShadowServerBase.cleanup(self)
@@ -62,7 +87,7 @@ def start_refresh(self, wid):
         self.start_capture()
 
     def start_capture(self):
-        self.dbus_request_screenscast()
+        pass
 
     def screen_cast_call(self, method, callback, *args, options={}):
         #generate a new token and path:
@@ -92,57 +117,67 @@ def dbus_request_screenscast(self):
             )
 
     def on_create_session_response(self, response, results):
-        if response != 0:
-            log.error("Error: failed to create the session:")
-            log.error(f" {response}, {results}")
+        r = int(response)
+        res = typedict(dbus_to_native(results))
+        if r:
+            log.error("on_create_session_response", (response, results))
+            log.error(f"Error {r} creating the session")
             self.quit(ExitCode.UNSUPPORTED)
             return
-        self.session_handle = results.get("session_handle")
-        log("on_create_session_response%s session_handle=%s", (response, results), self.session_handle)
+        self.session_handle = res.strget("session_handle")
+        log("on_create_session_response%s session_handle=%s", (r, res), self.session_handle)
         if not self.session_handle:
-            log.error("Error: failed to create the session:")
-            log.error(" missing session handle")
+            log.error("Error: missing session handle creating the session")
             self.quit(ExitCode.UNSUPPORTED)
             return
         from dbus.types import UInt32
         options = {
             "multiple"  : self.multi_window,
             "types"     : UInt32(AvailableSourceTypes.WINDOW | AvailableSourceTypes.MONITOR),
             }
-        log(f"on_create_session_response calling {self.portal_interface.SelectSources} with options={options}")
+        log(f"on_create_session_response calling SelectSources with options={options}")
         self.screen_cast_call(
             self.portal_interface.SelectSources,
             self.on_select_sources_response,
             self.session_handle,
             options=options)
 
     def on_select_sources_response(self, response, results):
-        if response != 0:
-            log.error("Error: failed to select sources:")
-            log.error(f" {response}, {results}")
+        r = int(response)
+        res = typedict(dbus_to_native(results))
+        if r:
+            log("on_select_sources_response%s", (response, results))
+            log.error(f"Error {r} selecting sources")
             self.quit(ExitCode.UNSUPPORTED)
             return
-        log(f"on_select_sources_response sources selected, results={results}")
+        log(f"on_select_sources_response sources selected, results={res}")
         self.screen_cast_call(
             self.portal_interface.Start,
             self.on_start_response,
             self.session_handle,
             "")
 
     def on_start_response(self, response, results):
-        if response != 0:
-            log.error("Error: failed to start capture:")
-            log.error(f" {response}, {results}")
+        r = int(response)
+        res = typedict(dbus_to_native(results))
+        if r:
+            log.error("on_start_response%s", (response, results))
+            log.error(f"Error {r} starting the screen capture")
             self.quit(ExitCode.UNSUPPORTED)
             return
-        streams = results.get("streams")
+        streams = res.tupleget("streams")
         if not streams:
             log.error("Error: failed to start capture:")
             log.error(" missing streams")
             self.quit(ExitCode.UNSUPPORTED)
             return
+        log(f"on_start_response starting pipewire capture for {streams}")
         for node_id, props in streams:
-            self.start_pipewire_capture(node_id, props)
+            #start_thread(self.start_pipewire_capture,
+            #             f"start-pipewire-capture-{node_id}",
+            #             daemon=True,
+            #             args = (node_id, typedict(props)))
+            self.start_pipewire_capture(node_id, typedict(props))
 
     def start_pipewire_capture(self, node_id, props):
         log(f"start_pipewire_capture({node_id}, {props})")
@@ -152,20 +187,49 @@ def start_pipewire_capture(self, node_id, props):
                                                              empty_dict,
                                                              dbus_interface=SCREENCAST_IFACE)
         fd = fd_object.take()
+        #from gi.repository import Gst
+        #pipeline = Gst.parse_launch('pipewiresrc fd=%d path=%u ! videoconvert ! xvimagesink'%(fd, node_id))
+        #pipeline.set_state(Gst.State.PLAYING)
+        #def on_gst_message(*args):
+        #    log.info(f"on_gst_message{args}")
+        #pipeline.get_bus().connect('message', on_gst_message)
+        #return
+        x, y = props.inttupleget("position", (0, 0))
+        w, h = props.inttupleget("size", (0, 0))
         el = f"pipewiresrc fd={fd} path={node_id}"
-        self.capture = Capture(el)
-        self.capture.start()
+        self.capture = Capture(el, pixel_format="BGRX", width=w, height=h)
         self.capture.connect("state-changed", self.capture_state_changed)
         self.capture.connect("error", self.capture_error)
+        self.capture.connect("new-image", self.capture_new_image)
+        self.capture.start()
+        source_type = props.intget("source_type")
+        title = f"{AvailableSourceTypes(source_type)} {node_id}"
+        geometry = (x, y, w, h)
+        model = RootWindowModel(self.root, self.capture, title, geometry)
+        #must be called from the main thread:
+        log(f"new model: {model}")
+        self.idle_add(self._add_new_window, model)
+
+    def capture_new_image(self, capture, frame):
+        log(f"capture_new_image({capture}, {frame})")
 
     def capture_error(self, *args):
         log.warn(f"capture_error{args}")
+        self.quit(ExitCode.INTERNAL_ERROR)
 
     def capture_state_changed(self, *args):
-        log.warn(f"capture_state_changed{args}")
+        log.info(f"capture_state_changed{args}")
 
     def stop_capture(self):
         c = self.capture
         if c:
             self.capture = None
             c.clean()
+
+
+    def _move_pointer(self, device_id, wid, pos, props=None):
+        #x, y = pos
+        pass
+
+    def do_process_button_action(self, *args):
+        pass