#3872 add 'stream' encoding option

xpra/client/gtk_base/gtk_tray_menu_base.py

@@ -651,6 +651,8 @@ def get_encoding_options(self):
         #auto at the very top:
         client_encodings.insert(0, "auto")
         server_encodings.insert(0, "auto")
+        client_encodings.insert(1, "stream")
+        server_encodings.insert(1, "stream")
         return client_encodings, server_encodings
 
     def make_encodingssubmenu(self):

xpra/client/mixins/encodings.py

@@ -5,9 +5,9 @@
 # later version. See the file COPYING for details.
 
 import os
-from typing import Dict, Any
+from typing import Dict, Any, Tuple
 
-from xpra.codecs.codec_constants import preforder
+from xpra.codecs.codec_constants import preforder, STREAM_ENCODINGS
 from xpra.codecs.loader import load_codec, codec_versions, has_codec, get_codec
 from xpra.codecs.video_helper import getVideoHelper
 from xpra.scripts.config import parse_bool_or_int
@@ -77,17 +77,17 @@ def __init__(self):
         self.video_scaling = None
         self.video_max_size = VIDEO_MAX_SIZE
 
-        self.server_encodings = []
-        self.server_core_encodings = []
-        self.server_encodings_with_speed = ()
-        self.server_encodings_with_quality = ()
-        self.server_encodings_with_lossless_mode = ()
+        self.server_encodings : Tuple[str,...] = ()
+        self.server_core_encodings : Tuple[str,...] = ()
+        self.server_encodings_with_speed : Tuple[str,...] = ()
+        self.server_encodings_with_quality : Tuple[str,...] = ()
+        self.server_encodings_with_lossless_mode : Tuple[str,...] = ()
 
         #what we told the server about our encoding defaults:
         self.encoding_defaults = {}
 
 
-    def init(self, opts):
+    def init(self, opts) -> None:
         self.allowed_encodings = opts.encodings
         self.encoding = opts.encoding
         if opts.video_scaling.lower() in ("auto", "on"):
@@ -118,23 +118,23 @@ def init(self, opts):
         vh.init()
 
 
-    def cleanup(self):
+    def cleanup(self) -> None:
         try:
             getVideoHelper().cleanup()
         except Exception:   # pragma: no cover
             log.error("error on video cleanup", exc_info=True)
 
 
-    def init_authenticated_packet_handlers(self):
+    def init_authenticated_packet_handlers(self) -> None:
         self.add_packet_handler("encodings", self._process_encodings, False)
 
 
-    def _process_encodings(self, packet):
+    def _process_encodings(self, packet) -> None:
         caps = typedict(packet[1])
         self._parse_server_capabilities(caps)
 
 
-    def get_info(self):
+    def get_info(self) -> Dict[str,Any]:
         return {
             "encodings" : {
                 "core"          : self.get_core_encodings(),
@@ -167,7 +167,7 @@ def parse_server_capabilities(self, c : typedict) -> bool:
         self._parse_server_capabilities(c)
         return True
 
-    def _parse_server_capabilities(self, c):
+    def _parse_server_capabilities(self, c) -> None:
         self.server_encodings = c.strtupleget("encodings", DEFAULT_ENCODINGS)
         self.server_core_encodings = c.strtupleget("encodings.core", self.server_encodings)
         #old servers only supported x264:
@@ -270,7 +270,7 @@ def get_encodings_caps(self) -> Dict[str,Any]:
         log("encoding capabilities: %s", caps)
         return caps
 
-    def get_encodings(self):
+    def get_encodings(self) -> Tuple[str,...]:
         """
             Unlike get_core_encodings(), this method returns "rgb" for both "rgb24" and "rgb32".
             That's because although we may support both, the encoding chosen is plain "rgb",
@@ -280,6 +280,8 @@ def get_encodings(self):
         cenc = [{"rgb32" : "rgb", "rgb24" : "rgb"}.get(x, x) for x in self.get_core_encodings()]
         if "grayscale" not in cenc and "png/L" in cenc:
             cenc.append("grayscale")
+        if any(x in cenc for x in STREAM_ENCODINGS):
+            cenc.append("stream")
         return preforder(cenc)
 
     def get_cursor_encodings(self):

xpra/codecs/codec_constants.py

@@ -24,7 +24,14 @@
     "h265", "av1",
     "scroll",
     "grayscale",
+    "stream",
     )
+STREAM_ENCODINGS : Tuple[str,...] = (
+    "h264", "vp9", "vp8", "mpeg4",
+    "mpeg4+mp4", "h264+mp4", "vp8+webm", "vp9+webm",
+    "h265", "av1",
+    )
+
 #encoding order for edges (usually one pixel high or wide):
 EDGE_ENCODING_ORDER : Tuple[str, ...] = (
     "rgb24", "rgb32",
@@ -34,6 +41,7 @@
 
 HELP_ORDER : Tuple[str, ...] = (
     "auto",
+    "stream",
     "grayscale",
     "h264", "h265", "av1", "vp8", "vp9", "mpeg4",
     "png", "png/P", "png/L", "webp", "avif",

xpra/codecs/loader.py

@@ -54,7 +54,7 @@ def filt(*values) -> Tuple[str,...]:
 
 codec_errors : Dict[str,str] = {}
 codecs = {}
-def codec_import_check(name, description, top_module, class_module, classnames):
+def codec_import_check(name:str, description:str, top_module, class_module, classnames):
     log(f"{name}:")
     log(" codec_import_check%s", (name, description, top_module, class_module, classnames))
     if any(name.find(s)>=0 for s in SKIP_LIST):
@@ -142,7 +142,7 @@ def codec_import_check(name, description, top_module, class_module, classnames):
                      name, description, exc_info=True)
     return None
 codec_versions : Dict[str,Tuple[Any, ...]]= {}
-def add_codec_version(name, top_module, version="get_version()", alt_version="__version__"):
+def add_codec_version(name:str, top_module, version:str="get_version()", alt_version:str="__version__"):
     try:
         fieldnames = [x for x in (version, alt_version) if x is not None]
         for fieldname in fieldnames:
@@ -174,7 +174,7 @@ def add_codec_version(name, top_module, version="get_version()", alt_version="__
         log.warn("", exc_info=True)
     return None
 
-def xpra_codec_import(name, description, top_module, class_module, classname):
+def xpra_codec_import(name:str, description:str, top_module, class_module, classname:str):
     xpra_top_module = f"xpra.codecs.{top_module}"
     xpra_class_module = f"{xpra_top_module}.{class_module}"
     if codec_import_check(name, description, xpra_top_module, xpra_class_module, classname):
@@ -185,7 +185,7 @@ def xpra_codec_import(name, description, top_module, class_module, classname):
 
 platformname = sys.platform.rstrip("0123456789")
 
-CODEC_OPTIONS = {
+CODEC_OPTIONS : Dict[str,Tuple[str,str,str,str]] = {
     #encoders:
     "enc_rgb"       : ("RGB encoder",       "argb",         "encoder", "encode"),
     "enc_pillow"    : ("Pillow encoder",    "pillow",       "encoder", "encode"),
@@ -226,7 +226,7 @@ def xpra_codec_import(name, description, top_module, class_module, classname):
     "nvfbc"         : ("NVIDIA Capture SDK","nvidia.nvfbc", f"fbc_capture_{platformname}", "NvFBC_SysCapture"),
     }
 
-NOLOAD = []
+NOLOAD : List[str] = []
 if OSX:
     #none of the nvidia codecs are available on MacOS,
     #so don't bother trying:
@@ -253,10 +253,9 @@ def load_codec(name):
     return get_codec(name)
 
 
-def load_codecs(encoders=True, decoders=True, csc=True, video=True, sources=False):
+def load_codecs(encoders=True, decoders=True, csc=True, video=True, sources=False) -> Tuple[str,...]:
     log("loading codecs")
-
-    loaded = []
+    loaded : List[str] = []
     def load(*names):
         for name in names:
             if has_codec(name):
@@ -281,9 +280,9 @@ def load(*names):
     if sources:
         load(*SOURCES)
     log("done loading codecs: %s", loaded)
-    return loaded
+    return tuple(loaded)
 
-def show_codecs(show:Tuple[str,...]=()):
+def show_codecs(show:Tuple[str,...]=()) -> None:
     #print("codec_status=%s" % codecs)
     for name in sorted(show or ALL_CODECS):
         log(f"* {name.ljust(20)} : {str(name in codecs).ljust(10)} {codecs.get(name, '')}")
@@ -318,9 +317,10 @@ def get_rgb_compression_options() -> List[str]:
         RGB_COMP_OPTIONS  += ["/".join(compressors)]
     return RGB_COMP_OPTIONS
 
-def get_encoding_name(encoding):
-    ENCODINGS_TO_NAME = {
+def get_encoding_name(encoding:str) -> str:
+    ENCODINGS_TO_NAME : Dict[str,str] = {
           "auto"    : "automatic",
+          "stream"  : "video stream",
           "h264"    : "H.264",
           "h265"    : "H.265",
           "mpeg4"   : "MPEG4",
@@ -338,7 +338,7 @@ def get_encoding_name(encoding):
         }
     return ENCODINGS_TO_NAME.get(encoding, encoding)
 
-def get_encoding_help(encoding):
+def get_encoding_help(encoding:str) -> str:
     # pylint: disable=import-outside-toplevel
     from xpra.net import compression
     compressors = [x for x in compression.get_enabled_compressors()
@@ -348,6 +348,7 @@ def get_encoding_help(encoding):
         compressors_str = ", may be compressed using "+(" or ".join(compressors))+" "
     return {
           "auto"    : "automatic mode (recommended)",
+          "stream"  : "video stream",
           "grayscale" : "same as 'auto' but in grayscale mode",
           "h264"    : "H.264 video codec",
           "h265"    : "H.265 (HEVC) video codec (not recommended)",
@@ -368,7 +369,7 @@ def get_encoding_help(encoding):
           }.get(encoding)
 
 
-def encodings_help(encodings):
+def encodings_help(encodings) -> List[str]:
     h = []
     for e in HELP_ORDER:
         if e in encodings:

xpra/scripts/main.py

@@ -1514,14 +1514,13 @@ def get_client_gui_app(error_cb, opts, request_mode, extra_args, mode:str):
         if opts.encoding=="auto":
             opts.encoding = ""
         if opts.encoding:
-            encodings = app.get_encodings()
-            err = opts.encoding and (opts.encoding not in encodings)
-            einfo = ""
-            if err and opts.encoding!="help":
+            encodings = list(app.get_encodings())+["auto", "stream"]
+            err = opts.encoding not in encodings
+            ehelp = opts.encoding=="help"
+            if err and not ehelp:
                 einfo = f"invalid encoding: {opts.encoding}\n"
-            if opts.encoding=="help" or err:
+            if err or ehelp:
                 from xpra.codecs.loader import encodings_help
-                encodings = ["auto"] + list(encodings)
                 raise InitInfo(einfo+"%s xpra client supports the following encodings:\n * %s" %
                                (app.client_toolkit(), "\n * ".join(encodings_help(encodings))))
         def handshake_complete(*_args):