Blame for audio_processor.py

# =============================================================
# audio_processor.py  —  LAS GPU専用版 v3.2
# =============================================================
import os, io, threading, time, logging
import numpy as np
import soundfile as sf
from pathlib import Path

logger = logging.getLogger(__name__)

try:
    import torchaudio
    TORCHAUDIO_OK = True
except ImportError:
    TORCHAUDIO_OK = False

LIBROSA_OK = False

try:
    import torch
    TORCH_OK = True
    _force_gpu = os.environ.get("FORCE_GPU","0").strip() in ("1","true","True","yes")
    DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
    if DEVICE == "cuda":
        torch.backends.cudnn.benchmark        = True
        torch.backends.cudnn.deterministic    = False
        torch.backends.cuda.matmul.allow_tf32 = True
        torch.backends.cudnn.allow_tf32       = True
        _props  = torch.cuda.get_device_properties(0)
        _vram   = _props.total_memory // 1024**2
        GPU_VRAM_MB = _vram
        logger.info(f"[GPU] {_props.name} | VRAM {_vram} MB")
    else:
        GPU_VRAM_MB = 0
        logger.warning("[GPU] ⚠️  CUDA利用不可 — CPU動作")
        if _force_gpu:
            raise RuntimeError("FORCE_GPU=1 だが CUDA が利用できません")
except ImportError:
    TORCH_OK = False; DEVICE = "cpu"; GPU_VRAM_MB = 0

try:
    import imageio_ffmpeg
    FFMPEG_EXE = imageio_ffmpeg.get_ffmpeg_exe()
except ImportError:
    FFMPEG_EXE = "ffmpeg"

# ─── Demucs設定 ───────────────────────────────────────────────
DEMUCS_MODELS   = {"lite":"htdemucs_ft","pro":"htdemucs_ft","ultra":"htdemucs_6s"}
DEMUCS_QUALITY  = {
    "lite":  {"shifts":1,"overlap":0.25,"num_workers":0,"segment":None},
    "pro":   {"shifts":2,"overlap":0.50,"num_workers":0,"segment":None},
    "ultra": {"shifts":4,"overlap":0.75,"num_workers":0,"segment":None},
}
STEM_NAMES      = {
    "lite":  ["vocals","no_vocals"],
    "pro":   ["vocals","drums","bass","other"],
    "ultra": ["vocals","drums","bass","piano","guitar","other"],
}
STEM_LABELS_JP  = {
    "vocals":"ボーカル","no_vocals":"インスト",
    "drums":"ドラム","bass":"ベース",
    "piano":"ピアノ","guitar":"ギター","other":"その他",
}
STEM_COLORS     = {
    "vocals":"#c0c0c0","no_vocals":"#8080ff","drums":"#4040ff",
    "bass":"#ff4080","piano":"#40ffff","guitar":"#ffff40","other":"#ff8040",
}

_MODEL_CACHE: dict = {}
_MODEL_LOCK = threading.Lock()

# GPU推定時間（秒）: プログレスシミュレーション用
_INFER_EST_GPU = {"lite":14,"pro":28,"ultra":55}
_INFER_EST_CPU = {"lite":50,"pro":90,"ultra":180}


def _get_demucs_model(model_name: str):
    with _MODEL_LOCK:
        if model_name not in _MODEL_CACHE:
            from demucs.pretrained import get_model
            logger.info(f"[DEMUCS] ロード: {model_name} → {DEVICE}")
            model = get_model(model_name)
            model.to(DEVICE); model.eval()
            _MODEL_CACHE[model_name] = model
    return _MODEL_CACHE[model_name]


def preload_models():
    seen = set()
    for name in DEMUCS_MODELS.values():
        if name not in seen:
            seen.add(name)
            try: _get_demucs_model(name)
            except Exception as e: logger.warning(f"[PRELOAD] {name} 失敗: {e}")


def _load_audio_gpu(filepath, target_sr, target_ch):
    if TORCHAUDIO_OK:
        try:
            wav, sr = torchaudio.load(filepath)
            if sr != target_sr:
                wav = wav.to(DEVICE)
                wav = torchaudio.functional.resample(wav, sr, target_sr)
            else:
                wav = wav.to(DEVICE)
            if wav.shape[0] == 1 and target_ch == 2:
                wav = wav.expand(2,-1).contiguous()
            elif wav.shape[0] > target_ch:
                wav = wav[:target_ch]
            elif wav.shape[0] < target_ch:
                wav = wav.expand(target_ch,-1).contiguous()
            return wav.float()
        except Exception as e:
            logger.warning(f"[AUDIO] torchaudio失敗→soundfile: {e}")
    data, sr = sf.read(filepath, always_2d=True)
    wav = torch.from_numpy(data.T.astype(np.float32)).to(DEVICE)
    if sr != target_sr and TORCHAUDIO_OK:
        wav = torchaudio.functional.resample(wav, sr, target_sr)
    if wav.shape[0] == 1 and target_ch == 2:
        wav = wav.expand(2,-1).contiguous()
    elif wav.shape[0] > target_ch:
        wav = wav[:target_ch]
    return wav


class AudioProcessor:
    def get_audio_info(self, filepath):
        try:
            info = sf.info(filepath)
            return {"duration":float(info.duration),"sample_rate":int(info.samplerate),"channels":int(info.channels)}
        except Exception:
            if TORCHAUDIO_OK:
                info = torchaudio.info(filepath)
                return {"duration":float(info.num_frames/info.sample_rate),"sample_rate":int(info.sample_rate),"channels":int(info.num_channels)}
            raise

    def get_waveform_data(self, filepath, num_points=200):
        if TORCHAUDIO_OK:
            try:
                wav, sr = torchaudio.load(filepath)
                target_sr = 8000
                if sr != target_sr:
                    wav = torchaudio.functional.resample(wav, sr, target_sr)
                y = wav.mean(dim=0).to(DEVICE)
                max_samples = target_sr * 120
                if y.shape[0] > max_samples: y = y[:max_samples]
            except Exception:
                y = torch.zeros(num_points, device=DEVICE)
        else:
            data, sr = sf.read(filepath)
            if data.ndim > 1: data = data.mean(axis=1)
            y = torch.from_numpy(data[:8000*120].astype(np.float32)).to(DEVICE)
        n = len(y); fl = max(1, n // num_points)
        pad = (fl - n % fl) % fl
        y_p = torch.nn.functional.pad(y, (0, pad))
        rms = y_p.reshape(-1,fl).pow(2).mean(dim=1).sqrt()[:num_points].cpu().tolist()
        peak = max(rms) if rms and max(rms) > 0 else 1.0
        return [min(1.0, v/peak) for v in rms]


    def get_bpm_key(self, filepath):
        """BPM とキーを推定（torchaudio + 自前 autocorrelation）"""
        try:
            import numpy as np
            if TORCHAUDIO_OK:
                wav, sr = torchaudio.load(filepath)
                y = wav.mean(dim=0).numpy().astype(np.float32)
            else:
                y, sr = sf.read(filepath, dtype="float32")
                if y.ndim > 1: y = y.mean(axis=1)
            # 先頭60秒のみ使用
            y = y[:int(sr * 60)]

            # ─── BPM: onset strength + autocorrelation ───────────
            hop = 512
            frame_len = 2048
            # onset strength: energy differences
            frames = []
            for i in range(0, len(y) - frame_len, hop):
                seg = y[i:i+frame_len]
                frames.append(float(np.sqrt(np.mean(seg**2))))
            env = np.array(frames, dtype=np.float32)
            # diff → onset
            onset = np.maximum(0, np.diff(np.concatenate([[0], env])))
            # autocorrelation
            max_lag = int(sr / hop * 2.5)   # max ~2.5s = ~24bpm
            min_lag = int(sr / hop / 5.0)   # min 0.2s = ~300bpm
            if len(onset) < max_lag * 2:
                return {"bpm": None, "key": None}
            ac = np.correlate(onset, onset, mode="full")
            ac = ac[len(ac)//2:]  # positive lags only
            window = ac[min_lag:max_lag]
            if window.max() <= 0:
                return {"bpm": None, "key": None}
            peak_lag = int(np.argmax(window)) + min_lag
            bpm = round(60.0 / (peak_lag * hop / sr))
            # Octave correction
            for mult in [2, 0.5]:
                b2 = bpm * mult
                if 60 <= b2 <= 200:
                    bpm = round(b2)
            bpm = max(60, min(200, bpm))

            # ─── Key: chroma chromagram ───────────────────────────
            # Simple pitch-class profile from FFT
            N = min(len(y), sr * 30)
            fft = np.abs(np.fft.rfft(y[:N], n=sr))[:sr//2]
            freqs = np.fft.rfftfreq(sr, 1/sr)[:sr//2]
            chroma = np.zeros(12, dtype=np.float32)
            A4 = 440.0
            for i, freq in enumerate(freqs):
                if freq < 80 or freq > 2000: continue
                midi = 12 * np.log2(freq / A4) + 69
                pc = int(round(midi)) % 12
                if 0 <= pc < 12:
                    chroma[pc] += fft[i]
            chroma /= (chroma.sum() + 1e-8)

            # Krumhansl-Schmuckler key profiles
            major = np.array([6.35,2.23,3.48,2.33,4.38,4.09,2.52,5.19,2.39,3.66,2.29,2.88])
            minor = np.array([6.33,2.68,3.52,5.38,2.60,3.53,2.54,4.75,3.98,2.69,3.34,3.17])
            keys = ['C','C#','D','D#','E','F','F#','G','G#','A','A#','B']
            best_score, best_key = -999, "C"
            for i in range(12):
                rot_chroma = np.roll(chroma, -i)
                sm = float(np.dot(rot_chroma, major))
                if sm > best_score: best_score, best_key = sm, keys[i] + " メジャー"
                sm = float(np.dot(rot_chroma, minor))
                if sm > best_score: best_score, best_key = sm, keys[i] + " マイナー"
            return {"bpm": bpm, "key": best_key}
        except Exception as e:
            logger.debug(f"[BPM] 検出失敗: {e}")
            return {"bpm": None, "key": None}

    # =========================================================
    # メインパイプライン
    # =========================================================
    def process_full_pipeline(self, filepath, plan, effect, output_format, output_dir, progress_callback=None,
                               noise_reduce=False, pitch_semitones=0.0, tempo_rate=1.0, stem_volumes=None):
        """
        分離 → ノイズ除去 → エフェクト → ピッチ/テンポ → 書き出し
        stem_volumes: {"vocals":80, "drums":120, ...}  各ステムの音量(%)
        """
        if progress_callback: progress_callback(3)
        stem_tensors, sr = self._separate_to_tensors(filepath, plan, progress_callback)
        if progress_callback: progress_callback(75)

        # ── ノイズ除去（テンソルレベル） ──
        if noise_reduce:
            try:
                stem_tensors = self._noise_reduce_tensors(stem_tensors, sr)
                logger.info("[PIPE] ノイズ除去 適用")
            except Exception as e:
                logger.warning(f"[PIPE] ノイズ除去失敗、スキップ: {e}")

        # ── エフェクト適用 ──
        if effect not in ("none", None, ""):
            logger.info(f"[PIPE] エフェクト: {effect}")
            stem_tensors = self._apply_effects_on_tensors(stem_tensors, effect, sr)
        if progress_callback: progress_callback(85)

        # ── ピッチシフト ──
        if pitch_semitones and abs(float(pitch_semitones)) > 0.01:
            try:
                stem_tensors = self._pitch_shift_tensors(stem_tensors, sr, float(pitch_semitones))
                logger.info(f"[PIPE] ピッチシフト {pitch_semitones:+.1f}半音 適用")
            except Exception as e:
                logger.warning(f"[PIPE] ピッチシフト失敗、スキップ: {e}")

        # ── テンポ変更 ──
        if tempo_rate and abs(float(tempo_rate) - 1.0) > 0.01:
            try:
                stem_tensors = self._tempo_change_tensors(stem_tensors, sr, float(tempo_rate))
                logger.info(f"[PIPE] テンポ変更 ×{tempo_rate:.2f} 適用")
            except Exception as e:
                logger.warning(f"[PIPE] テンポ変更失敗、スキップ: {e}")

        # ── ステム別音量調整 ──
        if stem_volumes:
            stem_tensors = self._apply_stem_volumes(stem_tensors, stem_volumes)

        if progress_callback: progress_callback(90)
        final = self._write_final_output(stem_tensors, sr, output_format, output_dir, progress_callback)
        del stem_tensors
        if DEVICE == "cuda": torch.cuda.empty_cache()
        if progress_callback: progress_callback(100)
        return final

    # ── ノイズ除去（スペクトル減算） ──────────────────────────────
    def _noise_reduce_tensors(self, stem_tensors, sr):
        """先頭0.3秒をノイズプロファイルとしてスペクトル減算"""
        import numpy as np
        result = {}
        for key, t in stem_tensors.items():
            arr = t.cpu().float().numpy()  # (C, L)
            if arr.shape[-1] < sr:         # 短すぎる場合はスキップ
                result[key] = t; continue
            n_fft, hop = 1024, 256
            profile_len = int(sr * 0.3)
            reduced_channels = []
            for ch in range(arr.shape[0]):
                x = arr[ch]
                # STFT
                win = np.hanning(n_fft)
                frames = np.lib.stride_tricks.sliding_window_view(
                    np.pad(x, (n_fft//2, n_fft//2)), n_fft)[::hop]
                S = np.fft.rfft(frames * win, axis=-1)
                mag, phase = np.abs(S), np.angle(S)
                # ノイズプロファイル（先頭フレームの平均）
                noise_frames = int(profile_len / hop) + 1
                noise_mag = mag[:noise_frames].mean(axis=0, keepdims=True)
                # スペクトル減算（Wiener風）
                reduced = np.maximum(mag - 1.2 * noise_mag, 0.08 * mag)
                # ISTFT
                y_frames = np.fft.irfft(reduced * np.exp(1j * phase), axis=-1) * win
                y_out = np.zeros(len(x) + n_fft)
                for i, frame in enumerate(y_frames):
                    y_out[i*hop:i*hop+n_fft] += frame
                y_out = y_out[n_fft//2:n_fft//2+len(x)]
                reduced_channels.append(y_out)
            arr_new = np.stack(reduced_channels)
            result[key] = torch.from_numpy(arr_new).to(t.device)
        return result

    # ── ピッチシフト（位相ボコーダ） ──────────────────────────────
    def _pitch_shift_tensors(self, stem_tensors, sr, semitones):
        semitones = max(-12.0, min(12.0, semitones))
        try:
            import librosa, numpy as np
            result = {}
            for key, t in stem_tensors.items():
                arr = t.cpu().float().numpy()
                shifted = np.stack([
                    librosa.effects.pitch_shift(arr[ch], sr=sr, n_steps=semitones)
                    for ch in range(arr.shape[0])
                ])
                result[key] = torch.from_numpy(shifted).to(t.device)
            return result
        except ImportError:
            logger.warning("[PIPE] librosaなし、ピッチシフトスキップ")
            return stem_tensors

    # ── テンポ変更（タイムストレッチ） ────────────────────────────
    def _tempo_change_tensors(self, stem_tensors, sr, rate):
        rate = max(0.5, min(2.0, rate))
        try:
            import librosa, numpy as np
            result = {}
            for key, t in stem_tensors.items():
                arr = t.cpu().float().numpy()
                stretched = np.stack([
                    librosa.effects.time_stretch(arr[ch], rate=rate)
                    for ch in range(arr.shape[0])
                ])
                result[key] = torch.from_numpy(stretched).to(t.device)
            return result
        except ImportError:
            logger.warning("[PIPE] librosaなし、テンポ変更スキップ")
            return stem_tensors

    # ── ステム別音量 ──────────────────────────────────────────────
    def _apply_stem_volumes(self, stem_tensors, volumes):
        """volumes: {"vocals": 80, "drums": 120}  単位:%"""
        result = {}
        for key, t in stem_tensors.items():
            gain = float(volumes.get(key, 100)) / 100.0
            result[key] = (t * gain).clamp(-1.0, 1.0)
        return result

    # ─── Demucs推論（スムーズ進捗シミュレーション付き） ────────
    def _separate_to_tensors(self, filepath, plan, progress_callback=None):
        try:
            from demucs.apply import apply_model
        except ImportError:
            logger.warning("[DEMUCS] Python API なし → subprocess")
            paths = self._subprocess_separate(filepath, plan, None, progress_callback)
            tensors, sr_used = {}, 44100
            for k, p in paths.items():
                d, sr_used = sf.read(p, always_2d=True)
                tensors[k] = torch.from_numpy(d.T.astype(np.float32)).to(DEVICE)
            return tensors, sr_used

        quality    = DEMUCS_QUALITY[plan]
        model_name = DEMUCS_MODELS[plan]
        model      = _get_demucs_model(model_name)
        target_sr  = model.samplerate if hasattr(model,"samplerate") else 44100
        target_ch  = model.audio_channels if hasattr(model,"audio_channels") else 2

        wav = _load_audio_gpu(filepath, target_sr, target_ch)
        mix = wav.unsqueeze(0)

        if progress_callback: progress_callback(10)

        # ── 推論中にプログレスを 10→80 でシミュレーション ──
        _done_evt  = threading.Event()
        _est_secs  = (_INFER_EST_GPU if DEVICE=="cuda" else _INFER_EST_CPU).get(plan, 30)

        def _sim():
            t0 = time.time()
            while not _done_evt.is_set():
                elapsed = time.time() - t0
                pct = 10 + int(min(70, (elapsed / _est_secs) * 70))
                if progress_callback:
                    try: progress_callback(pct)
                    except Exception: pass
                time.sleep(0.25)

        sim_t = None
        if progress_callback:
            sim_t = threading.Thread(target=_sim, daemon=True)
            sim_t.start()

        try:
            with torch.inference_mode():
                sources = apply_model(
                    model, mix,
                    segment=quality["segment"], shifts=quality["shifts"],
                    overlap=quality["overlap"], num_workers=quality["num_workers"],
                    progress=False,
                )
        finally:
            _done_evt.set()
            if sim_t: sim_t.join(timeout=1.0)

        if progress_callback: progress_callback(80)

        sources      = sources[0]
        source_names = list(model.sources)
        stem_tensors = {}
        if plan == "lite":
            vi = source_names.index("vocals")
            nv = sum(sources[i] for i in range(len(source_names)) if i != vi)
            stem_tensors["vocals"]    = sources[vi].float()
            stem_tensors["no_vocals"] = nv.float()
        else:
            for i, name in enumerate(source_names):
                stem_tensors[name] = sources[i].float()

        del sources, wav, mix
        return stem_tensors, target_sr

    # ─── GPUバッチエフェクト ──────────────────────────────────
    # ─────────────────────────────────────────────────────────────
    #  ソフトリミッター / 正規化ユーティリティ
    # ─────────────────────────────────────────────────────────────
    @staticmethod
    def _soft_limit(x: "torch.Tensor", ceiling: float = 0.98) -> "torch.Tensor":
        """
        Soft-knee リミッター（tanh ベース）
        音割れの原因となるハードクリップを避け、滑らかに頭打ちさせる。
        ceiling: 最終ピーク値 (0〜1)
        """
        # tanh(-3)≈-0.9951, tanh(3)≈0.9951 → ±3 に正規化してから tanh
        peak = x.abs().amax().clamp(min=1e-8)
        if peak > ceiling:
            # 超えた分だけ tanh で圧縮
            x = torch.tanh(x / peak * 3.0) * (ceiling / 0.9951)
        return x

    @staticmethod
    def _loudness_norm(x: "torch.Tensor", target_rms: float = 0.20) -> "torch.Tensor":
        """
        RMS ベースの音量正規化（ラウドネス維持）
        target_rms: 目標 RMS 振幅 (0〜1)。0.20 ≈ -14 dBFS
        """
        rms  = x.pow(2).mean().sqrt().clamp(min=1e-8)
        gain = target_rms / rms
        # 過剰なゲインアップは禁止（+6 dB まで）
        gain = gain.clamp(max=2.0)
        return x * gain

    def _apply_effects_on_tensors(self, stem_tensors, effect, sr):
        if not TORCH_OK or DEVICE != "cuda":
            return self._apply_effects_cpu_tensors(stem_tensors, effect, sr)
        keys     = list(stem_tensors.keys())
        tensors  = [stem_tensors[k] for k in keys]
        max_len  = max(t.shape[-1] for t in tensors)
        orig_lens= [t.shape[-1] for t in tensors]

        # ★ float32 で処理（float16 は中間値オーバーフローの原因）
        batch = torch.zeros(len(keys), 2, max_len, dtype=torch.float32, device=DEVICE)
        for i, t in enumerate(tensors):
            src = t.float().to(DEVICE)
            if src.shape[0] == 1: src = src.expand(2,-1)
            batch[i,:,:src.shape[-1]] = src

        # 入力ピーク記録（後でラウドネスを揃えるため）
        in_rms = batch.pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)

        if   effect == "3d":        batch = self._bfx_3d(batch, sr)
        elif effect == "8d":        batch = self._bfx_rotate(batch, sr, 0.10)
        elif effect == "16d":       batch = self._bfx_rotate(batch, sr, 0.045)
        elif effect == "cathedral": batch = self._bfx_reverb(batch, sr, 3.5, 0.35)
        elif effect == "lofi":      batch = self._bfx_lofi(batch, sr)
        elif effect == "orbit":     batch = self._bfx_orbit(batch, sr)
        elif effect == "spatial":   batch = self._bfx_spatial(batch, sr)
        elif effect == "hall":      batch = self._bfx_hall(batch, sr)
        elif effect == "echo":      batch = self._bfx_echo(batch, sr)
        elif effect == "vinyl":     batch = self._bfx_vinyl(batch, sr)
        elif effect == "tunnel":    batch = self._bfx_tunnel(batch, sr)

        # ★ ラウドネス整合 — エフェクト後の RMS を入力と揃える
        out_rms = batch.pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)
        gain    = (in_rms / out_rms).clamp(min=0.1, max=4.0)
        batch   = batch * gain

        # ★ ソフトリミッター（tanh） → 最終ハードシーリング -0.5 dBTP
        CEILING = 0.944   # -0.5 dBTP
        peak = batch.abs().amax(dim=(1,2), keepdim=True).clamp(min=1e-8)
        over = (peak > CEILING)
        if over.any():
            ratio        = torch.where(over, CEILING / peak, torch.ones_like(peak))
            batch_scaled = batch * ratio
            # tanh ソフトニー: 超えた部分のみ圧縮
            batch = torch.where(
                over.expand_as(batch),
                torch.tanh(batch_scaled / CEILING * 2.0) * (CEILING / torch.tanh(torch.tensor(2.0, device=DEVICE))),
                batch,
            )
        # 最終クリップ保護
        batch = batch.clamp(-CEILING, CEILING)

        result = {}
        for i, key in enumerate(keys):
            result[key] = batch[i,:,:orig_lens[i]]
        del batch
        return result

    # ─── 最終出力（書き込み進捗95→99） ──────────────────────
    def _write_final_output(self, stem_tensors, sr, output_format, output_dir, progress_callback=None):
        os.makedirs(output_dir, exist_ok=True)
        final_stems  = {}
        keys         = list(stem_tensors.keys())
        total        = len(keys)

        for idx, (key, tensor) in enumerate(stem_tensors.items()):
            arr      = tensor.cpu().float().numpy().T
            out_path = os.path.join(output_dir, f"{key}.{output_format}")

            if output_format == "wav":
                sf.write(out_path, arr, sr, subtype="PCM_16")
            elif output_format in ("mid","midi"):
                tmp = out_path + "_tmp.wav"
                sf.write(tmp, arr, sr, subtype="PCM_16")
                self.export_midi(tmp, out_path)
                try: os.remove(tmp)
                except OSError: pass
            else:
                buf = io.BytesIO()
                sf.write(buf, arr, sr, format="WAV", subtype="PCM_16")
                buf.seek(0)
                try:
                    import subprocess as _sp
                    qa  = {"mp3":["-q:a","2"],"flac":["-compression_level","5"]}
                    cmd = [FFMPEG_EXE,"-y","-f","wav","-i","pipe:0"
                           ] + qa.get(output_format,[]) + ["-f",output_format,"pipe:1","-loglevel","error"]
                    proc = _sp.run(cmd, input=buf.read(), capture_output=True, timeout=120)
                    if proc.returncode == 0 and proc.stdout:
                        with open(out_path,"wb") as f: f.write(proc.stdout)
                    else:
                        logger.warning(f"[FFMPEG] パイプ失敗→WAV")
                        out_path = out_path.rsplit(".",1)[0] + ".wav"
                        sf.write(out_path, arr, sr, subtype="PCM_16")
                except Exception as e:
                    logger.warning(f"[FFMPEG] エラー→WAV: {e}")
                    out_path = out_path.rsplit(".",1)[0] + ".wav"
                    sf.write(out_path, arr, sr, subtype="PCM_16")

            final_stems[key] = out_path
            if progress_callback:
                pct = 90 + int((idx+1)/total * 9)
                progress_callback(pct)

        return final_stems

    # ─── GPU エフェクト実装 ───────────────────────────────────
    def _bfx_3d(self, b, sr):
        d = int(sr * 0.00035); r = b.clone()
        r[:,1,d:] = b[:,1,:-d]; r[:,1,:d] = 0; return r

    def _bfx_rotate(self, b, sr, hz):
        n = b.shape[2]
        t = torch.linspace(0, n/sr, n, device=b.device, dtype=torch.float32)
        pan = ((torch.sin(2*torch.pi*hz*t)+1)*0.5).to(b.dtype)
        r = b.clone(); r[:,0] = b[:,0]*(1-pan); r[:,1] = b[:,1]*pan; return r

    def _bfx_reverb(self, b, sr, decay, wet):
        """
        高品質リバーブ。
        - wet 信号を入力と同エネルギーに RMS 正規化してから wet/dry 合成
          → エネルギーが爆発せず音割れ・ノイズを完全防止
        - ceiling=0.92 でさらに安全マージン確保
        """
        N, C, S = b.shape; ir_len = int(sr * decay)
        torch.manual_seed(42)
        ir = torch.randn(ir_len, device=b.device, dtype=torch.float32)
        env = torch.exp(-torch.arange(ir_len, device=b.device, dtype=torch.float32) * 6.9 / ir_len)
        ir  = ir * env
        ir[0] = 1.0
        # IR を L2 正規化（畳み込み後のエネルギーを制御）
        ir_l2 = ir.pow(2).sum().sqrt().clamp(min=1e-8)
        ir     = ir / ir_l2   # ||ir||_2 = 1 → 出力 RMS ≈ 入力 RMS

        n  = S + ir_len - 1
        bf = b.float().reshape(N * C, S)
        Y  = torch.fft.rfft(bf, n=n)
        IR = torch.fft.rfft(ir, n=n)
        c  = torch.fft.irfft(Y * IR, n=n)[:, :S].reshape(N, C, S)

        # ★ wet 信号のエネルギーを dry と揃えて安全に混合
        dry_rms = b.float().pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)
        wet_rms = c.pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)
        c_norm  = c * (dry_rms / wet_rms)   # wet 信号を dry と同エネルギーに

        mixed = (1.0 - wet) * b.float() + wet * c_norm
        # 安全クリップ
        return mixed.clamp(-0.92, 0.92)

    def _bfx_lofi(self, b, sr):
        N,C,S = b.shape
        if TORCHAUDIO_OK:
            bf = b.float().reshape(N*C,S)
            lpf = torchaudio.functional.lowpass_biquad(bf, sample_rate=sr, cutoff_freq=8000.0).reshape(N,C,S)
        else:
            lpf = self._gpu_fir(b.float(), 8000.0/sr)
        torch.manual_seed(0)
        return (lpf + torch.randn_like(lpf)*0.004).to(b.dtype)

    def _gpu_fir(self, b, cutoff):
        N,C,S = b.shape; taps = 65
        t = torch.arange(-(taps//2), taps//2+1, device=b.device, dtype=torch.float32)
        h = torch.where(t==0, torch.tensor(2*cutoff, device=b.device),
                        torch.sin(2*torch.pi*cutoff*t)/(torch.pi*t))
        h *= torch.hann_window(taps, device=b.device); h /= h.sum()
        out = torch.nn.functional.conv1d(b.reshape(N*C,1,S), h.view(1,1,-1), padding=taps//2)
        return out.reshape(N,C,S)

    def _bfx_orbit(self, b, sr):
        n = b.shape[2]
        t = torch.linspace(0, n/sr, n, device=b.device, dtype=torch.float32)
        pan = ((torch.sin(2*torch.pi*0.08*t)*torch.cos(2*torch.pi*0.03*t)+1)*0.5).to(b.dtype)
        r = b.clone(); r[:,0] = b[:,0]*(1-pan); r[:,1] = b[:,1]*pan; return r

    def _bfx_spatial(self, b, sr):
        return self._bfx_rotate(self._bfx_reverb(b, sr, 2.0, 0.25), sr, 0.05)

    # ── Ultra専用エフェクト ────────────────────────────────────

    def _bfx_hall(self, b, sr):
        """コンサートホール: 適度なリバーブ + 高音ロールオフ（クリップなし）"""
        reverbed = self._bfx_reverb(b, sr, 3.8, 0.32)
        if TORCHAUDIO_OK:
            N, C, S = reverbed.shape
            bf       = reverbed.float().reshape(N * C, S)
            filtered = torchaudio.functional.lowpass_biquad(bf, sample_rate=sr, cutoff_freq=14000.0).reshape(N, C, S)
            return filtered.clamp(-0.92, 0.92)
        return reverbed

    def _bfx_echo(self, b, sr):
        """エコーチェンバー: 280ms遅延 + フィードバック（エネルギー厳密制御）"""
        N, C, S = b.shape
        delay    = int(sr * 0.28)   # 280ms
        fb       = 0.32             # フィードバック量（抑えめ）
        r = b.float().clone()
        echo_energy = 0.0
        for i in range(1, 6):
            d = delay * i
            if d >= S: break
            coef = fb ** i
            echo_energy += coef ** 2
            r[:, :, d:] = r[:, :, d:] + b.float()[:, :, :S - d] * coef
        # エネルギー正規化（累積エコーで音量が上がりすぎないよう）
        norm = (1.0 + echo_energy) ** 0.5
        return (r / norm).clamp(-0.92, 0.92)

    def _bfx_vinyl(self, b, sr):
        """ビニールレコード: ローファイ + クラックル + ワウフラッター"""
        lofi = self._bfx_lofi(b, sr)
        N,C,S = lofi.shape
        # ワウフラッター（振幅変調で近似）
        t   = torch.linspace(0, S/sr, S, device=b.device, dtype=torch.float32)
        wow = (1.0 + 0.022 * torch.sin(2*torch.pi*0.5*t)).to(b.dtype)
        lofi = lofi * wow.view(1,1,-1)
        # ヴィンテージノイズ（クラックル、控えめ）
        torch.manual_seed(7)
        crack = (torch.rand(N,1,S, device=b.device, dtype=b.dtype) < 0.0002).float() * 0.12
        result = lofi + crack
        # エネルギー正規化してクリップ防止
        dry_rms    = b.float().pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)
        result_rms = result.float().pow(2).mean(dim=(1,2), keepdim=True).sqrt().clamp(min=1e-8)
        result = result * (dry_rms / result_rms)
        return result.clamp(-0.92, 0.92)

    def _bfx_tunnel(self, b, sr):
        """トンネルリバーブ: タイトなリバーブ + フラッターエコー（クリップ制御）"""
        tight = self._bfx_reverb(b, sr, 0.8, 0.28)
        N, C, S = b.shape
        # 60ms ショートエコー×3 — 係数を抑えてエネルギー制御
        echo_energy = 0.0
        for d_ms, coef in [(60, 0.18), (120, 0.10), (180, 0.06)]:
            d = int(sr * d_ms / 1000)
            if d < S:
                tight[:, :, d:] = tight[:, :, d:] + b.float()[:, :, :S - d] * coef
                echo_energy += coef ** 2
        # ステレオ広がり（L/R微小ディレイ）
        spread = int(sr * 0.0015)
        r = tight.clone()
        if spread < S:
            r[:, 0, spread:] = (tight[:, 0, spread:] + tight[:, 1, :-spread]) / 2
        # エネルギー正規化
        norm = (1.0 + echo_energy) ** 0.5
        return (r / norm).clamp(-0.92, 0.92)

    # ─── CPUフォールバック ─────────────────────────────────────
    def _apply_effects_cpu_tensors(self, stem_tensors, effect, sr):
        result = {}
        for k, t in stem_tensors.items():
            y = t.cpu().float().numpy()
            if y.shape[0] == 1: y = np.vstack([y, y])
            in_rms  = float(np.sqrt(np.mean(y ** 2)) + 1e-8)
            y       = self._cpu_effect(y, sr, effect)
            out_rms = float(np.sqrt(np.mean(y ** 2)) + 1e-8)
            # ラウドネス整合
            gain = min(in_rms / out_rms, 4.0)
            y    = y * gain
            # tanh ソフトリミッター
            CEIL = 0.944
            peak = float(np.abs(y).max() + 1e-8)
            if peak > CEIL:
                y = np.tanh(y / peak * 2.0) * (CEIL / np.tanh(np.float32(2.0)))
            y = np.clip(y, -CEIL, CEIL)
            result[k] = torch.from_numpy(y)
        return result

    def _cpu_effect(self, y, sr, effect):
        if effect == "3d":
            d = int(sr*0.00035); y[1] = np.pad(y[1],(d,0))[:y.shape[1]]
        elif effect in ("8d","16d"):
            hz = 0.10 if effect=="8d" else 0.045
            t  = np.linspace(0,y.shape[1]/sr,y.shape[1],dtype=np.float32)
            p  = (np.sin(2*np.pi*hz*t)+1)*0.5
            y[0] *= (1-p); y[1] *= p
        elif effect == "cathedral":
            y = self._cpu_reverb(y, sr, 3.5, 0.40)
        elif effect == "lofi":
            try:
                from scipy.signal import butter, sosfilt
                sos = butter(4,8000,fs=sr,output="sos")
                for ch in range(y.shape[0]): y[ch] = sosfilt(sos,y[ch]).astype(np.float32)
            except ImportError: pass
            y += np.random.default_rng(0).standard_normal(y.shape).astype(np.float32)*0.004
        elif effect == "orbit":
            t = np.linspace(0,y.shape[1]/sr,y.shape[1],dtype=np.float32)
            p = (np.sin(2*np.pi*0.08*t)*np.cos(2*np.pi*0.03*t)+1)*0.5
            y[0] *= (1-p); y[1] *= p
        elif effect in ("spatial","hall"):
            y = self._cpu_reverb(y, sr, 4.5 if effect=="hall" else 2.0, 0.45 if effect=="hall" else 0.25)
            t = np.linspace(0,y.shape[1]/sr,y.shape[1],dtype=np.float32)
            p = (np.sin(2*np.pi*0.05*t)+1)*0.5
            y[0] *= (1-p); y[1] *= p
        elif effect == "echo":
            d = int(sr*0.25)
            for i in range(1,5):
                dd = d*i
                if dd < y.shape[1]:
                    y[:,dd:] += y[:,:y.shape[1]-dd] * (0.5**i)
        elif effect == "vinyl":
            try:
                from scipy.signal import butter, sosfilt
                sos = butter(4,8000,fs=sr,output="sos")
                for ch in range(y.shape[0]): y[ch] = sosfilt(sos,y[ch]).astype(np.float32)
            except ImportError: pass
            t = np.linspace(0,y.shape[1]/sr,y.shape[1],dtype=np.float32)
            y *= (1.0 + 0.025*np.sin(2*np.pi*0.5*t)).astype(np.float32)
        elif effect == "tunnel":
            y = self._cpu_reverb(y, sr, 0.8, 0.35)
        return y  # ソフトリミッターは呼び出し元で適用

    def _cpu_reverb(self, y, sr, decay, wet):
        try:
            from scipy.signal import fftconvolve
        except ImportError:
            return y
        rng = np.random.default_rng(42)
        ir  = rng.standard_normal(int(sr * decay)).astype(np.float32)
        env = np.exp(-np.arange(len(ir)) * 6.9 / len(ir)).astype(np.float32)
        ir  = ir * env
        ir[0] = 1.0
        # L2 正規化（エネルギー保存型）
        ir_l2 = float(np.sqrt(np.sum(ir ** 2)) + 1e-8)
        ir     = ir / ir_l2
        r = np.zeros_like(y)
        for ch in range(y.shape[0]):
            c = fftconvolve(y[ch], ir)[:y.shape[1]].astype(np.float32)
            # wet 信号のエネルギーを dry と揃える
            dry_rms = float(np.sqrt(np.mean(y[ch] ** 2)) + 1e-8)
            wet_rms = float(np.sqrt(np.mean(c ** 2)) + 1e-8)
            c_norm  = c * (dry_rms / wet_rms)
            r[ch]   = np.clip((1 - wet) * y[ch] + wet * c_norm, -0.92, 0.92)
        return r

    # ─── 後方互換 ─────────────────────────────────────────────
    def separate_audio(self, filepath, plan, output_dir, progress_callback=None):
        stem_tensors, sr = self._separate_to_tensors(filepath, plan, progress_callback)
        model_name = DEMUCS_MODELS[plan]
        stem_dir   = Path(output_dir) / model_name / Path(filepath).stem
        stem_dir.mkdir(parents=True, exist_ok=True)
        stems = {}
        for k, t in stem_tensors.items():
            p = str(stem_dir / f"{k}.wav")
            sf.write(p, t.cpu().float().numpy().T, sr, subtype="PCM_16")
            stems[k] = p
        del stem_tensors
        if DEVICE == "cuda": torch.cuda.empty_cache()
        return stems

    def apply_effect_batch(self, stems_wav, effect):
        keys, sr_ref = list(stems_wav.keys()), 44100
        tensors = {}
        for k in keys:
            d, sr_ref = sf.read(stems_wav[k], always_2d=True)
            y = d.T.astype(np.float32)
            if y.shape[0] == 1: y = np.vstack([y,y])
            tensors[k] = torch.from_numpy(y)
        out_t = self._apply_effects_on_tensors(tensors, effect, sr_ref)
        result = {}
        for k, t in out_t.items():
            p = str(Path(stems_wav[k]).with_suffix("")) + f"_{effect}.wav"
            sf.write(p, t.cpu().float().numpy().T, sr_ref, subtype="PCM_16")
            result[k] = p
        return result

    def convert_format(self, src, dst, fmt):
        import subprocess as _sp, shutil
        try:
            qa  = {"mp3":["-q:a","2"],"flac":["-compression_level","5"],"wav":[]}
            cmd = [FFMPEG_EXE,"-y","-i",src]+qa.get(fmt,[])+[dst,"-loglevel","error"]
            _sp.run(cmd, check=True, capture_output=True, text=True)
        except (_sp.CalledProcessError, FileNotFoundError) as e:
            logger.warning(f"[FFMPEG] 失敗: {e}"); shutil.copy(src, dst)
        return dst

    def _subprocess_separate(self, filepath, plan, output_dir, progress_callback=None):
        import subprocess, re as _re
        if output_dir is None: output_dir = str(Path(filepath).parent / "demucs_out")
        os.makedirs(output_dir, exist_ok=True)
        model = DEMUCS_MODELS[plan]; quality = DEMUCS_QUALITY[plan]
        file_id = Path(filepath).stem
        cmd = ["python","-m","demucs"]
        if plan == "lite": cmd += ["--two-stems","vocals"]
        cmd += ["-n",model,"--device",DEVICE,
                "--shifts",str(quality["shifts"]),
                "--overlap",str(quality["overlap"]),
                "--out",output_dir,filepath]
        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                             text=True, bufsize=1, env={**os.environ,"PYTHONUNBUFFERED":"1"})
        lines = []
        def _err():
            for line in p.stderr:
                lines.append(line.rstrip())
                m = _re.search(r"(\d+)%",line)
                if m and progress_callback: progress_callback(int(int(m.group(1))*0.75))
        t = threading.Thread(target=_err, daemon=True)
        t.start()
        for line in p.stdout:
            m = _re.search(r"(\d+)%",line)
            if m and progress_callback: progress_callback(int(int(m.group(1))*0.75))
        p.wait(); t.join(timeout=5)
        if p.returncode != 0:
            raise RuntimeError(f"demucs失敗({p.returncode})\n{''.join(lines[-20:])}")
        stem_dir = Path(output_dir) / model / file_id
        stems = {}
        for key in (["vocals","no_vocals"] if plan=="lite" else STEM_NAMES[plan]):
            f = stem_dir / f"{key}.wav"
            if f.exists(): stems[key] = str(f)
        return stems

    def export_midi(self, wav_path, output_path):
        """
        音声をMIDIに変換する。
        mido が利用可能な場合はピッチ検出でMIDIを生成する。
        利用不可の場合は最小限の有効なMIDIファイルを出力する。
        """
        try:
            import librosa
            import numpy as np
            import struct

            y, sr = librosa.load(wav_path, sr=22050, mono=True, duration=120.0)

            # ピッチ検出
            f0, voiced_flag, _ = librosa.pyin(
                y, fmin=librosa.note_to_hz("C2"),
                fmax=librosa.note_to_hz("C7"),
                sr=sr, frame_length=2048, hop_length=512
            )

            # MIDI note number に変換
            notes_on = []
            prev_note = None
            prev_start = 0
            hop_dur = 512 / sr  # 秒/フレーム

            for i, (f, v) in enumerate(zip(f0, voiced_flag)):
                if v and f and not np.isnan(f):
                    note = int(round(69 + 12 * np.log2(max(f, 1) / 440)))
                    note = max(0, min(127, note))
                    t = i * hop_dur
                    if note != prev_note:
                        if prev_note is not None:
                            notes_on.append((prev_start, t, prev_note, 80))
                        prev_note = note
                        prev_start = t
                else:
                    if prev_note is not None:
                        notes_on.append((prev_start, i * hop_dur, prev_note, 80))
                    prev_note = None

            if prev_note is not None:
                notes_on.append((prev_start, len(f0) * hop_dur, prev_note, 80))

            # --- バイナリMIDI生成 ---
            TICKS_PER_BEAT = 480
            TEMPO          = 500000  # 120BPM
            SEC_PER_TICK   = TEMPO / 1_000_000 / TICKS_PER_BEAT

            def to_vlq(n):
                b = [n & 0x7F]
                n >>= 7
                while n:
                    b.insert(0, (n & 0x7F) | 0x80)
                    n >>= 7
                return bytes(b)

            track_events = []
            prev_tick = 0
            for (start, end, note, vel) in sorted(notes_on):
                tick_on  = int(start / SEC_PER_TICK)
                tick_off = int(end   / SEC_PER_TICK)
                dt_on  = tick_on  - prev_tick; prev_tick = tick_on
                dt_off = tick_off - tick_on
                track_events += to_vlq(max(0, dt_on)) + bytes([0x90, note, vel])
                track_events += to_vlq(max(0, dt_off)) + bytes([0x80, note, 0])

            # テンポイベント
            tempo_event = bytes([0x00, 0xFF, 0x51, 0x03]) + struct.pack(">I", TEMPO)[1:]
            # End of track
            eot = bytes([0x00, 0xFF, 0x2F, 0x00])

            track_data = tempo_event + bytes(track_events) + eot

            # ヘッダ + トラック
            header = struct.pack(">4sI3H", b"MThd", 6, 0, 1, TICKS_PER_BEAT)
            track  = b"MTrk" + struct.pack(">I", len(track_data)) + track_data

            with open(output_path, "wb") as f:
                f.write(header + track)

            logger.info(f"[MIDI] ピッチ検出MIDI生成完了: {len(notes_on)}ノート → {output_path}")
            return output_path

        except Exception as e:
            logger.warning(f"[MIDI] ピッチ検出失敗、最小MIDIを出力: {e}")
            # フォールバック: 有効な空MIDIファイル
            import struct
            TICKS  = 480
            TEMPO  = 500000
            tempo_evt = bytes([0x00, 0xFF, 0x51, 0x03]) + struct.pack(">I", TEMPO)[1:]
            eot       = bytes([0x00, 0xFF, 0x2F, 0x00])
            track_data = tempo_evt + eot
            header = struct.pack(">4sI3H", b"MThd", 6, 0, 1, TICKS)
            track  = b"MTrk" + struct.pack(">I", len(track_data)) + track_data
            with open(output_path, "wb") as f:
                f.write(header + track)
            return output_path