1035 lines
39 KiB
Python
1035 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
import tempfile
|
|
from contextlib import redirect_stdout
|
|
from datetime import datetime, timedelta, timezone
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from tools.avisynth_render import AvisynthClipInfo, run_validation_script
|
|
from tools.avisynth_runner import script_requests_32bit, validate_runner_path
|
|
from tools.avisynth_validate import FrameIdentity
|
|
from tools.avisynth_workspace import (
|
|
_clear_source_root,
|
|
_format_relative_timestamp,
|
|
_hidden_source_script,
|
|
)
|
|
from tools.executables import require_executable
|
|
from tools.video_encode_output import (
|
|
AudioEncodeOptions,
|
|
VideoCodecOptions,
|
|
add_video_threads,
|
|
atempo_filters,
|
|
choose_video_pixel_format,
|
|
encoder_statistics_lines,
|
|
read_ffmpeg_progress,
|
|
speed_change_filters,
|
|
)
|
|
from tools.video_timestamp_encode import write_timecode_v2, x264_command
|
|
from tools.video_encode_plan import (
|
|
missing_source_frames_between_kept,
|
|
normal_deleted_frame_count,
|
|
y4m_command_for_timeline,
|
|
)
|
|
from tools.video_codec_constraints import (
|
|
VideoConstraintSpec,
|
|
available_levels,
|
|
available_profiles,
|
|
minimum_level,
|
|
minimum_profile,
|
|
resolve_level,
|
|
resolve_profile,
|
|
)
|
|
from tools.video_inputs import VideoInput
|
|
from tools.video_options import OutputNamingOptions
|
|
from tools.video_outputs import output_begin_timestamp, planned_output_path
|
|
from tools.video_batch_encode import (
|
|
_default_output_timezone,
|
|
_default_answer,
|
|
_encoding_steps,
|
|
_parse_answer,
|
|
_question_for_key,
|
|
)
|
|
from tools.video_probe import AudioStreamProbe, VideoProbe, _video_timezone, summarize_frame_timing
|
|
from tools.video_reporting import _format_probe_timestamp, _format_vfr_timing
|
|
from tools.video_reporting import _format_source_span, format_validation_summary_lines
|
|
from tools.console import PipelineProgressView, ProgressView, _progress_bar
|
|
from tools.video_encode_output import _update_renderer_progress
|
|
from tools.video_timeline import OutputFrameTiming, OutputTimeline, build_output_timeline
|
|
from video_encode import print_workspace_summary, wait_for_script_edits
|
|
|
|
|
|
def make_probe(
|
|
*,
|
|
timestamps: list[float],
|
|
duration: float,
|
|
metadata_end: datetime | None = None,
|
|
source_timezone: timezone | None = None,
|
|
) -> VideoProbe:
|
|
return VideoProbe(
|
|
path=Path("source.mp4"),
|
|
codec="h264",
|
|
width=1920,
|
|
height=1080,
|
|
pixel_format="yuv420p",
|
|
bit_depth=8,
|
|
video_bit_rate=10_000_000,
|
|
color_space="bt709",
|
|
color_matrix="Rec709",
|
|
expected_color_matrix="Rec709",
|
|
duration_seconds=duration,
|
|
frame_timestamps=timestamps,
|
|
stream_frame_count=len(timestamps),
|
|
audio_streams=[AudioStreamProbe("aac", 2, 48000, 192_000)],
|
|
audio_stream_count=1,
|
|
audio_sample_rate=48000,
|
|
avg_frame_rate="1/1",
|
|
r_frame_rate="1/1",
|
|
time_base="1/1000",
|
|
timing=summarize_frame_timing(timestamps),
|
|
metadata_end=metadata_end,
|
|
metadata_begin_tag=None,
|
|
source_timezone=source_timezone,
|
|
computed_begin=(
|
|
metadata_end - timedelta(seconds=duration)
|
|
if metadata_end is not None
|
|
else None
|
|
),
|
|
filename_begin=(
|
|
metadata_end - timedelta(seconds=round(duration))
|
|
if metadata_end is not None
|
|
else None
|
|
),
|
|
gps_latitude=None,
|
|
gps_longitude=None,
|
|
warnings=[],
|
|
)
|
|
|
|
|
|
class VideoCodecConstraintTests(unittest.TestCase):
|
|
def test_pixel_format_preserves_ten_bit_output(self) -> None:
|
|
formats = [
|
|
(8, "yuv420p"),
|
|
(10, "yuv420p10le"),
|
|
(12, "yuv420p12le"),
|
|
]
|
|
with patch(
|
|
"tools.video_encode_output.supported_pixel_formats",
|
|
return_value=formats,
|
|
):
|
|
selected = choose_video_pixel_format(
|
|
Path("ffmpeg"),
|
|
"libx265",
|
|
10,
|
|
"420",
|
|
)
|
|
|
|
self.assertEqual(selected, "yuv420p10le")
|
|
|
|
def test_x265_minimum_profile_tracks_depth_and_chroma(self) -> None:
|
|
self.assertEqual(
|
|
minimum_profile("libx265", VideoConstraintSpec(1920, 1080, 12, "420", 60)),
|
|
"main12",
|
|
)
|
|
self.assertEqual(
|
|
minimum_profile("libx265", VideoConstraintSpec(1920, 1080, 10, "422", 60)),
|
|
"main422-10",
|
|
)
|
|
self.assertEqual(
|
|
minimum_profile("libx265", VideoConstraintSpec(1920, 1080, 8, "444", 60)),
|
|
"main444-8",
|
|
)
|
|
|
|
def test_manual_x265_profile_is_upgraded_per_clip(self) -> None:
|
|
clip = VideoConstraintSpec(1920, 1080, 10, "422", 60)
|
|
self.assertEqual(resolve_profile("libx265", "main", clip), "main422-10")
|
|
self.assertEqual(resolve_profile("libx265", "main12", clip), "main422-12")
|
|
|
|
def test_manual_x264_profile_is_upgraded_per_clip(self) -> None:
|
|
clip = VideoConstraintSpec(1920, 1080, 10, "444", 60)
|
|
self.assertEqual(resolve_profile("libx264", "high", clip), "high444")
|
|
|
|
def test_profile_options_include_profiles_supported_by_some_clips(self) -> None:
|
|
clips = [
|
|
VideoConstraintSpec(1920, 1080, 8, "420", 30),
|
|
VideoConstraintSpec(1920, 1080, 10, "422", 30),
|
|
]
|
|
profiles = available_profiles("libx265", clips)
|
|
self.assertIn("main", profiles)
|
|
self.assertIn("main422-10", profiles)
|
|
self.assertNotIn("main444-12", profiles)
|
|
|
|
x264_profiles = available_profiles("libx264", [clips[0]])
|
|
self.assertEqual(x264_profiles, ["baseline", "main", "high"])
|
|
|
|
def test_minimum_hevc_level_and_manual_floor_are_per_clip(self) -> None:
|
|
hd = VideoConstraintSpec(1920, 1080, 10, "420", 60)
|
|
uhd = VideoConstraintSpec(3840, 2160, 10, "420", 60)
|
|
self.assertEqual(minimum_level("libx265", hd), "4.1")
|
|
self.assertEqual(minimum_level("libx265", uhd), "5.1")
|
|
self.assertEqual(resolve_level("libx265", "4.0", uhd), "5.1")
|
|
self.assertIn("4.1", available_levels("libx265", [hd, uhd]))
|
|
|
|
def test_none_omits_profile_and_level(self) -> None:
|
|
clip = VideoConstraintSpec(1920, 1080, 8, "420", 30)
|
|
self.assertIsNone(resolve_profile("libx265", None, clip))
|
|
self.assertIsNone(resolve_level("libx265", None, clip))
|
|
|
|
|
|
class WorkspaceSummaryTests(unittest.TestCase):
|
|
def test_source_cleanup_retries_transient_windows_lock(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
source_root = Path(temp_dir) / ".source"
|
|
source_root.mkdir()
|
|
with (
|
|
patch(
|
|
"tools.avisynth_workspace.shutil.rmtree",
|
|
side_effect=[PermissionError("locked"), None],
|
|
) as remove,
|
|
patch("tools.avisynth_workspace.time.sleep") as sleep,
|
|
):
|
|
_clear_source_root(source_root)
|
|
|
|
self.assertEqual(remove.call_count, 2)
|
|
sleep.assert_called_once_with(0.25)
|
|
|
|
def test_existing_scripts_are_not_repeated(self) -> None:
|
|
root = Path("workspace")
|
|
items = [
|
|
VideoInput(Path("a.mp4"), Path("a.mp4"), Path("a.mp4")),
|
|
VideoInput(Path("b.mp4"), Path("b.mp4"), Path("b.mp4")),
|
|
]
|
|
existing = [root / "a.mp4.avs", root / "b.mp4.avs"]
|
|
|
|
output = self._summary(root, items, existing)
|
|
|
|
self.assertIn("existing editable scripts:", output)
|
|
self.assertNotIn("new editable scripts:", output)
|
|
self.assertEqual(output.count("a.mp4.avs"), 1)
|
|
self.assertEqual(output.count("b.mp4.avs"), 1)
|
|
|
|
def test_mixed_scripts_are_split_into_existing_and_new(self) -> None:
|
|
root = Path("workspace")
|
|
items = [
|
|
VideoInput(Path("a.mp4"), Path("a.mp4"), Path("a.mp4")),
|
|
VideoInput(Path("b.mp4"), Path("b.mp4"), Path("b.mp4")),
|
|
]
|
|
|
|
output = self._summary(root, items, [root / "a.mp4.avs"])
|
|
|
|
self.assertIn("existing editable scripts:", output)
|
|
self.assertIn("new editable scripts:", output)
|
|
self.assertEqual(output.count("a.mp4.avs"), 1)
|
|
self.assertEqual(output.count("b.mp4.avs"), 1)
|
|
|
|
def test_new_only_uses_plain_editable_heading(self) -> None:
|
|
root = Path("workspace")
|
|
item = VideoInput(Path("a.mp4"), Path("a.mp4"), Path("a.mp4"))
|
|
|
|
output = self._summary(root, [item], [])
|
|
|
|
self.assertIn(" editable scripts:", output)
|
|
self.assertNotIn("new editable scripts:", output)
|
|
|
|
def test_validation_prompt_defaults_to_fast_and_accepts_real(self) -> None:
|
|
with (
|
|
patch("video_encode.sys.stdin.isatty", return_value=True),
|
|
patch("video_encode.prompt_input", return_value=""),
|
|
):
|
|
self.assertEqual(wait_for_script_edits(), "fast")
|
|
with (
|
|
patch("video_encode.sys.stdin.isatty", return_value=True),
|
|
patch("video_encode.prompt_input", return_value="r"),
|
|
):
|
|
self.assertEqual(wait_for_script_edits(), "real")
|
|
|
|
@staticmethod
|
|
def _summary(root, items, existing) -> str:
|
|
output = StringIO()
|
|
with redirect_stdout(output):
|
|
print_workspace_summary(root, items, [], existing)
|
|
return output.getvalue()
|
|
|
|
|
|
class VideoTimelineTests(unittest.TestCase):
|
|
def test_timecode_v2_includes_final_frame_end(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
path = Path(temp_dir) / "frames.txt"
|
|
write_timecode_v2(path, [0.04, 0.06, 0.05])
|
|
|
|
self.assertEqual(
|
|
path.read_text(encoding="utf-8").splitlines(),
|
|
[
|
|
"# timecode format v2",
|
|
"0.000000000",
|
|
"40.000000000",
|
|
"100.000000000",
|
|
"150.000000000",
|
|
],
|
|
)
|
|
|
|
def test_x264_vfr_command_uses_timecodes_and_output_format(self) -> None:
|
|
command = x264_command(
|
|
x264=Path("x264"),
|
|
timecodes=Path("frames.txt"),
|
|
output=Path("video.h264"),
|
|
options=VideoCodecOptions(
|
|
codec="libx264",
|
|
crf=18,
|
|
preset="slow",
|
|
profile="high10",
|
|
level="4.1",
|
|
threads=3,
|
|
),
|
|
pixel_format="yuv422p10le",
|
|
colorspace="bt709",
|
|
)
|
|
|
|
self.assertIn("--tcfile-in", command)
|
|
self.assertEqual(command[command.index("--output-depth") + 1], "10")
|
|
self.assertEqual(command[command.index("--output-csp") + 1], "i422")
|
|
self.assertEqual(command[command.index("--colormatrix") + 1], "bt709")
|
|
self.assertEqual(command[command.index("--threads") + 1], "3")
|
|
|
|
def test_encoder_threads_use_x265_worker_pools(self) -> None:
|
|
command: list[str] = []
|
|
|
|
add_video_threads(
|
|
command,
|
|
VideoCodecOptions(codec="libx265", crf=24, preset="slow", threads=3),
|
|
)
|
|
|
|
self.assertEqual(command, ["-x265-params", "pools=3"])
|
|
|
|
def test_encoder_statistics_keep_frame_and_reference_lines(self) -> None:
|
|
statistics = encoder_statistics_lines(
|
|
"x264 [info]: frame I:2 Avg QP:18.00\n"
|
|
"x264 [info]: frame P:5 Avg QP:20.00\n"
|
|
"x264 [info]: frame B:9 Avg QP:21.00\n"
|
|
"x264 [info]: consecutive B-frames: 10.0% 20.0%\n"
|
|
"x264 [info]: ref P L0: 75.0% 25.0%\n"
|
|
"x264 [info]: kb/s:1234.56\n"
|
|
)
|
|
|
|
self.assertEqual(
|
|
statistics,
|
|
[
|
|
"frame I:2 Avg QP:18.00",
|
|
"frame P:5 Avg QP:20.00",
|
|
"frame B:9 Avg QP:21.00",
|
|
"consecutive B-frames: 10.0% 20.0%",
|
|
"ref P L0: 75.0% 25.0%",
|
|
],
|
|
)
|
|
|
|
def test_timestamped_y4m_uses_timeline_average_rate(self) -> None:
|
|
probe = make_probe(timestamps=[0.0, 0.04, 0.10], duration=0.16)
|
|
timeline = build_output_timeline(
|
|
[
|
|
FrameIdentity(output_frame=0, source_id=1, source_frame=0),
|
|
FrameIdentity(output_frame=1, source_id=1, source_frame=1),
|
|
FrameIdentity(output_frame=2, source_id=1, source_frame=2),
|
|
],
|
|
probe=probe,
|
|
)
|
|
|
|
command = y4m_command_for_timeline(
|
|
Path("runner"), timeline, force_timeline_rate=True
|
|
)
|
|
|
|
self.assertEqual(command[:2], ["runner", "--y4m-skip-drop"])
|
|
self.assertEqual(command[2:], ["75", "4"])
|
|
|
|
def test_validation_progress_embeds_percent_at_fifty_percent_cutoff(self) -> None:
|
|
self.assertEqual(
|
|
_progress_bar(1.0, embedded_percent=True),
|
|
"[######## 100.0% ########]",
|
|
)
|
|
self.assertEqual(
|
|
_progress_bar(0.801, embedded_percent=True),
|
|
"[###### 80.1% ######-----]",
|
|
)
|
|
self.assertEqual(
|
|
_progress_bar(0.499, embedded_percent=True),
|
|
"[############-- 49.9% ---]",
|
|
)
|
|
|
|
def test_progress_eta_samples_ignore_duplicate_frame_updates(self) -> None:
|
|
progress = ProgressView(10, "Encoding", stream=StringIO())
|
|
|
|
progress.update(1)
|
|
progress.update(2)
|
|
progress.update(2)
|
|
progress.update(2)
|
|
|
|
self.assertEqual([frame for frame, _ in progress.samples], [1, 2])
|
|
self.assertEqual(
|
|
_progress_bar(0.5, embedded_percent=True),
|
|
"[## 50.0% ###------------]",
|
|
)
|
|
|
|
def test_encoding_progress_matches_persistent_validation_style(self) -> None:
|
|
class TtyBuffer(StringIO):
|
|
def isatty(self) -> bool:
|
|
return True
|
|
|
|
class Process:
|
|
stderr = iter(["frame=2\n", "frame=10\n", "progress=end\n"])
|
|
returncode = None
|
|
|
|
def wait(self) -> None:
|
|
self.returncode = 0
|
|
|
|
output = TtyBuffer()
|
|
with patch("tools.video_encode_output.sys.stdout", output):
|
|
read_ffmpeg_progress(
|
|
Process(),
|
|
total_frames=10,
|
|
label="Encoding script 1/1: clip.mp4",
|
|
)
|
|
|
|
rendered = output.getvalue()
|
|
self.assertIn("Encoding script 1/1: clip.mp4", rendered)
|
|
self.assertIn("100.0%", rendered)
|
|
self.assertIn("fps", rendered)
|
|
self.assertTrue(rendered.endswith("\n"))
|
|
|
|
def test_pipeline_progress_accepts_runner_render_protocol(self) -> None:
|
|
class TtyBuffer(StringIO):
|
|
def isatty(self) -> bool:
|
|
return True
|
|
|
|
output = TtyBuffer()
|
|
with patch("tools.console.sys.stdout", output):
|
|
progress = PipelineProgressView(
|
|
rendering_label="Rendering script 1/1: clip.mp4",
|
|
encoding_label="Encoding script 1/1: clip.mp4",
|
|
encoding_total=10,
|
|
stream=output,
|
|
)
|
|
self.assertTrue(
|
|
_update_renderer_progress(
|
|
progress,
|
|
"mbt_render,2,10,20.00,4.000,16.000\n",
|
|
)
|
|
)
|
|
progress.update_encoding(1)
|
|
progress.finish()
|
|
|
|
rendered = output.getvalue()
|
|
self.assertIn("Rendering script 1/1: clip.mp4", rendered)
|
|
self.assertIn("Encoding script 1/1: clip.mp4", rendered)
|
|
self.assertIn("2/10", rendered)
|
|
|
|
def test_drop_frame_extends_previous_frame_duration(self) -> None:
|
|
probe = make_probe(timestamps=[0.0, 1.0, 2.0, 3.0], duration=4.0)
|
|
timeline = build_output_timeline(
|
|
[
|
|
FrameIdentity(output_frame=0, source_id=1, source_frame=0),
|
|
FrameIdentity(output_frame=1, source_id=1, source_frame=1, drop_frame=True),
|
|
FrameIdentity(output_frame=2, source_id=1, source_frame=2),
|
|
FrameIdentity(output_frame=3, source_id=1, source_frame=3, drop_frame=True),
|
|
],
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertEqual(timeline.dropped_frame_count, 2)
|
|
self.assertEqual([frame.source_frame for frame in timeline.frames], [0, 2])
|
|
self.assertEqual([frame.duration for frame in timeline.frames], [2.0, 2.0])
|
|
self.assertEqual(timeline.duration_seconds, 4.0)
|
|
self.assertEqual(timeline.timing_kind, "cfr")
|
|
self.assertFalse(timeline.beginning_trimmed)
|
|
self.assertFalse(timeline.ending_trimmed)
|
|
|
|
def test_vfr_source_durations_are_preserved(self) -> None:
|
|
probe = make_probe(timestamps=[0.0, 0.04, 0.10], duration=0.16)
|
|
timeline = build_output_timeline(
|
|
[
|
|
FrameIdentity(output_frame=0, source_id=1, source_frame=0),
|
|
FrameIdentity(output_frame=1, source_id=1, source_frame=1),
|
|
FrameIdentity(output_frame=2, source_id=1, source_frame=2),
|
|
],
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertAlmostEqual(timeline.duration_seconds, 0.16)
|
|
self.assertEqual([round(frame.duration, 3) for frame in timeline.frames], [0.04, 0.06, 0.06])
|
|
self.assertEqual(timeline.timing_kind, "vfr")
|
|
|
|
def test_complex_source_span_is_summarized(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[float(index) for index in range(10)],
|
|
duration=10.0,
|
|
)
|
|
timeline = build_output_timeline(
|
|
[
|
|
FrameIdentity(index, 1, source_frame)
|
|
for index, source_frame in enumerate((0, 2, 4, 6, 8))
|
|
],
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertEqual(
|
|
_format_source_span(timeline),
|
|
"0 -> 8, 5 source frames in 5 spans of 10",
|
|
)
|
|
|
|
def test_validation_summary_compares_changed_values(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[float(index) for index in range(4)],
|
|
duration=4.0,
|
|
metadata_end=datetime(2026, 1, 1, 12, 0, 4, tzinfo=timezone.utc),
|
|
)
|
|
timeline = build_output_timeline(
|
|
[FrameIdentity(0, 1, 1), FrameIdentity(1, 1, 2)],
|
|
probe=probe,
|
|
)
|
|
clip_info = AvisynthClipInfo(
|
|
width=1280,
|
|
height=720,
|
|
chroma="420",
|
|
bit_depth=8,
|
|
frames=2,
|
|
fps_num=1,
|
|
fps_den=1,
|
|
has_audio=True,
|
|
audio_rate=48000,
|
|
audio_channels=1,
|
|
audio_samples=96000,
|
|
)
|
|
|
|
lines = format_validation_summary_lines(
|
|
output_frame_count=2,
|
|
dropped_frame_count=0,
|
|
timeline=timeline,
|
|
clip_info=clip_info,
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertIn("duration: 2.000s, 2 frames (4.000s, 4 frames)", lines)
|
|
self.assertIn("source span: 1 -> 2 of 4", lines)
|
|
self.assertIn(
|
|
"video: CFR 1.000 fps, 1280x720, 4:2:0, 8-bit, unknown "
|
|
"(h264, 1.000 fps, 1920x1080, Rec709, 10 Mbit/s)",
|
|
lines,
|
|
)
|
|
self.assertFalse(any(line.startswith("framerate:") for line in lines))
|
|
self.assertTrue(any("audio: 1 channel, 48000 Hz (aac, 2 channels, 192 kbit/s)" in line for line in lines))
|
|
self.assertIn("timestamp: 2026-01-01 12:00:01 (12:00:00)", lines)
|
|
|
|
def test_validation_summary_marks_source_vfr_when_output_is_cfr(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0, 0.02, 0.04, 0.08, 0.10],
|
|
duration=0.12,
|
|
)
|
|
timeline = build_output_timeline(
|
|
[FrameIdentity(0, 1, 0), FrameIdentity(1, 1, 1)],
|
|
probe=probe,
|
|
)
|
|
|
|
lines = format_validation_summary_lines(
|
|
output_frame_count=2,
|
|
dropped_frame_count=0,
|
|
timeline=timeline,
|
|
clip_info=None,
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertEqual(timeline.timing_kind, "cfr")
|
|
self.assertIn("h264, VFR 1.000 fps", lines[2])
|
|
|
|
def test_probe_timestamp_uses_recorded_timezone_and_one_date(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0, 1.0],
|
|
duration=7.0,
|
|
metadata_end=datetime(2026, 7, 4, 14, 13, 56, tzinfo=timezone.utc),
|
|
source_timezone=timezone(timedelta(hours=2)),
|
|
)
|
|
|
|
self.assertEqual(
|
|
_format_probe_timestamp(probe),
|
|
"2026-07-04 16:13:49 - 16:13:56 +02:00",
|
|
)
|
|
|
|
def test_android_timezone_is_accepted_as_source_metadata(self) -> None:
|
|
parsed = _video_timezone({"QuickTime:AndroidTimeZone": "+0200"})
|
|
|
|
self.assertIsNotNone(parsed)
|
|
self.assertEqual(parsed.utcoffset(None), timedelta(hours=2))
|
|
|
|
def test_encoding_timezone_prefers_one_recorded_offset(self) -> None:
|
|
probes = [
|
|
make_probe(timestamps=[0.0], duration=1.0),
|
|
make_probe(
|
|
timestamps=[0.0],
|
|
duration=1.0,
|
|
source_timezone=timezone(timedelta(hours=9)),
|
|
),
|
|
]
|
|
|
|
self.assertEqual(
|
|
_default_output_timezone(probes).utcoffset(None),
|
|
timedelta(hours=9),
|
|
)
|
|
|
|
def test_timestamp_naming_off_skips_timezone_step(self) -> None:
|
|
self.assertNotIn(
|
|
"timezone",
|
|
_encoding_steps({"timestamp_names": False}, has_speed_changes=False),
|
|
)
|
|
|
|
def test_threads_setting_follows_level_and_requires_positive_integer(self) -> None:
|
|
steps = _encoding_steps({}, has_speed_changes=False)
|
|
|
|
self.assertEqual(steps[steps.index("level") + 1], "threads")
|
|
self.assertEqual(_parse_answer("threads", "auto", {}, []), "auto")
|
|
self.assertEqual(_parse_answer("threads", "2", {}, []), 2)
|
|
with self.assertRaisesRegex(ValueError, "positive integer"):
|
|
_parse_answer("threads", "0", {}, [])
|
|
|
|
def test_default_crf_depends_on_codec(self) -> None:
|
|
self.assertEqual(_default_answer("crf", {"codec": "libx264"}, []), 16)
|
|
self.assertEqual(_default_answer("crf", {"codec": "libx265"}, []), 21)
|
|
|
|
def test_container_question_lists_available_options(self) -> None:
|
|
question = _question_for_key("container", {}, [])
|
|
|
|
self.assertIn("(mp4/mkv)", question)
|
|
|
|
def test_preset_question_lists_available_options(self) -> None:
|
|
question = _question_for_key("preset", {}, [])
|
|
|
|
self.assertIn("ultrafast/superfast/veryfast", question)
|
|
self.assertIn("slow/slower/veryslow/placebo", question)
|
|
|
|
def test_vfr_summary_places_milliseconds_in_heading(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0, 0.016667, 0.033334, 0.066667],
|
|
duration=0.1,
|
|
)
|
|
|
|
summary = _format_vfr_timing(probe)
|
|
|
|
self.assertIn("rounded deltas (ms):", summary)
|
|
self.assertIn("common deltas (ms):", summary)
|
|
self.assertIn("16.667 2x", summary)
|
|
self.assertNotIn(" ms", summary)
|
|
|
|
def test_drop_frames_do_not_require_audio_cuts(self) -> None:
|
|
timeline = OutputTimeline(
|
|
frames=[
|
|
OutputFrameTiming(0, 1, 0, 0.0, 2.0),
|
|
OutputFrameTiming(1, 1, 2, 2.0, 1.0),
|
|
],
|
|
audio_segments=[(0.0, 3.0)],
|
|
consumed_source_frames=[0, 1, 2],
|
|
duration_seconds=3.0,
|
|
timing_kind="vfr",
|
|
beginning_trimmed=False,
|
|
ending_trimmed=False,
|
|
dropped_frame_count=1,
|
|
source_frame_count=3,
|
|
matrix=None,
|
|
)
|
|
|
|
self.assertEqual(missing_source_frames_between_kept(timeline), 1)
|
|
self.assertEqual(normal_deleted_frame_count(timeline), 0)
|
|
|
|
def test_normal_frame_deletion_requires_audio_cut_segments(self) -> None:
|
|
timeline = OutputTimeline(
|
|
frames=[
|
|
OutputFrameTiming(0, 1, 0, 0.0, 1.0),
|
|
OutputFrameTiming(1, 1, 1, 1.0, 1.0),
|
|
OutputFrameTiming(2, 1, 3, 3.0, 1.0),
|
|
OutputFrameTiming(3, 1, 5, 5.0, 1.0),
|
|
OutputFrameTiming(4, 1, 6, 6.0, 1.0),
|
|
],
|
|
audio_segments=[(0.0, 2.0), (3.0, 1.0), (5.0, 2.0)],
|
|
consumed_source_frames=[0, 1, 3, 5, 6],
|
|
duration_seconds=5.0,
|
|
timing_kind="vfr",
|
|
beginning_trimmed=False,
|
|
ending_trimmed=False,
|
|
dropped_frame_count=0,
|
|
source_frame_count=7,
|
|
matrix=None,
|
|
)
|
|
|
|
self.assertEqual(missing_source_frames_between_kept(timeline), 2)
|
|
self.assertEqual(normal_deleted_frame_count(timeline), 2)
|
|
self.assertEqual(timeline.audio_segments, [(0.0, 2.0), (3.0, 1.0), (5.0, 2.0)])
|
|
|
|
def test_mixed_delete_and_drop_frame_audio_segments(self) -> None:
|
|
probe = make_probe(timestamps=[0.0, 1.0, 2.0, 3.0, 4.0], duration=5.0)
|
|
timeline = build_output_timeline(
|
|
[
|
|
FrameIdentity(output_frame=0, source_id=1, source_frame=0),
|
|
FrameIdentity(output_frame=1, source_id=1, source_frame=2),
|
|
FrameIdentity(output_frame=2, source_id=1, source_frame=3, drop_frame=True),
|
|
FrameIdentity(output_frame=3, source_id=1, source_frame=4),
|
|
],
|
|
probe=probe,
|
|
)
|
|
|
|
self.assertEqual([frame.source_frame for frame in timeline.frames], [0, 2, 4])
|
|
self.assertEqual([frame.duration for frame in timeline.frames], [1.0, 2.0, 1.0])
|
|
self.assertEqual(timeline.audio_segments, [(0.0, 1.0), (2.0, 3.0)])
|
|
self.assertEqual(timeline.consumed_source_frames, [0, 2, 3, 4])
|
|
self.assertEqual(normal_deleted_frame_count(timeline), 1)
|
|
|
|
|
|
class AudioFilterTests(unittest.TestCase):
|
|
def test_preserve_pitch_uses_atempo_chain(self) -> None:
|
|
self.assertEqual(atempo_filters(0.25), ["atempo=0.5", "atempo=0.50000000"])
|
|
self.assertEqual(
|
|
speed_change_filters(
|
|
2.0,
|
|
AudioEncodeOptions(mode="aac", preserve_pitch=True),
|
|
48000,
|
|
),
|
|
["atempo=2.00000000"],
|
|
)
|
|
|
|
def test_shift_pitch_changes_sample_rate_then_resamples(self) -> None:
|
|
self.assertEqual(
|
|
speed_change_filters(
|
|
0.5,
|
|
AudioEncodeOptions(mode="aac", preserve_pitch=False),
|
|
48000,
|
|
),
|
|
["asetrate=24000", "aresample=48000"],
|
|
)
|
|
|
|
|
|
class OutputNamingTests(unittest.TestCase):
|
|
def test_output_begin_uses_source_end_minus_source_duration_plus_first_frame(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0, 1.0, 2.0],
|
|
duration=10.0,
|
|
metadata_end=datetime(2026, 6, 2, 12, 0, 10, tzinfo=timezone.utc),
|
|
)
|
|
timeline = OutputTimeline(
|
|
frames=[
|
|
OutputFrameTiming(
|
|
output_index=0,
|
|
source_id=1,
|
|
source_frame=2,
|
|
start=2.0,
|
|
duration=1.0,
|
|
)
|
|
],
|
|
audio_segments=[(2.0, 1.0)],
|
|
consumed_source_frames=[2],
|
|
duration_seconds=1.0,
|
|
timing_kind="cfr",
|
|
beginning_trimmed=True,
|
|
ending_trimmed=True,
|
|
dropped_frame_count=0,
|
|
source_frame_count=3,
|
|
matrix=None,
|
|
)
|
|
|
|
self.assertEqual(
|
|
output_begin_timestamp(probe, timeline, timezone.utc),
|
|
datetime(2026, 6, 2, 12, 0, 2, tzinfo=timezone.utc),
|
|
)
|
|
|
|
def test_planned_output_path_uses_unique_timestamp_name(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0],
|
|
duration=1.0,
|
|
metadata_end=datetime(2026, 6, 2, 12, 0, 1, tzinfo=timezone.utc),
|
|
)
|
|
timeline = OutputTimeline(
|
|
frames=[
|
|
OutputFrameTiming(
|
|
output_index=0,
|
|
source_id=1,
|
|
source_frame=0,
|
|
start=0.0,
|
|
duration=1.0,
|
|
)
|
|
],
|
|
audio_segments=[(0.0, 1.0)],
|
|
consumed_source_frames=[0],
|
|
duration_seconds=1.0,
|
|
timing_kind="unknown",
|
|
beginning_trimmed=False,
|
|
ending_trimmed=False,
|
|
dropped_frame_count=0,
|
|
source_frame_count=1,
|
|
matrix=None,
|
|
)
|
|
|
|
path = planned_output_path(
|
|
script=Path("video_workspace/source.mp4.avs"),
|
|
timeline=timeline,
|
|
probe=probe,
|
|
naming=OutputNamingOptions(use_timestamps=True, timezone_value=timezone.utc),
|
|
extension=".mp4",
|
|
planned_outputs=set(),
|
|
)
|
|
|
|
self.assertEqual(path, Path("video_workspace/20260602_120000.mp4"))
|
|
|
|
def test_planned_output_path_adds_suffix_for_same_batch_collision(self) -> None:
|
|
probe = make_probe(
|
|
timestamps=[0.0],
|
|
duration=1.0,
|
|
metadata_end=datetime(2026, 6, 2, 12, 0, 1, tzinfo=timezone.utc),
|
|
)
|
|
timeline = OutputTimeline(
|
|
frames=[OutputFrameTiming(0, 1, 0, 0.0, 1.0)],
|
|
audio_segments=[(0.0, 1.0)],
|
|
consumed_source_frames=[0],
|
|
duration_seconds=1.0,
|
|
timing_kind="unknown",
|
|
beginning_trimmed=False,
|
|
ending_trimmed=False,
|
|
dropped_frame_count=0,
|
|
source_frame_count=1,
|
|
matrix=None,
|
|
)
|
|
planned_outputs: set[Path] = set()
|
|
first = planned_output_path(
|
|
script=Path("video_workspace/one.mp4.avs"),
|
|
timeline=timeline,
|
|
probe=probe,
|
|
naming=OutputNamingOptions(use_timestamps=True, timezone_value=timezone.utc),
|
|
extension=".mp4",
|
|
planned_outputs=planned_outputs,
|
|
)
|
|
second = planned_output_path(
|
|
script=Path("video_workspace/two.mp4.avs"),
|
|
timeline=timeline,
|
|
probe=probe,
|
|
naming=OutputNamingOptions(use_timestamps=True, timezone_value=timezone.utc),
|
|
extension=".mp4",
|
|
planned_outputs=planned_outputs,
|
|
)
|
|
|
|
self.assertEqual(first, Path("video_workspace/20260602_120000.mp4"))
|
|
self.assertEqual(second, Path("video_workspace/20260602_120000-1.mp4"))
|
|
|
|
def test_planned_output_path_avoids_existing_file_when_not_overwriting(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
directory = Path(temp_dir)
|
|
script = directory / "source.mp4.avs"
|
|
existing = directory / "20260602_120000.mp4"
|
|
existing.touch()
|
|
probe = make_probe(
|
|
timestamps=[0.0],
|
|
duration=1.0,
|
|
metadata_end=datetime(2026, 6, 2, 12, 0, 1, tzinfo=timezone.utc),
|
|
)
|
|
timeline = OutputTimeline(
|
|
frames=[OutputFrameTiming(0, 1, 0, 0.0, 1.0)],
|
|
audio_segments=[(0.0, 1.0)],
|
|
consumed_source_frames=[0],
|
|
duration_seconds=1.0,
|
|
timing_kind="unknown",
|
|
beginning_trimmed=False,
|
|
ending_trimmed=False,
|
|
dropped_frame_count=0,
|
|
source_frame_count=1,
|
|
matrix=None,
|
|
)
|
|
|
|
path = planned_output_path(
|
|
script=script,
|
|
timeline=timeline,
|
|
probe=probe,
|
|
naming=OutputNamingOptions(use_timestamps=True, timezone_value=timezone.utc),
|
|
extension=".mp4",
|
|
planned_outputs=set(),
|
|
overwrite_existing=False,
|
|
)
|
|
|
|
self.assertEqual(path, directory / "20260602_120000-1.mp4")
|
|
|
|
|
|
class AvisynthRunnerTests(unittest.TestCase):
|
|
def test_32bit_marker_allows_whitespace_and_hyphen(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
script = Path(temp_dir) / "edit.avs"
|
|
script.write_text("last\n\n # 32 - bit \n \n", encoding="utf-8")
|
|
|
|
self.assertTrue(script_requests_32bit(script))
|
|
|
|
def test_32bit_marker_only_matches_last_nonblank_line(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
script = Path(temp_dir) / "edit.avs"
|
|
script.write_text("#32bit\nlast\n", encoding="utf-8")
|
|
|
|
self.assertFalse(script_requests_32bit(script))
|
|
|
|
def test_bare_bundled_windows_runner_name_resolves(self) -> None:
|
|
runner = Path("tools/avisynth_runner/mbt_avs_runner.exe")
|
|
if not runner.is_file():
|
|
self.skipTest("bundled Windows runner is not present")
|
|
|
|
self.assertEqual(validate_runner_path("mbt_avs_runner.exe"), runner.resolve())
|
|
|
|
|
|
class ExecutableLookupTests(unittest.TestCase):
|
|
def test_windows_mkvmerge_default_is_used_before_prompt(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
executable = Path(temp_dir) / "mkvmerge.exe"
|
|
executable.touch()
|
|
output = StringIO()
|
|
with (
|
|
patch("tools.executables.which", return_value=None),
|
|
patch(
|
|
"tools.executables._windows_default_executable",
|
|
return_value=executable.resolve(),
|
|
),
|
|
patch("tools.executables.sys.stdin.isatty", return_value=True),
|
|
redirect_stdout(output),
|
|
):
|
|
self.assertEqual(require_executable("mkvmerge"), str(executable.resolve()))
|
|
self.assertIn("using", output.getvalue())
|
|
|
|
def test_missing_tool_accepts_an_absolute_path_for_this_run(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
executable = Path(temp_dir) / "mkvmerge.exe"
|
|
executable.touch()
|
|
output = StringIO()
|
|
with (
|
|
patch("tools.executables.which", return_value=None),
|
|
patch("tools.executables.sys.stdin.isatty", return_value=True),
|
|
patch("tools.executables.prompt_input", return_value=str(executable)),
|
|
redirect_stdout(output),
|
|
):
|
|
self.assertEqual(require_executable("mkvmerge"), str(executable.resolve()))
|
|
|
|
def test_missing_tool_rejects_relative_path_before_accepting_absolute_path(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
executable = Path(temp_dir) / "mkvmerge.exe"
|
|
executable.touch()
|
|
output = StringIO()
|
|
with (
|
|
patch("tools.executables.which", return_value=None),
|
|
patch("tools.executables.sys.stdin.isatty", return_value=True),
|
|
patch(
|
|
"tools.executables.prompt_input",
|
|
side_effect=["mkvmerge.exe", str(executable)],
|
|
),
|
|
redirect_stdout(output),
|
|
):
|
|
self.assertEqual(require_executable("mkvmerge"), str(executable.resolve()))
|
|
self.assertIn("Please enter an absolute path.", output.getvalue())
|
|
|
|
|
|
class AvisynthWorkspaceTests(unittest.TestCase):
|
|
def test_relative_timestamp_label_switches_to_minutes_and_hours(self) -> None:
|
|
self.assertEqual(_format_relative_timestamp(4.5), "4.500000")
|
|
self.assertEqual(
|
|
_format_relative_timestamp(291.029733),
|
|
"4:51.029733 (291.029733)",
|
|
)
|
|
self.assertEqual(
|
|
_format_relative_timestamp(3891.029733),
|
|
"1:04:51.029733 (3891.029733)",
|
|
)
|
|
|
|
def test_hidden_source_handles_missing_video_only_marker(self) -> None:
|
|
probe = make_probe(timestamps=[0.0, 0.04], duration=0.08)
|
|
probe = VideoProbe(
|
|
path=probe.path,
|
|
codec=probe.codec,
|
|
width=probe.width,
|
|
height=probe.height,
|
|
pixel_format=probe.pixel_format,
|
|
bit_depth=probe.bit_depth,
|
|
video_bit_rate=probe.video_bit_rate,
|
|
color_space=probe.color_space,
|
|
color_matrix=probe.color_matrix,
|
|
expected_color_matrix=probe.expected_color_matrix,
|
|
duration_seconds=probe.duration_seconds,
|
|
frame_timestamps=probe.frame_timestamps,
|
|
stream_frame_count=probe.stream_frame_count,
|
|
audio_streams=probe.audio_streams,
|
|
audio_stream_count=1,
|
|
audio_sample_rate=48000,
|
|
avg_frame_rate=probe.avg_frame_rate,
|
|
r_frame_rate=probe.r_frame_rate,
|
|
time_base=probe.time_base,
|
|
timing=probe.timing,
|
|
metadata_end=probe.metadata_end,
|
|
metadata_begin_tag=probe.metadata_begin_tag,
|
|
source_timezone=probe.source_timezone,
|
|
computed_begin=probe.computed_begin,
|
|
filename_begin=probe.filename_begin,
|
|
gps_latitude=probe.gps_latitude,
|
|
gps_longitude=probe.gps_longitude,
|
|
warnings=probe.warnings,
|
|
)
|
|
text = _hidden_source_script(
|
|
VideoInput(Path("source.mp4"), Path("source.mp4"), Path("source.mp4")),
|
|
probe,
|
|
0,
|
|
Path("workspace/.source/source.mp4.source.avs"),
|
|
Path("tools/avisynth_helpers.avs"),
|
|
None,
|
|
Path("workspace/.source/source.mp4.timestamps.txt"),
|
|
Path("workspace/.source/source.mp4.frame-times.txt"),
|
|
Path("workspace/.source/source.mp4.frame-time-labels.txt"),
|
|
Path("workspace/.source/source.mp4.durations.txt"),
|
|
Path("workspace/.source/source.mp4.duration-labels.txt"),
|
|
)
|
|
|
|
self.assertIn("try {\n mbt_video_only = mbt_video_only", text)
|
|
self.assertIn("} catch (err_msg) {\n mbt_video_only = false", text)
|
|
self.assertIn("mbt_validation_blank = mbt_validation_blank", text)
|
|
self.assertIn("BlankClip(video, color=$000000)", text)
|
|
self.assertNotIn('Defined("mbt_video_only")', text)
|
|
|
|
def test_validation_wrapper_keeps_audio_for_clip_summary(self) -> None:
|
|
from tools.avisynth_workspace import _validation_script
|
|
|
|
text = _validation_script(Path("edit.avs"), Path("source.validate.avs"))
|
|
|
|
self.assertNotIn("mbt_video_only = true", text)
|
|
self.assertNotIn("mbt_validation_blank", text)
|
|
|
|
|
|
class AvisynthRenderTests(unittest.TestCase):
|
|
def test_complete_csv_can_recover_from_renderer_cleanup_crash(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp = Path(temp_dir)
|
|
renderer = temp / "renderer.py"
|
|
csv_path = temp / "frames.csv"
|
|
script = temp / "validate.avs"
|
|
script.write_text("last\n", encoding="utf-8")
|
|
renderer.write_text(
|
|
"\n".join(
|
|
[
|
|
"from pathlib import Path",
|
|
"import sys",
|
|
"csv = Path(sys.argv[1])",
|
|
"csv.write_text('output_frame,source_id,source_frame,drop_frame\\n0,1,0,0\\n', encoding='utf-8')",
|
|
"print('clip_info,width=16,height=16,frames=1,fps_num=1,fps_den=1,has_audio=0,audio_rate=0,audio_samples=0')",
|
|
"print('rendered=1')",
|
|
"raise SystemExit(7)",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
run = run_validation_script(
|
|
renderer_command=["python3", str(renderer), str(csv_path)],
|
|
script=script,
|
|
csv_path=csv_path,
|
|
expected_source_ids={1},
|
|
)
|
|
|
|
self.assertTrue(run.recovered_from_renderer_error)
|
|
self.assertEqual(run.renderer_returncode, 7)
|
|
self.assertTrue(run.result.accepted)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|