음성 & 오디오 AI 완전 정복: Whisper, TTS, 화자 인식, 음악 생성까지

import librosa
import librosa.display
import numpy as np
import matplotlib.pyplot as plt

# 오디오 파일 로드 (16kHz 모노)
y, sr = librosa.load("speech.wav", sr=16000, mono=True)
print(f"샘플 수: {len(y)}, 샘플 레이트: {sr}Hz, 길이: {len(y)/sr:.2f}초")

# STFT (Short-Time Fourier Transform)
n_fft = 512        # FFT 윈도우 크기
hop_length = 128   # 윈도우 이동 간격

D = librosa.stft(y, n_fft=n_fft, hop_length=hop_length, window="hann")
magnitude = np.abs(D)

# 파워 스펙트로그램 (dB 스케일)
S_db = librosa.amplitude_to_db(magnitude, ref=np.max)

fig, axes = plt.subplots(3, 1, figsize=(12, 10))

# 파형
axes[0].plot(np.linspace(0, len(y)/sr, len(y)), y)
axes[0].set_title("파형 (Waveform)")
axes[0].set_xlabel("시간 (초)")

# 선형 스펙트로그램
librosa.display.specshow(S_db, sr=sr, hop_length=hop_length,
                         x_axis="time", y_axis="linear", ax=axes[1])
axes[1].set_title("선형 스펙트로그램 (Linear Spectrogram)")

# Mel 스펙트로그램
mel_spec = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=80,
                                           n_fft=n_fft, hop_length=hop_length)
mel_db = librosa.power_to_db(mel_spec, ref=np.max)
librosa.display.specshow(mel_db, sr=sr, hop_length=hop_length,
                         x_axis="time", y_axis="mel", ax=axes[2])
axes[2].set_title("Mel 스펙트로그램 (80 Mel bins)")

plt.tight_layout()
plt.savefig("spectrogram_comparison.png", dpi=150)

import librosa
import numpy as np

def extract_mfcc_features(audio_path, sr=16000, n_mfcc=13, n_mels=40):
    """
    MFCC 특징 추출 함수
    Returns: (39, T) 형태의 mfcc + delta + delta2 특징
    """
    y, sr = librosa.load(audio_path, sr=sr)

    # 프리엠퍼시스
    y_emphasized = np.append(y[0], y[1:] - 0.97 * y[:-1])

    # MFCC 추출
    mfcc = librosa.feature.mfcc(
        y=y_emphasized, sr=sr,
        n_mfcc=n_mfcc,
        n_mels=n_mels,
        n_fft=512,
        hop_length=160,   # 10ms (16kHz 기준)
        win_length=400,   # 25ms
        window="hann"
    )

    # Delta (1차 미분) — 동적 특징
    delta = librosa.feature.delta(mfcc)
    # Delta-Delta (2차 미분)
    delta2 = librosa.feature.delta(mfcc, order=2)

    # 39차원 특징 벡터 (13 + 13 + 13)
    features = np.vstack([mfcc, delta, delta2])

    # CMVN 정규화
    features = (features - features.mean(axis=1, keepdims=True)) / \
               (features.std(axis=1, keepdims=True) + 1e-8)

    return features  # shape: (39, T)

features = extract_mfcc_features("speech.wav")
print(f"MFCC 특징 shape: {features.shape}")

import whisper
import json

def transcribe_with_timestamps(audio_path, model_size="large-v3", language="ko"):
    """
    Whisper로 타임스탬프 포함 전사
    모델 크기: tiny, base, small, medium, large, large-v3
    """
    model = whisper.load_model(model_size)

    result = model.transcribe(
        audio_path,
        language=language,
        task="transcribe",           # "translate"로 바꾸면 영어 번역
        word_timestamps=True,        # 단어별 타임스탬프
        condition_on_previous_text=True,
        temperature=0,
        beam_size=5,
        verbose=False
    )

    print(f"전체 전사: {result['text']}")
    print(f"감지된 언어: {result['language']}")

    for seg in result["segments"]:
        start = f"{seg['start']:.2f}s"
        end   = f"{seg['end']:.2f}s"
        text  = seg["text"].strip()
        print(f"[{start} -> {end}] {text}")

        if "words" in seg:
            for word in seg["words"]:
                ws = f"{word['start']:.2f}s"
                we = f"{word['end']:.2f}s"
                print(f"  {word['word']}: {ws}~{we}")

    return result

result = transcribe_with_timestamps("meeting.wav", model_size="large-v3", language="ko")

with open("transcript.json", "w", encoding="utf-8") as f:
    json.dump(result, f, ensure_ascii=False, indent=2)

from TTS.api import TTS
import torch

def synthesize_speech_ko(text, output_path, speed=1.0):
    """
    Coqui TTS를 사용한 한국어 음성 합성
    모델: tts_models/ko/css10/vits
    """
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tts = TTS("tts_models/ko/css10/vits").to(device)

    tts.tts_to_file(
        text=text,
        file_path=output_path,
        speed=speed,
    )
    print(f"합성 완료: {output_path}")

# 한국어 TTS 예시
text_ko = "안녕하세요. 음성 AI 기술이 놀랍도록 발전했습니다."
synthesize_speech_ko(text_ko, "output_ko.wav")

# 다국어 TTS (XTTS-v2)
def synthesize_multilang(text, language, speaker_wav, output_path):
    """
    XTTS-v2: 화자 음성 클로닝 + 다국어 합성
    language: "ko", "ja", "en", "zh-cn" 등
    speaker_wav: 3~6초 참조 음성
    """
    tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2")
    tts.tts_to_file(
        text=text,
        language=language,
        speaker_wav=speaker_wav,
        file_path=output_path,
    )

synthesize_multilang(
    "목소리를 복제하여 다국어로 말할 수 있습니다.",
    language="ko",
    speaker_wav="reference_voice.wav",
    output_path="cloned_ko.wav"
)

from pyannote.audio import Pipeline
import torch

def speaker_diarization(audio_path, hf_token, num_speakers=None):
    """
    화자 분리: 누가 언제 말했는지 파악
    """
    pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )

    if torch.cuda.is_available():
        pipeline = pipeline.to(torch.device("cuda"))

    params = {}
    if num_speakers:
        params["num_speakers"] = num_speakers
    else:
        params["min_speakers"] = 1
        params["max_speakers"] = 10

    diarization = pipeline(audio_path, **params)

    segments = []
    for turn, _, speaker in diarization.itertracks(yield_label=True):
        seg = {
            "start": round(turn.start, 3),
            "end": round(turn.end, 3),
            "speaker": speaker
        }
        segments.append(seg)
        print(f"[{seg['start']:.3f}s ~ {seg['end']:.3f}s] {speaker}")

    with open("diarization.rttm", "w") as rttm:
        diarization.write_rttm(rttm)

    speakers_found = len(set(s["speaker"] for s in segments))
    print(f"\n총 {speakers_found}명의 화자 감지")
    return segments

segments = speaker_diarization("meeting.wav", hf_token="YOUR_HF_TOKEN")

import whisper
from pyannote.audio import Pipeline
import torch

def diarize_and_transcribe(audio_path, hf_token, language="ko"):
    """화자 분리 + 음성 인식 결합"""
    # 1. 화자 분리
    diarization_pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )
    diarization = diarization_pipeline(audio_path)

    # 2. Whisper 전사
    asr_model = whisper.load_model("large-v3")
    asr_result = asr_model.transcribe(audio_path, language=language,
                                       word_timestamps=True)

    # 3. 단어 타임스탬프와 화자 정보 매핑
    words_with_speakers = []
    for seg in asr_result["segments"]:
        for word in seg.get("words", []):
            word_mid = (word["start"] + word["end"]) / 2
            speaker = "UNKNOWN"
            for turn, _, spk in diarization.itertracks(yield_label=True):
                if turn.start <= word_mid <= turn.end:
                    speaker = spk
                    break
            words_with_speakers.append({
                "word": word["word"],
                "start": word["start"],
                "end": word["end"],
                "speaker": speaker
            })

    # 4. 화자별 발화 그룹핑
    grouped = []
    current_speaker = None
    current_text = []

    for item in words_with_speakers:
        if item["speaker"] != current_speaker:
            if current_text:
                grouped.append({
                    "speaker": current_speaker,
                    "text": "".join(current_text).strip()
                })
            current_speaker = item["speaker"]
            current_text = [item["word"]]
        else:
            current_text.append(item["word"])

    if current_text:
        grouped.append({"speaker": current_speaker,
                        "text": "".join(current_text).strip()})

    for entry in grouped:
        print(f"[{entry['speaker']}]: {entry['text']}")

    return grouped

from audiocraft.models import MusicGen
from audiocraft.data.audio import audio_write
import torch

def generate_music(descriptions, duration=10, model_size="medium"):
    """
    텍스트 설명으로 음악 생성
    모델 크기: small (300M), medium (1.5B), large (3.3B), melody
    """
    model = MusicGen.get_pretrained(f"facebook/musicgen-{model_size}")
    model.set_generation_params(
        duration=duration,     # 초 단위 (최대 30초)
        temperature=1.0,       # 창의성 제어
        top_k=250,
        cfg_coef=3.0,          # Classifier-Free Guidance 강도
    )

    wav = model.generate(descriptions)
    # shape: (batch, channels, samples)

    for i, (desc, audio) in enumerate(zip(descriptions, wav)):
        audio_write(
            f"music_{i}",
            audio.cpu(),
            model.sample_rate,
            strategy="loudness",
            loudness_compressor=True
        )
        print(f"생성 완료: music_{i}.wav")
        print(f"  설명: {desc}")

# 다양한 스타일 생성
descriptions = [
    "Upbeat K-pop with synth leads, 120 BPM, energetic and bright",
    "Calm Korean traditional music with gayageum and daegeum, peaceful",
    "Epic orchestral trailer music with powerful drums and brass",
    "Lo-fi hip hop beats with jazz piano, for studying",
]

generate_music(descriptions, duration=15, model_size="medium")

from audiocraft.models import AudioGen
from audiocraft.data.audio import audio_write

def generate_sound_effects(descriptions, duration=5):
    """환경음, 음향 효과 생성"""
    model = AudioGen.get_pretrained("facebook/audiogen-medium")
    model.set_generation_params(duration=duration, temperature=1.0)

    wav = model.generate(descriptions)

    for i, (desc, audio) in enumerate(zip(descriptions, wav)):
        audio_write(f"sfx_{i}", audio.cpu(), model.sample_rate)
        print(f"생성: sfx_{i}.wav — {desc}")

sound_descriptions = [
    "Rain falling on a tin roof with distant thunder",
    "Busy city street with cars and people talking",
    "Birds chirping in a forest at dawn",
    "Keyboard typing in a quiet office",
]

generate_sound_effects(sound_descriptions, duration=8)

import subprocess
import os

def separate_audio_tracks(audio_path, output_dir="separated", model="htdemucs"):
    """
    음악을 보컬/드럼/베이스/기타 등으로 분리
    모델: htdemucs (4-stem), htdemucs_6s (6-stem)
    """
    os.makedirs(output_dir, exist_ok=True)

    cmd = [
        "python", "-m", "demucs",
        "--name", model,
        "--out", output_dir,
        "--mp3",
        "--mp3-bitrate", "320",
        audio_path
    ]

    result = subprocess.run(cmd, capture_output=True, text=True)

    if result.returncode == 0:
        base_name = os.path.splitext(os.path.basename(audio_path))[0]
        stems_dir = os.path.join(output_dir, model, base_name)
        stems = os.listdir(stems_dir)
        print(f"분리 성공! 트랙: {stems}")
        # htdemucs: vocals.mp3, drums.mp3, bass.mp3, other.mp3
        # htdemucs_6s: + guitar.mp3, piano.mp3
    else:
        print(f"오류: {result.stderr}")

    return output_dir

separate_audio_tracks("song.mp3", model="htdemucs_6s")

import sounddevice as sd
import numpy as np
import whisper
import queue
import threading
from collections import deque

class RealTimeASR:
    """
    실시간 스트리밍 음성 인식
    look-ahead window로 정확도/지연시간 균형 조절
    """
    def __init__(self, model_size="small", language="ko",
                 chunk_duration=2.0, look_ahead=0.5, sample_rate=16000):
        self.model = whisper.load_model(model_size)
        self.language = language
        self.chunk_duration = chunk_duration
        self.look_ahead = look_ahead
        self.sample_rate = sample_rate
        self.audio_queue = queue.Queue()
        max_buf = int(sample_rate * (chunk_duration + look_ahead) * 3)
        self.buffer = deque(maxlen=max_buf)
        self.is_running = False

    def audio_callback(self, indata, frames, time, status):
        if status:
            print(f"오디오 상태: {status}")
        self.audio_queue.put(indata.copy())

    def process_audio(self):
        chunk_samples = int(self.sample_rate * self.chunk_duration)

        while self.is_running:
            audio_chunk = []
            while len(audio_chunk) < chunk_samples:
                try:
                    data = self.audio_queue.get(timeout=0.1)
                    audio_chunk.extend(data.flatten())
                except queue.Empty:
                    break

            if len(audio_chunk) < chunk_samples // 2:
                continue

            self.buffer.extend(audio_chunk)
            audio_array = np.array(list(self.buffer), dtype=np.float32)
            audio_array = audio_array / (np.max(np.abs(audio_array)) + 1e-8)

            result = self.model.transcribe(
                audio_array,
                language=self.language,
                temperature=0,
                no_speech_threshold=0.6,
            )

            if result["text"].strip():
                print(f"\r인식: {result['text'].strip()}", end="", flush=True)

    def start(self):
        self.is_running = True
        t = threading.Thread(target=self.process_audio, daemon=True)
        t.start()

        print("마이크 활성화 — 말씀하세요 (Ctrl+C로 종료)")
        with sd.InputStream(
            samplerate=self.sample_rate,
            channels=1,
            dtype="float32",
            blocksize=int(self.sample_rate * 0.1),
            callback=self.audio_callback
        ):
            try:
                while True:
                    sd.sleep(100)
            except KeyboardInterrupt:
                self.is_running = False
                print("\n인식 종료")

# 실행
asr = RealTimeASR(model_size="small", language="ko",
                  chunk_duration=2.0, look_ahead=0.5)
asr.start()

import whisper
from pyannote.audio import Pipeline
from openai import OpenAI

def auto_meeting_summary(audio_path, hf_token, openai_key, language="ko"):
    """
    회의 자동 요약: 화자 분리 + 전사 + LLM 요약
    """
    client = OpenAI(api_key=openai_key)

    # 1. 화자 분리
    print("화자 분리 중...")
    diarization_pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )
    diarization = diarization_pipeline(audio_path)

    # 2. 전사
    print("음성 인식 중...")
    asr_model = whisper.load_model("large-v3")
    result = asr_model.transcribe(audio_path, language=language,
                                   word_timestamps=True)

    # 3. 화자-텍스트 매핑 및 대화록 구성
    lines = []
    current_speaker = None
    current_words = []

    for seg in result["segments"]:
        for word in seg.get("words", []):
            mid = (word["start"] + word["end"]) / 2
            speaker = "UNKNOWN"
            for turn, _, spk in diarization.itertracks(yield_label=True):
                if turn.start <= mid <= turn.end:
                    speaker = spk
                    break

            if speaker != current_speaker:
                if current_words:
                    lines.append(f"[{current_speaker}]: {''.join(current_words).strip()}")
                current_speaker = speaker
                current_words = [word["word"]]
            else:
                current_words.append(word["word"])

    if current_words:
        lines.append(f"[{current_speaker}]: {''.join(current_words).strip()}")

    transcript = "\n".join(lines)

    # 4. LLM 요약
    print("요약 생성 중...")
    prompt = f"""다음 회의 전사록을 아래 항목으로 요약해주세요:

1. 주요 논의 사항
2. 결정된 사항
3. Action Items (담당자 포함)
4. 다음 회의 일정

전사록:
{transcript[:4000]}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )

    summary = response.choices[0].message.content
    print("\n=== 회의 요약 ===")
    print(summary)

    return {"transcript": transcript, "summary": summary}