見出し画像

GoogleColabにもAI機能が!おまけ付

こんにちは、高橋です。
最近使っていなかったのでもしかするとすごく今更な可能性もあるのですが、GoogleColabの右側にあったGeminiが中央下に移動してました。

押してみると…

こんな感じで、『何々をしたい』『このエラーをなおして』というと、コードを直接修正してくれるようになっていました。

GoogleColabは、音声生成や画像生成などのマシンパワーを要するコードを書く際に主に使用していたのですが、
よくある依存関係の問題もこの子なら一発でクリアできるようになっているのでは?と思い、以前から気になっていたParler-TTSちゃんを起動するためのコードをカキカキ✏

依存関係にぶつかったところで『このコード動くようにして!』とGeminiにお願いしたところ、確かにコードは書いてくれたのですが、エラーのままでした…。
環境が統合されているくらいなので精度も期待していたのですが、まだまだ用途によっては力を発揮できないようです。

結局手で絡んだ糸をほどきました。。。
この辺もできるといいんですけどね~
今後の進化に期待です!!

おまけ
Parler-TTSの実行コードです。
初めのユーザー設定あたりは、先にgoogledriveにフォルダを作るなど準備をしておいてください。

ランタイムバージョンは必ず2025.07で!
# ======================================================
#  Colab一発: 隔離env + 多参照 XTTS v2 日本語合成 + 後処理(無音/間短縮/NR/De-esser/正規化)
#  CPMLプロンプトは ModelManager.ask_tos を確実にパッチして無効化
#  出力:
#   - /MyDrive/xtts_finetune/outs/out_multi_ref.wav
#   - /MyDrive/xtts_finetune/outs/out_multi_ref_post.wav
# ======================================================

# ===== ユーザー設定(必要に応じて変更)=====
BASE      = "/content/drive/MyDrive/xtts_finetune"
REF_RAW   = f"{BASE}/raw_refs"   # 参照(wav/mp3)
REF_WAV   = f"{BASE}/wavs_refs"  # 前処理後
OUT_DIR   = f"{BASE}/outs"
TEXT      = "最後まで読んでくれてありがとうございました!"

# 参照選抜 & 長文分割
MAX_REFS      = 6      # 使う参照数(上限)
SAFE_MIN_DUR  = 3.0    # 参照として採用する最短長(秒)
CHUNK_MAX     = 120    # 1文の最大文字数(長文安定化)

# 後処理パラメータ
SIL_TH_DB     = -40.0  # 無音判定閾値(-35〜-45で調整)
MIN_VOICED_MS = 120    # これより短い音は捨てる(ブレス等)
MAX_PAUSE_MS  = 1200   # これより長い「間」は短縮
TARGET_GAP_MS = 150    # 短縮後の標準的な「間」
NOISE_HEAD_MS = 400    # 冒頭からノイズ推定に使う長さ(静かな冒頭がある時のみ)
DEESS_START_HZ= 6000   # デエッサー適用開始周波数
DEESS_GAIN    = 0.65   # 高域減衰係数(0.6〜0.85)
PEAK_DBFS     = -1.0   # 最終ピーク
CLEAR_CACHE   = False  # TrueならXTTSモデル再DL
# ============================================

import os, sys, subprocess, shutil, glob, json

# 0) Drive & 環境変数
from google.colab import drive
drive.mount('/content/drive')
os.environ["COQUI_TOS_ACCEPT"] = "1"       # CPML同意(非商用)。商用は別途ライセンス要
os.environ["PIP_DISABLE_PIP_VERSION_CHECK"] = "1"
os.environ["PYTHONNOUSERSITE"] = "1"
os.makedirs(REF_RAW, exist_ok=True)
os.makedirs(REF_WAV, exist_ok=True)
os.makedirs(OUT_DIR, exist_ok=True)

# 1) まっさら env
VENV = "/content/xtts_env"
if os.path.exists(VENV):
    print("[info] remove old env:", VENV)
    shutil.rmtree(VENV, ignore_errors=True)

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "virtualenv"])
subprocess.check_call([sys.executable, "-m", "virtualenv", "--clear", "--no-download", VENV])
PY  = f"{VENV}/bin/python"
PIP = f"{VENV}/bin/pip"

# 2) pip 最新化
subprocess.check_call([PY, "-m", "pip", "install", "-q", "--upgrade", "pip", "setuptools", "wheel"])
env = dict(os.environ, MPLBACKEND="Agg")  # 画面描画不要

# 3) PyTorch(CUDA12.1 → ダメなら CPU)
def _try(cmd):
    try:
        subprocess.check_call(cmd, env=env); return True
    except subprocess.CalledProcessError:
        return False

print("[info] Installing PyTorch (CUDA12.1 preferred)...")
ok = _try([PIP, "install", "-q", "--index-url", "https://download.pytorch.org/whl/cu121",
           "torch==2.5.1+cu121", "torchaudio==2.5.1+cu121", "torchvision==0.20.1+cu121"])
if not ok:
    print("[warn] CUDA wheels failed; installing CPU wheels.")
    subprocess.check_call([PIP, "install", "-q",
                           "torch==2.5.1", "torchaudio==2.5.1", "torchvision==0.20.1"], env=env)

# 4) 互換スタック + 基本依存(ABI衝突を避ける)
subprocess.check_call([PIP, "install", "-q",
    "numpy==1.26.4", "pandas==1.5.3", "networkx==2.8.8", "packaging>=23.2",
    "transformers==4.44.2", "librosa==0.10.2.post1", "soundfile", "ffmpeg-python"
], env=env)

# 5) Coqui TTS(--no-deps)+必要依存まとめ(テキスト処理/多言語等)
subprocess.check_call([PIP, "install", "-q", "--no-deps", "TTS==0.22.0"], env=env)
subprocess.check_call([PIP, "install", "-q",
    "encodec==0.1.1", "g2pkk==0.1.2", "Unidecode==1.3.8",
    "pysbd==0.3.4", "anyascii==0.3.2",
    "bangla==0.0.2", "bnnumerizer==0.0.2", "bnunicodenormalizer==0.1.7",
    "jamo==0.4.1", "pypinyin==0.49.0", "jieba==0.42.1",
    "hangul-romanize==0.1.0", "num2words==0.5.13",
    "cutlet", "fugashi", "unidic-lite", "pykakasi",
    "coqpit==0.0.17", "trainer==0.0.36",
    "gruut==2.2.3", "gruut-ipa==0.13.0",
    "matplotlib<3.9", "inflect==7.0.0", "spacy<3.9",
    # 後処理用
    "noisereduce==3.0.0"
], env=env)

# 6) (必要なら)XTTSモデルキャッシュ掃除
def clear_xtts_cache():
    cache = os.path.expanduser("~/.local/share/tts/tts_models--multilingual--multi-dataset--xtts_v2")
    shutil.rmtree(cache, ignore_errors=True)
    for p in glob.glob(os.path.expanduser("~/.local/share/tts/tokenizer*")):
        shutil.rmtree(p, ignore_errors=True)

if CLEAR_CACHE:
    print("[info] Clear model cache")
    clear_xtts_cache()

# 7) 実行スクリプト(合成 + 後処理)— CPMLプロンプト完全スキップ込み
run_code = f"""\
import os, glob, re, tempfile, shutil, warnings
import numpy as np, librosa, soundfile as sf, ffmpeg, torch
from noisereduce import reduce_noise

warnings.filterwarnings("ignore")

# ---- CPML 同意プロンプトを完全スキップ(ModelManager.ask_tos を上書き)----
os.environ.setdefault("COQUI_TOS_ACCEPT","1")
try:
    import TTS.utils.manage as _manage
    if hasattr(_manage, "ModelManager"):
        _manage.ModelManager.ask_tos = lambda self, output_path: True
    if hasattr(_manage, "Manager"):
        _manage.Manager.ask_tos = lambda self, output_path: True
except Exception as e:
    print("[warn] ask_tos patch failed:", e)
# --------------------------------------------------------------------------

from TTS.api import TTS

BASE      = {json.dumps(BASE)}
REF_RAW   = {json.dumps(REF_RAW)}
REF_WAV   = {json.dumps(REF_WAV)}
OUT_DIR   = {json.dumps(OUT_DIR)}
TEXT      = {json.dumps(TEXT)}
MAX_REFS  = {MAX_REFS}
CHUNK_MAX = {CHUNK_MAX}
MIN_DUR   = {SAFE_MIN_DUR}

# 後処理設定
SIL_TH_DB     = {SIL_TH_DB}
MIN_VOICED_MS = {MIN_VOICED_MS}
MAX_PAUSE_MS  = {MAX_PAUSE_MS}
TARGET_GAP_MS = {TARGET_GAP_MS}
NOISE_HEAD_MS = {NOISE_HEAD_MS}
DEESS_START_HZ= {DEESS_START_HZ}
DEESS_GAIN    = {DEESS_GAIN}
PEAK_DBFS     = {PEAK_DBFS}

os.makedirs(REF_RAW, exist_ok=True)
os.makedirs(REF_WAV, exist_ok=True)
os.makedirs(OUT_DIR, exist_ok=True)

print("参照を置くフォルダ:", REF_RAW)

def loudness_norm(y, target_rms=-20.0):
    eps=1e-9; rms=np.sqrt(np.mean(y**2)+eps)
    gain=(10**(target_rms/20.0))/(rms+eps)
    return np.clip(y*gain, -1.0, 1.0)

def highpass(y, sr, fc=70.0):
    c=np.tan(np.pi*fc/sr); a0=1/(1+c); a1=-a0; b1=(1-c)*a0
    out=np.zeros_like(y); xm1=ym1=0.0
    for i,x in enumerate(y):
        yv=a0*x+a1*xm1+b1*ym1; out[i]=yv; xm1,ym1=x,yv
    return out

def trim_silence_segments(y, sr, top_db=35):
    iv=librosa.effects.split(y, top_db=top_db)
    if not len(iv): return y
    y2=np.concatenate([y[s:e] for s,e in iv])
    pad=int(0.02*sr); return np.pad(y2,(pad,pad))

def process_ref(src, dst):
    tmp=dst+".tmp.wav"
    (ffmpeg.input(src).output(tmp, ac=1, ar="16000", format="wav")
           .overwrite_output().run(quiet=True))
    y, sr = sf.read(tmp); y=y[:,0] if y.ndim>1 else y
    y=trim_silence_segments(y,sr,35); y=highpass(y,sr,70.0); y=loudness_norm(y,-20.0)
    sf.write(dst, y, sr); os.remove(tmp)

# 参照のWAV化(未処理のみ)
raws = sorted(glob.glob(f"{{REF_RAW}}/*"))
assert raws, f"{{REF_RAW}} に wav/mp3 を入れてから再実行してください。"
for i, src in enumerate(raws):
    dst = os.path.join(REF_WAV, f"ref_{{i:03d}}.wav")
    if not os.path.isfile(dst):
        try: process_ref(src, dst); print("前処理OK:", os.path.basename(dst))
        except Exception as e: print("前処理NG:", os.path.basename(src), "->", e)

# 参照の選抜(MIN_DUR秒以上を優先)
cands = sorted(glob.glob(f"{{REF_WAV}}/*.wav"))
longs = []
for p in cands:
    try:
        y, sr = sf.read(p); y=y[:,0] if y.ndim>1 else y
        if sr>0 and len(y)/sr >= MIN_DUR:
            longs.append(p)
    except Exception:
        pass
refs = (longs or cands)[:MAX_REFS]
print("使用参照:", [os.path.basename(r) for r in refs])
assert refs, "参照が見つかりません(形式エラーの可能性)。別の音源でお試しください。"

# ---- テキスト分割
def split_text(txt, max_len=CHUNK_MAX):
    s=[]
    for chunk in re.split(r'(?<=[。!?!?])', txt):
        c=chunk.strip()
        if not c: continue
        if len(c)<=max_len: s.append(c); continue
        cur=""
        for t in re.split(r'(?<=[、,])', c):
            if len(cur)+len(t)<=max_len: cur+=t
            else:
                if cur.strip(): s.append(cur.strip()); cur=t
        if cur.strip(): s.append(cur.strip())
    return s

sents = split_text(TEXT, CHUNK_MAX)
print("分割文数:", len(sents), "→", sents)

# Torch 2.6+ の weights_only 対策(環境依存)
try:
    from packaging import version
    if version.parse(torch.__version__) >= version.parse("2.6"):
        from torch.serialization import add_safe_globals
        try:
            from TTS.tts.configs.xtts_config import XttsConfig
            add_safe_globals([XttsConfig])
        except Exception: pass
        try:
            from TTS.tts.models.xtts import XttsAudioConfig
            add_safe_globals([XttsAudioConfig])
        except Exception: pass
except Exception:
    pass

device = "cuda" if torch.cuda.is_available() else "cpu"
print("Torch:", torch.__version__, "| CUDA:", torch.version.cuda, "| device:", device)

# モデルロード(CPMLプロンプトはパッチで無効化済み)
tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device)

# ---- 合成(文ごと) → 連結して out_multi_ref.wav 保存
tmpdir = tempfile.mkdtemp(prefix="xtts_chunks_")
paths=[]
try:
    for i, sent in enumerate(sents):
        out = os.path.join(tmpdir, f"chunk_{{i:03d}}.wav")
        tts.tts_to_file(text=sent, file_path=out, speaker_wav=refs, language="ja")
        paths.append(out); print(f"[{{i+1}}/{{len(sents)}}] 生成OK")
    waves=[]; sr_t=None
    for p in paths:
        y, sr = sf.read(p); y=y[:,0] if y.ndim>1 else y
        if sr_t is None: sr_t=sr
        if waves: waves.append(np.zeros(int(0.05*sr_t), dtype=y.dtype))  # 50ms
        waves.append(y)
    final=np.concatenate(waves) if waves else np.array([], dtype=np.float32)
    out_wav=os.path.join(OUT_DIR, "out_multi_ref.wav")
    os.makedirs(OUT_DIR, exist_ok=True)
    sf.write(out_wav, final, sr_t)
    print("Saved (raw):", out_wav)

    # ======================== 後処理ブロック ========================
    IN_WAV  = out_wav
    OUT_WAV = os.path.join(OUT_DIR, "out_multi_ref_post.wav")

    def db_to_lin(db): return 10.0 ** (db / 20.0)

    def rms_frames(y, frame, hop):
        S = librosa.feature.rms(y=y, frame_length=frame, hop_length=hop, center=True)
        return S[0]

    def trim_leading_trailing(y, sr, thr_db):
        frame = int(0.025*sr); hop = int(0.010*sr)
        r = rms_frames(y, frame, hop)
        thr = np.median(r) * max(0.3, db_to_lin(thr_db))
        mask = (r > thr)
        if not mask.any():
            return y
        idx = np.where(mask)[0]
        start = max(0, idx[0]*hop - frame)
        end   = min(len(y), idx[-1]*hop + frame)
        return y[start:end]

    def segment_by_silence(y, sr, thr_db, min_voiced_ms):
        frame = int(0.02*sr); hop = int(0.01*sr)
        r = rms_frames(y, frame, hop)
        thr_rel = np.median(r)*0.5
        thr_abs = db_to_lin(thr_db)
        thr = max(thr_rel, thr_abs)
        voiced = (r > thr)
        segs = []
        i = 0
        while i < len(voiced):
            if voiced[i]:
                j = i+1
                while j < len(voiced) and voiced[j]:
                    j += 1
                s = max(0, i*hop - frame//2)
                e = min(len(y), j*hop + frame//2)
                if (e - s) >= int((min_voiced_ms/1000.0)*sr):
                    segs.append((s, e))
                i = j
            else:
                i += 1
        return segs

    def apply_deesser(y, sr, start_hz=6000, gain=0.7):
        n_fft = 1024; hop = 256
        S = librosa.stft(y, n_fft=n_fft, hop_length=hop, win_length=1024)
        freqs = librosa.fft_frequencies(sr=sr, n_fft=n_fft)
        S[freqs >= start_hz, :] *= gain
        y_out = librosa.istft(S, hop_length=hop, win_length=1024, length=len(y))
        return y_out

    def peak_normalize(y, target_db=-1.0):
        peak = np.max(np.abs(y)) + 1e-12
        target = db_to_lin(target_db)
        g = target / peak
        return np.clip(y*g, -1.0, 1.0)

    # 読み込み
    y, sr = sf.read(IN_WAV)
    if y.ndim > 1: y = y[:,0]

    # 冒頭が静かならノイズ低減(軽め/自動)
    head = y[:int(sr*NOISE_HEAD_MS/1000)]
    if head.size > 0 and np.mean(np.abs(head)) < 0.02:
        y = reduce_noise(y=y, sr=sr, stationary=True, prop_decrease=0.7, time_constant_s=0.4)

    # 先頭/末尾の無音トリム
    y = trim_leading_trailing(y, sr, SIL_TH_DB)

    # 無音で分割(短いブレスは除去)
    segs = segment_by_silence(y, sr, SIL_TH_DB, MIN_VOICED_MS)

    # 長すぎる間を短縮しながら連結
    gap = np.zeros(int(sr*TARGET_GAP_MS/1000), dtype=y.dtype)
    out = []
    prev_end = 0
    for (s, e) in segs:
        if out:
            pause = s - prev_end
            if pause > int(sr*MAX_PAUSE_MS/1000):
                out.append(gap.copy())
            else:
                out.append(y[prev_end:s])
        out.append(y[s:e])
        prev_end = e
    y = np.concatenate(out) if out else y

    # デエッサー & ピーク正規化
    y = apply_deesser(y, sr, DEESS_START_HZ, DEESS_GAIN)
    y = peak_normalize(y, PEAK_DBFS)

    sf.write(OUT_WAV, y, sr)
    print("Saved (post):", OUT_WAV)
    # ===================================================

finally:
    shutil.rmtree(tmpdir, ignore_errors=True)
"""
open("/content/run_xtts.py","w",encoding="utf-8").write(run_code)

# 8) スマート実行(不足モジュール→自動補完→再試行、キャッシュ掃除→最終リトライ)
smart_code = r"""
import os, sys, subprocess, time, runpy, re, traceback, glob, shutil

RUN = "/content/run_xtts.py"
PIP = sys.executable.replace("python","pip")
os.environ.setdefault("COQUI_TOS_ACCEPT","1")
os.environ.setdefault("MPLBACKEND","Agg")

ALIASES = {
    "anyascii":"anyascii", "cutlet":"cutlet", "fugashi":"fugashi",
    "unidic_lite":"unidic-lite", "Unidecode":"Unidecode",
    "gruut_ipa":"gruut-ipa", "hangul_romanize":"hangul-romanize",
    "noisereduce":"noisereduce",
}

def have(mod):
    try:
        import importlib.util
        return importlib.util.find_spec(mod) is not None
    except Exception:
        return False

# 事前に落ちやすいものをチェック
for m in ["matplotlib","inflect","spacy","noisereduce"]:
    if not have(m):
        print("[preflight] installing:", m, flush=True)
        subprocess.run([PIP,"install","-q",m], check=True)

def clear_xtts_cache():
    cache = os.path.expanduser("~/.local/share/tts/tts_models--multilingual--multi-dataset--xtts_v2")
    shutil.rmtree(cache, ignore_errors=True)
    for p in glob.glob(os.path.expanduser("~/.local/share/tts/tokenizer*")):
        shutil.rmtree(p, ignore_errors=True)

def run_once():
    try:
        runpy.run_path(RUN, run_name="__main__")
        return 0
    except ModuleNotFoundError as e:
        msg = str(e)
        m = re.search(r"No module named '([^']+)'", msg)
        name = m.group(1) if m else getattr(e, "name", None)
        if not name:
            print("[error] cannot parse missing module:", msg, file=sys.stderr)
            traceback.print_exc()
            return 2
        pkg = ALIASES.get(name, name)
        print(f"[auto-fix] installing missing: {name} -> {pkg}", flush=True)
        r = subprocess.run([PIP,"install","-q",pkg])
        if r.returncode != 0:
            print(f"[auto-fix] install failed: {pkg}", file=sys.stderr)
            return 2
        time.sleep(0.5)
        return 3  # retry
    except Exception:
        traceback.print_exc()
        return 1

print("[smart] run #1", flush=True)
rc = run_once()

tries = 1
while rc == 3 and tries < 5:
    tries += 1
    print(f"[smart] run #{tries}", flush=True)
    rc = run_once()

if rc != 0:
    print("[smart] clearing XTTS cache & final retry...", flush=True)
    clear_xtts_cache()
    print("[smart] run #final", flush=True)
    rc = run_once()

if rc == 0:
    print("[smart] success", flush=True)
else:
    print("[smart] failed with rc=", rc, file=sys.stderr)
    sys.exit(1)
"""
open("/content/smart_run.py","w").write(smart_code)

# 9) 実行(ログを表示)
print("[info] Running in isolated virtualenv ...")
res = subprocess.run([f"{VENV}/bin/python","-u","/content/smart_run.py"], env=env, capture_output=True, text=True)
print(res.stdout)
if res.returncode != 0:
    print(res.stderr, file=sys.stderr)
else:
    print("[done] All finished.")

いいなと思ったら応援しよう!

だいあろごす。 よろしければ応援お願いします!