from __future__ import annotations import os import sys from dataclasses import dataclass, replace from datetime import datetime, timezone from pathlib import Path from tools.avisynth_render import AvisynthClipInfo from tools.avisynth_runner import AvisynthRunnerSet from tools.avisynth_workspace import ( editable_audio_script_path, rendered_audio_wav_path, visible_script_path, ) from tools.console import clear_screen, light_red, prompt_input, prompt_yes_no, red_strikethrough from tools.filenames import format_rounded_filename_stem from tools.timezones import local_timezone, parse_timezone_offset from tools.timezones import timezone_to_string from tools.video_encode_output import ( ENCODER_PRESETS, AudioEncodeOptions, VideoCodecOptions, choose_video_pixel_format, default_audio_bitrate, encode_video_only_y4m, ffmpeg_colorspace_from_matrix, pixel_format_bit_depth, print_encoder_statistics, render_audio_wav, ) from tools.video_timestamp_encode import encode_video_with_timestamps from tools.video_encode_plan import ( audio_options_for_probe, audio_tempo_factor, avisynth_duration_differs, encoded_duration_seconds, has_avisynth_speed_changes, normal_deleted_frame_count, should_preserve_video_timestamps, should_use_audio_segments, timeline_frame_starts, y4m_command_for_timeline, ) from tools.video_codec_constraints import ( VideoConstraintSpec, available_levels, available_profiles, level_supports, profile_supports, resolve_level, resolve_profile, ) from tools.video_inputs import VideoInput from tools.video_formatting import chroma_family, matrix_name from tools.video_options import ( OutputNamingOptions, ask_audio_options, ask_output_naming_options, ask_video_codec_options, ) from tools.video_outputs import ( EncodedItem, copy_video_metadata, finalize_encoded_outputs, output_begin_timestamp, planned_output_path, validate_encoded_video, write_video_end_timestamp, ) from tools.video_probe import VideoProbe, require_tool from tools.video_reporting import format_validation_summary_lines from tools.video_timeline import OutputTimeline @dataclass(frozen=True) class EncodingOutputSpec: label: str width: int height: int bit_depth: int chroma: str fps: float begin_utc: str summary_lines: tuple[str, ...] @dataclass(frozen=True) class EncodingSettings: naming: OutputNamingOptions codec_options: VideoCodecOptions audio_options: AudioEncodeOptions def _plan_batch_outputs( *, inputs: list[VideoInput], root: Path, timelines: dict[Path, OutputTimeline], probes_by_path: dict[Path, VideoProbe], naming: OutputNamingOptions, extension: str, ) -> dict[Path, Path]: def plan(*, overwrite_existing: bool) -> dict[Path, Path]: reserved: set[Path] = set() return { item.path.resolve(): planned_output_path( script=visible_script_path(root, item), timeline=timelines[item.path.resolve()], probe=probes_by_path[item.path.resolve()], naming=naming, extension=extension, planned_outputs=reserved, overwrite_existing=overwrite_existing, ) for item in inputs } output_paths = plan(overwrite_existing=True) existing = sorted({path for path in output_paths.values() if path.exists()}) if not existing: return output_paths print("\nExisting output video file(s):") for path in existing: print(f" - {path}") overwrite = ( prompt_yes_no("Overwrite existing output video file(s)?", default=True) if sys.stdin.isatty() else True ) return output_paths if overwrite else plan(overwrite_existing=False) def encode_validated_video_only( root: Path, inputs: list[VideoInput], timelines: dict[Path, OutputTimeline], clip_infos: dict[Path, AvisynthClipInfo], probes_by_path: dict[Path, VideoProbe], exiftool: str, avs_runners: AvisynthRunnerSet, ffprobe: str, ) -> None: inputs = [item for item in inputs if item.path.resolve() in timelines] if not inputs: print("\nNo validated outputs to encode.") return ffmpeg = Path(require_tool("ffmpeg")).resolve() default_timezone = _default_output_timezone( [probes_by_path[item.path.resolve()] for item in inputs] ) if sys.stdin.isatty() and not _encoding_env_is_set(): settings = ask_interactive_encoding_settings( inputs=inputs, timelines=timelines, clip_infos=clip_infos, probes_by_path=probes_by_path, ) if settings is None: print("\nEncoding skipped.") return naming = settings.naming codec_options = settings.codec_options audio_options = settings.audio_options else: naming = ask_output_naming_options(default_timezone=default_timezone) codec_options = ask_video_codec_options() audio_options = ask_audio_options( has_speed_changes=has_avisynth_speed_changes(timelines, clip_infos), container_extension=codec_options.extension, ) specs = _encoding_specs(inputs, timelines, clip_infos, probes_by_path) if codec_options.codec == "libx264" and any(spec.bit_depth > 10 for spec in specs): raise RuntimeError("x264 cannot encode this batch because at least one output is above 10-bit") needs_timecodes = any( should_preserve_video_timestamps(timelines[item.path.resolve()]) for item in inputs ) mkvmerge = Path(require_tool("mkvmerge")).resolve() if needs_timecodes else None x264 = ( Path(require_tool("x264")).resolve() if needs_timecodes and codec_options.codec == "libx264" else None ) print(f"\nAvisynth runner: {avs_runners.default}") print(f"FFmpeg: {ffmpeg}") if mkvmerge is not None: print(f"mkvmerge: {mkvmerge}") if x264 is not None: print(f"x264: {x264}") print( "Output naming: " + ( f"timestamps in {timezone_to_string(naming.timezone_value)}" if naming.use_timestamps else "original names" ) ) print( "Codec: " f"{codec_options.codec}, CRF {codec_options.crf}, " f"preset {codec_options.preset}, " f"profile {codec_options.profile or 'none'}, " f"level {codec_options.level or 'none'}, " f"threads {codec_options.threads if codec_options.threads is not None else 'auto'}, " f"{codec_options.extension}" ) print(f"Audio: {_format_audio_options(audio_options)}") output_paths = _plan_batch_outputs( inputs=inputs, root=root, timelines=timelines, probes_by_path=probes_by_path, naming=naming, extension=codec_options.extension, ) encoded_items: list[EncodedItem] = [] for item_index, item in enumerate(inputs, start=1): timeline = timelines[item.path.resolve()] script = visible_script_path(root, item) avs_runner = avs_runners.for_script(script) print(f"\n{item.collapsed_relative}") if avs_runner != avs_runners.default: print(f" runner: {avs_runner}") clip_info = clip_infos.get(item.path.resolve()) output_bit_depth = clip_info.bit_depth if clip_info is not None else 8 pixel_format = choose_video_pixel_format( ffmpeg, codec_options.codec, output_bit_depth, clip_info.chroma if clip_info is not None else None, ) spec = _constraint_spec_for_item(item, timelines, clip_infos, probes_by_path) encoded_spec = VideoConstraintSpec( width=spec.width, height=spec.height, bit_depth=pixel_format_bit_depth(pixel_format), chroma=spec.chroma, fps=spec.fps, ) effective_codec_options = _resolve_codec_options(codec_options, encoded_spec) if effective_codec_options.profile or effective_codec_options.level: print( " constraints: " f"profile {effective_codec_options.profile or 'none'}, " f"level {effective_codec_options.level or 'none'}" ) colorspace = ffmpeg_colorspace_from_matrix(timeline.matrix) if output_bit_depth >= 10: encoded_bit_depth = pixel_format_bit_depth(pixel_format) print( f" video: {output_bit_depth}-bit Avisynth output -> " f"{encoded_bit_depth}-bit {pixel_format}" ) if colorspace is not None: print(f" color space: {matrix_name(timeline.matrix)} -> ffmpeg {colorspace}") normal_deleted_frames = normal_deleted_frame_count(timeline) preserve_video_timestamps = should_preserve_video_timestamps(timeline) if not preserve_video_timestamps and timeline.timing_kind != "cfr": print(" skipped encode: output timeline is not CFR") continue if not preserve_video_timestamps and timeline.dropped_frame_count and ( clip_info is not None and avisynth_duration_differs(clip_info, timeline) ): print( " skipped encode: drop_frame plus Avisynth speed change needs " "timestamp-aware muxing" ) continue probe = probes_by_path[item.path.resolve()] output_duration = ( timeline.duration_seconds if preserve_video_timestamps else encoded_duration_seconds(clip_info, timeline) ) audio_tempo = audio_tempo_factor(timeline, output_duration) output = output_paths[item.path.resolve()] effective_audio_options = audio_options_for_probe( audio_options, probe, timeline, output_duration, normal_deleted_frames=normal_deleted_frames, ) audio_segments = ( timeline.audio_segments if should_use_audio_segments(timeline, effective_audio_options) else None ) audio_source, source_has_audio = prepare_audio_source( avs_runners=avs_runners, root=root, item=item, original_has_audio=probe.audio_stream_count > 0, audio_options=effective_audio_options, ) encode_kwargs = dict( y4m_command=y4m_command_for_timeline( avs_runner, timeline, force_timeline_rate=preserve_video_timestamps, ), ffmpeg=ffmpeg, script=script, audio_source=audio_source, output=output, options=effective_codec_options, audio_options=effective_audio_options, audio_start=timeline.frames[0].start, audio_duration=timeline.duration_seconds, audio_segments=audio_segments, audio_tempo=audio_tempo, audio_sample_rate=probe.audio_sample_rate, source_has_audio=source_has_audio, allow_audio_copy=effective_audio_options.mode == "copy", pixel_format=pixel_format, colorspace=colorspace, frame_count=len(timeline.frames), progress_label=( f"script {item_index}/{len(inputs)}: {item.collapsed_relative}" ), ) if preserve_video_timestamps: assert mkvmerge is not None result = encode_video_with_timestamps( mkvmerge=mkvmerge, x264=x264, frame_durations=[frame.duration for frame in timeline.frames], **encode_kwargs, ) else: result = encode_video_only_y4m(**encode_kwargs) validate_encoded_video( ffprobe, result.output, timeline, expected_duration=output_duration, expected_timestamps=( timeline_frame_starts(timeline) if preserve_video_timestamps else None ), expected_audio=( effective_audio_options.mode != "none" and source_has_audio ), ) print_encoder_statistics(result.encoder_log) copy_video_metadata(exiftool, item.path, result.output) write_video_end_timestamp(exiftool, result.output, probe, timeline, output_duration) encoded_items.append(EncodedItem(input_item=item, output=result.output)) print(f" encoded output: {result.output}") if not encoded_items: print("\nNo outputs encoded.") return finalize_encoded_outputs(root, encoded_items) def ask_interactive_encoding_settings( *, inputs: list[VideoInput], timelines: dict[Path, OutputTimeline], clip_infos: dict[Path, AvisynthClipInfo], probes_by_path: dict[Path, VideoProbe], ) -> EncodingSettings | None: answers: dict[str, object] = { "_timezone_default": _default_output_timezone( [probes_by_path[item.path.resolve()] for item in inputs] ) } index = 0 has_speed_changes = has_avisynth_speed_changes(timelines, clip_infos) specs = _encoding_specs(inputs, timelines, clip_infos, probes_by_path) def answer(key: str, value: object) -> None: old_value = answers.get(key) answers[key] = value if key == "codec" and old_value is not None and old_value != value: _forget(answers, {"crf", "profile", "level"}) if key == "container" and old_value is not None and old_value != value: _forget(answers, {"audio", "audio_bitrate"}) if key == "timestamp_names" and value is False: _forget(answers, {"timezone"}) if key == "audio" and value not in {"aac", "opus"}: _forget(answers, {"audio_bitrate", "audio_pitch"}) while True: steps = _encoding_steps(answers, has_speed_changes) index = max(0, min(index, len(steps) - 1)) key = steps[index] _render_encoding_screen(answers, key, specs, has_speed_changes) raw = prompt_input(_question_for_key(key, answers, specs)).strip() if raw.lower() in {"b", "back"}: index = max(0, index - 1) continue try: value = _parse_answer(key, raw, answers, specs) except ValueError as exc: print(exc) prompt_input("Press Enter to retry...") continue answer(key, value) if key == "confirm" and value is False: return None if index == len(steps) - 1: break index += 1 codec = answers["codec"] assert isinstance(codec, str) audio_mode = answers.get("audio", "aac" if answers["container"] == ".mp4" else "opus") assert isinstance(audio_mode, str) return EncodingSettings( naming=OutputNamingOptions( use_timestamps=bool(answers["timestamp_names"]), timezone_value=answers.get( "timezone", answers["_timezone_default"], ), ), codec_options=VideoCodecOptions( codec=codec, crf=int(answers["crf"]), preset=str(answers["preset"]), extension=str(answers["container"]), profile=_none_if_none(str(answers["profile"])), level=_none_if_none(str(answers["level"])), threads=_threads_from_answer(answers["threads"]), ), audio_options=AudioEncodeOptions( mode=audio_mode, bitrate=str(answers.get("audio_bitrate") or default_audio_bitrate(audio_mode)), preserve_pitch=bool(answers.get("audio_pitch", True)), ), ) def _encoding_env_is_set() -> bool: names = { "MBT_VIDEO_TIMESTAMP_NAMES", "MBT_VIDEO_TIMEZONE", "MBT_VIDEO_CONTAINER", "MBT_VIDEO_CODEC", "MBT_VIDEO_CRF", "MBT_VIDEO_PRESET", "MBT_VIDEO_PROFILE", "MBT_VIDEO_LEVEL", "MBT_VIDEO_THREADS", "MBT_VIDEO_AUDIO", "MBT_VIDEO_AUDIO_BITRATE", "MBT_VIDEO_AUDIO_PITCH", } return any(os.environ.get(name) for name in names) def _encoding_steps(answers: dict[str, object], has_speed_changes: bool) -> list[str]: steps = ["timestamp_names"] if answers.get("timestamp_names", True): steps.append("timezone") steps.extend(["container", "codec", "crf", "preset", "profile", "level", "threads", "audio"]) if answers.get("audio") in {"aac", "opus"}: steps.append("audio_bitrate") if has_speed_changes and answers.get("audio") in {"aac", "opus"}: steps.append("audio_pitch") steps.append("confirm") return steps def _render_encoding_screen( answers: dict[str, object], current_key: str, specs: list[EncodingOutputSpec], has_speed_changes: bool, ) -> None: clear_screen() print("Validated clips:") for spec in specs: print(f"\n{spec.label}") for line in spec.summary_lines: print(f" {line}") print("") print("Encoding settings:") rows = [ ( key, _setting_label(key), _setting_value(key, _default_answer(key, answers, specs), answers, specs), ) for key in _encoding_table_keys(answers, has_speed_changes) ] width = max(len(label) for _, label, _ in rows) for key, label, value in rows: marker = ">" if key == current_key else " " print(f"{marker} {label.ljust(width)} {value}") print("\nType b to go back.\n") if current_key == "codec" and _x264_unavailable(specs): print(light_red("x264 is unavailable because this batch contains output above 10-bit.\n")) def _encoding_table_keys(answers: dict[str, object], has_speed_changes: bool) -> list[str]: keys = [ "timestamp_names", "container", "codec", "crf", "preset", "profile", "level", "threads", "audio", ] if _default_answer("timestamp_names", answers, []): keys.insert(1, "timezone") if _default_answer("audio", answers, []) in {"aac", "opus"}: keys.append("audio_bitrate") if has_speed_changes and _default_answer("audio", answers, []) in {"aac", "opus"}: keys.append("audio_pitch") return keys def _question_for_key( key: str, answers: dict[str, object], specs: list[EncodingOutputSpec], ) -> str: if key == "timestamp_names": default = bool(_default_answer(key, answers, specs)) preview = _filename_preview(answers, specs, limit=1) return ( "Name encoded files from edited beginning timestamp? " f"({preview}) ({_yes_no_suffix(default)}): " ) if key == "timezone": default = _default_answer(key, answers, specs) return f"Output filename timezone [{timezone_to_string(default)}]: " if key == "container": return ( f"Output container [{_container_label(_default_answer(key, answers, specs))}] " "(mp4/mkv): " ) if key == "codec": options = ["x264", "x265"] if _x264_unavailable(specs): options[0] = red_strikethrough(options[0]) return ( f"Video codec [{_codec_label(_default_answer(key, answers, specs))}] " f"({'/'.join(options)}): " ) if key == "crf": return f"CRF [{_default_answer(key, answers, specs)}]: " if key == "preset": default = _default_answer(key, answers, specs) return f"Encoder preset [{default}] ({'/'.join(ENCODER_PRESETS)}): " if key == "profile": codec = str(answers["codec"]) options = _profile_options(codec, specs) unsupported = _not_supported_by_all(codec, options, specs, profile=True) return "Profile " + _options_prompt( options, str(_default_answer(key, answers, specs)), unsupported, ) + ": " if key == "level": codec = str(answers["codec"]) options = _level_options(codec, specs) unsupported = _not_supported_by_all(codec, options, specs, profile=False) return "Level " + _options_prompt( options, str(_default_answer(key, answers, specs)), unsupported, ) + ": " if key == "threads": return f"Encoder threads [{_default_answer(key, answers, specs)}] (auto or positive integer): " if key == "audio": if answers.get("container") == ".mkv": return f"Audio [{_default_answer(key, answers, specs)}] (opus/aac/flac/copy/none): " return f"Audio [{_default_answer(key, answers, specs)}] (aac/copy/none): " if key == "audio_bitrate": return f"{str(answers['audio']).upper()} audio bitrate [{_default_answer(key, answers, specs)}]: " if key == "audio_pitch": return "Speed-changed audio pitch [preserve] (preserve/shift): " if key == "confirm": return "Start encoding with these settings? (Y/n): " raise ValueError(f"Unknown setting: {key}") def _parse_answer( key: str, raw: str, answers: dict[str, object], specs: list[EncodingOutputSpec], ) -> object: default = _default_answer(key, answers, specs) value = raw or str(default) lowered = value.strip().lower() if key == "confirm": return lowered not in {"n", "no", "0", "false", "off"} if key == "timestamp_names": return lowered not in {"n", "no", "0", "false", "off"} if key == "timezone": return parse_timezone_offset(value) if raw else default if key == "container": if lowered in {"mp4", ".mp4"}: return ".mp4" if lowered in {"m", "mkv", ".mkv"}: return ".mkv" raise ValueError("Container must be mp4 or mkv.") if key == "codec": if lowered in {"x264", "h264", "libx264"}: if _x264_unavailable(specs): raise ValueError( "x264 is unavailable because at least one validated output is above 10-bit." ) return "libx264" if lowered in {"x265", "h265", "hevc", "libx265"}: return "libx265" raise ValueError("Codec must be x264 or x265.") if key == "crf": try: return int(value) except ValueError as exc: raise ValueError("CRF must be an integer.") from exc if key == "preset": if lowered not in ENCODER_PRESETS: raise ValueError(f"Preset must be one of: {'/'.join(ENCODER_PRESETS)}.") return lowered if key == "profile": options = _profile_options(str(answers["codec"]), specs) return _parse_option("profile", value, options) if key == "level": options = _level_options(str(answers["codec"]), specs) return _parse_option("level", value, options) if key == "threads": if lowered == "auto": return "auto" try: threads = int(value) except ValueError as exc: raise ValueError("Threads must be auto or a positive integer.") from exc if threads < 1: raise ValueError("Threads must be auto or a positive integer.") return threads if key == "audio": return _parse_audio(value, str(answers["container"])) if key == "audio_bitrate": return value if key == "audio_pitch": if lowered in {"preserve", "p", "keep", "same"}: return True if lowered in {"shift", "s", "change", "speed"}: return False raise ValueError("Audio pitch must be preserve or shift.") raise ValueError(f"Unknown setting: {key}") def _default_answer( key: str, answers: dict[str, object], specs: list[EncodingOutputSpec], ) -> object: if key in answers: return answers[key] codec = str(answers.get("codec", "libx265")) if key == "confirm": return True if key == "timestamp_names": return True if key == "timezone": return answers.get("_timezone_default", local_timezone()) if key == "container": return ".mp4" if key == "codec": return "libx265" if key == "crf": return 16 if codec == "libx264" else 21 if key == "preset": return "slow" if key == "profile": return "minimum" if key == "level": return "minimum" if key == "threads": return "auto" if key == "audio": return "aac" if answers.get("container", ".mp4") == ".mp4" else "opus" if key == "audio_bitrate": return default_audio_bitrate(str(answers.get("audio", "aac"))) if key == "audio_pitch": return "preserve" raise ValueError(f"Unknown setting: {key}") def _encoding_specs( inputs: list[VideoInput], timelines: dict[Path, OutputTimeline], clip_infos: dict[Path, AvisynthClipInfo], probes_by_path: dict[Path, VideoProbe], ) -> list[EncodingOutputSpec]: specs: list[EncodingOutputSpec] = [] for item in inputs: timeline = timelines[item.path.resolve()] clip_info = clip_infos.get(item.path.resolve()) probe = probes_by_path[item.path.resolve()] width = clip_info.width if clip_info else (probe.width or 0) height = clip_info.height if clip_info else (probe.height or 0) bit_depth = clip_info.bit_depth if clip_info else (probe.bit_depth or 8) chroma = chroma_family(clip_info.chroma if clip_info else probe.pixel_format) fps = len(timeline.frames) / timeline.duration_seconds if timeline.duration_seconds > 0 else 0 specs.append( EncodingOutputSpec( label=str(item.collapsed_relative), width=width, height=height, bit_depth=bit_depth, chroma=chroma, fps=fps, begin_utc=_begin_utc(probe, timeline), summary_lines=tuple( format_validation_summary_lines( output_frame_count=( clip_info.frames if clip_info is not None else len(timeline.frames) + timeline.dropped_frame_count ), dropped_frame_count=timeline.dropped_frame_count, timeline=timeline, clip_info=clip_info, probe=probe, ) ), ) ) return specs def _profile_options(codec: str, specs: list[EncodingOutputSpec]) -> list[str]: constraints = [_constraint_spec(spec) for spec in specs] return ["none", "minimum", *available_profiles(codec, constraints)] def _level_options(codec: str, specs: list[EncodingOutputSpec]) -> list[str]: constraints = [_constraint_spec(spec) for spec in specs] return ["none", "minimum", *available_levels(codec, constraints)] def _not_supported_by_all( codec: str, options: list[str], specs: list[EncodingOutputSpec], *, profile: bool, ) -> set[str]: support = profile_supports if profile else level_supports constraints = [_constraint_spec(spec) for spec in specs] return { option for option in options if option not in {"none", "minimum"} and not all(support(codec, option, spec) for spec in constraints) } def _constraint_spec(spec: EncodingOutputSpec) -> VideoConstraintSpec: return VideoConstraintSpec( width=spec.width, height=spec.height, bit_depth=spec.bit_depth, chroma=spec.chroma, fps=spec.fps, ) def _constraint_spec_for_item( item: VideoInput, timelines: dict[Path, OutputTimeline], clip_infos: dict[Path, AvisynthClipInfo], probes_by_path: dict[Path, VideoProbe], ) -> VideoConstraintSpec: return _constraint_spec( _encoding_specs([item], timelines, clip_infos, probes_by_path)[0] ) def _resolve_codec_options( options: VideoCodecOptions, spec: VideoConstraintSpec, ) -> VideoCodecOptions: return replace( options, profile=resolve_profile(options.codec, options.profile, spec), level=resolve_level(options.codec, options.level, spec), ) def _x264_unavailable(specs: list[EncodingOutputSpec]) -> bool: return any(spec.bit_depth > 10 for spec in specs) def _begin_utc(probe: VideoProbe, timeline: OutputTimeline) -> str: try: return output_begin_timestamp(probe, timeline, timezone.utc).isoformat() except RuntimeError: return "unknown" def _parse_option(name: str, value: str, options: list[str]) -> str: lowered = value.strip().lower() matches = [option for option in options if option.lower() == lowered] if matches: return matches[0] raise ValueError(f"{name.title()} must be one of: {', '.join(options)}") def _parse_audio(value: str, container: str) -> str: lowered = value.strip().lower() if lowered in {"aac", "a"}: return "aac" if lowered in {"copy", "c"}: return "copy" if lowered in {"none", "no", "n"}: return "none" if container == ".mkv" and lowered in {"opus", "o"}: return "opus" if container == ".mkv" and lowered in {"flac", "f"}: return "flac" raise ValueError("Unsupported audio choice for this container.") def _yes_no_suffix(default: bool) -> str: return "Y/n" if default else "y/N" def _forget(answers: dict[str, object], keys: set[str]) -> None: for key in keys: answers.pop(key, None) def _setting_label(key: str) -> str: return { "timestamp_names": "timestamp names", "timezone": "timezone", "container": "container", "codec": "codec", "crf": "CRF", "preset": "preset", "profile": "profile", "level": "level", "threads": "threads", "audio": "audio", "audio_bitrate": "audio bitrate", "audio_pitch": "audio pitch", }.get(key, key) def _setting_value( key: str, value: object, answers: dict[str, object], specs: list[EncodingOutputSpec], ) -> str: if key == "timestamp_names": if not value: return "no (original filename)" return f"yes ({_filename_preview(answers, specs, limit=3)})" if key == "timezone": return timezone_to_string(value) if key == "container": return _container_label(value) if key == "codec": return _codec_label(value) if key in {"profile", "level"} and value not in {"none", "minimum"}: codec = str(answers.get("codec", "libx265")) unsupported = _not_supported_by_all( codec, [str(value)], specs, profile=key == "profile", ) return light_red(value) if unsupported else str(value) if key == "audio_pitch": return "preserve" if value else "shift" return str(value) def _filename_preview( answers: dict[str, object], specs: list[EncodingOutputSpec], *, limit: int, ) -> str: if not specs: return "unknown" timezone_value = _default_answer("timezone", answers, specs) assert hasattr(timezone_value, "utcoffset") stems = [ _filename_preview_for_timezone(spec, timezone_value) for spec in specs[:limit] ] if len(specs) > limit: stems.append("...") return ", ".join(stems) def _filename_preview_for_timezone( spec: EncodingOutputSpec, timezone_value: timezone, ) -> str: begin = _parse_iso_datetime(spec.begin_utc) if begin is None: return "unknown" return format_rounded_filename_stem(begin.astimezone(timezone_value)) def _parse_iso_datetime(value: str) -> datetime | None: if value == "unknown": return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None def _container_label(value: object) -> str: return "MP4" if value == ".mp4" else "MKV" def _default_output_timezone(probes: list[VideoProbe]) -> timezone: offsets = { probe.source_timezone.utcoffset(None) for probe in probes if probe.source_timezone is not None } if len(offsets) == 1: offset = offsets.pop() if offset is not None: return timezone(offset) return local_timezone() def _codec_label(value: object) -> str: return "x264" if value == "libx264" else "x265" def _options_prompt( options: list[str], default: str, unsupported: set[str] | None = None, ) -> str: unsupported = unsupported or set() displayed = [light_red(option) if option in unsupported else option for option in options] return f"[{default}] ({'/'.join(displayed)})" def _none_if_none(value: str) -> str | None: return None if value == "none" else value def _threads_from_answer(value: object) -> int | None: return None if str(value).strip().lower() == "auto" else int(value) def prepare_audio_source( *, avs_runners: AvisynthRunnerSet, root: Path, item: VideoInput, original_has_audio: bool, audio_options: AudioEncodeOptions, ) -> tuple[Path, bool]: if audio_options.mode == "none": return item.path, False audio_script = editable_audio_script_path(root, item) if not audio_script.exists(): return item.path, original_has_audio avs_runner = avs_runners.for_script(audio_script) wav_path = rendered_audio_wav_path(root, item) print(f" audio: rendering {audio_script.name}") render_audio_wav(avs_runner=avs_runner, script=audio_script, output=wav_path) return wav_path, True def _format_audio_options(options: AudioEncodeOptions) -> str: if options.mode == "none": return "none" if options.mode == "aac": pitch = "preserve pitch" if options.preserve_pitch else "shift pitch with speed" return f"AAC {options.bitrate}, {pitch}" if options.mode == "opus": pitch = "preserve pitch" if options.preserve_pitch else "shift pitch with speed" return f"Opus {options.bitrate}, {pitch}" if options.mode == "flac": return "FLAC" if options.mode == "copy": return "copy with best-effort beginning/end trim" return options.mode