# ============================================================= # 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