Add video stream copy remuxing

This commit is contained in:
ajp_anton
2026-07-22 22:51:00 +00:00
parent 49296d6f45
commit d60ce51b14
7 changed files with 404 additions and 68 deletions
+31 -11
View File
@@ -232,6 +232,30 @@ class VideoEncodeEndToEndTests(unittest.TestCase):
self.assertFalse(list((WORKSPACE / ".source").rglob("*.frames.csv")))
self.assertFalse(list((WORKSPACE / ".source").rglob("*.runner.log")))
def test_mp4_stream_copy_remuxes_video_and_audio(self) -> None:
with tempfile.TemporaryDirectory(prefix="mbt-video-e2e-") as temp_dir:
source = self.copy_fixture("20260602_222842.mp4", Path(temp_dir))
self.create_workspace(source)
self.run_video_encode(
source,
encode=True,
extra_env={
"MBT_VIDEO_CONTAINER": "mp4",
"MBT_VIDEO_CODEC": "copy",
"MBT_VIDEO_AUDIO": "copy",
},
)
output = WORKSPACE / "20260602_222842.mp4.mp4"
self.assertTrue(output.is_file(), output)
self.assert_video_stream_frames(output, 170)
streams = ffprobe_json(output, self.env)["streams"]
self.assertEqual(
next(stream["codec_name"] for stream in streams if stream["codec_type"] == "video"),
"hevc",
)
self.assert_audio_codec(output, "aac")
def test_vfr_mp4_trim_preserves_location_and_adjusts_end_timestamp(self) -> None:
with tempfile.TemporaryDirectory(prefix="mbt-video-e2e-") as temp_dir:
source = self.copy_fixture("20260602_222842.mp4", Path(temp_dir))
@@ -359,21 +383,17 @@ last
self.assertEqual(completed.returncode, 0, completed.stderr)
self.assertIn(expected, completed.stdout)
def test_simple_trim_can_copy_audio(self) -> None:
def test_incompatible_mp4_audio_copy_is_rejected(self) -> None:
with tempfile.TemporaryDirectory(prefix="mbt-video-e2e-") as temp_dir:
source = self.copy_fixture("C0011.MP4", Path(temp_dir))
script = self.create_workspace(source)
self.replace_user_script_body(script, "Trim(last, 10, 49)\nlast")
self.run_video_encode(
source,
encode=True,
extra_env={"MBT_VIDEO_CONTAINER": "mp4", "MBT_VIDEO_AUDIO": "copy"},
)
output = WORKSPACE / "C0011.MP4.mp4"
self.assertTrue(output.is_file(), output)
self.assert_video_stream_frames(output, 40)
self.assert_audio_codec(output, "pcm_s16be")
with self.assertRaises(subprocess.CalledProcessError):
self.run_video_encode(
source,
encode=True,
extra_env={"MBT_VIDEO_CONTAINER": "mp4", "MBT_VIDEO_AUDIO": "copy"},
)
def test_mkv_opus_output(self) -> None:
with tempfile.TemporaryDirectory(prefix="mbt-video-e2e-") as temp_dir: