""" Step-Audio-EditX - Audio Editing Demo using StepFun API """ import logging import gradio as gr from stepfun_api import get_api_token, process_audio # transcribe_audio # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def get_supported_edit_types(): """ 获取支持的编辑类型和选项 Returns: Dict[str, list]: Dictionary of edit types and their options """ return { "clone": [], "emotion": [ 'happy', 'angry', 'sad', 'humour', 'confusion', 'disgusted', 'empathy', 'embarrass', 'fear', 'surprised', 'excited', 'depressed', 'coldness', 'admiration', 'remove' ], "style": [ 'serious', 'arrogant', 'child', 'older', 'girl', 'pure', 'sister', 'sweet', 'ethereal', 'whisper', 'gentle', 'recite', 'generous', 'act_coy', 'warm', 'shy', 'comfort', 'authority', 'chat', 'radio', 'soulful', 'story', 'vivid', 'program', 'news', 'advertising', 'roar', 'murmur', 'shout', 'deeply', 'loudly', 'remove', 'exaggerated' ], "vad": [], "denoise": [], "paralinguistic": [], "speed": ["faster", "slower", "more faster", "more slower"], } class EditxTab: """Audio editing and voice cloning interface tab""" def __init__(self): self.edit_type_list = list(get_supported_edit_types().keys()) self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") def history_messages_to_show(self, messages): """Convert message history to gradio chatbot format""" show_msgs = [] for message in messages: edit_type = message['edit_type'] edit_info = message['edit_info'] source_text = message['source_text'] target_text = message['target_text'] raw_audio_path = message['raw_audio_path'] edit_audio_path = message['edit_audio_path'] type_str = f"{edit_type}-{edit_info}" if edit_info is not None else f"{edit_type}" show_msgs.extend([ {"role": "user", "content": f"任务类型：{type_str}\n文本：{source_text}"}, {"role": "user", "content": gr.Audio(value=raw_audio_path, interactive=False)}, {"role": "assistant", "content": f"输出音频：\n文本：{target_text}"}, {"role": "assistant", "content": gr.Audio(value=edit_audio_path, interactive=False)} ]) return show_msgs def generate_clone(self, prompt_text_input, prompt_audio_input, generated_text, edit_type, edit_info, state): """Generate cloned audio using API""" self.logger.info("Starting voice cloning via API") # Input validation if not prompt_text_input or prompt_text_input.strip() == "": error_msg = "[Error] Uploaded text cannot be empty." self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state if not prompt_audio_input: error_msg = "[Error] Uploaded audio cannot be empty." self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state if not generated_text or generated_text.strip() == "": error_msg = "[Error] Clone content cannot be empty." self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state if edit_type != "clone": error_msg = "[Error] CLONE button must use clone task." self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state try: # Call API output_path = process_audio( prompt_audio_input, prompt_text_input, generated_text, edit_type, edit_info ) # Create message for history cur_msg = { "edit_type": edit_type, "edit_info": edit_info, "source_text": prompt_text_input, "target_text": generated_text, "raw_audio_path": prompt_audio_input, "edit_audio_path": output_path, } state["history_audio"].append((output_path, generated_text)) state["history_messages"].append(cur_msg) show_msgs = self.history_messages_to_show(state["history_messages"]) self.logger.info("Voice cloning completed successfully") return show_msgs, state except Exception as e: error_msg = f"[Error] Clone failed: {str(e)}" self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state def generate_edit(self, prompt_text_input, prompt_audio_input, generated_text, edit_type, edit_info, state): """Generate edited audio using API""" self.logger.info("Starting audio editing via API") # Input validation if not prompt_audio_input: error_msg = "[Error] Uploaded audio cannot be empty." self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state try: # Determine which audio to use if len(state["history_audio"]) == 0: audio_to_edit = prompt_audio_input text_to_use = prompt_text_input else: audio_to_edit, previous_text = state["history_audio"][-1] text_to_use = previous_text # For para-linguistic, use generated_text; otherwise use source text if edit_type not in {"paralinguistic"}: generated_text = text_to_use # Call API output_path = process_audio( audio_to_edit, text_to_use, generated_text, edit_type, edit_info ) # Create message for history cur_msg = { "edit_type": edit_type, "edit_info": edit_info, "source_text": text_to_use, "target_text": generated_text, "raw_audio_path": audio_to_edit, "edit_audio_path": output_path, } state["history_audio"].append((output_path, generated_text)) state["history_messages"].append(cur_msg) show_msgs = self.history_messages_to_show(state["history_messages"]) self.logger.info("Audio editing completed successfully") return show_msgs, state except Exception as e: error_msg = f"[Error] Edit failed: {str(e)}" self.logger.error(error_msg) return self.history_messages_to_show(state["history_messages"]) + [{"role": "user", "content": error_msg}], state def clear_history(self, state): """Clear conversation history""" state["history_messages"] = [] state["history_audio"] = [] return [], state # def auto_transcribe_audio(self, audio_path, state): # """ # 自动转录音频文件，一次性返回最终结果 # Args: # audio_path: 音频文件路径 # state: 状态字典 # Returns: # 转录的文本内容和更新后的状态 # """ # if not audio_path: # return "", state # # 防止重复调用 - 简化逻辑 # if state.get("last_audio_path") == audio_path: # self.logger.debug(f"⚠️ Skipping duplicate transcription request for: {audio_path}") # return state.get("last_transcribed_text", ""), state # try: # # 更新音频路径 # state["last_audio_path"] = audio_path # self.logger.info(f"🎙️ Starting auto transcription for: {audio_path}") # # 使用stepfun_api中的transcribe_audio函数，不使用streaming模式 # transcribed_text = transcribe_audio(audio_path, streaming=False) # # 转录完成，缓存结果 # state["last_transcribed_text"] = transcribed_text # self.logger.info(f"✅ Auto transcription completed: {transcribed_text}") # return transcribed_text, state # except Exception as e: # error_msg = f"[转录失败: {str(e)}]" # self.logger.error(f"❌ Auto transcription failed: {str(e)}") # state["last_transcribed_text"] = error_msg # return error_msg, state def init_state(self): """Initialize conversation state""" return { "history_messages": [], "history_audio": [] # # ASR相关状态（已禁用） # "last_audio_path": None, # 用于防重复调用 # "last_transcribed_text": "" # 缓存最后的转录结果 } def update_edit_info(self, category): """Update sub-task dropdown based on main task selection""" category_items = get_supported_edit_types() choices = category_items.get(category, []) value = None if len(choices) == 0 else choices[0] return gr.Dropdown(label="Sub-task", choices=choices, value=value) def register_components(self): """Register gradio components - maintaining exact layout from original""" with gr.Tab("Editx"): with gr.Row(): with gr.Column(): self.model_input = gr.Textbox(label="Model Name", value="Step-Audio-EditX", scale=1) self.prompt_text_input = gr.Textbox(label="Prompt Text", value="", scale=1) self.prompt_audio_input = gr.Audio( sources=["upload", "microphone"], format="wav", type="filepath", label="Input Audio", ) self.generated_text = gr.Textbox(label="Target Text", lines=1, max_lines=200, max_length=1000) with gr.Column(): with gr.Row(): self.edit_type = gr.Dropdown(label="Task", choices=self.edit_type_list, value="clone") self.edit_info = gr.Dropdown(label="Sub-task", choices=[], value=None) self.chat_box = gr.Chatbot(label="History", type="messages", height=480*1) with gr.Row(): with gr.Column(): with gr.Row(): self.button_tts = gr.Button("CLONE", variant="primary") self.button_edit = gr.Button("EDIT", variant="primary") with gr.Column(): self.clean_history_submit = gr.Button("Clear History", variant="primary") gr.Markdown("---") gr.Markdown(""" **Button Description:** - CLONE: Synthesizes audio based on uploaded audio and text, only used for clone mode, will clear history information when used. - EDIT: Edits based on uploaded audio, or continues to stack edit effects based on the previous round of generated audio. """) gr.Markdown(""" **Operation Workflow:** - Upload the audio to be edited on the left side and fill in the corresponding text content of the audio; - If the task requires modifying text content (such as clone, para-linguistic), fill in the text to be synthesized in the "target text" field. For all other tasks, keep the uploaded audio text content unchanged; - Select tasks and subtasks on the right side (some tasks have no subtasks, such as vad, etc.); - Click the "CLONE" or "EDIT" button on the left side, and audio will be generated in the dialog box on the right side. """) gr.Markdown(""" **Para-linguistic Description:** - Supported tags include: [Breathing] [Laughter] [Surprise-oh] [Confirmation-en] [Uhm] [Surprise-ah] [Surprise-wa] [Sigh] [Question-ei] [Dissatisfaction-hnn] - Example: - Fill in "target text" field: "Great, the weather is so nice today." Click the "CLONE" button to get audio. - Change "target text" field to: "Great[Laughter], the weather is so nice today[Surprise-ah]." Click the "EDIT" button to get para-linguistic audio. """) def register_events(self): """Register event handlers""" state = gr.State(self.init_state()) self.button_tts.click( self.generate_clone, inputs=[self.prompt_text_input, self.prompt_audio_input, self.generated_text, self.edit_type, self.edit_info, state], outputs=[self.chat_box, state] ) self.button_edit.click( self.generate_edit, inputs=[self.prompt_text_input, self.prompt_audio_input, self.generated_text, self.edit_type, self.edit_info, state], outputs=[self.chat_box, state] ) self.clean_history_submit.click(self.clear_history, inputs=[state], outputs=[self.chat_box, state]) self.edit_type.change( fn=self.update_edit_info, inputs=self.edit_type, outputs=self.edit_info, ) # # 音频上传时自动转录 # self.prompt_audio_input.change( # fn=self.auto_transcribe_audio, # inputs=[self.prompt_audio_input, state], # outputs=[self.prompt_text_input, state] # ) def create_demo(): """Create and return the Gradio demo""" editx_tab = EditxTab() with gr.Blocks( theme=gr.themes.Soft(), title="🎙️ Step-Audio-EditX", css=""" :root { --font: "Helvetica Neue", Helvetica, Arial, sans-serif; --font-mono: "SFMono-Regular", Consolas, "Liberation Mono", Menlo, monospace; } """ ) as demo: gr.Markdown("## 🎙️ Step-Audio-EditX") gr.Markdown("Audio Editing and Zero-Shot Cloning using Step-Audio-EditX") editx_tab.register_components() editx_tab.register_events() return demo # Main entry point if __name__ == "__main__": logger.info("🚀 Starting Step-Audio-EditX Demo (API Mode)") logger.info(f"API Token configured: {'Yes' if get_api_token() else 'No'}") demo = create_demo() demo.queue().launch( server_name="0.0.0.0", server_port=7860, share=False )