Delete models/ai_effector.py

models/ai_effector.py

@@ -1,539 +0,0 @@
-"""
-AI Effector - DiffVox LLM 기반 이펙트 파라미터 예측
-===================================================
-V2: 학습과 동일한 CLAP 인코더 + 프롬프트 형식 사용
-"""
-
-import os
-import json
-import re
-import torch
-import numpy as np
-from typing import Dict, List, Optional, Any
-from pathlib import Path
-from datetime import datetime
-import warnings
-
-warnings.filterwarnings("ignore")
-
-# 기본 파라미터 (모델 로드 실패 시 사용)
-DEFAULT_PARAMETERS = {
-    "eq_peak1.params.freq": 1000.0,
-    "eq_peak1.params.gain": 0.0,
-    "eq_peak1.params.Q": 1.0,  # 대문자 Q (학습 데이터와 일치)
-    "eq_peak2.params.freq": 4000.0,
-    "eq_peak2.params.gain": 0.0,
-    "eq_peak2.params.Q": 1.0,
-    "eq_lowshelf.params.freq": 200.0,
-    "eq_lowshelf.params.gain": 0.0,
-    "eq_highshelf.params.freq": 8000.0,
-    "eq_highshelf.params.gain": 0.0,
-    "distortion_amount": 0.0,
-    "delay.delay_time": 0.02,
-    "delay.feedback": 0.3,
-    "delay.mix": 0.2,
-    "final_wet_mix": 0.5
-}
-
-# 스타일 프리셋 (AI 없이도 작동)
-STYLE_PRESETS = {
-    "warm": {
-        "eq_lowshelf.params.gain": 3.0,
-        "eq_highshelf.params.gain": -1.0,
-        "distortion_amount": 0.05,
-    },
-    "bright": {
-        "eq_highshelf.params.gain": 4.0,
-        "eq_peak2.params.gain": 2.0,
-        "eq_lowshelf.params.gain": -1.0,
-    },
-    "vintage": {
-        "eq_lowshelf.params.gain": 2.0,
-        "eq_highshelf.params.gain": -2.0,
-        "distortion_amount": 0.1,
-        "delay.mix": 0.15,
-    },
-    "modern": {
-        "eq_peak1.params.gain": 2.0,
-        "eq_peak2.params.gain": 3.0,
-        "eq_highshelf.params.gain": 2.0,
-    },
-    "spacious": {
-        "delay.delay_time": 0.05,
-        "delay.feedback": 0.4,
-        "delay.mix": 0.35,
-    },
-    "dry": {
-        "final_wet_mix": 0.2,
-        "delay.mix": 0.0,
-    },
-    "saturated": {
-        "distortion_amount": 0.15,
-        "eq_lowshelf.params.gain": 1.0,
-    }
-}
-
-
-class CLAPAudioEncoder:
-    """
-    CLAP 기반 오디오 인코더 (학습 시와 동일)
-    laion/larger_clap_music 모델 사용, 512→64 pooling
-    """
-    
-    def __init__(self, output_dim: int = 64, model_name: str = "laion/larger_clap_music"):
-        self.output_dim = output_dim
-        self.model_name = model_name
-        self.target_sr = 48000  # CLAP은 48kHz 사용
-        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        
-        self.model = None
-        self.processor = None
-        self._load_model()
-    
-    def _load_model(self):
-        """CLAP 모델 로드"""
-        try:
-            from transformers import ClapModel, ClapProcessor
-            
-            print(f"[CLAPEncoder] CLAP 모델 로딩 중: {self.model_name}")
-            
-            self.processor = ClapProcessor.from_pretrained(self.model_name)
-            self.model = ClapModel.from_pretrained(self.model_name)
-            self.model = self.model.to(self.device)
-            self.model.eval()
-            
-            print(f"[CLAPEncoder] ✅ CLAP 모델 로드 완료 (512→{self.output_dim} pooling)")
-            
-        except ImportError:
-            print("[CLAPEncoder] ❌ transformers 미설치")
-            print("   pip install transformers")
-        except Exception as e:
-            print(f"[CLAPEncoder] ❌ 모델 로드 실패: {e}")
-            import traceback
-            traceback.print_exc()
-    
-    def get_audio_features(self, audio_path: str) -> List[float]:
-        """
-        오디오 파일에서 64차원 특징 벡터 추출 (학습과 동일한 방식)
-        """
-        if self.model is None:
-            print("[CLAPEncoder] 모델이 로드되지 않음, 빈 특징 반환")
-            return [0.0] * self.output_dim
-        
-        try:
-            import librosa
-            
-            # 1. 오디오 로드 (48kHz로 리샘플링 - CLAP 요구사항)
-            audio, sr = librosa.load(audio_path, sr=self.target_sr, mono=True)
-            
-            # 2. CLAP 입력 준비
-            inputs = self.processor(
-                audios=audio,
-                sampling_rate=self.target_sr,
-                return_tensors="pt",
-                padding=True
-            ).to(self.device)
-            
-            # 3. 특징 추출
-            with torch.no_grad():
-                outputs = self.model.get_audio_features(**inputs)
-            
-            # [1, 512] 형태의 텐서
-            features_512 = outputs[0].cpu().numpy()
-            
-            # 4. 512 → 64 차원 축소 (평균 풀링, 학습과 동일)
-            features_64 = self._reduce_dimension(features_512)
-            
-            return features_64.tolist()
-            
-        except Exception as e:
-            print(f"[CLAPEncoder] 특징 추출 실패: {e}")
-            import traceback
-            traceback.print_exc()
-            return [0.0] * self.output_dim
-    
-    def _reduce_dimension(self, features: np.ndarray) -> np.ndarray:
-        """512차원 → 64차원 평균 풀링 (학습과 동일한 방식)"""
-        current_dim = len(features)
-        
-        if current_dim == self.output_dim:
-            return features
-        
-        # 평균 풀링: 8개씩 묶어서 평균 (512 / 64 = 8)
-        pool_size = current_dim // self.output_dim
-        remainder = current_dim % self.output_dim
-        
-        pooled = []
-        idx = 0
-        for i in range(self.output_dim):
-            size = pool_size + (1 if i < remainder else 0)
-            pooled.append(np.mean(features[idx:idx+size]))
-            idx += size
-        
-        return np.array(pooled)
-    
-    def is_loaded(self) -> bool:
-        return self.model is not None
-
-
-class AIEffector:
-    """AI 기반 이펙터 파라미터 예측 (V2: 학습과 동일한 설정)"""
-    
-    def __init__(
-        self,
-        model_repo_id: str = "heybaeheef/KU_SW_Academy",
-        model_subfolder: str = "checkpoints",
-        base_model_name: str = "Qwen/Qwen3-8B",
-        audio_feature_dim: int = 64,
-        use_huggingface: bool = True
-    ):
-        self.model_repo_id = model_repo_id
-        self.model_subfolder = model_subfolder
-        self.base_model_name = base_model_name
-        self.audio_feature_dim = audio_feature_dim
-        self.use_huggingface = use_huggingface
-        
-        self.model = None
-        self.tokenizer = None
-        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        
-        # ★★★ 핵심 수정: CLAP 오디오 인코더 사용 (학습과 동일) ★★★
-        print(f"[AIEffector] CLAP 오디오 인코더 초기화 중...")
-        self.audio_encoder = CLAPAudioEncoder(output_dim=audio_feature_dim)
-        
-        # 요청 카운터
-        self.request_count = 0
-        
-        # 모델 로드 시도
-        self._load_model()
-    
-    def _load_model(self):
-        """모델 로드"""
-        try:
-            from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
-            from peft import PeftModel
-            
-            print(f"[AIEffector] 베이스 모델 로딩 중: {self.base_model_name}")
-            
-            # 4bit 양자화 설정
-            if torch.cuda.is_available():
-                bnb_config = BitsAndBytesConfig(
-                    load_in_4bit=True,
-                    bnb_4bit_quant_type="nf4",
-                    bnb_4bit_compute_dtype=torch.float16,
-                    bnb_4bit_use_double_quant=True
-                )
-                base_model = AutoModelForCausalLM.from_pretrained(
-                    self.base_model_name,
-                    quantization_config=bnb_config,
-                    device_map="auto",
-                    trust_remote_code=True
-                )
-            else:
-                base_model = AutoModelForCausalLM.from_pretrained(
-                    self.base_model_name,
-                    torch_dtype=torch.float32,
-                    device_map="auto",
-                    trust_remote_code=True
-                )
-            
-            self.tokenizer = AutoTokenizer.from_pretrained(
-                self.base_model_name,
-                trust_remote_code=True
-            )
-            
-            if self.tokenizer.pad_token is None:
-                self.tokenizer.pad_token = self.tokenizer.eos_token
-            
-            print(f"[AIEffector] LoRA 어댑터 로딩 중...")
-            
-            if self.use_huggingface:
-                print(f"[AIEffector] HuggingFace에서 LoRA 로딩: {self.model_repo_id}/{self.model_subfolder}")
-                self.model = PeftModel.from_pretrained(
-                    base_model,
-                    self.model_repo_id,
-                    subfolder=self.model_subfolder,
-                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-                )
-            else:
-                local_path = os.path.join(self.model_repo_id, self.model_subfolder)
-                print(f"[AIEffector] 로컬에서 LoRA 어댑터 로딩: {local_path}")
-                self.model = PeftModel.from_pretrained(
-                    base_model,
-                    local_path,
-                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-                )
-            
-            self.model.eval()
-            print(f"[AIEffector] ✅ 모델 로드 성공!")
-            
-        except Exception as e:
-            print(f"[AIEffector] ❌ 모델 로드 실패: {e}")
-            import traceback
-            traceback.print_exc()
-            print(f"[AIEffector] 폴백 모드로 전환 (프리셋 기반)")
-            self.model = None
-            self.tokenizer = None
-    
-    def is_loaded(self) -> bool:
-        """모델 로드 여부"""
-        return self.model is not None
-    
-    def _apply_preset(self, prompt: str) -> Dict[str, float]:
-        """프롬프트에서 프리셋 매칭"""
-        params = DEFAULT_PARAMETERS.copy()
-        prompt_lower = prompt.lower()
-        
-        matched_presets = []
-        for style_name, style_params in STYLE_PRESETS.items():
-            if style_name in prompt_lower:
-                params.update(style_params)
-                matched_presets.append(style_name)
-        
-        if matched_presets:
-            print(f"    [Preset] 매칭된 프리셋: {matched_presets}")
-        
-        return params
-    
-    def _format_prompt(self, text_prompt: str, audio_features: List[float]) -> str:
-        """
-        ★★★ 핵심 수정: 학습 시와 동일한 프롬프트 형식 사용 ★★★
-        train_model.py의 243-246줄과 동일한 형식
-        """
-        audio_state_str = json.dumps(audio_features)
-        
-        # 학습 시와 완전히 동일한 형식!
-        prompt = f"""Task: Convert text to audio parameters.
-Audio: {audio_state_str}
-Text: {text_prompt}
-Parameters:"""
-        
-        return prompt
-    
-    def _parse_output(self, output_text: str) -> Dict[str, float]:
-        """LLM 출력에서 파라미터 추출 (향상된 버전)"""
-        
-        print(f"    [Parse] Raw output 길이: {len(output_text)} 문자")
-        
-        try:
-            text = output_text
-            
-            # 1. <think>...</think> 태그 제거 (Qwen3 thinking mode)
-            text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
-            
-            # 2. 마크다운 코드블록 추출
-            code_block_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', text)
-            if code_block_match:
-                text = code_block_match.group(1)
-                print(f"    [Parse] 코드블록에서 JSON 추출")
-            
-            # 3. JSON 객체 찾기 (중첩 브레이스 지원)
-            json_str = self._extract_json_object(text)
-            
-            if json_str:
-                print(f"    [Parse] 추출된 JSON (처음 200자):\n{json_str[:200]}...")
-                
-                # 4. JSON 전처리
-                json_str = self._preprocess_json(json_str)
-                
-                # 5. 파싱 시도
-                params = json.loads(json_str)
-                
-                # 6. 결과 검증 및 매핑
-                result = DEFAULT_PARAMETERS.copy()
-                for key, value in params.items():
-                    # 키 정규화 (대소문자 처리)
-                    normalized_key = self._normalize_key(key)
-                    if normalized_key in result:
-                        try:
-                            result[normalized_key] = float(value)
-                        except (ValueError, TypeError):
-                            pass
-                
-                print(f"    [Parse] ✅ 파싱 성공! {len(params)}개 파라미터 추출")
-                return result
-            else:
-                print(f"    [Parse] ❌ JSON 객체를 찾을 수 없음")
-                
-        except json.JSONDecodeError as e:
-            print(f"    [Parse] ❌ JSON 파싱 에러: {e}")
-            if json_str:
-                print(f"    [Parse] 문제 위치 근처: ...{json_str[max(0, e.pos-20):e.pos+20]}...")
-        except Exception as e:
-            print(f"    [Parse] ❌ 예외 발생: {e}")
-        
-        print(f"    [Parse] ⚠️ 기본값으로 폴백")
-        return DEFAULT_PARAMETERS.copy()
-    
-    def _normalize_key(self, key: str) -> str:
-        """파라미터 키 정규화 (대소문자 처리)"""
-        # Q/q 정규화
-        if key.endswith('.q'):
-            return key[:-2] + '.Q'
-        return key
-    
-    def _extract_json_object(self, text: str) -> Optional[str]:
-        """텍스트에서 JSON 객체 추출 (중첩 브레이스 지원)"""
-        start = text.find('{')
-        if start == -1:
-            return None
-        
-        depth = 0
-        for i, char in enumerate(text[start:], start):
-            if char == '{':
-                depth += 1
-            elif char == '}':
-                depth -= 1
-                if depth == 0:
-                    return text[start:i+1]
-        
-        return None
-    
-    def _preprocess_json(self, json_str: str) -> str:
-        """JSON 문자열 전처리"""
-        # Trailing comma 제거
-        json_str = re.sub(r',(\s*[}\]])', r'\1', json_str)
-        
-        # NaN, Infinity 처리
-        json_str = re.sub(r'\bNaN\b', '0', json_str)
-        json_str = re.sub(r'\bInfinity\b', '999999', json_str)
-        json_str = re.sub(r'-Infinity\b', '-999999', json_str)
-        
-        return json_str
-    
-    def predict(self, audio_path: str, text_prompt: str = "") -> Dict[str, float]:
-        """파라미터 예측"""
-        
-        self.request_count += 1
-        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        
-        print(f"\n{'='*60}")
-        print(f"[AIEffector] 🎵 요청 #{self.request_count} - {timestamp}")
-        print(f"{'='*60}")
-        print(f"  📂 오디오 파일: {Path(audio_path).name}")
-        print(f"  💬 텍스트 프롬프트: '{text_prompt}'")
-        print(f"  🤖 모델 상태: {'AI 모드' if self.is_loaded() else '프리셋 모드'}")
-        print(f"  🎧 인코더: CLAP (학습과 동일)")
-        
-        # 모델이 없으면 프리셋 사용
-        if not self.is_loaded():
-            print(f"\n  ⚠️ AI 모델 미로드 - 프리셋 모드 사용")
-            params = self._apply_preset(text_prompt)
-            self._log_parameters(params)
-            return self._convert_to_effect_chain_format(params)
-        
-        try:
-            # 1. CLAP 오디오 특징 추출 (학습과 동일)
-            print(f"\n  📊 [Step 1] CLAP 오디오 특징 추출 중...")
-            audio_features = self.audio_encoder.get_audio_features(audio_path)
-            
-            if not audio_features or all(f == 0 for f in audio_features):
-                print(f"    ⚠️ 특징 추출 실패, 프리셋으로 폴백")
-                params = self._apply_preset(text_prompt)
-                self._log_parameters(params)
-                return self._convert_to_effect_chain_format(params)
-            
-            print(f"    ✅ {len(audio_features)}차원 특징 추출 완료")
-            print(f"    - 특징 벡터 (처음 8개): {[round(v, 3) for v in audio_features[:8]]}")
-            
-            # 2. LLM 프롬프트 생성 (학습과 동일한 형식)
-            print(f"\n  🔤 [Step 2] LLM 프롬프트 생성 중 (학습 형식)...")
-            prompt = self._format_prompt(text_prompt, audio_features)
-            print(f"    - 프롬프트 길이: {len(prompt)} 문자")
-            
-            # 3. 토큰화
-            print(f"\n  🔢 [Step 3] 토큰화 중...")
-            inputs = self.tokenizer(
-                prompt,
-                return_tensors="pt",
-                truncation=True,
-                max_length=1500  # 학습 시와 동일
-            ).to(self.device)
-            print(f"    - 입력 토큰 수: {inputs['input_ids'].shape[1]}")
-            
-            # 4. LLM 생성
-            print(f"\n  🧠 [Step 4] LLM 추론 중...")
-            import time
-            start_time = time.time()
-            
-            with torch.no_grad():
-                outputs = self.model.generate(
-                    **inputs,
-                    max_new_tokens=500,
-                    do_sample=False,
-                    temperature=0.1,
-                    pad_token_id=self.tokenizer.pad_token_id,
-                    eos_token_id=self.tokenizer.eos_token_id,
-                )
-            
-            inference_time = time.time() - start_time
-            print(f"    - 추론 시간: {inference_time:.2f}초")
-            
-            # 5. 디코딩 (생성된 부분만)
-            print(f"\n  📝 [Step 5] 출력 디코딩 중...")
-            generated_tokens = outputs[0][inputs['input_ids'].shape[1]:]
-            output_text = self.tokenizer.decode(generated_tokens, skip_special_tokens=True).strip()
-            
-            print(f"    - LLM 출력 (처음 300자):\n{output_text[:300]}")
-            
-            # 6. 파싱
-            print(f"\n  🔧 [Step 6] 파라미터 파싱 중...")
-            params = self._parse_output(output_text)
-            
-            # 7. 결과 로깅
-            self._log_parameters(params)
-            
-            print(f"\n  ✅ AI 예측 완료!")
-            print(f"{'='*60}\n")
-            
-            # effect_chain.py 형식으로 변환
-            return self._convert_to_effect_chain_format(params)
-            
-        except Exception as e:
-            print(f"\n  ❌ 예측 실패: {e}")
-            import traceback
-            traceback.print_exc()
-            print(f"  ⚠️ 프리셋으로 폴백...")
-            params = self._apply_preset(text_prompt)
-            self._log_parameters(params)
-            return self._convert_to_effect_chain_format(params)
-    
-    def _convert_to_effect_chain_format(self, params: Dict[str, float]) -> Dict[str, float]:
-        """
-        학습 데이터 형식 → effect_chain.py 형식으로 변환
-        주로 Q/q 대소문자 처리
-        """
-        result = {}
-        for key, value in params.items():
-            # Q → q 변환 (effect_chain.py는 소문자 q 사용)
-            new_key = key.replace('.Q', '.q')
-            result[new_key] = value
-        return result
-    
-    def _log_parameters(self, params: Dict[str, float]):
-        """예측된 파라미터 로깅"""
-        print(f"\n  📋 예측된 파라미터:")
-        print(f"    [EQ Peak 1]")
-        print(f"      - Freq: {params.get('eq_peak1.params.freq', 0):.1f} Hz")
-        print(f"      - Gain: {params.get('eq_peak1.params.gain', 0):.2f} dB")
-        print(f"      - Q: {params.get('eq_peak1.params.Q', params.get('eq_peak1.params.q', 0)):.2f}")
-        
-        print(f"    [EQ Peak 2]")
-        print(f"      - Freq: {params.get('eq_peak2.params.freq', 0):.1f} Hz")
-        print(f"      - Gain: {params.get('eq_peak2.params.gain', 0):.2f} dB")
-        print(f"      - Q: {params.get('eq_peak2.params.Q', params.get('eq_peak2.params.q', 0)):.2f}")
-        
-        print(f"    [Low Shelf]")
-        print(f"      - Freq: {params.get('eq_lowshelf.params.freq', 0):.1f} Hz")
-        print(f"      - Gain: {params.get('eq_lowshelf.params.gain', 0):.2f} dB")
-        
-        print(f"    [High Shelf]")
-        print(f"      - Freq: {params.get('eq_highshelf.params.freq', 0):.1f} Hz")
-        print(f"      - Gain: {params.get('eq_highshelf.params.gain', 0):.2f} dB")
-        
-        print(f"    [Effects]")
-        print(f"      - Distortion: {params.get('distortion_amount', 0):.3f}")
-        print(f"      - Delay Time: {params.get('delay.delay_time', 0):.3f}s")
-        print(f"      - Delay Feedback: {params.get('delay.feedback', 0):.2f}")
-        print(f"      - Delay Mix: {params.get('delay.mix', 0):.2f}")
-        print(f"      - Final Wet Mix: {params.get('final_wet_mix', 0):.2f}")