Spaces:
Sleeping
Sleeping
| # ============================================================= | |
| # 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バッチエフェクト ────────────────────────────────── | |
| # ───────────────────────────────────────────────────────────── | |
| # ソフトリミッター / 正規化ユーティリティ | |
| # ───────────────────────────────────────────────────────────── | |
| 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 | |
| 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 | |