ensure_reference_videos_available(*, local_dir: Path | None = None, repo_id: str | None = None, repo_type: str | None = None, quality_tier: str = DEFAULT_OUTPUT_QUALITY_TIER) -> bool
Return True if downloaded from HF, False if already present locally.
Source code in fastvideo/tests/ssim/reference_videos_cli.py
| def ensure_reference_videos_available(
*,
local_dir: Path | None = None,
repo_id: str | None = None,
repo_type: str | None = None,
quality_tier: str = DEFAULT_OUTPUT_QUALITY_TIER,
) -> bool:
"""Return True if downloaded from HF, False if already present locally."""
if quality_tier not in QUALITY_TIERS:
raise ValueError(f"Unsupported quality tier: {quality_tier}")
target_dir = local_dir or _ssim_dir()
lock_path = target_dir / ".reference_videos_download.lock"
with _exclusive_download_lock(lock_path):
if _has_local_reference_videos(target_dir, quality_tier):
print(f"Reference videos ({quality_tier}) already available at {target_dir}")
return False
resolved_repo_id = repo_id or _default_repo_id()
resolved_repo_type = repo_type or _default_repo_type()
if not resolved_repo_id:
raise RuntimeError(
f"No local reference videos found and no HF repo configured.\nSet {HF_REPO_ENV_KEY} or pass --repo-id."
)
print(f"Repo ID: {resolved_repo_id}")
print(f"Quality tier: {quality_tier}")
print(f"No local {quality_tier} reference videos found under {target_dir}. Starting download...")
try:
download_reference_videos(
repo_id=resolved_repo_id,
repo_type=resolved_repo_type,
local_dir=target_dir,
quality_tiers=[quality_tier],
)
print(f"Download completed for {quality_tier} reference videos.")
except Exception as exc:
print(f"ERROR: Failed to download {quality_tier} reference videos from {resolved_repo_id}.")
print(
f"Suggested command to retry: "
f"python fastvideo/tests/ssim/reference_videos_cli.py download "
f"--quality-tier {quality_tier}"
)
raise
if not _has_local_reference_videos(target_dir, quality_tier):
raise RuntimeError(f"HF download completed but no {quality_tier} *_reference_videos content found.")
return True
|