from __future__ import annotations import subprocess import sys from pathlib import Path from tools.console import clear_screen from tools.avisynth_runner import AvisynthRunnerSet from tools.avisynth_render import AvisynthClipInfo, run_validation_script from tools.avisynth_workspace import ( helper_script_path, hidden_source_script_path, validation_csv_path, validation_script_path, visible_script_path, ) from tools.video_inputs import VideoInput from tools.video_probe import VideoProbe from tools.video_reporting import print_validation_summary from tools.video_timeline import OutputTimeline, build_output_timeline def run_validations( avs_runners: AvisynthRunnerSet, root: Path, inputs: list[VideoInput], probes_by_path: dict[Path, VideoProbe], *, blank_source: bool, ffms2_plugin_path: Path | None, ) -> tuple[dict[Path, OutputTimeline], dict[Path, AvisynthClipInfo]]: print(f"\nAvisynth runner: {avs_runners.default}") print( "Validation source: " + ("blank frames (fast)" if blank_source else "real frames (slow)") ) rejected: list[Path] = [] timelines: dict[Path, OutputTimeline] = {} clip_infos: dict[Path, AvisynthClipInfo] = {} summaries: list[tuple[VideoInput, bool, int, int, OutputTimeline | None, AvisynthClipInfo | None, VideoProbe, list[str]]] = [] for source_id, item in enumerate(inputs): probe = probes_by_path[item.path.resolve()] visible_script = visible_script_path(root, item) avs_runner = avs_runners.for_script(visible_script) progress_label = ( f"Validating script {source_id + 1}/{len(inputs)}: " f"{item.collapsed_relative}" ) if not sys.stdout.isatty(): print(f"\n{progress_label}") if avs_runner != avs_runners.default: print(f" runner: {avs_runner}") csv_path = validation_csv_path(root, item) try: run = run_validation_script( renderer_command=[ str(avs_runner), "--validate" if blank_source else "--validate-real", ], script=validation_script_path(root, item), csv_path=csv_path, expected_source_ids={source_id}, progress_label=progress_label, ) except RuntimeError as exc: diagnostics = _diagnostics_for_error( avs_runner, root, item, ffms2_plugin_path, exc, ) if diagnostics: raise RuntimeError(f"{exc}\n\n{diagnostics}") from exc raise _remove_validation_artifacts(csv_path) result = run.result if run.recovered_from_renderer_error: print( " warning: Avisynth runner exited with code " f"{run.renderer_returncode} after writing a complete frame table; " "continuing validation" ) issues = list(result.issues) unsupported_depth = run.clip_info is not None and run.clip_info.bit_depth > 12 if unsupported_depth: issues.append( f"Avisynth output is {run.clip_info.bit_depth}-bit; maximum supported output is 12-bit" ) if issues: summaries.append((item, False, result.frame_count, result.dropped_frame_count, None, run.clip_info, probe, issues)) if result.issues: rejected.append(item.collapsed_relative) continue timeline = build_output_timeline(run.frames, probe=probe) timelines[item.path.resolve()] = timeline if run.clip_info is not None: clip_infos[item.path.resolve()] = run.clip_info summaries.append((item, result.accepted, result.frame_count, result.dropped_frame_count, timeline, run.clip_info, probe, [])) clear_screen() for item, accepted, frame_count, dropped_count, timeline, clip_info, probe, issues in summaries: print(f"\n{item.collapsed_relative}") if issues: print(" not valid") for issue in issues: print(f" issue: {issue}") continue assert timeline is not None print_validation_summary( accepted=accepted, output_frame_count=frame_count, dropped_frame_count=dropped_count, timeline=timeline, clip_info=clip_info, probe=probe, ) if rejected: raise RuntimeError( "Rejected Avisynth output identity for: " + ", ".join(str(path) for path in rejected) ) return timelines, clip_infos def _diagnostics_for_error( runner: Path, root: Path, item: VideoInput, ffms2_plugin_path: Path | None, error: RuntimeError, ) -> str: if "exit code 3221225477" not in str(error): return "" try: return _run_import_diagnostics(runner, root, item, ffms2_plugin_path) except Exception as exc: return f"Avisynth import diagnostics failed: {exc}" def _run_import_diagnostics( runner: Path, root: Path, item: VideoInput, ffms2_plugin_path: Path | None, ) -> str: source_dir = validation_script_path(root, item).parent stem = item.collapsed_relative.name helper = helper_script_path() hidden_source = hidden_source_script_path(root, item) visible = visible_script_path(root, item) plugin_path = str(ffms2_plugin_path) if ffms2_plugin_path else None source_path = str(item.path) cases: list[tuple[str, str]] = [ ( "blank", 'BlankClip(length=1, width=16, height=16, pixel_type="YV12")\n', ), ( "import-helper", f'Import("{_avs_path(str(helper.resolve()))}")\n' 'BlankClip(length=1, width=16, height=16, pixel_type="YV12")\n', ), ] if plugin_path: cases.append( ( "load-ffms2", f'LoadPlugin("{_avs_path(str(Path(plugin_path).resolve()))}")\n' 'BlankClip(length=1, width=16, height=16, pixel_type="YV12")\n', ) ) cases.extend( [ ( "ffvideo", _optional_load_plugin(plugin_path) + f'FFVideoSource("{_avs_path(str(Path(source_path).resolve()))}", ' + f'cachefile="{_avs_path(str((source_dir / (stem + ".diag.ffindex")).resolve()))}")\n', ), ( "hidden-source", "mbt_video_only = true\n" f'Import("{_avs_path(str(hidden_source.resolve()))}")\n' "last\n", ), ( "editable", "mbt_video_only = true\n" f'Import("{_avs_path(str(visible.resolve()))}")\n' "last\n", ), ] ) lines = [f"Avisynth import diagnostics with {runner}:"] for name, script_text in cases: script = source_dir / f"{stem}.diag.{name}.avs" csv = source_dir / f"{stem}.diag.{name}.csv" log = Path(f"{csv}.runner.log") for path in (csv, log): if path.exists(): path.unlink() script.write_text(script_text, encoding="utf-8") completed = subprocess.run( [str(runner), "--validate", str(script), str(csv)], capture_output=True, text=True, ) lines.append(f"- {name}: exit {completed.returncode}, csv={'yes' if csv.exists() else 'no'}") if completed.stderr.strip(): lines.append(_indent("stderr: " + completed.stderr.strip())) if completed.stdout.strip(): lines.append(_indent("stdout: " + _tail(completed.stdout.strip()))) if log.exists(): lines.append(_indent("runner log:\n" + _tail(log.read_text(encoding="utf-8", errors="replace").strip()))) return "\n".join(lines) def _remove_validation_artifacts(csv_path: Path) -> None: for path in (csv_path, Path(f"{csv_path}.runner.log")): path.unlink(missing_ok=True) def _optional_load_plugin(plugin_path: str | None) -> str: if not plugin_path: return "" return f'LoadPlugin("{_avs_path(str(Path(plugin_path).resolve()))}")\n' def _avs_path(path: str) -> str: return path.replace("\\", "/").replace('"', '\\"') def _indent(text: str) -> str: return "\n".join(" " + line for line in text.splitlines()) def _tail(text: str, max_lines: int = 12) -> str: lines = text.splitlines() if len(lines) <= max_lines: return text return "\n".join(lines[-max_lines:])