""" 🌿 Haven Kitchen OS - Streamlit Interface The main chat interface for Olivia & Brie """ import streamlit as st import asyncio from datetime import datetime from dotenv import load_dotenv import sys import os # Add package directory to path for imports when running as subprocess _pkg_dir = os.path.dirname(os.path.abspath(__file__)) if _pkg_dir not in sys.path: sys.path.insert(0, _pkg_dir) # Now import modules (works both as package and standalone) from src.config import SystemConfig from src.agents.brain import KitchenBrain from src.utils.card_gen import generate_luxury_card from speech import speak, olivia_handoff_to_brie, celebrate_completion, start_listening from recipe_manager import get_recipe_session, parse_recipe from recipe_cast import cast_recipe, generate_qr_code, get_cast_url, start_cast_server from animations import animate, idle_mode from audio_recorder_streamlit import audio_recorder import speech_recognition as sr import io load_dotenv() st.set_page_config(page_title="Haven", page_icon="🌿", layout="wide") # 🎤 Voice-to-Text Helper def transcribe_audio(audio_bytes): """Convert audio bytes to text using Google Speech Recognition.""" if not audio_bytes: return None recognizer = sr.Recognizer() try: audio_file = io.BytesIO(audio_bytes) with sr.AudioFile(audio_file) as source: audio_data = recognizer.record(source) text = recognizer.recognize_google(audio_data) if text: text = text[0].upper() + text[1:] if len(text) > 1 else text.upper() return text except sr.UnknownValueError: st.warning("🎤 Couldn't understand audio. Please try again.") return None except sr.RequestError as e: st.error(f"🎤 Speech recognition error: {e}") return None except Exception as e: st.error(f"🎤 Audio error: {e}") return None # Start cast server on app load start_cast_server() # --- THE SAGE & LATTE THEME --- st.markdown(""" """, unsafe_allow_html=True) def get_greeting(): h = datetime.now().hour return "Good Morning" if 5<=h<12 else "Good Afternoon" if 12<=h<17 else "Good Evening" TIME_OPTIONS = [datetime.strptime(f"{h}:{m}", "%H:%M").strftime("%I:%M %p") for h in range(24) for m in (0, 15, 30, 45)] def show_cooking_mode_banner(recipe_title: str, current_step: int, total_steps: int, is_active: bool): """Display the cooking mode status banner.""" if is_active: st.markdown(f"""

👩🏻‍🍳 Cooking: {recipe_title}

Step {current_step + 1} of {total_steps}
🗣️ "Next" 🗣️ "Repeat" 🗣️ "Back" 🗣️ "Ingredients"
""", unsafe_allow_html=True) else: st.markdown(f"""

👩🏻‍🍳 Recipe Ready: {recipe_title}

🗣️ "Let's cook" to start 🗣️ "Ingredients" for the list
""", unsafe_allow_html=True) def main(): if "brain" not in st.session_state: st.session_state.brain = KitchenBrain() if "recipe_session" not in st.session_state: st.session_state.recipe_session = get_recipe_session() config = st.session_state.brain.cfg recipe_session = st.session_state.recipe_session st.markdown("""
👩🏻‍💼   🌿   👩🏻‍🍳
Olivia & Brie
The Haven System
""", unsafe_allow_html=True) tab_home, tab_settings = st.tabs(["🌿 Living Room", "⚙️ Settings"]) with tab_home: # Show cooking mode banner if recipe is active if recipe_session.has_active_recipe(): recipe = recipe_session.current_recipe show_cooking_mode_banner( recipe.title, recipe.current_step, recipe.total_steps, recipe_session.cooking_mode ) if "messages" not in st.session_state: history = st.session_state.brain.memory.load_history() if history: st.session_state.messages = history else: greeting = get_greeting() st.session_state.messages = [{ "role": "assistant", "content": f"**{greeting}, {config.user_name}.**\n\nWelcome home. I'm reviewing the daily agenda, and Brie is standing by in the kitchen.", "avatar": "👩🏻‍💼" }] for msg in st.session_state.messages: with st.chat_message(msg["role"], avatar=msg.get("avatar", "👩🏻‍💼")): clean = msg["content"].replace("**Brie:**", "").strip() st.markdown(clean) # Show cast button for recipe messages if msg.get("has_recipe") and msg.get("cast_url"): col1, col2 = st.columns([3, 1]) with col1: st.markdown(f"📱 **Send to your phone:** `{msg['cast_url']}`") with col2: if msg.get("qr_code"): st.image(msg["qr_code"], width=80) st.markdown("
", unsafe_allow_html=True) # 🎤 Voice Input - integrated into input bar if "last_audio_hash" not in st.session_state: st.session_state.last_audio_hash = None if "voice_input" not in st.session_state: st.session_state.voice_input = "" if "input_key" not in st.session_state: st.session_state.input_key = 0 st.markdown(""" """, unsafe_allow_html=True) # Input bar with mic on left, text input on right mic_col, input_col, send_col = st.columns([1, 10, 1]) with mic_col: audio_bytes = audio_recorder( text="", recording_color="#e74c3c", neutral_color="#6B8E6B", icon_name="microphone", icon_size="lg", pause_threshold=2.0, sample_rate=16000 ) # Process voice input if we got NEW audio prompt = None voice_text = "" if audio_bytes: audio_hash = hash(audio_bytes) if audio_hash != st.session_state.last_audio_hash: st.session_state.last_audio_hash = audio_hash transcribed = transcribe_audio(audio_bytes) if transcribed: voice_text = transcribed prompt = transcribed st.toast(f"🎤 \"{prompt}\"") with input_col: typed_prompt = st.text_input( "message", placeholder="Chat with Olivia... (or click 🎤 to speak)", label_visibility="collapsed", key=f"text_input_{st.session_state.input_key}" ) with send_col: send_clicked = st.button("➤", use_container_width=True) # Handle send button or Enter key if typed_prompt and send_clicked: prompt = typed_prompt st.session_state.input_key += 1 elif typed_prompt: if "last_typed" not in st.session_state: st.session_state.last_typed = "" if typed_prompt != st.session_state.last_typed: prompt = typed_prompt st.session_state.last_typed = typed_prompt st.session_state.input_key += 1 if prompt: st.session_state.messages.append({"role": "user", "content": prompt, "avatar": config.user_avatar}) with st.chat_message("user", avatar=config.user_avatar): st.markdown(prompt) # 🎙️ Check for recipe voice commands FIRST voice_response = recipe_session.handle_command(prompt) if voice_response: if any(word in prompt.lower() for word in ["next", "done", "finished"]): animate("step_complete", persona="Brie") with st.chat_message("assistant", avatar="👩🏻‍🍳"): st.markdown(voice_response) speak(voice_response, "Brie") st.session_state.messages.append({ "role": "assistant", "content": voice_response, "avatar": "👩🏻‍🍳" }) st.session_state.brain.memory.save_interaction("assistant", voice_response, "👩🏻‍🍳") st.session_state.last_typed = "" st.rerun() # Normal routing to Olivia/Brie start_listening() persona, handoff, generator = asyncio.run(st.session_state.brain.route_and_process(prompt)) av = "👩🏻‍🍳" if persona == "Brie" else "👩🏻‍💼" full_response = "" with st.chat_message("assistant", avatar=av): if handoff: olivia_handoff_to_brie() st.markdown(handoff) full_response += handoff + "\n\n" placeholder = st.empty() streamed_text = "" async def consume(): nonlocal streamed_text async for chunk in generator: streamed_text += chunk placeholder.markdown(streamed_text + "▌") placeholder.markdown(streamed_text) asyncio.run(consume()) full_response += streamed_text # 📖 Check if this is a recipe response from Brie recipe = None cast_url = None qr_bytes = None if persona == "Brie": recipe = recipe_session.load_recipe(full_response) if recipe: animate("excited", persona="Brie") speak(recipe.get_intro(), "Brie") cast_url = cast_recipe(recipe.to_dict()) qr_bytes = generate_qr_code(cast_url) st.markdown("---") st.info("🗣️ **Ready to cook?** Just say **'let's cook'** to start, or **'ingredients'** for the shopping list!") col1, col2 = st.columns([2, 1]) with col1: st.markdown(f"📱 **View on your phone:**") st.code(cast_url, language=None) with col2: if qr_bytes: st.image(qr_bytes, caption="Scan me!", width=100) try: card_img = generate_luxury_card(recipe.title, full_response.replace("**", "")) st.download_button("📄 Save Recipe Card", card_img, f"{recipe.title}.png", "image/png") except: pass else: speak(streamed_text, persona) else: speak(streamed_text, persona) msg_data = { "role": "assistant", "content": full_response, "avatar": av, "has_recipe": recipe is not None, "cast_url": cast_url, "qr_code": qr_bytes } st.session_state.brain.memory.save_interaction("assistant", full_response, av) st.session_state.messages.append(msg_data) st.session_state.last_typed = "" st.rerun() with tab_settings: st.header("Preferences") c1, c2 = st.columns(2) with c1: new_name = st.text_input("Name", value=config.user_name) avatars = ["👩", "👩🏻", "👩🏼", "👩🏽", "👩🏾", "👩🏿", "👨", "👨🏻", "👨🏼", "👨🏽", "👨🏾", "👨🏿"] try: idx = avatars.index(config.user_avatar) except: idx = 0 new_avatar = st.selectbox("Avatar", options=avatars, index=idx) with c2: new_loc = st.text_input("Location", value=config.location) new_diet = st.text_area("Diet", value=config.dietary_restrictions) st.subheader("Meal Times") def t_idx(t): return TIME_OPTIONS.index(t) if t in TIME_OPTIONS else 0 cb, cl, cd = st.columns(3) with cb: st.markdown("**Breakfast**") bs = st.selectbox("Start", TIME_OPTIONS, index=t_idx(config.meal_times['breakfast']['start']), key="bs") be = st.selectbox("End", TIME_OPTIONS, index=t_idx(config.meal_times['breakfast']['end']), key="be") with cl: st.markdown("**Lunch**") ls = st.selectbox("Start", TIME_OPTIONS, index=t_idx(config.meal_times['lunch']['start']), key="ls") le = st.selectbox("End", TIME_OPTIONS, index=t_idx(config.meal_times['lunch']['end']), key="le") with cd: st.markdown("**Dinner**") ds = st.selectbox("Start", TIME_OPTIONS, index=t_idx(config.meal_times['dinner']['start']), key="ds") de = st.selectbox("End", TIME_OPTIONS, index=t_idx(config.meal_times['dinner']['end']), key="de") if st.button("💾 Save Settings", type="primary"): config.user_name = new_name config.user_avatar = new_avatar config.location = new_loc config.dietary_restrictions = new_diet config.meal_times = {"breakfast": {"start": bs, "end": be}, "lunch": {"start": ls, "end": le}, "dinner": {"start": ds, "end": de}} config.save_preferences() st.success("Saved!") st.rerun() if __name__ == "__main__": main()