""" title: Beatoven.ai Music Generator Pipe author: Professor Patterns description: Generate music tracks using the Beatoven.ai API version: 0.1.0 license: MIT """ import json import time import httpx from pydantic import BaseModel, Field from typing import Optional, Dict, Any, List, Callable, Awaitable, AsyncGenerator import asyncio class Pipe: class Valves(BaseModel): BEATOVEN_API_KEY: str = Field( default="", description="Your Beatoven.ai API key for authentication" ) BEATOVEN_API_BASE_URL: str = Field( default="https://public-api.beatoven.ai", description="Base URL for Beatoven.ai API", ) AUDIO_FORMAT: str = Field( default="wav", description="Format for generated audio (wav, mp3, aac)" ) ENABLE_LOOPING: bool = Field( default=False, description="Enable higher amount of looping in the track" ) CHECK_INTERVAL: int = Field( default=5, description="Interval in seconds to check for task completion" ) MAX_CHECKS: int = Field( default=60, description="Maximum number of status checks before giving up" ) def __init__(self): self.valves = self.Valves() self.id = "beatoven_music_generator" self.type = "manifold" self.name = "Beatoven.ai: " self.emitter = None def pipes(self) -> List[dict]: return [ { "id": "beatoven-music", "name": "Beatoven.ai Music Generator", } ] async def pipe( self, body: dict, __event_emitter__: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None, ) -> AsyncGenerator[str, None]: """Generate music using Beatoven.ai API""" self.emitter = __event_emitter__ # Validate API key if not self.valves.BEATOVEN_API_KEY: yield json.dumps( {"error": "Beatoven API key is not configured"}, ensure_ascii=False ) return # Extract the last user message to use as prompt last_user_message = self._get_last_user_message(body.get("messages", [])) if not last_user_message: yield json.dumps({"error": "No user message found"}, ensure_ascii=False) return try: # Send status update await self._emit_status("Sending request to Beatoven.ai API...") # Prepare request parameters headers = { "Authorization": f"Bearer {self.valves.BEATOVEN_API_KEY}", "Content-Type": "application/json", } payload = { "prompt": {"text": last_user_message}, "format": self.valves.AUDIO_FORMAT, "looping": self.valves.ENABLE_LOOPING, } # Start composition async with httpx.AsyncClient() as client: response = await client.post( f"{self.valves.BEATOVEN_API_BASE_URL}/api/v1/tracks/compose", json=payload, headers=headers, ) if response.status_code != 200: yield json.dumps( {"error": f"Error starting composition: {response.text}"}, ensure_ascii=False, ) return task_data = response.json() task_id = task_data.get("task_id") if not task_id: yield json.dumps( {"error": "No task ID received from Beatoven API"}, ensure_ascii=False, ) return # Notify user that composition has started yield f"🎵 Music generation has started with prompt: '{last_user_message}'\n\n" yield f"Task ID: `{task_id}`\n\n" # Poll for completion checks = 0 status = "composing" track_data = None while ( status not in ["composed", "failed", "error"] and checks < self.valves.MAX_CHECKS ): checks += 1 await self._emit_status( f"Checking composition status... (Attempt {checks})" ) # Wait before checking await asyncio.sleep(self.valves.CHECK_INTERVAL) # Check status status_response = await client.get( f"{self.valves.BEATOVEN_API_BASE_URL}/api/v1/tasks/{task_id}", headers=headers, ) if status_response.status_code != 200: yield json.dumps( {"error": f"Error checking status: {status_response.text}"}, ensure_ascii=False, ) return task_status = status_response.json() status = task_status.get("status", "unknown") # Provide status updates to the user if status == "composing": yield "⏳ Your track is queued for composition...\n" elif status == "running": yield "🔄 Composition in progress...\n" elif status == "composed": track_data = task_status.get("meta", {}) break # Final results if status == "composed" and track_data: track_url = track_data.get("track_url", "") stems = track_data.get("stems_url", {}) yield f"✅ Music generation complete!\n\n" yield f"## Download Links\n\n" yield f"- [Download Full Track]({track_url})\n\n" if stems: yield f"## Individual Stems\n\n" for stem_name, stem_url in stems.items(): yield f"- [{stem_name.capitalize()}]({stem_url})\n" # Add instructions for usage yield f"\n### How to use\n" yield f"1. Click the links above to download your track and stems\n" yield f"2. You can use these files in your projects under Beatoven.ai's licensing terms\n" elif checks >= self.valves.MAX_CHECKS: yield f"⚠️ Timed out waiting for composition to complete. You can check the status later with task ID: `{task_id}`\n" else: yield f"❌ Composition failed or encountered an error.\n" # Final status update await self._emit_status("Music generation process complete", done=True) except Exception as e: yield json.dumps( {"error": f"An error occurred: {str(e)}"}, ensure_ascii=False ) await self._emit_status(f"Error: {str(e)}", done=True) def _get_last_user_message(self, messages): """Extract the last user message from the conversation""" for message in reversed(messages): if message.get("role") == "user": content = message.get("content") if isinstance(content, str): return content elif isinstance(content, list): # Handle content that might be a list of parts for part in content: if part.get("type") == "text": return part.get("text", "") return "" async def _emit_status( self, description: str, done: bool = False ) -> Awaitable[None]: """Send status updates""" if self.emitter: return await self.emitter( { "type": "status", "data": { "description": description, "done": done, }, } ) return None