#3964 choose a csc mode and expose it

we need python3-gstreamer to make the gi bindings work

packaging/debian/xpra/control

@@ -287,6 +287,7 @@ Recommends:
         ,gstreamer-plugins-ugly
         ,gstreamer1.0-vaapi
         ,gstreamer1.0-pipewire
+        ,python3-gst-1.0
 Description: Extra picture and video codecs
  for the xpra server and client.
 

packaging/rpm/xpra.spec

@@ -236,6 +236,7 @@ BuildRequires:		pkgconfig(libavif)
 %endif
 #for gstreamer video encoder and decoder:
 Recommends:			gstreamer1
+Recommends:			python3-gstreamer1
 #appsrc, videoconvert:
 Recommends:			gstreamer1-plugins-base
 #vaapi:

xpra/codecs/gstreamer/capture.py

@@ -23,7 +23,7 @@
     get_version, get_type, get_info,
     init_module, cleanup_module,
     get_video_encoder_caps, get_video_encoder_options,
-    get_gst_encoding,
+    get_gst_encoding, get_encoder_info, get_gst_rgb_format,
 )
 from xpra.codecs.image import ImageWrapper
 from xpra.log import Logger
@@ -169,6 +169,13 @@ def choose_video_encoder(encodings: Iterable[str]) -> str:
     return ""
 
 
+def choose_csc(modes: Iterable[str]) -> str:
+    prefer = "YUV420P"
+    if not modes or prefer in modes:
+        return prefer
+    return modes[0]
+
+
 class CaptureAndEncode(Capture):
     """
     Uses a GStreamer pipeline to capture the screen
@@ -188,7 +195,10 @@ def create_pipeline(self, capture_element: str = "ximagesrc") -> None:
             "speed": 100,
             "quality": 100,
         })
-        self.profile = get_profile(options, encoding, csc_mode="YUV444P",
+        einfo = get_encoder_info(encoder)
+        log(f"{encoder}: {einfo=}")
+        self.csc_mode = choose_csc(einfo.get("format", ()))
+        self.profile = get_profile(options, encoding, csc_mode=self.csc_mode,
                                    default_profile="high" if encoder == "x264enc" else "")
         eopts = get_video_encoder_options(encoder, self.profile, options)
         vcaps = get_video_encoder_caps(encoder)
@@ -199,8 +209,9 @@ def create_pipeline(self, capture_element: str = "ximagesrc") -> None:
         gst_encoding = get_gst_encoding(encoding)  # ie: "hevc" -> "video/x-h265"
         elements = [
             f"{capture_element} name=capture",  # ie: ximagesrc or pipewiresrc
-            "videoconvert",
             "queue leaky=2",
+            "videoconvert",
+            "video/x-raw,format=%s" % get_gst_rgb_format(self.csc_mode),
             get_element_str(encoder, eopts),
             get_caps_str(gst_encoding, vcaps),
             get_element_str("appsink", get_default_appsink_attributes()),
@@ -227,6 +238,7 @@ def on_new_sample(self, _bus) -> int:
             self.frames += 1
             client_info = self.extra_client_info
             client_info["frame"] = self.frames
+            client_info["csc"] = self.csc_mode
             self.extra_client_info = {}
             self.emit("new-image", self.pixel_format, data, client_info)
         if SAVE_TO_FILE:

xpra/codecs/gstreamer/common.py

@@ -11,7 +11,7 @@
 from xpra.util.parsing import parse_simple_dict
 from xpra.util.objects import typedict
 from xpra.util.env import envint
-from xpra.os_util import OSX
+from xpra.os_util import OSX, gi_import
 from xpra.gstreamer.common import import_gst, GST_FLOW_OK
 from xpra.gstreamer.pipeline import Pipeline
 from xpra.log import Logger
@@ -161,27 +161,109 @@ def cleanup_module() -> None:
     log("gstreamer.cleanup_module()")
 
 
+IDENTICAL_PIXEL_FORMATS = (
+    "NV12",
+    "RGBA", "BGRA", "ARGB", "ABGR",
+    "RGB", "BGR",
+    "RGB15", "RGB16", "BGR15",
+    "r210",
+    "BGRP", "RGBP",
+)
+XPRA_TO_GSTREAMER = {
+    "YUV420P": "I420",
+    "YUV444P": "Y444",
+    "BGRX": "BGRx",
+    "XRGB": "xRGB",
+    "XBGR": "xBGR",
+    "YUV400": "GRAY8",
+    # "RGB8P"
+}
+GSTREAMER_TO_XPRA = dict((v, k) for k, v in XPRA_TO_GSTREAMER.items())
+
+
 def get_gst_rgb_format(rgb_format: str) -> str:
-    if rgb_format in (
-            "NV12",
-            "RGBA", "BGRA", "ARGB", "ABGR",
-            "RGB", "BGR",
-            "RGB15", "RGB16", "BGR15",
-            "r210",
-            "BGRP", "RGBP",
-    ):
-        # identical name:
+    if rgb_format in IDENTICAL_PIXEL_FORMATS:
         return rgb_format
     # translate to gstreamer name:
-    return {
-        "YUV420P": "I420",
-        "YUV444P": "Y444",
-        "BGRX": "BGRx",
-        "XRGB": "xRGB",
-        "XBGR": "xBGR",
-        "YUV400": "GRAY8",
-        # "RGB8P"
-    }[rgb_format]
+    return XPRA_TO_GSTREAMER.get(rgb_format, "")
+
+
+def get_xpra_rgb_format(rgb_format: str) -> str:
+    if rgb_format in IDENTICAL_PIXEL_FORMATS:
+        return rgb_format
+    # translate to gstreamer name:
+    return GSTREAMER_TO_XPRA.get(rgb_format, "")
+
+
+def gst_to_native(value: Any) -> tuple | str:
+    if isinstance(value, Gst.ValueList):
+        return tuple(value.get_value(value, i) for i in range(value.get_size(value)))
+    if isinstance(value, Gst.IntRange):
+        return value.range.start, value.range.stop
+    if isinstance(value, Gst.FractionRange):
+
+        def preferint(v):
+            if int(v) == v:
+                return int(v)
+            return v
+
+        def numdenom(fraction):
+            return preferint(fraction.num), preferint(fraction.denom)
+        return numdenom(value.start), numdenom(value.stop)
+    # print(f"value={value} ({type(value)} - {dir(value)}")
+    return Gst.value_serialize(value)
+
+
+def get_encoder_info(element="vp8enc") -> dict:
+    """
+    Get the element's input information,
+    we only retrieve the "SINK" pad
+    if it accepts `video/x-raw`.
+    Convert Gst types into native types,
+    and convert pixel formats to the names used in xpra.
+    """
+    factory = Gst.ElementFactory.find(element)
+    if not factory:
+        return {}
+    if factory.get_num_pad_templates() == 0:
+        return {}
+    pads = factory.get_static_pad_templates()
+    info = {}
+    GLib = gi_import("GLib")
+    log(f"get_encoder_info({element}) pads={pads}")
+    for pad in pads:
+        if pad.direction != Gst.PadDirection.SINK:
+            log(f"ignoring {pad.direction}")
+            continue
+        padtemplate = pad.get()
+        caps = padtemplate.get_caps()
+        if not caps:
+            log(f"pad template {padtemplate} has no caps!")
+            continue
+        if caps.is_any():
+            continue
+        if caps.is_empty():
+            continue
+        for i in range(caps.get_size()):
+            structure = caps.get_structure(i)
+            if structure.get_name() != "video/x-raw":
+                continue
+
+            def add_cap(field, value):
+                fname = GLib.quark_to_string(field)
+                native_value = gst_to_native(value)
+                log(f"{fname}={native_value} ({type(native_value)})")
+                if fname == "format":
+                    if isinstance(native_value, str):
+                        # convert it to a tuple:
+                        native_value = native_value,
+                    if isinstance(native_value, tuple):
+                        xpra_formats = tuple(get_xpra_rgb_format(x) for x in native_value)
+                        native_value = tuple(fmt for fmt in xpra_formats if fmt)
+                info[fname] = native_value
+                return True
+            structure.foreach(add_cap)
+    return info
 
 
 def get_video_encoder_caps(encoder: str = "x264enc") -> dict[str, Any]:

xpra/codecs/gstreamer/encoder.py

@@ -192,6 +192,8 @@ def create_pipeline(self, options: typedict):
             raise ValueError(f"invalid encoding {self.encoding!r}")
         self.dst_formats = options.strtupleget("dst-formats")
         gst_rgb_format = get_gst_rgb_format(self.colorspace)
+        if not gst_rgb_format:
+            raise ValueError(f"unable to map {self.colorspace} to a gstreamer pixel format")
         vcaps: dict[str, Any] = {
             "width": self.width,
             "height": self.height,