Function install_demucs_reference

tools/model_manager.py:1568–1635

(
    package: ModelPackage,
    models_root: Path,
    overwrite: bool,
    source_dir_arg: str | None,
    variant_arg: str | None,
)

            chunk = response.read(1 << 20)
            if not chunk:
                break
            handle.write(chunk)
            written += len(chunk)
    if expected_size is not None and written != expected_size:
        raise RuntimeError(f"downloaded size mismatch for {target}: {written} != {expected_size}")
    return written


def validate_required_files(package: ModelPackage, root: Path) -> None:
    missing = [relative for relative in package.required_files if not (root / relative).exists()]
    if missing:
        raise RuntimeError(f"installed package is missing required files: {missing}")


def validate_composite_required_files(package: ModelPackage, staged_root: Path, final_root: Path) -> None:
    missing = [
        relative
        for relative in package.required_files
        if not normalized_join(staged_root, relative).exists() and not normalized_join(final_root, relative).exists()
    ]
    if missing:
        raise RuntimeError(f"installed package is missing required files: {missing}")


def validate_required_files_list(required_files: Iterable[str], root: Path, label: str) -> None:
    missing = [relative for relative in required_files if not (root / relative).exists()]
    if missing:
        raise RuntimeError(f"{label} is missing required files: {missing}")


def install_snapshot_into_dir(
    source: SnapshotSource,
    destination_root: Path,
    required_files: Iterable[str],
    *,
    validate: bool = True,
) -> None:
    files = list_hf_files(source)
    for remote, relative, expected_size in files:
        destination = destination_root / relative
        destination.parent.mkdir(parents=True, exist_ok=True)
        print(f"download {remote}")
        download_file(hf_resolve_url(source, remote), destination, expected_size)
    if validate:
        validate_required_files_list(required_files, destination_root, source.repo_id)


def write_vevo2_whisper_stats(target_path: Path, source_file: Path) -> None:
    payload = torch.load(source_file, map_location="cpu", weights_only=False)
    state = checkpoint_state_dict(payload)
    tensors = tensor_state_dict(state)
    if set(tensors.keys()) != {"mean", "std"}:
        raise RuntimeError(f"Vevo2 whisper stats must contain mean/std only: {sorted(tensors.keys())}")
    save_file(tensors, str(target_path))


def whisper_dims(payload: dict[str, Any]) -> dict[str, int]:
    raw_dims = payload.get("dims")