Create app.py

app.py

@@ -0,0 +1,144 @@
+import os
+import json
+import asyncio
+import base64
+import numpy as np
+from flask import Flask, request, send_file, jsonify
+from io import BytesIO
+from utils import decode, pcm_to_wav
+from datetime import datetime
+
+app = Flask(__name__)
+
+# Mock GoogleGenAI class (replace with actual SDK if available)
+class GoogleGenAI:
+    def __init__(self, config):
+        self.api_key = config['apiKey']
+        self.api_version = config['apiVersion']
+        self.live = MockLiveMusic()
+
+class MockLiveMusic:
+    def __init__(self):
+        self.music = MockMusic()
+
+class MockMusic:
+    async def connect(self, config):
+        return MockLiveMusicSession(config['model'])
+
+class MockLiveMusicSession:
+    def __init__(self, model):
+        self.model = model
+        self.callbacks = None
+        self.is_playing = False
+
+    async def setWeightedPrompts(self, params):
+        print(f"Setting prompts: {params['weightedPrompts']}")
+
+    async def setMusicGenerationConfig(self, params):
+        print(f"Setting config: {params['musicGenerationConfig']}")
+
+    def play(self):
+        self.is_playing = True
+        print("Starting music generation")
+
+    def close(self):
+        self.is_playing = False
+        if self.callbacks and self.callbacks.get('onclose'):
+            self.callbacks['onclose']()
+
+# Initialize AI client
+ai = GoogleGenAI({
+    'apiKey': os.getenv('GEMINI_API_KEY', 'PLACEHOLDER_API_KEY'),
+    'apiVersion': 'v1alpha'
+})
+model = 'lyria-realtime-exp'
+sample_rate = 48000
+
+@app.route('/generate', methods=['POST'])
+async def generate_music():
+    try:
+        data = request.get_json()
+        if not data:
+            return jsonify({'error': 'No JSON data provided'}), 400
+
+        prompts = data.get('prompts', [])
+        config = data.get('config', {
+            'temperature': 1.1,
+            'topK': 40,
+            'guidance': 4.0
+        })
+
+        if not prompts:
+            return jsonify({'error': 'At least one prompt is required'}), 400
+
+        weighted_prompts = [
+            {
+                'promptId': f"prompt-{i}",
+                'text': prompt['text'],
+                'weight': prompt.get('weight', 1.0),
+                'color': prompt.get('color', '#9900ff')
+            } for i, prompt in enumerate(prompts)
+        ]
+
+        collected_chunks = []
+        total_duration = 0
+        target_duration = 600  # 10 minutes in seconds
+
+        async def mock_generate_chunks():
+            nonlocal collected_chunks, total_duration
+            chunk_duration = 2  # seconds
+            samples_per_chunk = int(sample_rate * chunk_duration * 2)  # stereo, 16-bit
+            while total_duration < target_duration:
+                chunk = np.random.randint(-32768, 32768, samples_per_chunk, dtype=np.int16)
+                encoded_chunk = base64.b64encode(chunk.tobytes()).decode('utf-8')
+                decoded_chunk = decode(encoded_chunk)
+                collected_chunks.append(decoded_chunk)
+                total_duration += chunk_duration
+                yield {'serverContent': {'audioChunks': [{'data': encoded_chunk}]}}
+                await asyncio.sleep(0.1)  # Simulate network delay
+            yield {'close': True}
+
+        session = await ai.live.music.connect({
+            'model': model,
+            'callbacks': {
+                'onmessage': lambda e: None,
+                'onerror': lambda e: print(f"Error: {e}"),
+                'onclose': lambda: print("Session closed")
+            }
+        })
+
+        await session.setWeightedPrompts({'weightedPrompts': weighted_prompts})
+        await session.setMusicGenerationConfig({'musicGenerationConfig': config})
+        session.play()
+
+        async for message in mock_generate_chunks():
+            if message.get('close'):
+                break
+            if message.get('serverContent', {}).get('audioChunks'):
+                chunk_data = message['serverContent']['audioChunks'][0]['data']
+                decoded_chunk = decode(chunk_data)
+                collected_chunks.append(decoded_chunk)
+                chunk_duration = len(decoded_chunk) / (sample_rate * 2 * 2)
+                total_duration += chunk_duration
+
+        total_length = sum(len(chunk) for chunk in collected_chunks)
+        combined_pcm = np.concatenate([np.frombuffer(chunk, dtype=np.int16) for chunk in collected_chunks])
+        combined_pcm_bytes = combined_pcm.tobytes()
+        wav_blob = pcm_to_wav(combined_pcm_bytes, 2, sample_rate, 16)
+
+        output = BytesIO(wav_blob)
+        timestamp = datetime.now().isoformat().replace(':', '-')
+        filename = f"prompt-dj-music-10min-{timestamp}.wav"
+
+        return send_file(
+            output,
+            mimetype='audio/wav',
+            as_attachment=True,
+            download_name=filename
+        )
+
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=5000)