"""
🌿 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()