Spaces:
Sleeping
Sleeping
| import os, re, uuid, torch, scipy.io.wavfile, edge_tts, asyncio | |
| import numpy as np | |
| import gradio as gr | |
| from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline | |
| from groq import Groq | |
| # λͺ¨λΈ κ΄λ¦¬ μ±κΈν€ | |
| class ModelManager: | |
| _llm_pipeline = None | |
| _music_pipeline = None | |
| _groq_client = None | |
| def get_qwen(cls): | |
| if cls._llm_pipeline is None: | |
| model_id = "Qwen/Qwen2.5-0.5B-Instruct" | |
| tokenizer = AutoTokenizer.from_pretrained(model_id) | |
| model = AutoModelForCausalLM.from_pretrained(model_id, device_map="cpu", torch_dtype=torch.float32) | |
| cls._llm_pipeline = pipeline("text-generation", model=model, tokenizer=tokenizer) | |
| return cls._llm_pipeline | |
| def get_groq(cls): | |
| key = os.getenv("GROQ_API_KEY") | |
| if cls._groq_client is None and key: | |
| cls._groq_client = Groq(api_key=key) | |
| return cls._groq_client | |
| def get_music(cls): | |
| if cls._music_pipeline is None: | |
| cls._music_pipeline = pipeline("text-to-audio", "facebook/musicgen-small", device="cpu") | |
| return cls._music_pipeline | |
| # λ©€λ²λ³ κ³ μ μ±κ²©/λ§ν¬ νλ₯΄μλ | |
| PERSONA_MAP = { | |
| "μμ€ (Korea)": "μμ μ μ΄λ©΄μλ κ°λ ¬ν K-Rock 리λ. μ§μ§νκ³ μ² νμ μ΄λ©° λ°λ»νκ² μ‘°μΈν¨.", | |
| "Chloe (USA)": "μμ λΆλ°©ν μΊλ¦¬ν¬λμ νν¬ λ½μ€ν. κ±°μΉ¨μκ³ μΏ¨νλ©° μλμ ν±ν λ§ν¬.", | |
| "Naomi (Japan)": "μλ²½μ£Όμ J-Rock κΈ°ν리μ€νΈ. μ¬μΈνκ³ μμ λ°λ₯΄λ©° μμ μ΄λ‘ μ λ μΉ΄λ‘μ.", | |
| "Beatrice (Brazil)": "μ΄μ μ μΈ μΌλ° λ½μ»€. κΈμ μ μ΄κ³ 리λ¬κ°μ΄ λμΉλ©° μΈμμ μ¦κ±°μμ κ°μ‘°ν¨.", | |
| "Elena (Spain)": "λλΌλ§ν±ν κ°μ±μ μμ μ. νλΌλ©©μ½μ μ μ΄κ³Ό λ½μ νκ΄΄λ ₯μ λμμ κ°μ§.", | |
| "Amira (Egypt)": "μ λΉλ‘μ΄ μ€λ¦¬μν λ½μ»€. κΉμ μ§νμ κ³ μ μ μΈ λ¬΄κ²κ°μ λ΄μ μ‘°μΈν¨.", | |
| "Liwei (China)": "μ ν΅κ³Ό νλμ μ‘°νλ₯Ό μ€μνλ ν¨μ λ½μ»€. μ μ λκ³ ν μλ λ§ν¬.", | |
| "Sophie (France)": "μμ μ μμ‘΄μ¬μ΄ κ°ν μλ°©κ°λ₯΄λ λ½μ»€. μμ μ΄κ³ μΈλ ¨λ ννμ μ¦κ²¨ μ." | |
| } | |
| # μΈμ΄λ³ λ©€λ²/보μ΄μ€ λ§€ν | |
| LANG_MEMBER_MAP = { | |
| "Korean": {"name": "μμ€ (Korea)", "voice": "ko-KR-SunHiNeural"}, | |
| "English": {"name": "Chloe (USA)", "voice": "en-US-AriaNeural"}, | |
| "Japanese": {"name": "Naomi (Japan)", "voice": "ja-JP-NanamiNeural"}, | |
| "Portuguese": {"name": "Beatrice (Brazil)", "voice": "pt-BR-FranciscaNeural"}, | |
| "Spanish": {"name": "Elena (Spain)", "voice": "es-ES-ElviraNeural"}, | |
| "Arabic": {"name": "Amira (Egypt)", "voice": "ar-EG-SalmaNeural"}, | |
| "Chinese": {"name": "Liwei (China)", "voice": "zh-CN-XiaoxiaoNeural"}, | |
| "French": {"name": "Sophie (France)", "voice": "fr-FR-DeniseNeural"} | |
| } | |
| async def band_consulting(user_input, selected_lang, g_inst, b_inst, d_inst, chords): | |
| try: | |
| req_id = str(uuid.uuid4())[:8] | |
| voice_path = f"/tmp/v_{req_id}.mp3" | |
| music_path = f"/tmp/m_{req_id}.wav" | |
| m_info = LANG_MEMBER_MAP.get(selected_lang, LANG_MEMBER_MAP["Korean"]) | |
| persona = PERSONA_MAP.get(m_info['name'], "μ΄μ μ μΈ λ½μ€ν") | |
| # λ³Έλ¬Έ μμ½ λ° μμΈ ν λΆλ¦¬ μ§μ ν둬ννΈ | |
| system_prompt = f"""You are the rock star '{m_info['name']}'. Respond ONLY in {selected_lang}. | |
| Persona: {persona}. | |
| CRITICAL RULES: | |
| 1. MAIN ADVICE: Provide exactly 3 to 5 deep, soulful sentences for the main chat. | |
| 2. [TAB] SECTION: Put ALL long explanations, music theory, Drum patterns (e.g., H|x-x-x-x| S|--o---o-|), and detailed Guitar Solo tabs here. This must be very technical and professional. | |
| 3. [TRANSLATION]: English translation of your 3-5 sentence MAIN ADVICE only. | |
| 4. [MUSIC]: English prompt for MusicGen reflecting: Guitar:{g_inst}, Bass:{b_inst}, Drums:{d_inst}, Chords:{chords} | |
| """ | |
| ai_text_raw = "" | |
| groq_client = ModelManager.get_groq() | |
| if groq_client: | |
| try: | |
| res = groq_client.chat.completions.create( | |
| messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": user_input}], | |
| model="llama-3.3-70b-versatile" | |
| ) | |
| ai_text_raw = res.choices[0].message.content | |
| except Exception as e: | |
| print(f"Groq Error: {e}") | |
| if not ai_text_raw: | |
| qwen = ModelManager.get_qwen() | |
| input_t = f"<|im_start|>system\n{system_prompt}<|im_end|>\n<|im_start|>user\n{user_input}<|im_end|>\nassistant\n" | |
| out = qwen(input_t, max_new_tokens=1024) | |
| ai_text_raw = out[0]['generated_text'].split("assistant\n")[-1] | |
| # νμ± λ‘μ§ | |
| tab_match = re.search(r'\[TAB\](.*?)(\[|$)', ai_text_raw, re.DOTALL | re.IGNORECASE) | |
| music_match = re.search(r'\[MUSIC:(.*?)\]', ai_text_raw, re.IGNORECASE) | |
| trans_match = re.search(r'\[TRANSLATION\](.*?)(\[|$)', ai_text_raw, re.DOTALL | re.IGNORECASE) | |
| tab_display = tab_match.group(1).strip() if tab_match else "No Detailed Data" | |
| eng_translation = trans_match.group(1).strip() if trans_match else "" | |
| # μμ±μμ λ²μλ¬Έ/ν/μμ ν둬ννΈ μλ²½ μ κ±° | |
| speech_text = re.sub(r'\[TAB\].*?(\[|$)', '', ai_text_raw, flags=re.DOTALL | re.IGNORECASE) | |
| speech_text = re.sub(r'\[MUSIC:.*?\]', '', speech_text, flags=re.IGNORECASE) | |
| speech_text = re.sub(r'\[TRANSLATION\].*?(\[|$)', '', speech_text, flags=re.DOTALL | re.IGNORECASE).strip() | |
| # TTS μμ± | |
| tts_text = re.sub(r'[\*\#\-\_\~\|]', '', speech_text) | |
| communicate = edge_tts.Communicate(tts_text, m_info["voice"]) | |
| await communicate.save(voice_path) | |
| # MusicGen μμ μμ± | |
| music_gen = ModelManager.get_music() | |
| music_p = music_match.group(1).strip() if music_match else "rock music" | |
| music_output = music_gen(music_p, forward_params={"max_new_tokens": 512}) | |
| audio_data = np.squeeze(music_output["audio"]) | |
| audio_int16 = (audio_data * 32767).astype(np.int16) | |
| scipy.io.wavfile.write(music_path, music_output["sampling_rate"], audio_int16) | |
| return speech_text, voice_path, music_path, tab_display, eng_translation | |
| except Exception as e: | |
| print(f"Final Error: {e}") | |
| return f"λ°΄λ μμ€ν μλ¬: {str(e)}", None, None, "No Data", "Error occurred" | |
| with gr.Blocks() as demo: | |
| inputs = [gr.Textbox(visible=False) for _ in range(6)] | |
| outputs = [ | |
| gr.Textbox(visible=False), | |
| gr.Audio(visible=False), | |
| gr.Audio(visible=False), | |
| gr.Textbox(visible=False), | |
| gr.Textbox(visible=False) | |
| ] | |
| btn = gr.Button("API", visible=False) | |
| btn.click(band_consulting, inputs, outputs, api_name="predict") | |
| demo.queue().launch() |