1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 |
- import urllib.request
- from pathlib import Path
- from threading import Thread
- from urllib.error import HTTPError
- from tqdm import tqdm
- default_models = {
- "encoder": ("https://drive.google.com/uc?export=download&id=1q8mEGwCkFy23KZsinbuvdKAQLqNKbYf1", 17090379),
- "synthesizer": ("https://drive.google.com/u/0/uc?id=1EqFMIbvxffxtjiVrtykroF6_mUh-5Z3s&export=download&confirm=t", 370554559),
- "vocoder": ("https://drive.google.com/uc?export=download&id=1cf2NO6FtI0jDuy8AV3Xgn6leO6dHjIgu", 53845290),
- }
- class DownloadProgressBar(tqdm):
- def update_to(self, b=1, bsize=1, tsize=None):
- if tsize is not None:
- self.total = tsize
- self.update(b * bsize - self.n)
- def download(url: str, target: Path, bar_pos=0):
- # Ensure the directory exists
- target.parent.mkdir(exist_ok=True, parents=True)
- desc = f"Downloading {target.name}"
- with DownloadProgressBar(unit="B", unit_scale=True, miniters=1, desc=desc, position=bar_pos, leave=False) as t:
- try:
- urllib.request.urlretrieve(url, filename=target, reporthook=t.update_to)
- except HTTPError:
- return
- def ensure_default_models(models_dir: Path):
- # Define download tasks
- jobs = []
- for model_name, (url, size) in default_models.items():
- target_path = models_dir / "default" / f"{model_name}.pt"
- if target_path.exists():
- if target_path.stat().st_size != size:
- print(f"File {target_path} is not of expected size, redownloading...")
- else:
- continue
- thread = Thread(target=download, args=(url, target_path, len(jobs)))
- thread.start()
- jobs.append((thread, target_path, size))
- # Run and join threads
- for thread, target_path, size in jobs:
- thread.join()
- assert target_path.exists() and target_path.stat().st_size == size, \
- f"Download for {target_path.name} failed. You may download models manually instead.\n" \
- f"https://drive.google.com/drive/folders/1fU6umc5uQAVR2udZdHX-lDgXYzTyqG_j"
|