diff --git a/.vscode/settings.json b/.vscode/settings.json index 0e1aabb2..1bfe7599 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -29,6 +29,7 @@ "brainicism", "brotlicffi", "buildcache", + "caplog", "Cfmrc", "choco", "cifs", diff --git a/FAQ.md b/FAQ.md index b0ba036b..645e92fc 100644 --- a/FAQ.md +++ b/FAQ.md @@ -28,7 +28,7 @@ or the `environment:` section in `compose.yaml` file. | YTP_HOST | Which IP address to bind to | `0.0.0.0` | | YTP_PORT | Which port to bind to | `8081` | | YTP_LOG_LEVEL | Log level | `info` | -| YTP_STREAMER_VCODEC | The video codec to use for in-browser streaming | `libx264` | +| YTP_STREAMER_VCODEC | The video encoding codec, default to gpi and fallback to software | `""` | | YTP_STREAMER_ACODEC | The audio codec to use for in-browser streaming | `aac` | | YTP_ACCESS_LOG | Whether to log access to the web server | `true` | | YTP_DEBUG | Whether to turn on debug mode | `false` | diff --git a/app/library/Segments.py b/app/library/Segments.py index ef1edee7..ff2a1780 100644 --- a/app/library/Segments.py +++ b/app/library/Segments.py @@ -3,18 +3,44 @@ import hashlib import logging import os import subprocess +import sys import tempfile +from collections.abc import Iterable from pathlib import Path +from typing import TYPE_CHECKING, ClassVar from aiohttp import web from .config import Config from .ffprobe import ffprobe +if TYPE_CHECKING: + from asyncio.subprocess import Process + + from app.library.ffprobe import FFProbeResult + LOG: logging.Logger = logging.getLogger("player.segments") class Segments: + """ + Build and stream MPEG-TS segments using ffmpeg with optional on-the-fly + transcoding. The class can auto-detect an available hardware encoder on the + first video segment and cache that choice for subsequent segments to avoid + repeated probing. + + Supported encoder families (preference order): + - amd -> h264_amf + - nvidia -> h264_nvenc + - intel -> h264_qsv + - apple -> h264_videotoolbox + - software -> libx264 + """ + + # Cache of selected video encoder across all segment instances + _cached_vcodec: ClassVar[str | None] = None + _cache_initialized: ClassVar[bool] = False + def __init__(self, download_path: str, index: int, duration: float, vconvert: bool, aconvert: bool): config: Config = Config.get_instance() self.download_path: str = download_path @@ -27,6 +53,8 @@ class Segments: "Whether to convert video." self.aconvert = bool(aconvert) "Whether to convert audio." + + # Default to configured codec; can be replaced by auto-detection self.vcodec: str = config.streamer_vcodec "The video codec to use." self.acodec: str = config.streamer_acodec @@ -38,13 +66,122 @@ class Segments: self.aconvert = True "Whether to convert audio." + @staticmethod + def _encoder_map() -> dict[str, str]: + """Return mapping from logical encoder family to ffmpeg codec name.""" + return { + "intel": "h264_qsv", + "nvidia": "h264_nvenc", + "amd": "h264_amf", + "apple": "h264_videotoolbox", + "software": "libx264", + } + + @staticmethod + def _preferred_families(configured: str) -> list[str]: + """ + Build a preference list based on configuration value. + + Accepts either a logical family name (amd|nvidia|intel|apple|software), + a concrete ffmpeg encoder (e.g., h264_nvenc/libx264), or any other + string. If the value is not recognized, we try GPU families first and + fall back to software. + """ + families: list[str] = ["intel", "nvidia", "amd", "apple", "software"] + enc_map: dict[str, str] = Segments._encoder_map() + + # If user configured a concrete ffmpeg encoder, prioritize that and + # include software as fallback. + if configured in enc_map.values(): + # Map concrete to a family for consistent probing order + family: str | None = next((k for k, v in enc_map.items() if v == configured), None) + return [family, "software"] if family else ["software"] + + cfg_lower: str = configured.strip().lower() + if cfg_lower in enc_map: + # Specific family requested; then software fallback + return [cfg_lower, "software"] + + # Unknown/legacy value (e.g., libx264 default): try GPUs first, then software + return families + + @staticmethod + def _ffmpeg_encoders() -> set[str]: + """ + Return a set of available ffmpeg encoder names by parsing + `ffmpeg -encoders` output. On any error, return an empty set to signal + that probing failed (callers should fall back to software). + """ + try: + result: subprocess.CompletedProcess[str] = subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-encoders"], # noqa: S607 + capture_output=True, + text=True, + check=False, + creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, + ) + out: str = (result.stdout or "") + (result.stderr or "") + except Exception: + return set() + + encoders: set[str] = set() + if not out: + return encoders + + # Each encoder appears as a token in the list; match known names + for name in Segments._encoder_map().values(): + if name in out: + encoders.add(name) + + return encoders + + @staticmethod + def _select_encoder(preferred: Iterable[str]) -> str: + """ + Select the first available encoder according to the preferred family + order. Falls back to software if probing fails or none found. + """ + enc_map: dict[str, str] = Segments._encoder_map() + + # Only on Linux, if no /dev/dri device, fall back to software. + # On macOS/Windows, this check is not applicable. + if sys.platform.startswith("linux") and not Segments._has_dri_devices(): + return enc_map["software"] + + available: set[str] = Segments._ffmpeg_encoders() + + # If probing failed, prefer software to minimize failure risk + if not available: + return enc_map["software"] + + for family in preferred: + name: str | None = enc_map.get(family) + if name and name in available: + return name + + # Final fallback + return enc_map["software"] + + @staticmethod + def _has_dri_devices() -> bool: + """Check if /dev/dri exists and has entries (card0, renderD128, etc).""" + try: + dri = Path("/dev/dri") + if not dri.exists() or not dri.is_dir(): + return False + for _ in dri.iterdir(): + return True + return False + except Exception: + return False + async def build_ffmpeg_args(self, file: Path) -> list[str]: try: - ff = await ffprobe(file) + ff: FFProbeResult = await ffprobe(file) except UnicodeDecodeError: pass - tmpFile = Path(tempfile.gettempdir()).joinpath( + tmpFile: Path = Path(tempfile.gettempdir()).joinpath( f"ytptube_stream.{hashlib.sha256(str(file).encode()).hexdigest()}" ) @@ -53,7 +190,7 @@ class Segments: startTime: str = f"{0:.6f}" if self.index == 0 else f"{self.duration * self.index:.6f}" - fargs = [ + fargs: list[str] = [ "-xerror", "-hide_banner", "-loglevel", @@ -69,20 +206,52 @@ class Segments: "-1", ] - if ff and ff.has_video(): - fargs += [ - "-pix_fmt", - "yuv420p", + # Decide on encoder for the first segment that actually has video + if ff and hasattr(ff, "has_video") and ff.has_video(): + # Initialize cached encoder only once, on first video segment + if not Segments._cache_initialized and self.index == 0 and self.vconvert: + preferred: list[str] = Segments._preferred_families(self.vcodec) + Segments._cached_vcodec = Segments._select_encoder(preferred) + Segments._cache_initialized = True + self.vcodec = Segments._cached_vcodec + LOG.info(f"Selected streaming video encoder: {self.vcodec}") + elif Segments._cache_initialized and Segments._cached_vcodec: + self.vcodec = Segments._cached_vcodec + + # Base video args + v_args: list[str] = [ "-g", "52", "-map", "0:v:0", "-strict", "-2", - "-codec:v", - self.vcodec if self.vconvert else "copy", ] + # For hardware encoders, avoid forcing pix_fmt yuv420p. + if self.vconvert: + if self.vcodec in ("h264_nvenc", "h264_amf", "h264_videotoolbox"): + # Let ffmpeg pick appropriate conversions; do not force pix_fmt + pass + elif self.vcodec == "h264_qsv": + # On Linux with /dev/dri present, initialize QSV device and ensure + # frames are properly aligned and uploaded to GPU. + if sys.platform.startswith("linux") and Segments._has_dri_devices(): + v_args += [ + "-init_hw_device", + "qsv=hw:/dev/dri/renderD128", + "-filter_hw_device", + "hw", + "-vf", + "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64", + ] + else: + # Software path: ensure broad compatibility + v_args = ["-pix_fmt", "yuv420p", *v_args] + + v_args += ["-codec:v", self.vcodec if self.vconvert else "copy"] + fargs += v_args + if ff and ff.has_audio(): fargs += ["-map", "0:a:0", "-codec:a", self.acodec if self.aconvert else "copy"] @@ -92,47 +261,155 @@ class Segments: async def stream(self, file: Path, resp: web.StreamResponse): ffmpeg_args: list[str] = await self.build_ffmpeg_args(file) - proc = await asyncio.create_subprocess_exec( - "ffmpeg", - *ffmpeg_args, - stdin=asyncio.subprocess.DEVNULL, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, - ) + async def _run_and_stream(args: list[str]) -> tuple[bool, int, bool, str]: + proc: Process = await asyncio.create_subprocess_exec( + "ffmpeg", + *args, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, + ) - client_disconnected = False + client_disconnected_local = False + wrote_any = False + stderr_buf = bytearray() - LOG.debug(f"Streaming '{file}' segment '{self.index}'. ffmpeg: {' '.join(ffmpeg_args)}") - - try: - while True: - chunk: bytes = await proc.stdout.read(1024 * 64) - if not chunk: - break + async def _drain_stderr() -> None: try: - await resp.write(chunk) - except (asyncio.CancelledError, ConnectionResetError, BrokenPipeError, ConnectionError): - LOG.warning("Client disconnected or connection reset while writing.") - client_disconnected = True - break - except asyncio.CancelledError: - LOG.warning("Client disconnected. Terminating ffmpeg.") - client_disconnected = True - proc.terminate() + assert proc.stderr is not None + while True: + chunk = await proc.stderr.read(4096) + if not chunk: + break + stderr_buf.extend(chunk) + except Exception: + # best-effort only + pass + + stderr_task = asyncio.create_task(_drain_stderr()) + + LOG.debug(f"Streaming '{file}' segment '{self.index}'. ffmpeg: {' '.join(args)}") + try: - await asyncio.wait_for(proc.wait(), timeout=5) - except TimeoutError: - LOG.error("ffmpeg process did not terminate in time. Killing it.") - proc.kill() - raise - except ConnectionResetError: - LOG.warning("Connection reset by peer. Skipping further writes.") - client_disconnected = True - finally: - if not client_disconnected: + while True: + chunk: bytes = await proc.stdout.read(1024 * 64) + if not chunk: + break + wrote_any = True + try: + await resp.write(chunk) + except (asyncio.CancelledError, ConnectionResetError, BrokenPipeError, ConnectionError): + LOG.warning("Client disconnected or connection reset while writing.") + client_disconnected_local = True + break + except asyncio.CancelledError: + LOG.warning("Client disconnected. Terminating ffmpeg.") + client_disconnected_local = True + proc.terminate() try: - await resp.write_eof() - except (ConnectionResetError, RuntimeError): - LOG.warning("Failed to write EOF; client already disconnected.") - await proc.wait() + await asyncio.wait_for(proc.wait(), timeout=5) + except TimeoutError: + LOG.error("ffmpeg process did not terminate in time. Killing it.") + proc.kill() + raise + except ConnectionResetError: + LOG.warning("Connection reset by peer. Skipping further writes.") + client_disconnected_local = True + finally: + # Ensure process is stopped when client disconnected, to avoid hangs + if client_disconnected_local: + proc.terminate() + try: + rc: int = await asyncio.wait_for(proc.wait(), timeout=5) + except TimeoutError: + LOG.error("ffmpeg process did not terminate in time after disconnect. Killing it.") + proc.kill() + try: + rc = await asyncio.wait_for(proc.wait(), timeout=5) + except Exception: + rc = -1 + else: + # Normal termination path with a safety timeout + try: + rc = await asyncio.wait_for(proc.wait(), timeout=30) + except TimeoutError: + LOG.error("ffmpeg process wait timed out. Killing it.") + proc.kill() + try: + rc = await asyncio.wait_for(proc.wait(), timeout=5) + except Exception: + rc = -1 + try: + await asyncio.wait_for(stderr_task, timeout=1) + except Exception: + pass + + return (wrote_any, rc, client_disconnected_local, stderr_buf.decode("utf-8", errors="ignore").strip()) + + wrote, rc, client_disconnected, stderr_text = await _run_and_stream(ffmpeg_args) + + is_software: bool = self.vcodec == Segments._encoder_map()["software"] + if not is_software and not wrote and rc != 0 and not client_disconnected: + msg: str = stderr_text[:500] if stderr_text else "no error output" + LOG.warning(f"Hardware encoder failed (rc={rc}): {msg}. Retrying with software (libx264).") + Segments._cached_vcodec = Segments._encoder_map()["software"] + Segments._cache_initialized = True + + sw_args: list[str] = list(ffmpeg_args) + # Replace codec with software + try: + idx: int = sw_args.index("-codec:v") + sw_args[idx + 1] = Segments._encoder_map()["software"] + except ValueError: + sw_args += ["-codec:v", Segments._encoder_map()["software"]] + + # Remove QSV-specific flags if present prior to software fallback. + # -init_hw_device qsv=... + try: + while True: + i_idx = sw_args.index("-init_hw_device") + del sw_args[i_idx : min(i_idx + 2, len(sw_args))] + except ValueError: + pass + # -filter_hw_device + try: + while True: + f_idx = sw_args.index("-filter_hw_device") + del sw_args[f_idx : min(f_idx + 2, len(sw_args))] + except ValueError: + pass + # -vf QSV chain + try: + vf_idx = sw_args.index("-vf") + if ( + vf_idx + 1 < len(sw_args) + and sw_args[vf_idx + 1] + == "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64" + ): + del sw_args[vf_idx : vf_idx + 2] + except ValueError: + pass + + # Ensure software path uses yuv420p + if "-pix_fmt" in sw_args: + try: + pf_idx = sw_args.index("-pix_fmt") + if pf_idx + 1 < len(sw_args): + sw_args[pf_idx + 1] = "yuv420p" + except Exception: + pass + else: + # Insert near the start, before mapping/gop where possible + insert_pos = 0 + if "-g" in sw_args: + insert_pos = max(0, sw_args.index("-g") - 2) + sw_args[insert_pos:insert_pos] = ["-pix_fmt", "yuv420p"] + + wrote, rc, client_disconnected, _ = await _run_and_stream(sw_args) + + if not client_disconnected: + try: + await resp.write_eof() + except (ConnectionResetError, RuntimeError): + LOG.warning("Failed to write EOF; client already disconnected.") diff --git a/app/library/config.py b/app/library/config.py index 946df2a1..411569c4 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -72,8 +72,8 @@ class Config(metaclass=Singleton): max_workers: int = 1 """The maximum number of workers to use for downloading.""" - streamer_vcodec: str = "libx264" - """The video codec to use for streaming.""" + streamer_vcodec: str = "" + """The video codec to use for streaming. If empty, auto-detect.""" streamer_acodec: str = "aac" """The audio codec to use for streaming.""" diff --git a/app/tests/test_segments.py b/app/tests/test_segments.py index 5b809aeb..51c23859 100644 --- a/app/tests/test_segments.py +++ b/app/tests/test_segments.py @@ -1,5 +1,6 @@ import asyncio import hashlib +import logging import tempfile from pathlib import Path from typing import Any @@ -38,9 +39,7 @@ async def test_build_ffmpeg_args_video_and_audio(tmp_path: Path, monkeypatch: py args = await seg.build_ffmpeg_args(media) # Compute expected symlink path used by Segments - tmpFile = Path(tempfile.gettempdir()).joinpath( - f"ytptube_stream.{hashlib.sha256(str(media).encode()).hexdigest()}" - ) + tmpFile = Path(tempfile.gettempdir()).joinpath(f"ytptube_stream.{hashlib.sha256(str(media).encode()).hexdigest()}") # Start time is duration * index with 6 decimals for non-zero index assert "-ss" in args @@ -105,9 +104,10 @@ class _FakeProc: self.stderr = _FakeStdout([]) self.terminated = False self.killed = False + self._rc = 0 async def wait(self) -> int: - return 0 + return self._rc def terminate(self) -> None: self.terminated = True @@ -131,6 +131,154 @@ class _FakeResp: self.eof = True +class _FakeProcFail(_FakeProc): + def __init__(self, err: bytes = b"") -> None: + # no stdout data, immediate failure + super().__init__([b""]) + self.stderr = _FakeStdout([err, b""]) + self._rc = 1 + + +@pytest.mark.asyncio +async def test_build_ffmpeg_args_no_dri_falls_back_to_software(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + media = tmp_path / "file.mp4" + media.write_bytes(b"data") + + async def fake_ffprobe(_file: Path): + return DummyFF(v=True, a=True) + + monkeypatch.setattr("app.library.Segments.ffprobe", fake_ffprobe) + # Simulate no /dev/dri present but GPU encoders otherwise available + monkeypatch.setattr("app.library.Segments.Segments._has_dri_devices", lambda: False) + monkeypatch.setattr( + "app.library.Segments.Segments._ffmpeg_encoders", lambda: {"h264_nvenc", "h264_qsv", "h264_amf"} + ) + + # reset encoder cache to ensure clean selection in this test + from app.library.Segments import Segments as _Seg + + _Seg._cached_vcodec = None + _Seg._cache_initialized = False + + seg = Segments(download_path=str(tmp_path), index=0, duration=1.0, vconvert=True, aconvert=True) + # Make preferred list try GPUs first + seg.vcodec = "" # empty configured value triggers GPU->software preference + + args = await seg.build_ffmpeg_args(media) + + # Expect software encoder selected + assert "-codec:v" in args + assert args[args.index("-codec:v") + 1] == "libx264" + + +@pytest.mark.asyncio +async def test_stream_gpu_failure_falls_back_to_software( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + async def fake_ffprobe(_file: Path): + return DummyFF(v=True, a=True) + + monkeypatch.setattr("app.library.Segments.ffprobe", fake_ffprobe) + # Allow GPU usage and advertise an NVENC encoder so first pick is GPU + monkeypatch.setattr("app.library.Segments.Segments._has_dri_devices", lambda: True) + monkeypatch.setattr("app.library.Segments.Segments._ffmpeg_encoders", lambda: {"h264_nvenc"}) + + # First process fails (no data, rc=1), second succeeds and outputs bytes + proc_fail = _FakeProcFail(err=b"nvenc failure: encoder not available") + proc_ok = _FakeProc([b"gpu-fallback-", b"ok"]) # after fallback we stream this + + calls: list[int] = [] + + async def fake_create_subprocess_exec(*_args: Any, **_kwargs: Any): + calls.append(1) + return proc_fail if len(calls) == 1 else proc_ok + + monkeypatch.setattr("asyncio.create_subprocess_exec", fake_create_subprocess_exec) + + # reset encoder cache to ensure we try GPU first + from app.library.Segments import Segments as _Seg + + _Seg._cached_vcodec = None + _Seg._cache_initialized = False + + seg = Segments(download_path=str(tmp_path), index=0, duration=1.0, vconvert=True, aconvert=True) + # Encourage GPU preference + seg.vcodec = "" # empty -> try GPUs first + resp = _FakeResp() + with caplog.at_level(logging.WARNING, logger="player.segments"): + await seg.stream(tmp_path / "file.mp4", resp) + + # Ensure fallback path streamed data and closed properly + assert bytes(resp.data) == b"gpu-fallback-ok" + assert resp.eof is True + # Ensure we logged the reason for GPU failure + assert any("Hardware encoder failed" in r.message for r in caplog.records) + assert any("nvenc failure" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_stream_gpu_fallback_switches_to_software_no_hw_flags( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + async def fake_ffprobe(_file: Path): + return DummyFF(v=True, a=True) + + monkeypatch.setattr("app.library.Segments.ffprobe", fake_ffprobe) + # Only QSV advertised so initial build sets QSV + monkeypatch.setattr("app.library.Segments.Segments._has_dri_devices", lambda: True) + monkeypatch.setattr("app.library.Segments.Segments._ffmpeg_encoders", lambda: {"h264_qsv"}) + + # Fail first, succeed second + proc_fail = _FakeProcFail(err=b"qsv failure") + proc_ok = _FakeProc([b"ok"]) # after fallback we stream this + + captured_args: list[list[str]] = [] + + async def fake_create_subprocess_exec(*args: Any, **_kwargs: Any): + captured_args.append(list(args[1:])) + return proc_fail if len(captured_args) == 1 else proc_ok + + monkeypatch.setattr("asyncio.create_subprocess_exec", fake_create_subprocess_exec) + + # reset cache + from app.library.Segments import Segments as _Seg + + _Seg._cached_vcodec = None + _Seg._cache_initialized = False + + seg = Segments(download_path=str(tmp_path), index=0, duration=1.0, vconvert=True, aconvert=True) + seg.vcodec = "intel" + resp = _FakeResp() + await seg.stream(tmp_path / "file.mp4", resp) + + # First call had QSV codec and QSV flags + first = captured_args[0] + assert "-codec:v" in first + assert first[first.index("-codec:v") + 1] == "h264_qsv" + assert "-init_hw_device" in first + assert "qsv=hw:/dev/dri/renderD128" in first + 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] == ( + "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64" + ) + + # Second call (fallback) must be software without any hardware-specific flags + second = captured_args[1] + assert "-codec:v" in second + assert second[second.index("-codec:v") + 1] == "libx264" + assert "-init_hw_device" not in second + assert "-filter_hw_device" not in second + if "-vf" in second: + assert ( + second[second.index("-vf") + 1] + != "scale=trunc(iw/2)*2:trunc(ih/2)*2,format=nv12,hwupload=extra_hw_frames=64" + ) + assert "-pix_fmt" in second + assert second[second.index("-pix_fmt") + 1] == "yuv420p" + + @pytest.mark.asyncio async def test_stream_normal_flow(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: async def fake_ffprobe(_file: Path): @@ -195,6 +343,6 @@ async def test_stream_cancelled(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) # Fail with CancelledError on write to hit inner disconnection branch resp = _FakeResp(fail_with=asyncio.CancelledError()) await seg.stream(tmp_path / "file.mp4", resp) - # Inner branch treats it as client disconnected; no EOF and no termination + # Inner branch treats it as client disconnected; no EOF and we terminate ffmpeg assert resp.eof is False - assert proc.terminated is False + assert proc.terminated is True