Fix qsv for low power mode.

Dockerfile

@@ -46,11 +46,12 @@ ENV PYTHONFAULTHANDLER=1
 
 ARG DEBIAN_FRONTEND=noninteractive
 
-RUN mkdir /config /downloads && ln -snf /usr/share/zoneinfo/${TZ} /etc/localtime && echo ${TZ} > /etc/timezone && \
+RUN sed -i -E '/^Suites:[[:space:]]*trixie[[:space:]]+trixie-updates$/ {n; s/^(Components:[[:space:]]*)main([[:space:]]*|$)/\1main contrib non-free\2/}' /etc/apt/sources.list.d/debian.sources && \
+  mkdir /config /downloads && ln -snf /usr/share/zoneinfo/${TZ} /etc/localtime && echo ${TZ} > /etc/timezone && \
   apt-get update && \
   ARCH="$(dpkg --print-architecture)" && \
   EXTRA_PACKAGES="" && \
-  if [ "$ARCH" = "amd64" ]; then EXTRA_PACKAGES="intel-media-va-driver i965-va-driver libmfx-gen1.2"; fi && \
+  if [ "$ARCH" = "amd64" ]; then EXTRA_PACKAGES="intel-media-va-driver-non-free i965-va-driver libmfx-gen1.2"; fi && \
   apt-get install -y --no-install-recommends locales \
   bash mkvtoolnix patch aria2 curl ca-certificates xz-utils git sqlite3 tzdata file libmagic1 vainfo ${EXTRA_PACKAGES} \
   && useradd -u ${USER_ID:-1000} -U -d /app -s /bin/bash app && \
@@ -87,7 +88,7 @@ USER app
 
 WORKDIR /tmp
 
-HEALTHCHECK --interval=10s --timeout=20s --start-period=10s --retries=3 CMD [ "/usr/local/bin/healthcheck" ]
+HEALTHCHECK --interval=10s --timeout=20s --start-period=60s --retries=3 CMD [ "/usr/local/bin/healthcheck" ]
 
 ENTRYPOINT ["/entrypoint.sh"]
 

app/library/SegmentEncoders.py

@@ -4,10 +4,57 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+from functools import lru_cache
 from pathlib import Path
 from typing import Any, Protocol
 
 
+@lru_cache(maxsize=1)
+def detect_qsv_capabilities() -> dict[str, dict[str, bool]]:
+    """
+    Detects QSV encode capability for each relevant codec.
+    Returns a dict where keys are codec names (e.g. "h264", "hevc", "vp9") and
+    values are dicts with keys "full" and "lp" booleans.
+    """
+    try:
+        result: subprocess.CompletedProcess[str] = subprocess.run(
+            ["vainfo"],  # noqa: S607
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        out: str = result.stdout + result.stderr
+    except Exception:
+        return {}
+
+    caps = {}
+    for line in out.splitlines():
+        line: str = line.strip()
+        parts: list[str] = line.split(":")
+        if len(parts) < 2:
+            continue
+        prof, ep_str = parts[0].strip(), parts[1].strip()
+        if "H264" in prof:
+            key = "h264"
+        elif "HEVC" in prof:
+            key = "hevc"
+        elif "VP9" in prof:
+            key = "vp9"
+        else:
+            continue
+        if key not in caps:
+            caps[key] = {"full": False, "lp": False}
+
+        if "EncSlice " in ep_str:
+            caps[key]["full"] = True
+
+        if "EncSliceLP" in ep_str:
+            caps[key]["lp"] = True
+
+    return caps
+
+
+@lru_cache(maxsize=1)
 def has_dri_devices() -> bool:
     """
     Check if there are any /dev/dri devices.
@@ -29,6 +76,7 @@ def has_dri_devices() -> bool:
         return False
 
 
+@lru_cache(maxsize=1)
 def ffmpeg_encoders() -> set[str]:
     """
     Return a set of available ffmpeg encoders.
@@ -38,6 +86,7 @@ def ffmpeg_encoders() -> set[str]:
 
     """
     from .config import SUPPORTED_CODECS
+
     try:
         result: subprocess.CompletedProcess[str] = subprocess.run(
             ["ffmpeg", "-hide_banner", "-loglevel", "error", "-encoders"],  # noqa: S607
@@ -74,6 +123,7 @@ def select_encoder(configured: str) -> str:
 
     """
     from .config import SUPPORTED_CODECS
+
     configured = (configured or "").strip()
 
     avail: set[str] = ffmpeg_encoders()
@@ -109,7 +159,7 @@ class SoftwareBuilder(_BaseBuilder):
     codec_name = "libx264"
 
     def add_video_args(self, args: list[str], ctx: dict[str, Any] | None = None) -> list[str]:
-        return super().add_video_args(["-pix_fmt", "yuv420p", *args])
+        return super().add_video_args(["-pix_fmt", "yuv420p", *args], ctx)
 
 
 class NvencBuilder(_BaseBuilder):
@@ -164,29 +214,60 @@ class QsvBuilder(_BaseBuilder):
 
     def input_args(self, ctx: dict[str, Any] | None = None) -> list[str]:
         ctx = ctx or {}
+        args = []
         is_linux: bool = bool(ctx.get("is_linux", sys.platform.startswith("linux")))
         has_dri: bool = bool(ctx.get("has_dri", False))
         device: str = ctx.get("vaapi_device", "/dev/dri/renderD128")
         if is_linux and has_dri:
-            return ["-init_hw_device", f"qsv=hw:{device}", "-filter_hw_device", "hw"]
+            args: list[str] = [
-        return []
+                "-init_hw_device",
+                f"qsv=hw:{device}",
+                "-hwaccel",
+                "qsv",
+                "-hwaccel_output_format",
+                "qsv",
+                "-filter_hw_device",
+                "hw",
+            ]
+
+        return args
 
     def add_video_args(self, args: list[str], ctx: dict[str, Any] | None = None) -> list[str]:
         ctx = ctx or {}
-        new_args: list[str] = list(args)
         is_linux: bool = bool(ctx.get("is_linux", sys.platform.startswith("linux")))
         has_dri: bool = bool(ctx.get("has_dri", False))
         if is_linux and has_dri:
-            new_args += [
+            if ctx.get("qsv", {}).get("full", False):
+                return [
+                    *args,
+                    "-vf",
+                    "vpp_qsv=w=trunc(iw/2)*2:h=trunc(ih/2)*2:format=nv12",
+                    "-codec:v",
+                    self.codec_name,
+                    "-global_quality",
+                    "24",
+                    "-rc_mode",
+                    "cqp",
+                ]
+            return [
+                *args,
                 "-vf",
-                "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64",
+                "vpp_qsv=w=trunc(iw/2)*2:h=trunc(ih/2)*2:format=nv12",
-                # Use ICQ (Intelligent Constant Quality) ratecontrol mode with quality level
+                "-codec:v",
-                "-q:v",
+                self.codec_name,
-                "23",
+                "-low_power",
-                "-look_ahead",
+                "1",
-                "0",
+                "-rc_mode",
+                "cbr",
+                "-b:v",
+                "3M",
+                "-maxrate",
+                "3M",
+                "-bufsize",
+                "6M",
             ]
-        return super().add_video_args(new_args)
+
+        return super().add_video_args(args, ctx)
 
 
 def get_builder_for_codec(codec: str) -> EncoderBuilder:

app/library/Segments.py

@@ -6,13 +6,19 @@ import subprocess  # type: ignore
 import sys
 import tempfile
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, ClassVar
 
 from aiohttp import web
 
 from .config import SUPPORTED_CODECS, Config
 from .ffprobe import ffprobe
-from .SegmentEncoders import encoder_fallback_chain, get_builder_for_codec, has_dri_devices, select_encoder
+from .SegmentEncoders import (
+    detect_qsv_capabilities,
+    encoder_fallback_chain,
+    get_builder_for_codec,
+    has_dri_devices,
+    select_encoder,
+)
 
 if TYPE_CHECKING:
     from asyncio.subprocess import Process
@@ -80,27 +86,28 @@ class Segments:
 
         startTime: str = f"{0:.6f}" if self.index == 0 else f"{self.duration * self.index:.6f}"
 
+        ctx = {
+            "is_linux": sys.platform.startswith("linux"),
+            "has_dri": has_dri_devices(),
+            "vaapi_device": Config.get_instance().vaapi_device,
+            "qsv": {"full": False, "lp": False},
+        }
+
+        caps: dict[str, dict[str, bool]] = detect_qsv_capabilities()
+        base_codec: str = s_codec.split("_")[0]
+        codec_caps: dict[str, bool] = caps.get(base_codec, {"full": False, "lp": False})
+        ctx["qsv"] = codec_caps
+
         if self.vconvert and ff and hasattr(ff, "has_video") and ff.has_video():
             builder: EncoderBuilder = get_builder_for_codec(s_codec)
-            builder_ctx: dict[str, Any] = {
-                "is_linux": sys.platform.startswith("linux"),
-                "has_dri": has_dri_devices(),
-                "vaapi_device": Config.get_instance().vaapi_device,
-            }
         else:
             builder = None
-            builder_ctx = {}
+            ctx: dict = {}
 
         # Collect encoder-specific input/global flags that must precede the input
         input_args: list[str] = []
         if builder:
-            input_args = builder.input_args(
+            input_args = builder.input_args(ctx)
-                {
-                    "is_linux": sys.platform.startswith("linux"),
-                    "has_dri": has_dri_devices(),
-                    "vaapi_device": Config.get_instance().vaapi_device,
-                }
-            )
 
         fargs: list[str] = [
             "-xerror",
@@ -123,10 +130,7 @@ class Segments:
 
         v_args: list[str] = []
         if builder:
-            v_args = builder.add_video_args(
+            v_args = builder.add_video_args(["-g", "52", "-map", "0:v:0", "-strict", "-2"], ctx)
-                ["-g", "52", "-map", "0:v:0", "-strict", "-2"],
-                builder_ctx,
-            )
         else:
             v_args += ["-codec:v", "copy"]
 

app/tests/test_segments.py

@@ -314,9 +314,7 @@ async def test_stream_gpu_fallback_switches_codec(
     assert "-filter_hw_device" in first
     assert first[first.index("-filter_hw_device") + 1] == "hw"
     assert "-vf" in first
-    assert first[first.index("-vf") + 1] == (
+    assert "vpp_qsv" in first[first.index("-vf") + 1]
-        "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64"
-    )
 
     # Second call (fallback) must switch codec to a safe fallback
     second = captured_args[1]
@@ -329,8 +327,8 @@ async def test_stream_gpu_fallback_switches_codec(
         assert "-filter_hw_device" not in second
         if "-vf" in second:
             vf_val = second[second.index("-vf") + 1]
-            assert not vf_val.startswith("scale=trunc(")
+            assert "scale_qsv" not in vf_val
-            assert not vf_val.startswith("format=nv12,hwupload")
+            assert "hwupload" not in vf_val
         assert "-pix_fmt" in second
         assert second[second.index("-pix_fmt") + 1] == "yuv420p"
     else: