h264_bitrate streamer param

kvmd/apps/__init__.py

@@ -90,6 +90,7 @@ from ..validators.hid import valid_hid_mouse_move
 from ..validators.kvm import valid_stream_quality
 from ..validators.kvm import valid_stream_fps
 from ..validators.kvm import valid_stream_resolution
+from ..validators.kvm import valid_stream_h264_bitrate
 
 from ..validators.ugpio import valid_ugpio_driver
 from ..validators.ugpio import valid_ugpio_channel
@@ -359,6 +360,10 @@ def _get_config_scheme() -> Dict:
                 "resolution":  Option("", type=(lambda arg: (valid_stream_resolution(arg) if arg else ""))),
                 "available_resolutions": Option([], type=functools.partial(valid_string_list, subval=valid_stream_resolution)),
 
+                "h264_bitrate":     Option(0,     type=(lambda arg: (valid_stream_h264_bitrate(arg) if arg else 0))),
+                "h264_min_bitrate": Option(100,   type=valid_stream_h264_bitrate),
+                "h264_max_bitrate": Option(16000, type=valid_stream_h264_bitrate),
+
                 "host":    Option("localhost", type=valid_ip_or_host),
                 "port":    Option(0,   type=valid_port),
                 "unix":    Option("",  type=valid_abs_path, only_if="!port", unpack_as="unix_path"),

kvmd/apps/kvmd/server.py

@@ -55,6 +55,7 @@ from ...validators.basic import valid_bool
 from ...validators.kvm import valid_stream_quality
 from ...validators.kvm import valid_stream_fps
 from ...validators.kvm import valid_stream_resolution
+from ...validators.kvm import valid_stream_h264_bitrate
 
 from ... import aiotools
 from ... import aioproc
@@ -103,6 +104,11 @@ class StreamerResolutionNotSupported(OperationError):
         super().__init__("This streamer does not support resolution settings")
 
 
+class StreamerH264NotSupported(OperationError):
+    def __init__(self) -> None:
+        super().__init__("This streamer does not support H264")
+
+
 # =====
 @dataclasses.dataclass(frozen=True)
 class _Component:  # pylint: disable=too-many-instance-attributes
@@ -220,6 +226,7 @@ class KvmdServer(HttpServer):  # pylint: disable=too-many-arguments,too-many-ins
             ("quality", valid_stream_quality, StreamerQualityNotSupported),
             ("desired_fps", valid_stream_fps, None),
             ("resolution", valid_stream_resolution, StreamerResolutionNotSupported),
+            ("h264_bitrate", valid_stream_h264_bitrate, StreamerH264NotSupported),
         ]:
             value = request.query.get(name)
             if value:

kvmd/apps/kvmd/streamer.py

@@ -62,33 +62,46 @@ class _StreamerParams:
     __RESOLUTION = "resolution"
     __AVAILABLE_RESOLUTIONS = "available_resolutions"
 
+    __H264_BITRATE = "h264_bitrate"
+
     def __init__(
         self,
         desired_fps: int,
         max_fps: int,
+
         quality: int,
+
         resolution: str,
         available_resolutions: List[str],
+
+        h264_bitrate: int,
+        h264_min_bitrate: int,
+        h264_max_bitrate: int,
     ) -> None:
 
         self.__has_quality = bool(quality)
         self.__has_resolution = bool(resolution)
+        self.__has_h264 = bool(h264_bitrate)
 
-        self.__params: Dict = {
+        self.__params: Dict = {self.__DESIRED_FPS: min(desired_fps, max_fps)}
-            self.__DESIRED_FPS: desired_fps,
+        self.__limits: Dict = {self.__MAX_FPS: max_fps}
-            **({self.__QUALITY: quality} if self.__has_quality else {}),
-            **({self.__RESOLUTION: resolution} if self.__has_resolution else {}),
-        }
 
-        self.__limits: Dict = {
+        if self.__has_quality:
-            self.__MAX_FPS: max_fps,
+            self.__params[self.__QUALITY] = quality
-            **({self.__AVAILABLE_RESOLUTIONS: available_resolutions} if self.__has_resolution else {}),
+
-        }
+        if self.__has_resolution:
+            self.__params[self.__RESOLUTION] = resolution
+            self.__limits[self.__AVAILABLE_RESOLUTIONS] = available_resolutions
+
+        if self.__has_h264:
+            self.__params[self.__H264_BITRATE] = min(max(h264_bitrate, h264_min_bitrate), h264_max_bitrate)
+            self.__limits[self.__H264_BITRATE] = {"min": h264_min_bitrate, "max": h264_max_bitrate}
 
     def get_features(self) -> Dict:
         return {
             self.__QUALITY: self.__has_quality,
             self.__RESOLUTION: self.__has_resolution,
+            "h264": self.__has_h264,
         }
 
     def get_limits(self) -> Dict:
@@ -113,6 +126,12 @@ class _StreamerParams:
             if params[self.__RESOLUTION] in self.__limits[self.__AVAILABLE_RESOLUTIONS]:
                 new_params[self.__RESOLUTION] = params[self.__RESOLUTION]
 
+        if self.__H264_BITRATE in params and self.__has_h264:
+            min_bitrate = self.__limits[self.__H264_BITRATE]["min"]
+            max_bitrate = self.__limits[self.__H264_BITRATE]["max"]
+            if min_bitrate <= params[self.__H264_BITRATE] <= max_bitrate:
+                new_params[self.__H264_BITRATE] = params[self.__H264_BITRATE]
+
         self.__params = new_params
 
 

kvmd/validators/kvm.py

@@ -74,3 +74,7 @@ def valid_stream_resolution(arg: Any) -> str:
     width = int(valid_number(parts[0], min=1, name=f"{name} (width)"))
     height = int(valid_number(parts[1], min=1, name=f"{name} (height)"))
     return f"{width}x{height}"
+
+
+def valid_stream_h264_bitrate(arg: Any) -> int:
+    return int(valid_number(arg, min=100, max=16000, name="stream H264 bitrate"))

testenv/tests/validators/test_kvm.py

@@ -32,6 +32,7 @@ from kvmd.validators.kvm import valid_log_seek
 from kvmd.validators.kvm import valid_stream_quality
 from kvmd.validators.kvm import valid_stream_fps
 from kvmd.validators.kvm import valid_stream_resolution
+from kvmd.validators.kvm import valid_stream_h264_bitrate
 
 
 # =====
@@ -126,3 +127,17 @@ def test_ok__valid_stream_resolution(arg: Any) -> None:
 def test_fail__valid_stream_resolution(arg: Any) -> None:
     with pytest.raises(ValidatorError):
         print(valid_stream_resolution(arg))
+
+
+# =====
+@pytest.mark.parametrize("arg", ["100", " 16000 ", 5000])
+def test_ok__valid_stream_h264_bitrate(arg: Any) -> None:
+    value = valid_stream_h264_bitrate(arg)
+    assert type(value) == int  # pylint: disable=unidiomatic-typecheck
+    assert value == int(str(arg).strip())
+
+
+@pytest.mark.parametrize("arg", ["0", "-1", "100.0", 5000.1, None, ""])
+def test_fail__valid_stream_h264_bitrate(arg: Any) -> None:
+    with pytest.raises(ValidatorError):
+        print(valid_stream_h264_bitrate(arg))