import os import logging try: import gradio as gr except ModuleNotFoundError: # pragma: no cover - local test environments may omit gradio gr = None import matplotlib matplotlib.use('Agg') import tempfile import time from rhythma import RhythmaModulationEngine, RhythmaSymphAICore logging.basicConfig(level=logging.INFO) LOGGER = logging.getLogger(__name__) # --- 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: LOGGER.warning( "GROQ_API_KEY not found. Advanced LLM analysis and transcription are disabled." ) else: LOGGER.info("GROQ_API_KEY found. Enabling Groq features.") # --- End Environment Variable Check --- # --- Initialize Core Components --- try: symphai_core = RhythmaSymphAICore(use_groq=use_groq) except Exception as e: LOGGER.exception("Could not initialize RhythmaSymphAICore: %s", e) symphai_core = None # --- End Initialization --- # --- Core Functions --- def analyze_input(input_text=None, audio_input=None): if symphai_core is None: return {"error": "Analysis Core failed to initialize."} audio_filepath = audio_input if isinstance(audio_input, str) else None return symphai_core.analyze_input(input_text or "", audio_filepath) def generate_modulated_experience(analysis_result, base_freq=None, modulation_type="sine", rhythm_pattern=None, duration=5): if not isinstance(analysis_result, dict): error_msg = "Internal error: analysis result is not in the expected format." LOGGER.error(error_msg) return error_msg, None, None, None, None if analysis_result.get("error"): error_msg = f"Analysis Error: {analysis_result['error']}" LOGGER.error(error_msg) return error_msg, None, None, None, None emotional_state = analysis_result.get("emotional_state", "neutral") session_profile = build_runtime_session_profile( analysis_result, base_freq=base_freq, modulation_type=modulation_type, rhythm_pattern=rhythm_pattern, ) final_rhythm_pattern = ( session_profile.get("pattern") or rhythm_pattern or analysis_result.get("rhythm_pattern") or "calm" ) final_modulation_type = session_profile.get("modulation_type") or modulation_type final_base_freq = session_profile.get("tone_center") if not final_base_freq: final_base_freq = base_freq if base_freq and base_freq > 0 else None try: engine = RhythmaModulationEngine( base_freq=final_base_freq, modulation_type=final_modulation_type, rhythm_pattern=final_rhythm_pattern, emotional_state=emotional_state if not final_base_freq else None, ) if session_profile: engine.generate_modulated_wave = ( lambda requested_duration, _engine=engine, _profile=dict(session_profile): _engine.render_session(_profile, requested_duration) ) timestamp = int(time.time()) temp_dir = tempfile.gettempdir() os.makedirs(temp_dir, exist_ok=True) audio_file = os.path.join(temp_dir, f"rhythma_{timestamp}.wav") saved_audio_path = engine.save_audio(duration, audio_file) if not saved_audio_path: raise RuntimeError("Failed to save generated audio file.") fig = engine.visualize_waveform(duration) waveform_image = engine.get_waveform_image() if session_profile: analysis_text = "\n\n".join( [ f"Session: {session_profile.get('title', 'Session')}", f"Tone: {session_profile.get('emotional_tone', emotional_state.title())}", f"Listening Path: {session_profile.get('guidance', '')}", ] ) symbolic = session_profile.get("reflection") or engine.get_symbolic_interpretation() else: analysis_text = engine.get_complete_analysis() symbolic = engine.get_symbolic_interpretation() return analysis_text, saved_audio_path, fig, waveform_image, symbolic except Exception as e: error_msg = f"Error during Rhythma generation: {e}" LOGGER.exception(error_msg) return error_msg, None, None, None, None def coerce_frequency(value): try: numeric_value = float(value) except (TypeError, ValueError): return 0.0 return numeric_value if numeric_value > 0 else 0.0 def normalize_rhythm_override(value): if value in (None, "", "Automatic"): return None return value SESSION_EXAMPLES = [ ["I need something steady before a difficult conversation.", None, None, "sine", "Automatic", 12], ["I want to feel grounded and open as the evening slows down.", None, None, "sine", "Automatic", 18], ["I need a clear horizon for deep work.", None, None, "sine", "focused", 20], ["Everything feels loud and I want a softer landing.", None, None, "sine", "Automatic", 14], ["I feel bright and want a livelier pulse without losing calm.", None, None, "pulse", "active", 10], ["Give me a long unwind after a heavy day.", None, None, "sine", "relaxed", 30], ["I want a gentle session for a low-energy morning.", None, None, "sine", "Automatic", 16], ] def build_runtime_session_profile( analysis_result, base_freq=None, modulation_type="sine", rhythm_pattern=None, ): analysis_result = analysis_result if isinstance(analysis_result, dict) else {} profile = analysis_result.get("session_profile") or {} if not profile: return {} if symphai_core is not None: return symphai_core.apply_profile_overrides( profile, tone_center=base_freq, modulation_type=modulation_type, session_pattern=rhythm_pattern, ) shaped_profile = dict(profile) if base_freq is not None and base_freq > 0: shaped_profile["tone_center"] = base_freq if modulation_type: shaped_profile["modulation_type"] = modulation_type if rhythm_pattern: shaped_profile["pattern"] = rhythm_pattern return shaped_profile def build_session_copy(analysis_result): analysis_result = analysis_result if isinstance(analysis_result, dict) else {} profile = analysis_result.get("session_profile") or {} emotional_state = analysis_result.get("emotional_state", "neutral") session_name = profile.get("title") or f"{emotional_state.title()} Session" emotional_tone = profile.get("emotional_tone") or "Measured and receptive" guidance = profile.get("guidance") or "Stay with the pulse until your breath settles into its own pace." reflection = profile.get("reflection") or "This session offers a gentle reset without pushing for intensity." return { "session_name": session_name, "emotional_tone": emotional_tone, "listening_path": f"{emotional_tone}. {guidance}", "session_reflection": reflection, "tone_center": profile.get("tone_center"), "session_pattern": profile.get("pattern"), } def rhythma_experience( input_text, audio_input, override_freq=None, override_modulation="sine", override_rhythm=None, duration=5 ): input_text = input_text.strip() if input_text else "" freq_override_value = coerce_frequency(override_freq) analysis = analyze_input(input_text, audio_input) session_copy = build_session_copy(analysis) analysis_text, audio_file, fig, waveform_image, symbolic = generate_modulated_experience( analysis, base_freq=freq_override_value, modulation_type=override_modulation, rhythm_pattern=normalize_rhythm_override(override_rhythm), duration=duration ) transcription = analysis.get("transcription", "") if isinstance(analysis, dict) else "" plot_output = fig if fig else None listening_path = session_copy["listening_path"] session_reflection = session_copy["session_reflection"] if audio_file is None and isinstance(analysis_text, str) and analysis_text: fallback_message = analysis_text or symbolic or "Session generation is unavailable right now." listening_path = fallback_message session_reflection = fallback_message return ( f"### {session_copy['session_name']}", session_copy["emotional_tone"], listening_path, audio_file, plot_output, waveform_image, session_reflection, transcription, ) # --- Create the Gradio Interface --- def create_interface(): if gr is None: raise ModuleNotFoundError("gradio is required to create the Rhythma interface.") with gr.Blocks(theme=gr.themes.Soft(primary_hue="rose", secondary_hue="stone"), title="Rhythma") as demo: gr.Markdown("# Rhythma") gr.Markdown("### An artful wellness companion for reflective listening.") if not use_groq: gr.Warning( "Groq analysis is unavailable. Text-led sessions still work, but live voice transcription is off." ) with gr.Row(): with gr.Column(scale=1): gr.Markdown("**1. Share what you're carrying**") input_text = gr.Textbox( label="How are you feeling, or what intention would you like to hold?", placeholder="e.g., 'I need something steady before a conversation' or 'I need room to chill after a long day.'", lines=4 ) gr.Markdown("**Optional: add a voice note (requires Groq)**") audio_input = gr.Audio( sources=["microphone"], type="filepath", label="Record or Upload a Voice Note" if use_groq else "Voice Note (Disabled)", interactive=use_groq ) with gr.Accordion("Session shaping controls", open=False): override_freq = gr.Number( value=None, minimum=0, precision=0, label="Tone Center (Hz)", placeholder="Automatic", info="Leave blank to let Rhythma choose a tone center from your session profile." ) override_modulation = gr.Dropdown( choices=["sine", "pulse", "chirp"], value="sine", label="Texture Shape" ) available_patterns = list(RhythmaModulationEngine().rhythm_configs.keys()) override_rhythm = gr.Dropdown( choices=["Automatic"] + available_patterns, value="Automatic", label="Session Pattern", info="Leave on Automatic to follow the pattern inferred from your session profile." ) duration = gr.Slider( minimum=3, maximum=60, value=10, step=1, label="Session Length (seconds)" ) generate_button = gr.Button("Begin Exploration", variant="primary", scale=2) with gr.Column(scale=2): gr.Markdown("**2. Receive your listening session**") gr.Markdown( "_Rhythma shapes a named listening path, then renders the audio, reflection, and waveform around it._" ) session_name_output = gr.Markdown("### Session") emotional_tone_output = gr.Markdown("Measured and receptive") listening_path_output = gr.Textbox(label="Your Listening Path", lines=6, interactive=False) with gr.Row(): audio_output = gr.Audio(label="Session Audio", type="filepath", interactive=False) waveform_simple = gr.Image(label="Tone Center", interactive=False, height=100, width=200) waveform_plot = gr.Plot(label="Session Pattern") symbolic_output = gr.Textbox(label="Session Reflection", interactive=False) transcription_output = gr.Textbox( label="Transcribed Voice Note", interactive=False, visible=use_groq ) generate_button.click( fn=rhythma_experience, inputs=[ input_text, audio_input, override_freq, override_modulation, override_rhythm, duration ], outputs=[ session_name_output, emotional_tone_output, listening_path_output, audio_output, waveform_plot, waveform_simple, symbolic_output, transcription_output ] ) gr.Examples( examples=SESSION_EXAMPLES, inputs=[input_text, audio_input, override_freq, override_modulation, override_rhythm, duration], outputs=[ session_name_output, emotional_tone_output, listening_path_output, audio_output, waveform_plot, waveform_simple, symbolic_output, transcription_output ], fn=rhythma_experience, cache_examples=False ) gr.Markdown("---") gr.Markdown(""" ## About Rhythma Rhythma is an artful wellness companion that turns a felt state into a reflective listening session. It uses optional AI analysis, session profiling, and rhythmic sound design to shape a tone center, pattern, and guided path for the moment you are in. **Note:** Rhythma is for reflective listening and personal wellbeing rituals. It is not medical advice or a clinical treatment. © 2026 Vers3Dynamics """) return demo # --- Run the Gradio App --- if __name__ == "__main__": if symphai_core is None: LOGGER.error("Cannot launch Gradio app because RhythmaSymphAICore failed to initialize.") elif gr is None: LOGGER.error("Cannot launch Gradio app because gradio is not installed.") else: app_demo = create_interface() app_demo.launch()