232 lines
6.6 KiB
Python
232 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from tools.console import ProgressView
|
|
from tools.avisynth_validate import (
|
|
FrameIdentity,
|
|
IdentityValidationResult,
|
|
read_frame_identity_csv,
|
|
validate_frame_identities,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AvisynthClipInfo:
|
|
width: int
|
|
height: int
|
|
chroma: str
|
|
bit_depth: int
|
|
frames: int
|
|
fps_num: int
|
|
fps_den: int
|
|
has_audio: bool
|
|
audio_rate: int
|
|
audio_channels: int
|
|
audio_samples: int
|
|
|
|
@property
|
|
def duration_seconds(self) -> float | None:
|
|
if self.fps_num <= 0:
|
|
return None
|
|
return self.frames * self.fps_den / self.fps_num
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AvisynthValidationRun:
|
|
script: Path
|
|
csv_path: Path
|
|
frames: list[FrameIdentity]
|
|
result: IdentityValidationResult
|
|
renderer_stdout: str
|
|
renderer_stderr: str
|
|
renderer_returncode: int
|
|
recovered_from_renderer_error: bool
|
|
clip_info: AvisynthClipInfo | None
|
|
|
|
|
|
_progress_view: ProgressView | None = None
|
|
|
|
|
|
def run_validation_script(
|
|
*,
|
|
renderer_command: list[str],
|
|
script: Path,
|
|
csv_path: Path,
|
|
expected_source_ids: set[int],
|
|
progress_label: str = "Validating",
|
|
) -> AvisynthValidationRun:
|
|
global _progress_view
|
|
if csv_path.exists():
|
|
csv_path.unlink()
|
|
log_path = _runner_log_path(csv_path)
|
|
if log_path.exists():
|
|
log_path.unlink()
|
|
|
|
process = subprocess.Popen(
|
|
[*renderer_command, str(script), str(csv_path)],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
assert process.stderr is not None
|
|
_progress_view = None
|
|
stderr_lines: list[str] = []
|
|
for line in process.stderr:
|
|
if line.startswith("mbt_progress,"):
|
|
_print_progress(line, progress_label)
|
|
else:
|
|
stderr_lines.append(line)
|
|
stdout, _ = process.communicate()
|
|
_clear_progress_line(keep=True)
|
|
completed = subprocess.CompletedProcess(
|
|
process.args,
|
|
process.returncode,
|
|
stdout,
|
|
"".join(stderr_lines),
|
|
)
|
|
if not csv_path.exists():
|
|
raise RuntimeError(
|
|
f"Validation renderer did not create CSV: {csv_path}"
|
|
f"{_renderer_failure_suffix(completed, csv_path)}"
|
|
)
|
|
|
|
frames = read_frame_identity_csv(csv_path)
|
|
clip_info = parse_clip_info(completed.stdout)
|
|
recovered = completed.returncode != 0
|
|
if recovered and not _has_complete_validation_output(
|
|
frames=frames,
|
|
clip_info=clip_info,
|
|
rendered_count=parse_rendered_count(completed.stdout),
|
|
):
|
|
raise RuntimeError(
|
|
"Validation renderer failed before producing a complete frame table"
|
|
f"{_renderer_failure_suffix(completed, csv_path)}"
|
|
)
|
|
|
|
result = validate_frame_identities(frames, expected_source_ids=expected_source_ids)
|
|
return AvisynthValidationRun(
|
|
script=script,
|
|
csv_path=csv_path,
|
|
frames=frames,
|
|
result=result,
|
|
renderer_stdout=completed.stdout,
|
|
renderer_stderr=completed.stderr,
|
|
renderer_returncode=completed.returncode,
|
|
recovered_from_renderer_error=recovered,
|
|
clip_info=clip_info,
|
|
)
|
|
|
|
|
|
def parse_clip_info(stdout: str) -> AvisynthClipInfo | None:
|
|
for line in stdout.splitlines():
|
|
if not line.startswith("clip_info,"):
|
|
continue
|
|
values: dict[str, str] = {}
|
|
for part in line.split(",")[1:]:
|
|
if "=" not in part:
|
|
continue
|
|
key, value = part.split("=", 1)
|
|
values[key] = value
|
|
try:
|
|
return AvisynthClipInfo(
|
|
width=int(values["width"]),
|
|
height=int(values["height"]),
|
|
chroma=values.get("chroma", "unknown"),
|
|
bit_depth=int(values.get("bit_depth", "8")),
|
|
frames=int(values["frames"]),
|
|
fps_num=int(values["fps_num"]),
|
|
fps_den=int(values["fps_den"]),
|
|
has_audio=bool(int(values["has_audio"])),
|
|
audio_rate=int(values["audio_rate"]),
|
|
audio_channels=int(values.get("audio_channels", "0")),
|
|
audio_samples=int(values["audio_samples"]),
|
|
)
|
|
except (KeyError, ValueError) as exc:
|
|
raise RuntimeError(f"Invalid Avisynth clip_info line: {line}") from exc
|
|
return None
|
|
|
|
|
|
def parse_rendered_count(stdout: str) -> int | None:
|
|
for line in stdout.splitlines():
|
|
if not line.startswith("rendered="):
|
|
continue
|
|
try:
|
|
return int(line.split("=", 1)[1])
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"Invalid Avisynth rendered line: {line}") from exc
|
|
return None
|
|
|
|
|
|
def _has_complete_validation_output(
|
|
*,
|
|
frames: list[FrameIdentity],
|
|
clip_info: AvisynthClipInfo | None,
|
|
rendered_count: int | None,
|
|
) -> bool:
|
|
if rendered_count is not None:
|
|
return len(frames) == rendered_count
|
|
if clip_info is not None:
|
|
return len(frames) == clip_info.frames
|
|
return False
|
|
|
|
|
|
def _renderer_failure_suffix(
|
|
completed: subprocess.CompletedProcess[str],
|
|
csv_path: Path,
|
|
) -> str:
|
|
parts = [f" (exit code {completed.returncode})"]
|
|
if completed.stderr.strip():
|
|
parts.append(f"\nstderr:\n{completed.stderr.strip()}")
|
|
if completed.stdout.strip():
|
|
parts.append(f"\nstdout:\n{completed.stdout.strip()}")
|
|
log_path = _runner_log_path(csv_path)
|
|
if log_path.exists():
|
|
parts.append(f"\nrunner log ({log_path}):\n{log_path.read_text(encoding='utf-8', errors='replace').strip()}")
|
|
return "".join(parts)
|
|
|
|
|
|
def _runner_log_path(csv_path: Path) -> Path:
|
|
return Path(f"{csv_path}.runner.log")
|
|
|
|
|
|
def _print_progress(line: str, label: str) -> None:
|
|
global _progress_view
|
|
if not sys.stdout.isatty():
|
|
return
|
|
parts = line.strip().split(",")
|
|
if len(parts) != 6:
|
|
return
|
|
try:
|
|
current = int(parts[1])
|
|
total = int(parts[2])
|
|
float(parts[3])
|
|
elapsed = float(parts[4])
|
|
eta = float(parts[5])
|
|
except ValueError:
|
|
return
|
|
if _progress_view is None or _progress_view.total != total:
|
|
_progress_view = ProgressView(
|
|
total,
|
|
label,
|
|
embedded_percent=True,
|
|
show_rate=True,
|
|
)
|
|
_progress_view.update_external(
|
|
current,
|
|
elapsed=elapsed,
|
|
eta=eta if current >= 2 else None,
|
|
)
|
|
|
|
|
|
def _clear_progress_line(*, keep: bool) -> None:
|
|
global _progress_view
|
|
if not sys.stdout.isatty():
|
|
return
|
|
if _progress_view is not None:
|
|
_progress_view.finish(keep=keep)
|
|
_progress_view = None
|