Update app.py

app.py

@@ -1,17 +1,16 @@
 """
-Script ini dibuat oleh __drat dan BF667 di github aja cik
+Script ini dibuat oleh __drat dan BF667 di GitHub.
 Petunjuk:
-1. Script ini digunakan untuk mengkonversi teks menjadi suara menggunakan teknologi Edge TTS dan Retrieval-based Voice Conversion (RVC).
-2. Teknologi yang digunakan meliputi model text-to-speech (TTS) yang canggih dengan konversi teks ke fonem (G2P).
-3. Model yang dipakai dilatih khusus untuk bahasa Indonesia, Jawa, dan Sunda.
-4. Antarmuka dibuat dengan menggunakan Gradio dengan tema kustom bernama IndonesiaTheme.
+1. Mengkonversi teks menjadi suara menggunakan Edge TTS dan Retrieval-based Voice Conversion (RVC).
+2. Mendukung model text-to-speech (TTS) untuk bahasa Indonesia, Jawa, dan Sunda.
+3. Antarmuka menggunakan Gradio dengan tema kustom IndonesiaTheme.
 
 Cara Menggunakan:
-1. Pilih model suara dari dropdown yang tersedia.
-2. Atur parameter seperti kecepatan bicara, metode ekstraksi pitch, dan tingkat perlindungan.
-3. Masukkan teks yang ingin dikonversi menjadi suara.
-4. Klik tombol "Convert" untuk memulai proses konversi.
-5. Dengarkan hasil konversi melalui komponen audio yang tersedia.
+1. Pilih model suara dari dropdown.
+2. Atur parameter (kecepatan bicara, pitch, dll.).
+3. Masukkan teks untuk dikonversi.
+4. Klik "Convert" untuk menghasilkan suara.
+5. Dengarkan hasil melalui komponen audio.
 """
 
 import asyncio
@@ -20,12 +19,15 @@ import logging
 import os
 import time
 import traceback
-import warnings  # Untuk menangani peringatan
+import warnings
+from pathlib import Path
 
 import edge_tts
 import gradio as gr
 import librosa
 import torch
+import tqdm
+import requests
 
 from config import Config
 from lib.infer_pack.models import (
@@ -37,188 +39,140 @@ from lib.infer_pack.models import (
 from rmvpe import RMVPE
 from vc_infer_pipeline import VC
 
-# Menonaktifkan semua peringatan
+# Konfigurasi awal
 warnings.filterwarnings("ignore")
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+for logger_name in ["fairseq", "numba", "markdown_it", "urllib3", "matplotlib"]:
+    logging.getLogger(logger_name).setLevel(logging.ERROR)
 
-# Mengatur level logging untuk berbagai pustaka
-logging.getLogger("fairseq").setLevel(logging.ERROR)
-logging.getLogger("numba").setLevel(logging.ERROR)
-logging.getLogger("markdown_it").setLevel(logging.ERROR)
-logging.getLogger("urllib3").setLevel(logging.ERROR)
-logging.getLogger("matplotlib").setLevel(logging.ERROR)
-
-# Memeriksa apakah ada batasan sistem (contoh: menjalankan di HuggingFace Spaces)
-limitation = os.getenv("SYSTEM") == "spaces"
-
-# Memuat konfigurasi
 config = Config()
-BASE_DIR = os.getcwd()
+BASE_DIR = Path.cwd()
+MODEL_ROOT = BASE_DIR / "weights"
+EDGE_OUTPUT_FILENAME = "edge_output.mp3"
+LIMITATION = os.getenv("SYSTEM") == "spaces"
 
-# Edge TTS
-edge_output_filename = "edge_output.mp3"
-tts_voice_list = asyncio.get_event_loop().run_until_complete(edge_tts.list_voices())
+# Memuat daftar suara Edge TTS
+tts_voice_list = asyncio.run(edge_tts.list_voices())
 tts_voices = [f"{v['ShortName']}-{v['Gender']}" for v in tts_voice_list]
 
-# Memuat model RVC dari direktori "weights"
-model_root = "weights"
-models = [d for d in os.listdir(model_root) if os.path.isdir(f"{model_root}/{d}")]
-models.sort()
+# Memuat model RVC dari direktori weights
+models = sorted([d for d in MODEL_ROOT.iterdir() if d.is_dir()])
+
+def model_data(model_name: str):
+    """Memuat data model berdasarkan nama model."""
+    try:
+        pth_path = next(MODEL_ROOT / model_name).glob("*.pth")
+        logging.info(f"Memuat model: {pth_path}")
+        cpt = torch.load(pth_path, map_location="cpu")
+        tgt_sr = cpt["config"][-1]
+        cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0]
+        if_f0 = cpt.get("f0", 1)
+        version = cpt.get("version", "v1")
 
-# Fungsi untuk memuat data model berdasarkan nama model
-def model_data(model_name):
-    # Memuat file model (.pth)
-    pth_path = [
-        f"{model_root}/{model_name}/{f}"
-        for f in os.listdir(f"{model_root}/{model_name}")
-        if f.endswith(".pth")
-    ][0]
-    print(f"Memuat {pth_path}")
-    cpt = torch.load(pth_path, map_location="cpu")
-    tgt_sr = cpt["config"][-1]
-    cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0]  # n_spk
-    if_f0 = cpt.get("f0", 1)
-    version = cpt.get("version", "v1")
+        # Pilih model berdasarkan versi dan f0
+        model_classes = {
+            ("v1", 1): SynthesizerTrnMs256NSFsid,
+            ("v1", 0): SynthesizerTrnMs256NSFsid_nono,
+            ("v2", 1): SynthesizerTrnMs768NSFsid,
+            ("v2", 0): SynthesizerTrnMs768NSFsid_nono,
+        }
+        model_class = model_classes.get((version, if_f0))
+        if not model_class:
+            raise ValueError(f"Versi model tidak valid: {version}, f0: {if_f0}")
 
-    # Memilih model berdasarkan versi dan konfigurasi f0
-    if version == "v1":
-        if if_f0 == 1:
-            net_g = SynthesizerTrnMs256NSFsid(*cpt["config"], is_half=config.is_half)
-        else:
-            net_g = SynthesizerTrnMs256NSFsid_nono(*cpt["config"])
-    elif version == "v2":
-        if if_f0 == 1:
-            net_g = SynthesizerTrnMs768NSFsid(*cpt["config"], is_half=config.is_half)
-        else:
-            net_g = SynthesizerTrnMs768NSFsid_nono(*cpt["config"])
-    else:
-        raise ValueError("Versi tidak diketahui")
-    
-    # Menghapus bagian encoder
-    del net_g.enc_q
-    net_g.load_state_dict(cpt["weight"], strict=False)
-    print("Model dimuat")
-    net_g.eval().to(config.device)
-    
-    # Mengatur tipe data model
-    if config.is_half:
-        net_g = net_g.half()
-    else:
-        net_g = net_g.float()
-    
-    vc = VC(tgt_sr, config)
+        net_g = model_class(*cpt["config"], is_half=config.is_half)
+        del net_g.enc_q
+        net_g.load_state_dict(cpt["weight"], strict=False)
+        net_g.eval().to(config.device)
+        net_g = net_g.half() if config.is_half else net_g.float()
 
-    # Memuat file indeks jika ada
-    index_files = [
-        f"{model_root}/{model_name}/{f}"
-        for f in os.listdir(f"{model_root}/{model_name}")
-        if f.endswith(".index")
-    ]
-    if len(index_files) == 0:
-        print("Tidak ada file indeks ditemukan")
-        index_file = ""
-    else:
-        index_file = index_files[0]
-        print(f"File indeks ditemukan: {index_file}")
+        vc = VC(tgt_sr, config)
+        index_file = next((MODEL_ROOT / model_name).glob("*.index"), "")
+        logging.info(f"File indeks: {index_file or 'Tidak ditemukan'}")
 
-    return tgt_sr, net_g, vc, version, index_file, if_f0
+        return tgt_sr, net_g, vc, version, str(index_file), if_f0
+    except Exception as e:
+        logging.error(f"Error memuat model: {e}")
+        raise
 
-# Fungsi untuk memuat model Hubert
 def load_hubert():
-    from fairseq import fairseq
-    
-    forward_dml = fairseq.GradMultiply.forward 
+    """Memuat model Hubert."""
+    try:
+        from fairseq import checkpoint_utils
+        models, _, _ = checkpoint_utils.load_model_ensemble_and_task(
+            [str(BASE_DIR / "hubert_base.pt")], arg_overrides={"data": str(BASE_DIR)}
+        )
+        hubert_model = models[0].to(config.device)
+        hubert_model = hubert_model.half() if config.is_half else hubert_model.float()
+        return hubert_model.eval()
+    except Exception as e:
+        logging.error(f"Error memuat Hubert: {e}")
+        raise
 
-    
-    models, _, _ = fairseq.load_model(
-        f"{BASE_DIR}/hubert_base.pt",
-    )
-    
-    hubert_model = models[0]
-    hubert_model = hubert_model.to(config.device)
-    if config.is_half:
-        hubert_model = hubert_model.half()
-    else:
-        hubert_model = hubert_model.float()
-    return hubert_model.eval()
+def download_file(url: str, output_path: str = None):
+    """Mengunduh file dari URL dengan progress bar."""
+    try:
+        url = url.replace("/blob/", "/resolve/").replace("?download=true", "").strip()
+        output_path = Path(output_path or os.path.basename(url))
+        response = requests.get(url, stream=True, timeout=300)
+        response.raise_for_status()
+
+        total_size = int(response.headers.get("content-length", 0))
+        with open(output_path, "wb") as f, tqdm.tqdm(
+            desc=output_path.name, total=total_size, unit="B", unit_scale=True
+        ) as pbar:
+            for chunk in response.iter_content(chunk_size=10 * 1024 * 1024):
+                f.write(chunk)
+                pbar.update(len(chunk))
+        return str(output_path)
+    except Exception as e:
+        logging.error(f"Error mengunduh file: {e}")
+        raise
 
-# Fungsi utama TTS yang menggabungkan Edge TTS dan RVC
 def tts(
-    model_name,
-    speed,
-    tts_text,
-    tts_voice,
-    f0_up_key,
-    index_rate,
-    protect,
-    filter_radius=3,
-    resample_sr=0,
-    rms_mix_rate=0.25,
+    model_name: str,
+    speed: int,
+    tts_text: str,
+    tts_voice: str,
+    f0_up_key: int,
+    index_rate: float,
+    protect: float,
+    filter_radius: int = 3,
+    resample_sr: int = 0,
+    rms_mix_rate: float = 0.25,
 ):
-    print("RVC TTS V2")
-    print("------------------")
-    print(datetime.datetime.now())
-    print("Teks TTS:")
-    print(tts_text)
-    print(f"Suara TTS: {tts_voice}, kecepatan: {speed}")
-    print(f"Nama model: {model_name}")
-    print(f"Key: {f0_up_key}\n, Index: {index_rate}\n, Protect: {protect}")
+    """Fungsi utama untuk konversi teks ke suara."""
+    logging.info(f"Memulai TTS: {model_name}, teks: {tts_text[:50]}...")
     try:
-        # Batasan panjang teks jika ada batasan sistem
-        if limitation and len(tts_text) > 500:
-            print("Error: Teks terlalu panjang")
-            return (
-                f"Teks harus kurang dari 500 karakter di space ini, tetapi didapatkan {len(tts_text)} karakter.",
-                None,
-                None,
-            )
-        
+        if LIMITATION and len(tts_text) > 500:
+            return f"Teks terlalu panjang: {len(tts_text)} karakter (>500).", None, None
+
         t0 = time.time()
-        # Mengatur kecepatan bicara
-        if speed >= 0:
-            speed_str = f"+{speed}%"
-        else:
-            speed_str = f"{speed}%"
-        
-        # Menggunakan Edge TTS untuk menghasilkan file suara sementara
+        speed_str = f"+{speed}%" if speed >= 0 else f"{speed}%"
         asyncio.run(
             edge_tts.Communicate(
                 tts_text, "-".join(tts_voice.split("-")[:-1]), rate=speed_str
-            ).save(edge_output_filename)
+            ).save(EDGE_OUTPUT_FILENAME)
         )
-        t1 = time.time()
-        edge_time = t1 - t0
-        
-        # Memuat file suara dan menghitung durasi
-        audio, sr = librosa.load(edge_output_filename, sr=16000, mono=True)
+        edge_time = time.time() - t0
+
+        audio, sr = librosa.load(EDGE_OUTPUT_FILENAME, sr=16000, mono=True)
         duration = len(audio) / sr
-        print(f"Durasi audio: {duration}s")
-        
-        # Batasan durasi audio jika ada batasan sistem
-        if limitation and duration >= 50:
-            print("Error: Audio terlalu panjang")
-            return (
-                f"Audio harus kurang dari 50 detik di space ini, tetapi didapatkan {duration}s.",
-                edge_output_filename,
-                None,
-            )
-        
-        f0_up_key = int(f0_up_key)
+        if LIMITATION and duration >= 50:
+            return f"Audio terlalu panjang: {duration}s (>50s).", EDGE_OUTPUT_FILENAME, None
 
-        # Memuat model data
         tgt_sr, net_g, vc, version, index_file, if_f0 = model_data(model_name)
         vc.model_rmvpe = rmvpe_model
         times = [0, 0, 0]
-        f0_method = "rmvpe"
-        # Menggunakan pipeline RVC untuk menghasilkan file suara akhir
         audio_opt = vc.pipeline(
             hubert_model,
             net_g,
             0,
             audio,
-            edge_output_filename,
+            EDGE_OUTPUT_FILENAME,
             times,
             f0_up_key,
-            f0_method,
+            "rmvpe",
             index_file,
             index_rate,
             if_f0,
@@ -230,150 +184,76 @@ def tts(
             protect,
             None,
         )
-        
-        # Meresample jika diperlukan
-        if tgt_sr != resample_sr >= 16000:
-            tgt_sr = resample_sr
-        
-        info = f"Berhasil. Waktu: edge-tts: {edge_time}s, npy: {times[0]}s, f0: {times[1]}s, infer: {times[2]}s"
-        print(info)
-        return (
-            info,
-            edge_output_filename,
-            (tgt_sr, audio_opt),
-        )
-    except EOFError:
-        info = (
-            "Sepertinya output edge-tts tidak valid. "
-            "Ini bisa terjadi jika teks input dan pembicara tidak cocok. "
-            "Misalnya, mungkin Anda memasukkan teks dalam bahasa Jepang (tanpa huruf alfabet) tetapi memilih pembicara non-Jepang?"
-        )
-        print(info)
-        return info, None, None
-    except:
-        info = traceback.format_exc()
-        print(info)
-        return info, None, None
 
-# Memuat model Hubert
-print("Memuat model hubert...")
-hubert_model = load_hubert()
-print("Model hubert dimuat.")
+        tgt_sr = resample_sr if resample_sr >= 16000 else tgt_sr
+        info = f"Berhasil. Waktu: edge-tts: {edge_time:.2f}s, npy: {times[0]:.2f}s, f0: {times[1]:.2f}s, infer: {times[2]:.2f}s"
+        return info, EDGE_OUTPUT_FILENAME, (tgt_sr, audio_opt)
+    except Exception as e:
+        error_msg = f"Error: {str(e)}\n{traceback.format_exc()}"
+        logging.error(error_msg)
+        return error_msg, None, None
 
-# Memuat model RMVPE
-print("Memuat model rmvpe...")
+# Memuat model
+logging.info("Memuat model Hubert...")
+hubert_model = load_hubert()
+logging.info("Memuat model RMVPE...")
 rmvpe_model = RMVPE("rmvpe.pt", config.is_half, config.device)
-print("Model rmvpe dimuat.")
-
 
-def download_model(url, model_name):
-    from tools.huggingface import HF_download_file
-    model_folder = "weights"
-    output_path = os.path.join(model_folder, model_name)
-    HF_download_file(url, output_path=None)
+def download_model(url: str, model_name: str):
+    """Mengunduh dan menyimpan model ke direktori weights."""
+    output_path = MODEL_ROOT / model_name
+    output_path.mkdir(exist_ok=True)
+    return download_file(url, output_path / Path(url).name)
 
-
-# Initial markdown text untuk ditampilkan di antarmuka
+# Antarmuka Gradio
 initial_md = """
-<h1 align="center"><b> TTS RVC Indonesia 🎵 </b></h1>
-</div>
-
-
-Pembuktian algoritma **Retrieval-based Voice Conversion (RVC)** dan teknologi **Edge TTS** yang dapat membuat clone dari suara artis & selebriti di Indonesia. 
-
-**Perhatian:** Harap tidak menyalahgunakan teknologi ini. **Limitasi:** Teks 500, Audio 50 detik.
+<h1 align="center"><b>TTS RVC Indonesia 🎵</b></h1>
+<p align="center">Konversi teks ke suara menggunakan Edge TTS dan RVC untuk suara artis Indonesia.</p>
+<p><b>Perhatian:</b> Jangan menyalahgunakan teknologi ini. <b>Limitasi:</b> Teks maks. 500 karakter, audio maks. 50 detik.</p>
 """
 
-# Membuat aplikasi Gradio
-app = gr.Blocks(theme="Thatguy099/Sonix", title="TTS-RVC-Artis Indonesia")
-with app:
+with gr.Blocks(theme="Thatguy099/Sonix", title="TTS-RVC Indonesia") as app:
     gr.Markdown(initial_md)
-    model_name = gr.Dropdown(
-        label="Model",
-        choices=models,
-        value=models[0],
-    )
-    f0_key_up = gr.Number(
-        label="Tune (+12 = 1 oktaf dari edge-tts, nilai terbaik tergantung pada model dan pembicara)",
-        value=2,
-    )
+    with gr.Row():
+        model_name = gr.Dropdown(label="Model", choices=models, value=models[0])
+        f0_key_up = gr.Number(label="Tune (oktaf dari edge-tts)", value=2)
     with gr.Column():
         with gr.Row():
             with gr.Tab("Unduh Model"):
-                url = gr.Textbox(label="model URL")
-                model_nae = gr.Textbox(label=" Nama Model") 
+                url = gr.Textbox(label="URL Model")
+                model_nae = gr.Textbox(label="Nama Model")
                 dlm = gr.Button("Unduh Model")
                 dlm.click(fn=download_model, inputs=[url, model_nae], outputs=None)
-        with gr.Column():
-            
-            index_rate = gr.Slider(
-                minimum=0,
-                maximum=1,
-                label="Tingkat indeks",
-                value=0.5,
-                interactive=True,
-            )
-            protect0 = gr.Slider(
-                minimum=0,
-                maximum=0.5,
-                label="Perlindungan",
-                value=0.33,
-                step=0.01,
-                interactive=True,
-            )
-        with gr.Column():
-            tts_voice = gr.Dropdown(
-                label="Pembicara Edge-tts (format: bahasa-Negara-Nama-Jenis Kelamin), pastikan jenis kelamin cocok dengan model",
-                choices=tts_voices,
-                allow_custom_value=False,
-                value="id-ID-ArdiNeural-Male",  # Set nilai default
-            )
-            speed = gr.Slider(
-                minimum=-100,
-                maximum=100,
-                label="Kecepatan bicara (%)",
-                value=0,
-                step=10,
-                interactive=True,
-            )
-            tts_text = gr.Textbox(label="Teks Input", value="Konversi dari teks ke suara dalam bahasa Indonesia.")
-        with gr.Column():
-            with gr.Row():
-                but0 = gr.Button("Konversi", variant="primary")
-            info_text = gr.Textbox(label="Informasi Output")
-        with gr.Column():
-            with gr.Row():
-                edge_tts_output = gr.Audio(label="Suara Edge", type="filepath")
-                tts_output = gr.Audio(label="Hasil")
+        index_rate = gr.Slider(minimum=0, maximum=1, label="Tingkat Indeks", value=0.5)
+        protect0 = gr.Slider(minimum=0, maximum=0.5, label="Perlindungan", value=0.33, step=0.01)
+        tts_voice = gr.Dropdown(
+            label="Pembicara Edge-TTS (bahasa-Negara-Nama-Jenis Kelamin)",
+            choices=tts_voices,
+            value="id-ID-ArdiNeural-Male",
+        )
+        speed = gr.Slider(minimum=-100, maximum=100, label="Kecepatan Bicara (%)", value=0, step=10)
+        tts_text = gr.Textbox(label="Teks Input", value="Konversi teks ke suara dalam bahasa Indonesia.")
+        but0 = gr.Button("Konversi", variant="primary")
+        info_text = gr.Textbox(label="Informasi Output")
+        with gr.Row():
+            edge_tts_output = gr.Audio(label="Suara Edge", type="filepath")
+            tts_output = gr.Audio(label="Hasil")
         but0.click(
             tts,
-            [
-                model_name,
-                speed,
-                tts_text,
-                tts_voice,
-                f0_key_up,
-                index_rate,
-                protect0,
-            ],
+            [model_name, speed, tts_text, tts_voice, f0_key_up, index_rate, protect0],
             [info_text, edge_tts_output, tts_output],
         )
-    with gr.Row():
-        examples = gr.Examples(
-            examples_per_page=100,
-            examples=[
-                ["Ini adalah demo percobaan menggunakan Bahasa Indonesia untuk pria.", "id-ID-ArdiNeural-Male"],
-                ["Ini adalah teks percobaan menggunakan Bahasa Indonesia pada wanita.", "id-ID-GadisNeural-Female"],
-            ],
-            inputs=[tts_text, tts_voice],
-        )
-    
-    # Tambahkan footer di bagian bawah
+    gr.Examples(
+        examples=[
+            ["Ini adalah demo percobaan menggunakan Bahasa Indonesia untuk pria.", "id-ID-ArdiNeural-Male"],
+            ["Ini adalah teks percobaan menggunakan Bahasa Indonesia pada wanita.", "id-ID-GadisNeural-Female"],
+        ],
+        inputs=[tts_text, tts_voice],
+    )
     gr.HTML("""
     <footer style="text-align: center; margin-top: 20px; color:silver;">
         Energi Semesta Digital © 2024 __drat. | 🇮🇩 Untuk Indonesia Jaya!
     </footer>
     """)
 
-# Meluncurkan aplikasi
-app.launch()
+app.launch()