"""硬件后端探测与整片、HLS 转码的失败回退。""" import asyncio import os import subprocess import tempfile import unittest from pathlib import Path from unittest.mock import Mock, patch from peach.ffmpeg import BinaryChoice from peach.hardware_transcode import HardwareEncoder, HardwareTranscoder from peach.segments import HlsSegmentService from peach.streaming import StreamSessionRegistry from peach.transcodes import TranscodeService, _MediaProfile class HardwareDetectionTests(unittest.TestCase): def setUp(self): environment = patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "auto", "PEACH_VAAPI_DEVICE": ""}) environment.start() self.addCleanup(environment.stop) def test_linux_render_nodes_are_tried_until_one_encodes(self): commands = [] def run(command, **kwargs): commands.append(command) self.assertEqual(kwargs["timeout"], 10) device = command[command.index("-vaapi_device") + 1] return subprocess.CompletedProcess(command, int(device.endswith("128")), stderr=b"no display") hardware = HardwareTranscoder() with patch("peach.hardware_transcode.sys.platform", "linux"), \ patch.object(Path, "exists", return_value=False), \ patch.object(Path, "glob", return_value=[Path("/dev/dri/renderD128"), Path("/dev/dri/renderD129")]), \ patch("peach.hardware_transcode.subprocess.run", side_effect=run): encoders = hardware.encoders(Path("ffmpeg")) self.assertEqual(encoders, (HardwareEncoder("vaapi", "/dev/dri/renderD129"),)) self.assertEqual(hardware.encoders(Path("ffmpeg")), encoders) self.assertEqual(len(commands), 2) self.assertTrue(all("h264_vaapi" in command for command in commands)) self.assertTrue(all("format=nv12,hwupload" in command for command in commands)) self.assertTrue(all(command[-2:] == ("null", "-") for command in commands)) def test_explicit_device_and_backend_are_probed_without_gpu_listing(self): with patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "vaapi", "PEACH_VAAPI_DEVICE": "/dev/dri/renderD130"}): hardware = HardwareTranscoder() with patch("peach.hardware_transcode.subprocess.run", return_value=Mock(returncode=0)) as run: self.assertEqual(hardware.encoders(Path("ffmpeg")), (HardwareEncoder("vaapi", "/dev/dri/renderD130"),)) self.assertIn("/dev/dri/renderD130", run.call_args.args[0]) self.assertNotIn("h264_nvenc", run.call_args.args[0]) def test_probe_errors_and_timeouts_are_cached_as_unavailable(self): for failure in (PermissionError("denied"), subprocess.TimeoutExpired("ffmpeg", 10)): with self.subTest(failure=failure), patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "nvenc"}): hardware = HardwareTranscoder() with patch("peach.hardware_transcode.subprocess.run", side_effect=failure) as run: self.assertEqual(hardware.encoders(Path("ffmpeg")), ()) self.assertEqual(hardware.encoders(Path("ffmpeg")), ()) self.assertEqual(run.call_count, 1) def test_software_mode_and_disabled_preference_do_not_probe(self): for enabled, backend in ((True, "software"), (False, "auto")): with self.subTest(enabled=enabled, backend=backend), \ patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": backend}), \ patch("peach.hardware_transcode.subprocess.run") as run: self.assertEqual(HardwareTranscoder(enabled=enabled).encoders(Path("ffmpeg")), ()) run.assert_not_called() def test_linux_without_devices_has_no_automatic_candidates(self): for platform in ("linux",): with self.subTest(platform=platform), patch("peach.hardware_transcode.sys.platform", platform), \ patch.object(Path, "exists", return_value=False), patch.object(Path, "glob", return_value=[]): self.assertEqual(HardwareTranscoder()._candidates(), ()) def test_macos_probes_videotoolbox_and_uses_no_software_encoder_in_probe(self): with patch("peach.hardware_transcode.sys.platform", "darwin"), \ patch("peach.hardware_transcode.subprocess.run", return_value=Mock(returncode=0)) as run: self.assertEqual(HardwareTranscoder().encoders(Path("ffmpeg")), (HardwareEncoder("videotoolbox"),)) command = run.call_args.args[0] self.assertIn("h264_videotoolbox", command) self.assertEqual(command[command.index("-allow_sw") + 1], "0") def test_macos_av1_selects_native_decoder_for_hardware_and_keeps_software_decode_fallback(self): attempts = HardwareEncoder("videotoolbox").attempts("av1") self.assertEqual([attempt.name for attempt in attempts], ["videotoolbox", "videotoolbox-encode"]) self.assertEqual(attempts[0].before_input, ("-hwaccel", "videotoolbox", "-c:v", "av1")) self.assertEqual(attempts[1].before_input, ()) self.assertEqual(len(HardwareEncoder("videotoolbox").attempts("vc1")), 1) def test_linux_nvidia_device_enables_nvenc_probe(self): with patch("peach.hardware_transcode.sys.platform", "linux"), \ patch.object(Path, "exists", return_value=True), patch.object(Path, "glob", return_value=[]), \ patch("peach.hardware_transcode.subprocess.run", return_value=Mock(returncode=0)) as run: self.assertEqual(HardwareTranscoder().encoders(Path("ffmpeg")), (HardwareEncoder("nvenc"),)) self.assertIn("h264_nvenc", run.call_args.args[0]) def test_invalid_backend_has_an_actionable_error(self): with patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "typo"}): with self.assertRaisesRegex(ValueError, "PEACH_TRANSCODE_BACKEND"): HardwareTranscoder() def test_device_preference_can_switch_without_recreating_the_hardware_service(self): from peach.media_processing_policy import save_preferences with tempfile.TemporaryDirectory() as directory, \ patch("peach.hardware_transcode.sys.platform", "darwin"), \ patch("peach.hardware_transcode.subprocess.run", return_value=Mock(returncode=0)) as run: root = Path(directory).resolve() hardware = HardwareTranscoder(state_root=root) self.assertTrue(hardware.encoders(Path("ffmpeg"))) save_preferences(root, {'playback_acceleration': 'software'}) self.assertEqual(hardware.encoders(Path("ffmpeg")), ()) save_preferences(root, {'preview_acceleration': 'software'}) self.assertEqual(hardware.encoders(Path("ffmpeg")), ()) save_preferences(root, {'playback_acceleration': 'auto'}) self.assertTrue(hardware.encoders(Path("ffmpeg"))) self.assertEqual(run.call_count, 1) def test_linux_frame_capture_uses_the_first_vaapi_node_that_initializes(self): from peach import hardware_transcode probes = [] def probe(_ffmpeg, _label, arguments): probes.append(arguments) return arguments[1] == "vaapi=va:/dev/dri/renderD129" hardware_transcode._vaapi_device.cache_clear() self.addCleanup(hardware_transcode._vaapi_device.cache_clear) with tempfile.TemporaryDirectory() as directory: ffmpeg = Path(directory) / "ffmpeg" ffmpeg.write_bytes(b"") ffmpeg.with_name("ffprobe").write_bytes(b"") source = Path(directory) / "clip.mkv" source.write_bytes(b"fixture") with patch("peach.hardware_transcode.sys.platform", "linux"), patch.object(Path, "glob", return_value=[Path("/dev/dri/renderD128"), Path("/dev/dri/renderD129")]), patch.object(hardware_transcode, "_run_probe", side_effect=probe), patch("peach.media_processing_policy.acceleration", return_value="auto"): for codec, expected in (("hevc", ("-hwaccel", "vaapi", "-hwaccel_device", "/dev/dri/renderD129")), ("vc1", ())): with self.subTest(codec=codec), patch.object(hardware_transcode, "_frame_codec", return_value=codec): self.assertEqual(hardware_transcode.frame_decode_options(str(ffmpeg), str(source)), expected) self.assertEqual([arguments[:2] for arguments in probes], [("-init_hw_device", "vaapi=va:/dev/dri/renderD128"), ("-init_hw_device", "vaapi=va:/dev/dri/renderD129")], "渲染节点探测按 FFmpeg 缓存，第二部片不再探测") self.assertNotIn("-hwaccel_output_format", expected) def test_linux_frame_capture_without_render_nodes_stays_software_without_probing(self): from peach import hardware_transcode hardware_transcode._vaapi_device.cache_clear() self.addCleanup(hardware_transcode._vaapi_device.cache_clear) with tempfile.TemporaryDirectory() as directory: ffmpeg = Path(directory) / "ffmpeg" ffmpeg.write_bytes(b"") with patch("peach.hardware_transcode.sys.platform", "linux"), patch.object(Path, "glob", return_value=[]), patch.object(hardware_transcode, "_run_probe") as probe, patch("peach.media_processing_policy.acceleration", return_value="auto"): self.assertEqual(hardware_transcode.frame_decode_options(str(ffmpeg), str(ffmpeg)), ()) probe.assert_not_called() class WindowsBackendTests(unittest.TestCase): def setUp(self): for active in (patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "auto"}), patch("peach.hardware_transcode.sys.platform", "win32")): active.start() self.addCleanup(active.stop) @staticmethod def _machine(working): """按命令里的编码器或设备判定成败，模拟一台 NVIDIA 默认、AMD 在第二个序号的机器。""" def run(command, **kwargs): joined = " ".join(command) return subprocess.CompletedProcess(command, 0 if any(item in joined for item in working) else 1, stderr=b"unavailable") return run def test_automatic_mode_probes_every_windows_encoder_and_d3d11_decode(self): self.assertEqual(HardwareTranscoder()._candidates(), ( HardwareEncoder("nvenc"), HardwareEncoder("amf"), HardwareEncoder("qsv"), HardwareEncoder("d3d11va"))) with patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "amf"}): self.assertEqual(HardwareTranscoder()._candidates(), (HardwareEncoder("amf"),)) def test_amf_binds_to_the_adapter_it_can_derive_and_keeps_software_decode(self): working = ("-pix_fmt nv12", "d3d11va=peach:1", "-init_hw_device d3d11va -f") hardware = HardwareTranscoder() with patch("peach.hardware_transcode.subprocess.run", side_effect=self._machine(working)) as run: encoders = hardware.encoders(Path("ffmpeg")) self.assertEqual(hardware.decode_options(Path("ffmpeg"), "hevc"), ("-hwaccel", "d3d11va")) self.assertIn(HardwareEncoder("amf", "1"), encoders) self.assertNotIn(HardwareEncoder("d3d11va"), encoders) self.assertFalse(any("d3d11va=peach:2" in " ".join(call.args[0]) for call in run.call_args_list)) attempts = HardwareEncoder("amf", "1").attempts("hevc") self.assertEqual([attempt.name for attempt in attempts], ["d3d11va-amf", "amf"]) self.assertEqual(attempts[0].before_input, ("-hwaccel", "d3d11va", "-hwaccel_device", "1")) self.assertEqual(attempts[1].before_input, ()) self.assertEqual([attempt.name for attempt in HardwareEncoder("amf", "1").attempts("vc1")], ["amf"]) def test_d3d11_decode_feeds_libx264_when_no_encoder_works(self): with tempfile.TemporaryDirectory() as directory, \ patch("peach.hardware_transcode.subprocess.run", side_effect=self._machine(("-init_hw_device d3d11va -f",))): root = Path(directory).resolve() source = root / "clip.mkv" source.write_bytes(b"fixture") resolver = Mock(ffmpeg=lambda: BinaryChoice(Path("ffmpeg"), "test"), ffprobe=lambda: None) service = TranscodeService(resolver, root / "full") for codec, expected in (("hevc", ["d3d11va-libx264", "libx264"]), ("vc1", ["libx264"])): with self.subTest(codec=codec): attempts = list(service._attempts(Path("ffmpeg"), source, root / "out.mp4", _MediaProfile(codec, "yuv420p", "aac"))) self.assertEqual([attempt.name for attempt in attempts], expected) command = attempts[0].command self.assertEqual("-hwaccel" in command, codec == "hevc") self.assertNotIn("-hwaccel_output_format", command) self.assertIn("libx264", command) hls = HlsSegmentService(resolver, root / "hls", hardware=service.hardware) names = [name for name, _ in hls._attempts(Path("ffmpeg"), source, 12, 6, True, "hevc")] self.assertEqual(names, ["d3d11va-libx264", "libx264"]) command = hls._attempts(Path("ffmpeg"), source, 12, 6, True, "hevc")[0][1](root / "seg.ts") self.assertLess(command.index("d3d11va"), command.index("-i")) self.assertEqual(command[command.index("-threads") + 1], "2") def test_software_playback_setting_turns_off_d3d11_decode_too(self): from peach.media_processing_policy import save_preferences with tempfile.TemporaryDirectory() as directory, \ patch("peach.hardware_transcode.subprocess.run", return_value=Mock(returncode=0)) as run: root = Path(directory).resolve() save_preferences(root, {"playback_acceleration": "software"}) hardware = HardwareTranscoder(state_root=root) self.assertEqual(hardware.decode_options(Path("ffmpeg"), "hevc"), ()) self.assertEqual(hardware.encoders(Path("ffmpeg")), ()) run.assert_not_called() def test_frame_capture_uses_d3d11_only_for_listed_codecs_and_follows_preview_setting(self): from peach import hardware_transcode with tempfile.TemporaryDirectory() as directory: ffmpeg = Path(directory) / "ffmpeg.exe" ffmpeg.write_bytes(b"") ffmpeg.with_name("ffprobe.exe").write_bytes(b"") source = Path(directory) / "clip.mkv" source.write_bytes(b"fixture") for codec, preview, expected in (("hevc", "auto", ("-hwaccel", "d3d11va")), ("vc1", "auto", ()), ("hevc", "software", ())): with self.subTest(codec=codec, preview=preview), \ patch.object(hardware_transcode, "_d3d11va_ready", return_value=True), \ patch.object(hardware_transcode, "_frame_codec", return_value=codec) as probe, \ patch("peach.media_processing_policy.acceleration", return_value=preview): self.assertEqual(hardware_transcode.frame_decode_options(str(ffmpeg), str(source)), expected) if preview == "auto": self.assertEqual(Path(probe.call_args.args[0]).name, "ffprobe.exe") class VideoToolboxConsumerTests(unittest.TestCase): def test_full_file_and_hls_fall_back_from_decode_to_encode_to_software(self): with tempfile.TemporaryDirectory() as directory, \ patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "videotoolbox"}), \ patch.object(HardwareTranscoder, "_probe", return_value=True): root = Path(directory).resolve() source = root / "sample.mkv" source.write_bytes(b"fixture") resolver = Mock(ffmpeg=lambda: BinaryChoice(Path("ffmpeg"), "test"), ffprobe=lambda: None) for failures in (0, 1, 2): with self.subTest(failures=failures): service = TranscodeService(resolver, root / f"full-{failures}") captured = [] def execute(command, *, captured=captured, failures=failures, **kwargs): captured.append((kwargs['label'], command)) if len(captured) <= failures: return 1, b"", b"unavailable" Path(command[-1]).write_bytes(b"result") return 0, b"", b"" with patch.object(service, "_probe", return_value=_MediaProfile("av1", "yuv420p", "aac")), \ patch.object(service, "_execute", side_effect=execute): service.browser_path(1, source) self.assertEqual([name for name, _ in captured], ["videotoolbox", "videotoolbox-encode", "libx264"][:failures + 1]) self.assertIn("av1", captured[0][1]) if failures: self.assertNotIn("-hwaccel", captured[1][1]) async def drive(failures=failures): hls = HlsSegmentService(resolver, root / f"hls-{failures}", hardware=HardwareTranscoder()) commands = [] async def run(command, session, registry): commands.append(command) if len(commands) <= failures: return 1, b"unavailable" Path(command[-1]).write_bytes(b"segment") return 0, b"" with patch.object(hls, "video_codec", return_value="av1"), patch.object(hls, "_run", side_effect=run): await hls.generate(source, 12, 6, asset_id=1, index=2, session="test", registry=StreamSessionRegistry(), transcode=True) self.assertEqual(len(commands), failures + 1) for command in commands: self.assertEqual(command[command.index("-output_ts_offset") + 1], "12.000") self.assertIn("aac", command) asyncio.run(drive()) class VaapiConsumerTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name).resolve() self.source = self.root / "clip.mkv" self.source.write_bytes(b"source") self.resolver = Mock(ffmpeg=lambda: BinaryChoice(Path("ffmpeg"), "test"), ffprobe=lambda: None) environment = patch.dict(os.environ, {"PEACH_TRANSCODE_BACKEND": "vaapi", "PEACH_VAAPI_DEVICE": "/dev/dri/renderD128"}) environment.start() self.addCleanup(environment.stop) probe = patch.object(HardwareTranscoder, "_probe", return_value=True) probe.start() self.addCleanup(probe.stop) def test_full_file_tries_vaapi_then_hardware_encode_then_software(self): for failures, expected in ((0, ["vaapi"]), (1, ["vaapi", "vaapi-encode"]), (2, ["vaapi", "vaapi-encode", "libx264"])): with self.subTest(failures=failures): service = TranscodeService(self.resolver, self.root / f"cache{failures}") captured = [] def execute(command, *, captured=captured, failures=failures, **kwargs): captured.append((kwargs["label"], command)) if len(captured) <= failures: return 1, b"", b"hardware failed" Path(command[-1]).write_bytes(b"mp4") return 0, b"", b"" profile = _MediaProfile("hevc", "yuv420p10le", "aac") with patch.object(service, "_probe", return_value=profile), \ patch.object(service, "_execute", side_effect=execute): target, converted = service.browser_path(95, self.source) self.assertTrue(converted) self.assertEqual(target.read_bytes(), b"mp4") self.assertEqual([name for name, _ in captured], expected) command = captured[0][1] self.assertLess(command.index("-hwaccel"), command.index("-i")) self.assertIn("scale_vaapi=format=nv12", command) self.assertIn("h264_vaapi", command) self.assertIn("+faststart", command) if failures: self.assertNotIn("-hwaccel", captured[1][1]) self.assertIn("format=nv12,hwupload", captured[1][1]) def test_video_copy_does_not_probe_hardware(self): service = TranscodeService(self.resolver, self.root / "cache") commands = service._attempts(Path("ffmpeg"), self.source, self.root / "out.mp4", _MediaProfile("h264", "yuv420p", "aac")) with patch.object(service.hardware, "encoders") as encoders: self.assertEqual(next(commands).name, "video-copy") encoders.assert_not_called() def test_hls_vaapi_failures_preserve_seek_timestamps_and_audio(self): async def drive(failures): service = HlsSegmentService(self.resolver, self.root / f"hls{failures}", hardware=HardwareTranscoder()) captured = [] async def run(command, session, registry): captured.append(command) if len(captured) <= failures: return 1, b"hardware failed" Path(command[-1]).write_bytes(b"ts") return 0, b"" with patch.object(service, "_run", side_effect=run), \ patch.object(service, "video_codec", return_value="hevc"): target = await service.generate(self.source, 12, 6, asset_id=95, index=2, session="test", registry=StreamSessionRegistry(), transcode=True) self.assertEqual(target.read_bytes(), b"ts") self.assertEqual(len(captured), failures + 1) for command in captured: self.assertEqual(command[command.index("-ss") + 1], "12.000") self.assertEqual(command[command.index("-t") + 1], "6.000") self.assertEqual(command[command.index("-output_ts_offset") + 1], "12.000") self.assertEqual(command[command.index("-c:a") + 1], "aac") self.assertEqual(command[command.index("-f") + 1], "mpegts") self.assertIn("scale_vaapi=format=nv12", captured[0]) if failures: self.assertIn("format=nv12,hwupload", captured[1]) if failures == 2: self.assertIn("libx264", captured[2]) for failures in (0, 1, 2): with self.subTest(failures=failures): asyncio.run(drive(failures)) def test_hls_remux_does_not_probe_hardware(self): service = HlsSegmentService(self.resolver, self.root / "hls", hardware=HardwareTranscoder()) with patch.object(service.hardware, "encoders") as encoders: attempts = service._attempts(Path("ffmpeg"), self.source, 0, 6, False) encoders.assert_not_called() self.assertEqual([name for name, _ in attempts], ["copy"])