|
|
""" |
|
|
Event Handlers |
|
|
|
|
|
Handlers for all Gradio UI events. |
|
|
Connects UI to services with error handling. |
|
|
""" |
|
|
|
|
|
import asyncio |
|
|
from typing import Optional, Tuple |
|
|
|
|
|
from ..services.music import MusicService |
|
|
from ..services.image import ImageService |
|
|
from ..services.video import VideoService |
|
|
from ..api.client import StackNetClient |
|
|
|
|
|
|
|
|
def format_error(error: Exception) -> str: |
|
|
"""Format exception as user-friendly message.""" |
|
|
msg = str(error) |
|
|
|
|
|
|
|
|
if "timeout" in msg.lower(): |
|
|
return "The operation timed out. Please try again with a simpler prompt." |
|
|
if "network" in msg.lower() or "connection" in msg.lower(): |
|
|
return "Network error. Please check your connection and try again." |
|
|
if "rate limit" in msg.lower() or "429" in msg.lower(): |
|
|
return "Too many requests. Please wait a moment and try again." |
|
|
|
|
|
return f"Error: {msg}" |
|
|
|
|
|
|
|
|
class Handlers: |
|
|
""" |
|
|
Collection of event handlers for the Gradio UI. |
|
|
|
|
|
All handlers hide API complexity and provide clean feedback. |
|
|
""" |
|
|
|
|
|
@staticmethod |
|
|
def generate_music( |
|
|
prompt: str, |
|
|
tags: str, |
|
|
instrumental: bool, |
|
|
lyrics: str, |
|
|
title: str, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], str]: |
|
|
"""Handle text-to-music generation.""" |
|
|
if not prompt.strip(): |
|
|
return None, "Please enter a description for your music." |
|
|
if not api_key.strip(): |
|
|
return None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = MusicService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
clips = loop.run_until_complete( |
|
|
service.generate_music( |
|
|
prompt=prompt, |
|
|
title=title if title.strip() else None, |
|
|
tags=tags if tags.strip() else None, |
|
|
lyrics=lyrics if lyrics.strip() and not instrumental else None, |
|
|
instrumental=instrumental |
|
|
) |
|
|
) |
|
|
|
|
|
if clips: |
|
|
audio_path = loop.run_until_complete( |
|
|
service.download_clip(clips[0]) |
|
|
) |
|
|
return audio_path, "Music generation complete!" |
|
|
else: |
|
|
return None, "No music was generated. Please try a different prompt." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def create_cover( |
|
|
audio_file: str, |
|
|
style_prompt: str, |
|
|
tags: str, |
|
|
title: str, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], str]: |
|
|
"""Handle cover song creation.""" |
|
|
if not audio_file: |
|
|
return None, "Please upload an audio file." |
|
|
if not style_prompt.strip(): |
|
|
return None, "Please describe the style for your cover." |
|
|
if not api_key.strip(): |
|
|
return None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = MusicService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
audio_url = f"file://{audio_file}" |
|
|
|
|
|
clips = loop.run_until_complete( |
|
|
service.create_cover( |
|
|
audio_url=audio_url, |
|
|
style_prompt=style_prompt, |
|
|
title=title if title.strip() else None, |
|
|
tags=tags if tags.strip() else None |
|
|
) |
|
|
) |
|
|
|
|
|
if clips: |
|
|
audio_path = loop.run_until_complete( |
|
|
service.download_clip(clips[0]) |
|
|
) |
|
|
return audio_path, "Cover created successfully!" |
|
|
else: |
|
|
return None, "No cover was generated. Please try again." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def extract_stems( |
|
|
audio_file: str, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str], str]: |
|
|
"""Handle stem extraction.""" |
|
|
if not audio_file: |
|
|
return None, None, None, None, "Please upload an audio file." |
|
|
if not api_key.strip(): |
|
|
return None, None, None, None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = MusicService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
audio_url = f"file://{audio_file}" |
|
|
|
|
|
stems = loop.run_until_complete( |
|
|
service.extract_stems(audio_url=audio_url) |
|
|
) |
|
|
|
|
|
return ( |
|
|
stems.vocals_path, |
|
|
stems.drums_path, |
|
|
stems.bass_path, |
|
|
stems.other_path, |
|
|
"Stems extracted successfully!" |
|
|
) |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, None, None, None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def generate_image( |
|
|
prompt: str, |
|
|
format_type: str, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], Optional[str], str]: |
|
|
"""Handle text-to-image generation. |
|
|
|
|
|
Returns: |
|
|
Tuple of (image_url, model_url, status) |
|
|
- For image/multi: (url, None, status) |
|
|
- For 3d: (None, url, status) |
|
|
""" |
|
|
if not prompt.strip(): |
|
|
return None, None, "Please enter a description for your image." |
|
|
if not api_key.strip(): |
|
|
return None, None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = ImageService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
results = loop.run_until_complete( |
|
|
service.generate_image( |
|
|
prompt=prompt, |
|
|
format_type=format_type |
|
|
) |
|
|
) |
|
|
|
|
|
if results: |
|
|
url = results[0].image_url |
|
|
if format_type == "3d": |
|
|
|
|
|
return None, url, "3D model generated successfully!" |
|
|
else: |
|
|
|
|
|
return url, None, "Image generated successfully!" |
|
|
else: |
|
|
return None, None, "No result was generated. Please try a different prompt." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def edit_image( |
|
|
input_image: str, |
|
|
edit_prompt: str, |
|
|
strength: float, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], str]: |
|
|
"""Handle image-to-image editing.""" |
|
|
if not input_image: |
|
|
return None, "Please upload an image." |
|
|
if not edit_prompt.strip(): |
|
|
return None, "Please describe how you want to edit the image." |
|
|
if not api_key.strip(): |
|
|
return None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = ImageService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
image_url = f"file://{input_image}" |
|
|
|
|
|
images = loop.run_until_complete( |
|
|
service.edit_image( |
|
|
image_url=image_url, |
|
|
edit_prompt=edit_prompt, |
|
|
strength=strength |
|
|
) |
|
|
) |
|
|
|
|
|
if images: |
|
|
|
|
|
return images[0].image_url, "Image edited successfully!" |
|
|
else: |
|
|
return None, "No edited image was generated. Please try again." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def generate_video( |
|
|
prompt: str, |
|
|
duration: int, |
|
|
style: str, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], str]: |
|
|
"""Handle text-to-video generation.""" |
|
|
if not prompt.strip(): |
|
|
return None, "Please enter a description for your video." |
|
|
if not api_key.strip(): |
|
|
return None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = VideoService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
videos = loop.run_until_complete( |
|
|
service.generate_video( |
|
|
prompt=prompt, |
|
|
duration=int(duration), |
|
|
style=style |
|
|
) |
|
|
) |
|
|
|
|
|
if videos: |
|
|
|
|
|
return videos[0].video_url, "Video generated successfully!" |
|
|
else: |
|
|
return None, "No video was generated. Please try a different prompt." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
|
|
@staticmethod |
|
|
def animate_image( |
|
|
input_image: str, |
|
|
motion_prompt: str, |
|
|
duration: int, |
|
|
api_key: str = "" |
|
|
) -> Tuple[Optional[str], str]: |
|
|
"""Handle image-to-video animation.""" |
|
|
if not input_image: |
|
|
return None, "Please upload an image." |
|
|
if not motion_prompt.strip(): |
|
|
return None, "Please describe the motion you want." |
|
|
if not api_key.strip(): |
|
|
return None, "Please enter your API key in the Settings section." |
|
|
|
|
|
client = StackNetClient(api_key=api_key.strip()) |
|
|
service = VideoService(client=client) |
|
|
|
|
|
try: |
|
|
loop = asyncio.new_event_loop() |
|
|
asyncio.set_event_loop(loop) |
|
|
|
|
|
try: |
|
|
image_url = f"file://{input_image}" |
|
|
|
|
|
videos = loop.run_until_complete( |
|
|
service.animate_image( |
|
|
image_url=image_url, |
|
|
motion_prompt=motion_prompt, |
|
|
duration=int(duration) |
|
|
) |
|
|
) |
|
|
|
|
|
if videos: |
|
|
|
|
|
return videos[0].video_url, "Image animated successfully!" |
|
|
else: |
|
|
return None, "No video was generated. Please try again." |
|
|
|
|
|
finally: |
|
|
loop.close() |
|
|
|
|
|
except Exception as e: |
|
|
return None, format_error(e) |
|
|
|
|
|
finally: |
|
|
service.cleanup() |
|
|
|
