Update app.py

app.py

@@ -1,115 +1,291 @@
-# app.py
 import os
-import tempfile
-import base64
-import io
+import gradio as gr
+import numpy as np
+import matplotlib
+matplotlib.use('Agg') # Set backend BEFORE importing pyplot
 import matplotlib.pyplot as plt
-from fastapi import FastAPI, UploadFile, File, Form
-from fastapi.responses import JSONResponse
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.staticfiles import StaticFiles
-import uvicorn
-
-# Hugging Face Spaces specific setup
-os.environ["HF_HOME"] = "/tmp/hf_cache"
-os.environ["SENTENCE_TRANSFORMERS_HOME"] = "/tmp/hf_cache"
-os.environ["TRANSFORMERS_CACHE"] = "/tmp/hf_cache"
-
-from rhythma import RhythmaSymphAICore, RhythmaModulationEngine
-
-app = FastAPI(title="Rhythma: The Living Modulation Engine")
-
-# Enable CORS for the frontend
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# Mount static files so index.html is accessible
-app.mount("/static", StaticFiles(directory="."), name="static")
-
-# Initialize the core components
-symphai = RhythmaSymphAICore(use_groq=True)
-
-@app.post("/generate")
-async def generate(
-    input_text: str = Form(""),
-    audio: UploadFile = File(None),
-    override_freq: float = Form(0.0),
-    override_modulation: str = Form("sine"),
-    override_rhythm: str = Form("auto"),
-    duration: int = Form(10),
-):
-    audio_path = None
-    if audio and audio.filename:
-        # Save uploaded audio to temp file
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
-            tmp.write(await audio.read())
-            audio_path = tmp.name
+from PIL import Image
+import soundfile as sf
+import tempfile
+import time
+from rhythma import RhythmaModulationEngine, RhythmaSymphAICore # Assuming rhythma.py is in the same directory
 
-    try:
-        # Step 1: Analyze input using your SymphAI Core
-        analysis = symphai.analyze_input(
-            input_text.strip() or None,
-            audio_path
-        )
+# --- Environment Variable Check ---
+GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
+use_groq = bool(GROQ_API_KEY) # True only if key exists and is not empty
+
+if not use_groq:
+    print("*"*40)
+    print("⚠️ WARNING: GROQ_API_KEY not found or empty in environment variables.")
+    print("   Groq LLM analysis and audio transcription features will be disabled.")
+    print("   Falling back to local analysis methods.")
+    print("*"*40)
+else:
+    print("✅ GROQ_API_KEY found. Enabling Groq features.")
+# --- End Environment Variable Check ---
+
+
+# --- Initialize Core Components ---
+try:
+    # Pass the determined use_groq flag to the core
+    symphai_core = RhythmaSymphAICore(use_groq=use_groq)
+except Exception as e:
+     print(f"❌ FATAL ERROR: Could not initialize RhythmaSymphAICore: {e}")
+     # Handle fatal error appropriately - maybe exit or disable functionality
+     symphai_core = None # Indicate failure
+# --- End Initialization ---
+
+
+# --- Core Functions ---
+def analyze_input(input_text=None, audio_input=None):
+    """Analyze user input using the SymphAI Core."""
+    if symphai_core is None:
+        return {"error": "Analysis Core failed to initialize."}
+
+    # Ensure audio_input is a filepath string or None
+    audio_filepath = audio_input if isinstance(audio_input, str) else None
+
+    # Pass to SymphAI Core for analysis
+    # Add default empty string for input_text if None, as core expects string or None
+    return symphai_core.analyze_input(input_text or "", audio_filepath)
 
-        # Step 2: Generate modulated experience
+
+def generate_modulated_experience(analysis_result, base_freq=None, modulation_type="sine", rhythm_pattern=None, duration=5):
+    """Generate a complete modulated experience based on analysis and parameters."""
+    print(f"DEBUG: generate_modulated_experience received analysis: {analysis_result}")
+    print(f"DEBUG: Overrides - Freq: {base_freq}, Mod: {modulation_type}, Rhythm: {rhythm_pattern}, Dur: {duration}")
+
+    # --- Input Validation ---
+    if not isinstance(analysis_result, dict):
+        error_msg = "Internal Error: Analysis result is not in the expected format."
+        print(f"❌ {error_msg}")
+        return error_msg, None, None, None, None
+
+    if "error" in analysis_result and analysis_result["error"]:
+        error_msg = f"Analysis Error: {analysis_result['error']}"
+        print(f"❌ {error_msg}")
+        # Return the error message clearly for the analysis output
+        return error_msg, None, None, None, None
+
+    # Ensure required keys exist, even if defaults were used in analysis
+    emotional_state = analysis_result.get("emotional_state", "neutral") # Default if missing
+    rhythm_pattern_from_analysis = analysis_result.get("rhythm_pattern", "calm") # Default if missing
+
+    # --- Determine Final Parameters ---
+    # Use manual override if provided and valid, otherwise use analysis result
+    final_rhythm_pattern = rhythm_pattern if rhythm_pattern else rhythm_pattern_from_analysis
+    # Use manual frequency override ONLY if it's > 0
+    final_base_freq = base_freq if base_freq and base_freq > 0 else None # Pass None to let engine use emotion/default
+
+    print(f"DEBUG: Engine Params - Emotion: {emotional_state}, Freq Override: {final_base_freq}, Rhythm: {final_rhythm_pattern}, Mod: {modulation_type}")
+
+    try:
+        # --- Initialize the Rhythma Engine ---
         engine = RhythmaModulationEngine(
-            base_freq=override_freq if override_freq > 0 else None,
-            modulation_type=override_modulation,
-            rhythm_pattern=override_rhythm if override_rhythm != "auto" else None,
-            emotional_state=analysis.get("emotional_state")
+            base_freq=final_base_freq,  # Engine handles None: uses emotional_state or default
+            modulation_type=modulation_type,
+            rhythm_pattern=final_rhythm_pattern,
+            emotional_state=emotional_state if not final_base_freq else None # Pass emotion only if freq isn't overridden
         )
 
-        # Generate audio file
-        timestamp = int(os.times()[4] * 1000)
-        audio_file = f"rhythma_{timestamp}.wav"
+        # --- Generate Outputs ---
+        timestamp = int(time.time())
+        temp_dir = tempfile.gettempdir()
+        # Ensure temp_dir exists (useful in some restricted environments)
+        os.makedirs(temp_dir, exist_ok=True)
+        audio_file = os.path.join(temp_dir, f"rhythma_{timestamp}.wav")
+
+        # Generate and save audio
         saved_audio_path = engine.save_audio(duration, audio_file)
+        if not saved_audio_path: # Check if saving failed
+             raise RuntimeError("Failed to save generated audio file.")
 
-        # Generate visualizations
-        waveform_pil = engine.get_waveform_image()
+        # Generate waveform visualization (Plot)
         fig = engine.visualize_waveform(duration)
 
-        # Convert matplotlib figure to base64
-        buf = io.BytesIO()
-        fig.savefig(buf, format="png", bbox_inches="tight", dpi=220)
-        buf.seek(0)
-        plot_base64 = base64.b64encode(buf.read()).decode("utf-8")
-        plt.close(fig)
-
-        # Convert simple waveform PIL image to base64
-        buf = io.BytesIO()
-        waveform_pil.save(buf, format="PNG")
-        buf.seek(0)
-        simple_wave_base64 = base64.b64encode(buf.read()).decode("utf-8")
-
-        # Return all data to the frontend
-        return {
-            "analysis_text": engine.get_complete_analysis(),
-            "audio_base64": base64.b64encode(open(saved_audio_path, "rb").read()).decode("utf-8"),
-            "plot_base64": plot_base64,
-            "waveform_base64": simple_wave_base64,
-            "symbolic_text": engine.get_symbolic_interpretation(),
-            "transcription": analysis.get("transcription", ""),
-            "emotional_state": analysis.get("emotional_state"),
-            "rhythm_pattern": analysis.get("rhythm_pattern")
-        }
+        # Get simple waveform image (PIL Image)
+        waveform_image = engine.get_waveform_image()
+
+        # Get complete analysis text from the engine's perspective
+        analysis_text = engine.get_complete_analysis()
+
+        # Get symbolic interpretation
+        symbolic = engine.get_symbolic_interpretation()
+
+        print("✅ Modulation experience generated successfully.")
+        return analysis_text, saved_audio_path, fig, waveform_image, symbolic
 
     except Exception as e:
+        error_msg = f"Error during Rhythma generation: {e}"
+        print(f"❌ {error_msg}")
         import traceback
         traceback.print_exc()
-        return JSONResponse(status_code=500, content={"error": str(e)})
+        # Return error message for analysis, and None for other outputs
+        return error_msg, None, None, None, None
+
+
+def rhythma_experience(
+    input_text, audio_input,
+    override_freq=None,
+    override_modulation="sine",
+    override_rhythm=None,
+    duration=5
+):
+    """Complete Rhythma experience pipeline: Analysis -> Generation"""
+    print("\n--- Starting New Rhythma Experience ---")
+    # Clean up input text
+    input_text = input_text.strip() if input_text else ""
+
+    # --- Step 1: Analyze input ---
+    # Ensure override_freq is float or None
+    try:
+         freq_override_value = float(override_freq) if override_freq is not None else 0.0
+    except (ValueError, TypeError):
+         freq_override_value = 0.0 # Default to 0 if invalid input
+
+    analysis = analyze_input(input_text, audio_input)
+
+    # --- Step 2: Generate modulated experience ---
+    # Pass analysis results and overrides to the generation function
+    analysis_text, audio_file, fig, waveform_image, symbolic = generate_modulated_experience(
+        analysis,
+        base_freq=freq_override_value, # Pass the validated float/int
+        modulation_type=override_modulation,
+        rhythm_pattern=override_rhythm if override_rhythm else None, # Pass None if dropdown default is selected
+        duration=duration
+    )
+
+    # --- Step 3: Prepare Outputs ---
+    # Get transcription from analysis result (will be empty string if no audio/transcription)
+    transcription = analysis.get("transcription", "") if isinstance(analysis, dict) else ""
+    # If analysis itself failed, analysis_text will contain the error message from generate_modulated_experience
+    # If only transcription failed, it might be in the transcription field
+
+    # Handle potential None figure if generation failed
+    plot_output = fig if fig else None # Gradio handles None for Plot output
+
+    print("--- Rhythma Experience Complete ---")
+    # Return all outputs for Gradio interface
+    return analysis_text, audio_file, plot_output, waveform_image, symbolic, transcription
+
+# --- Create the Gradio Interface ---
+def create_interface():
+    with gr.Blocks(theme=gr.themes.Soft(), title="Rhythma: The Living Modulation Engine") as demo:
+        gr.Markdown("# Rhythma: The Living Modulation Engine")
+        gr.Markdown("### Dynamic rhythm-based sound modulation for wellbeing from Vers3Dynamics")
+
+        if not use_groq:
+            gr.Warning("Running with limited functionality: GROQ_API_KEY not found. "
+                       "Advanced AI analysis and audio transcription are disabled.")
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                gr.Markdown("**1. Describe Your State or Intention**")
+                input_text = gr.Textbox(
+                    label="How are you feeling, or what is your intention?",
+                    placeholder="e.g., 'feeling stressed about work', 'want to relax', 'need focus'...",
+                    lines=3
+                )
+
+                gr.Markdown("**Optional: Use Your Voice (Requires Groq API Key)**")
+                audio_input = gr.Audio(
+                    sources=["microphone"], # Prioritize microphone
+                    type="filepath", # RhythmaSymphAICore expects a filepath
+                    label="Record or Upload Audio" if use_groq else "Audio Input (Disabled)",
+                    interactive=use_groq # Disable if Groq not available
+                )
+
+                with gr.Accordion("Advanced Settings (Optional Overrides)", open=False):
+                    override_freq = gr.Slider(
+                        minimum=0, maximum=1000, value=0, step=1,
+                        label="Override Frequency (Hz)",
+                        info="Leave at 0 to use automatic frequency based on analysis."
+                    )
+                    override_modulation = gr.Dropdown(
+                        choices=["sine", "pulse", "chirp"],
+                        value="sine",
+                        label="Override Modulation Type"
+                    )
+                    # Get available patterns from the engine instance
+                    available_patterns = list(RhythmaModulationEngine().rhythm_configs.keys())
+                    override_rhythm = gr.Dropdown(
+                        choices=[None] + available_patterns, # Add None option for automatic
+                        value=None, # Default to automatic
+                        label="Override Rhythm Pattern",
+                        info="Leave blank to use automatic pattern based on analysis."
+                    )
+                    duration = gr.Slider(
+                        minimum=3, maximum=60, value=10, step=1,
+                        label="Duration (seconds)"
+                    )
+
+                generate_button = gr.Button("Generate Rhythma Experience", variant="primary", scale=2)
+
+            with gr.Column(scale=2):
+                gr.Markdown("**2. Experience Your Rhythma Soundscape**")
+                analysis_output = gr.Textbox(label="Rhythma Analysis & Guidance", lines=8, interactive=False)
+                with gr.Row():
+                    audio_output = gr.Audio(label="Modulated Audio", type="filepath", interactive=False)
+                    waveform_simple = gr.Image(label="Base Waveform", interactive=False, height=100, width=200)
+                waveform_plot = gr.Plot(label="Detailed Waveform & Spectrogram")
+                symbolic_output = gr.Textbox(label="Symbolic Interpretation", interactive=False)
+                # Conditionally visible transcription output
+                transcription_output = gr.Textbox(
+                    label="Transcribed Audio (If Provided)",
+                    interactive=False,
+                    visible=use_groq # Only show if Groq is potentially usable
+                )
+
+        # Define button action
+        generate_button.click(
+            fn=rhythma_experience,
+            inputs=[
+                input_text, audio_input,
+                override_freq, override_modulation, override_rhythm,
+                duration
+            ],
+            outputs=[
+                analysis_output, audio_output,
+                waveform_plot, waveform_simple, symbolic_output,
+                transcription_output
+            ]
+        )
+
+        # Add Examples
+        gr.Examples(
+            examples=[
+                ["I'm feeling anxious about my upcoming presentation.", None, 0, "sine", None, 10],
+                ["I feel at peace and grounded today.", None, 0, "sine", None, 15],
+                ["I need to focus on my work but keep getting distracted.", None, 0, "sine", None, 20],
+                ["Feeling overwhelmed with responsibilities.", None, 0, "sine", None, 10],
+                ["Excited about my vacation next week!", None, 0, "sine", None, 10],
+                ["Just want to relax after a long day.", None, 0, "sine", "relaxed", 30], # Example with override
+                ["Feeling sad and low energy.", None, 0, "sine", None, 15],
+            ],
+            inputs=[input_text, audio_input, override_freq, override_modulation, override_rhythm, duration],
+            outputs=[analysis_output, audio_output, waveform_plot, waveform_simple, symbolic_output, transcription_output],
+            fn=rhythma_experience, # Ensure examples also run the main function
+            cache_examples=False # Maybe disable caching during development
+        )
 
-    finally:
-        # Clean up temporary audio file
-        if audio_path and os.path.exists(audio_path):
-            os.unlink(audio_path)
+        gr.Markdown("---")
+        gr.Markdown("""
+        ## About Rhythma
+        Rhythma creates personalized soundscapes using frequency modulation based on your described emotional state or intention.
+        It leverages AI analysis (enhanced with Groq if available) and principles of rhythmic sound design.
+        **Note:** This is an experimental tool. The frequencies and interpretations are based on various theories and are not medical advice.
+        © 2025 Vers3Dynamics 
+        """)
 
+    return demo
 
-# For local testing only (Hugging Face uses the Dockerfile CMD)
+# --- Run the Gradio App ---
 if __name__ == "__main__":
-    uvicorn.run("app:app", host="0.0.0.0", port=7860)
+    if symphai_core is None:
+         print("\n❌ Cannot launch Gradio app because RhythmaSymphAICore failed to initialize.\n")
+    else:
+        print("\n🚀 Launching Rhythma Gradio Interface...")
+        app_demo = create_interface()
+        # Set share=True if you need a public link (useful for testing deployment)
+        # Set debug=True for more verbose logs during development
+        app_demo.launch()#debug=True)