""" LifeOS β€” Streamlit UI (File 12 of 15) Adaptive Student Schedule Optimizer with multi-agent AI backend. Features: - Pre-loads SAMPLE_TASKS by default (no typing required β€” Section 6 rule 10) - GROQ_API_KEY validation with red error banner - Sidebar with dynamic task editor + preferences - Tabbed layout: Schedule Results | Before vs After Training - 3-column main area: iteration history | reward chart | best schedule - Reward chart via matplotlib + st.pyplot() - Spinner wrapping entire run_lifeos() call (Section 6 rule 6) - Coloured badges for study/break/review slots GAP 6: "Before vs After Training" tab with side-by-side component comparison. IMPROVEMENT 3: Per-component horizontal bar chart in iteration history. """ import json import os import sys from datetime import date, timedelta from pathlib import Path import re # SUPER IMPORTANT: Force Python to look in the current directory for modules # This permanently fixes "No module named 'env'" on Hugging Face Spaces sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import streamlit as st from dotenv import load_dotenv load_dotenv() # --------------------------------------------------------------------------- # Page config β€” must be first Streamlit call # --------------------------------------------------------------------------- st.set_page_config( page_title="LifeOS β€” Study Schedule Optimizer", page_icon="πŸ“š", layout="wide", ) # --------------------------------------------------------------------------- # Global styling # --------------------------------------------------------------------------- st.markdown(""" """, unsafe_allow_html=True) # --------------------------------------------------------------------------- # Hero header # --------------------------------------------------------------------------- st.markdown('

πŸ“š LifeOS

', unsafe_allow_html=True) st.markdown( '

Adaptive Student Schedule Optimizer  Β·  ' 'Multi-agent AI  Β·  Powered by Groq LLaMA 3.1

', unsafe_allow_html=True, ) st.divider() # --------------------------------------------------------------------------- # GROQ_API_KEY validation # --------------------------------------------------------------------------- GROQ_KEY = os.getenv("GROQ_API_KEY", "") if not GROQ_KEY or GROQ_KEY == "your_groq_api_key_here": st.error( "**GROQ_API_KEY not configured.** \n" "1. Create `.env` in the project root \n" "2. Add `GROQ_API_KEY=gsk_your_key_here` \n" "3. Get a free key at [console.groq.com](https://console.groq.com) \n\n" "*The app will still run using the heuristic fallback β€” " "but LLM-powered scheduling and memory insights require a key.*" ) # --------------------------------------------------------------------------- # Session state bootstrap # --------------------------------------------------------------------------- if "results" not in st.session_state: st.session_state.results = [] if "best_summary" not in st.session_state: st.session_state.best_summary = {"best_reward": None, "best_plan": None, "best_iter": None} if "tasks" not in st.session_state: # Pre-load sample tasks so the demo works out of the box from data.sample_tasks import SAMPLE_TASKS as _DEFAULT st.session_state.tasks = [dict(t) for t in _DEFAULT] # --------------------------------------------------------------------------- # Main layout β€” left controls, right results (matches reference UI) # --------------------------------------------------------------------------- left_col, right_col = st.columns([1, 2], gap="large") with left_col: st.markdown('
', unsafe_allow_html=True) st.markdown('

Tasks

', unsafe_allow_html=True) st.markdown('

Define your study tasks and deadlines

', unsafe_allow_html=True) tool_a, tool_b = st.columns(2) with tool_a: if st.button("πŸ“¦ Load Sample Scenario", use_container_width=True, type="secondary"): from data.sample_tasks import SAMPLE_TASKS st.session_state.tasks = [dict(t) for t in SAMPLE_TASKS] st.session_state.results = [] st.success("Sample scenario loaded!") with tool_b: if st.button("βž• Add Task", use_container_width=True): next_id = max((t.get("id", 0) for t in st.session_state.tasks), default=0) + 1 st.session_state.tasks.append({ "id": next_id, "name": f"New Task {next_id}", "deadline": str(date.today() + timedelta(days=5)) + " 23:59", "duration_hrs": 2.0, "priority": "medium", "subject": "General", }) for idx, task in enumerate(st.session_state.tasks): with st.container(border=True): top_l, top_r = st.columns([8, 1]) with top_l: st.session_state.tasks[idx]["name"] = st.text_input( "Task name", value=task.get("name", ""), key=f"n_{idx}", label_visibility="collapsed", placeholder="Task name", ) with top_r: if st.button("πŸ—‘", key=f"rm_{idx}", use_container_width=True): st.session_state.tasks.pop(idx) st.rerun() row_1a, row_1b = st.columns(2) with row_1a: st.session_state.tasks[idx]["subject"] = st.text_input( "Subject", value=task.get("subject", ""), key=f"s_{idx}") with row_1b: st.session_state.tasks[idx]["priority"] = st.selectbox( "Priority", ["high", "medium", "low"], index=["high", "medium", "low"].index(task.get("priority", "medium")), key=f"p_{idx}", ) row_2a, row_2b = st.columns(2) with row_2a: dl_str = task.get("deadline", str(date.today()) + " 23:59") dl_date = date.fromisoformat(dl_str.split(" ")[0]) new_dl = st.date_input("Deadline", value=dl_date, key=f"d_{idx}") st.session_state.tasks[idx]["deadline"] = str(new_dl) + " 23:59" with row_2b: st.session_state.tasks[idx]["duration_hrs"] = st.number_input( "Duration (hrs)", min_value=0.5, max_value=24.0, value=float(task.get("duration_hrs", 2.0)), step=0.5, key=f"h_{idx}", ) st.markdown('
', unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) st.markdown('

Preferences

', unsafe_allow_html=True) st.markdown('

Customize your study preferences

', unsafe_allow_html=True) max_daily_hours = st.number_input( "Max Daily Hours", min_value=1, max_value=16, value=8) min_break = st.number_input( "Break Gap (minutes)", min_value=30, max_value=180, value=90) end_by = st.text_input("End By", value="22:00") max_iter = st.slider("Max Iterations", min_value=1, max_value=5, value=3) prefers_morning = st.checkbox("Prefer morning study sessions", value=True) generate_btn = st.button("β–· Generate Schedule", type="primary", use_container_width=True) st.markdown('
', unsafe_allow_html=True) # --------------------------------------------------------------------------- # Run orchestrator # --------------------------------------------------------------------------- if generate_btn: if not st.session_state.tasks: st.warning("Please add at least one task first.") st.stop() preferences = { "prefers_morning": prefers_morning, "max_daily_hours": max_daily_hours, "min_break_gap_mins": min_break, "end_by": end_by, } # Build a 5-day calendar starting from next Monday today = date.today() days_to_monday = (7 - today.weekday()) % 7 or 7 week_start = today + timedelta(days=days_to_monday) calendar = {} for i in range(5): d = week_start + timedelta(days=i) calendar[str(d)] = { "slots": ["09:00-12:00", "14:00-17:00", "19:00-21:00"], "day_name": d.strftime("%A"), } with st.spinner("πŸ€– Multi-agent negotiation in progress… (Planner β†’ Critic β†’ Memory β†’ Reward)"): try: from orchestrator import run_lifeos results = run_lifeos( tasks=st.session_state.tasks, calendar=calendar, preferences=preferences, max_iterations=max_iter, verbose=False, ) st.session_state.results = results if results: best_result = max(results, key=lambda r: float(r.get("reward", float("-inf")))) st.session_state.best_summary = { "best_reward": float(best_result.get("reward", 0.0)), "best_plan": best_result.get("plan", {}), "best_iter": int(best_result.get("iteration", 0)), } except Exception as e: st.error(f"Error during generation: {e}") st.stop() # --------------------------------------------------------------------------- # Helper: per-component horizontal bar chart HTML # --------------------------------------------------------------------------- def _component_bars_html(reward_breakdown: dict) -> str: """Generate HTML for per-component mini bar chart.""" from agents.reward import COMPONENT_MAX, COMPONENT_NAMES lines = [] for comp in COMPONENT_NAMES: val = float(reward_breakdown.get(comp, 0.0)) max_val = COMPONENT_MAX.get(comp, 20) pct = max(0, min(100, (abs(val) / max_val * 100) if max_val > 0 else 0)) comp_short = comp.replace("_score", "") if comp_short == "load" and val < 0: color_cls = "comp-bar-red" elif val > 0: color_cls = "comp-bar-green" else: color_cls = "comp-bar-gray" short_name = comp_short[:8] lines.append( f'
' f'{short_name}' f'
' f'
' f'
' f'{val:+.0f}' f'
' ) return "".join(lines) def show_reward_breakdown(reward_breakdown, iteration_num): try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: st.warning("Install matplotlib: `pip install matplotlib`") return components = { 'deadline': reward_breakdown.get('deadline_score', 0), 'break': reward_breakdown.get('break_score', 0), 'load': reward_breakdown.get('load_score', 0), 'energy': reward_breakdown.get('energy_score', 0), 'revision': reward_breakdown.get('revision_score', 0), 'format': reward_breakdown.get('format_score', 0), } labels = list(components.keys()) values = list(components.values()) colors = ['#E24B4A' if v < 0 else '#639922' for v in values] fig, ax = plt.subplots(figsize=(6, 4)) bars = ax.barh(labels, values, color=colors) # Add value labels for bar, val in zip(bars, values): ax.text(val + (1 if val >= 0 else -1), bar.get_y() + bar.get_height()/2, str(int(val)), va='center', ha='left' if val >= 0 else 'right', fontweight='bold') ax.axvline(x=0, color='black', linewidth=0.8) ax.set_title(f'Reward Components β€” Iteration {iteration_num}') ax.grid(True, axis='x', alpha=0.3) plt.tight_layout() st.pyplot(fig) plt.close() def _parse_slot_time_range(time_str: str): """Parse HH:MM-HH:MM from slot time and return (start_h, start_m, end_h, end_m).""" if not isinstance(time_str, str): return None match = re.search(r"(\d{1,2}):(\d{2})\s*-\s*(\d{1,2}):(\d{2})", time_str) if not match: return None sh, sm, eh, em = map(int, match.groups()) return sh, sm, eh, em def _slot_warning_style(slot: dict): """Return warning style for a schedule slot.""" slot_type = slot.get("type", "") if slot_type != "study": return "", "" parsed = _parse_slot_time_range(slot.get("time", "")) if not parsed: return "", "" sh, sm, eh, em = parsed start_mins = sh * 60 + sm end_mins = eh * 60 + em if end_mins < start_mins: end_mins += 24 * 60 duration_hours = (end_mins - start_mins) / 60 ends_after_21 = (eh > 21) or (eh == 21 and em > 0) long_block = duration_hours > 3.0 if ends_after_21: return "color:#b91c1c; background:#fee2e2; padding:4px 8px; border-radius:6px;", " ⚠️ Late-night slot" if long_block: return "color:#c2410c; background:#ffedd5; padding:4px 8px; border-radius:6px;", " ⚠️ Long block (>3h)" return "", "" # --------------------------------------------------------------------------- # Dialogs for detailed views # --------------------------------------------------------------------------- @st.dialog("πŸ”„ Iteration History", width="large") def show_iteration_history(results): from orchestrator import format_plan_for_display prev_issue_count = None for r in results: rew = r["reward"] score_cls = "score-good" if rew >= 80 else ("score-mid" if rew >= 40 else "score-bad") icon = "βœ…" if rew >= 80 else ("⚠️" if rew >= 50 else "❌") with st.expander( f"Iteration {r['iteration']} | Reward: {rew:.0f} {icon}", expanded=(r["iteration"] == len(results)), ): st.markdown( f"{rew:.0f}", unsafe_allow_html=True, ) drift_event = r.get("drift_event") if drift_event: st.markdown( f"
" f"Schema Drift Active: {drift_event.get('id', '')} β€” " f"{drift_event.get('description', '')}
", unsafe_allow_html=True, ) breakdown = r.get("reward_breakdown", {}) if breakdown: st.markdown("**Component Scores:**") st.markdown(_component_bars_html(breakdown), unsafe_allow_html=True) if r.get("process_bonus", 0) > 0: st.markdown(f"
πŸ”„ Process bonus: +{r['process_bonus']:.0f}
", unsafe_allow_html=True) current_issue_count = len(r.get("issues", [])) if prev_issue_count is None: st.markdown( f"
Issues: {current_issue_count} (starting point)
", unsafe_allow_html=True, ) else: trend_color = "#16a34a" if current_issue_count <= prev_issue_count else "#dc2626" st.markdown( f"
" f"Issues: {prev_issue_count} β†’ {current_issue_count}
", unsafe_allow_html=True, ) if r["issues"]: st.markdown("**Critic Issues:**") for iss in r["issues"]: st.markdown(f"
⚠ {iss}
", unsafe_allow_html=True) else: st.markdown("
βœ… All checks passed!
", unsafe_allow_html=True) st.code(format_plan_for_display(r["plan"]), language=None) prev_issue_count = current_issue_count @st.dialog("πŸ“ˆ Reward Details", width="large") def show_reward_details(results, best): try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: st.warning("Install matplotlib: `pip install matplotlib`") return st.subheader("Reward Curve") iters = [r["iteration"] for r in results] rewards = [r["reward"] for r in results] fig, ax = plt.subplots(figsize=(6, 3.8)) best_so_far = [max(rewards[:i + 1]) for i in range(len(rewards))] ax.plot(iters, rewards, color="royalblue", linewidth=2.5, linestyle="-", marker="o", markersize=8, label="Iteration reward") ax.plot(iters, best_so_far, color="forestgreen", linewidth=2.2, linestyle="--", marker=None, label="Best so far") for x, y in zip(iters, rewards): ax.annotate(f"{y:.0f}", xy=(x, y), xytext=(0, 9), textcoords="offset points", ha="center", fontsize=9, fontweight="bold", color="royalblue") ax.axhline(50, color="darkorange", linestyle="--", linewidth=1.3, alpha=0.85, label="Acceptable (50)") ax.axhline(80, color="red", linestyle="--", linewidth=1.3, alpha=0.85, label="Optimal (80)") ax.set_ylim(bottom=0, top=max(max(rewards)+25, 105)) ax.set_xticks(iters) ax.xaxis.set_major_locator(plt.MaxNLocator(integer=True)) ax.set_xlabel("Iteration", fontsize=10) ax.set_ylabel("Reward", fontsize=10) ax.set_title("Reward Improvement", fontsize=11, fontweight="bold") ax.legend(fontsize=8, loc="best") ax.grid(axis="y", alpha=0.3, linestyle=":") ax.spines["top"].set_visible(False) ax.spines["right"].set_visible(False) plt.tight_layout() st.pyplot(fig) plt.close(fig) st.divider() st.subheader("Reward Breakdown") rb = best.get('reward_breakdown', {}) iteration_num = best.get('iteration', 1) components = { 'deadline': rb.get('deadline_score', 0), 'break': rb.get('break_score', 0), 'load': rb.get('load_score', 0), 'energy': rb.get('energy_score', 0), 'revision': rb.get('revision_score', 0), 'format': rb.get('format_score', 0), } labels = list(components.keys()) values = list(components.values()) colors = ['#E24B4A' if v < 0 else '#2E8B57' for v in values] fig2, ax2 = plt.subplots(figsize=(6, 4)) bars = ax2.barh(labels, values, color=colors, height=0.5) for bar, val in zip(bars, values): ax2.text( val + (0.5 if val >= 0 else -0.5), bar.get_y() + bar.get_height()/2, str(int(val)), va='center', ha='left' if val >= 0 else 'right', fontweight='bold', fontsize=11 ) ax2.axvline(x=0, color='black', linewidth=0.8) ax2.set_title(f'Reward Components β€” Best Iteration {iteration_num}') ax2.set_facecolor('#1a1a2e') fig2.patch.set_facecolor('#1a1a2e') ax2.tick_params(colors='white') ax2.title.set_color('white') ax2.xaxis.label.set_color('white') for spine in ax2.spines.values(): spine.set_edgecolor('white') ax2.grid(True, axis='x', alpha=0.3, color='white') plt.tight_layout() st.pyplot(fig2) plt.close(fig2) # --------------------------------------------------------------------------- # Tabbed layout (right panel) # --------------------------------------------------------------------------- with right_col: tab_schedule, tab_comparison, tab_training = st.tabs(["Iterations", "Reward Curve", "Best Schedule"]) # ═══════════════════════════════════════════════════════════════════════════ # TAB 1: Iterations # ═══════════════════════════════════════════════════════════════════════════ with tab_schedule: results = st.session_state.results if not results: st.info( "πŸ‘ˆ **Configure your tasks** in the sidebar, then click **Generate Schedule** " "to start the multi-agent loop. Sample tasks are pre-loaded β€” just click Generate!" ) else: best_summary = st.session_state.get("best_summary", {}) if not best_summary.get("best_plan") and results: best_iter_dict = max(results, key=lambda r: r["reward"]) best_summary = { "best_reward": float(best_iter_dict.get("reward", 0.0)), "best_plan": best_iter_dict.get("plan", {}), "best_iter": int(best_iter_dict.get("iteration", 0)), } st.session_state.best_summary = best_summary best_reward = float(best_summary.get("best_reward", 0.0)) best_plan = best_summary.get("best_plan", {}) best_iter = int(best_summary.get("best_iter", 0)) best_iter_dict = max(results, key=lambda r: r["reward"]) # ── Top Section: Best Schedule & Buttons ──────────────────── col_title, col_btn1, col_btn2 = st.columns([1.6, 1.2, 1.2]) with col_title: st.markdown( f"

πŸ† Best: Iter {best_iter} | Reward: {best_reward:.0f}

", unsafe_allow_html=True, ) with col_btn1: st.write("") # Vertical alignment spacing if st.button("πŸ”„ View Iteration History", use_container_width=True): show_iteration_history(results) with col_btn2: st.write("") # Vertical alignment spacing if st.button("πŸ“ˆ View Reward Details", use_container_width=True): show_reward_details(results, best_iter_dict) st.divider() schedule = best_plan.get("schedule", {}) notes = best_plan.get("notes", "") badge = { "study": "STUDY", "break": "BREAK", "review": "REVIEW", } days = sorted(schedule.keys()) if days: for date_str in days: st.markdown(f"#### πŸ“… {date_str}") slots = schedule[date_str] for slot in (slots if isinstance(slots, list) else []): b = badge.get(slot.get("type", "study"), "") warn_style, warn_text = _slot_warning_style(slot) st.markdown( f"
" f"{b} {slot.get('time', '?')} " f"{slot.get('task', '?')}{warn_text}
", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True) if notes: st.info(f"πŸ’‘ {notes}") st.divider() # Summary footer banner first_reward = float(results[0].get("reward", 0.0)) delta = best_reward - first_reward pct = (delta / first_reward * 100) if first_reward > 0 else 0.0 issues_start = len(results[0].get("issues", [])) issues_end = len(best_iter_dict.get("issues", [])) if delta > 0: st.success(f"Started: {first_reward:.0f} β†’ Best: {best_reward:.0f} β†’ Improvement: +{pct:.0f}%") else: st.info(f"Started: {first_reward:.0f} β†’ Best: {best_reward:.0f} β†’ Issues resolved: {issues_start} β†’ {issues_end}") # ═══════════════════════════════════════════════════════════════════════════ # TAB 2: Reward Curve # ═══════════════════════════════════════════════════════════════════════════ with tab_comparison: st.subheader("Reward Progression") st.caption("Multi-agent learning curve") # Direct chart display - no dialog import matplotlib.pyplot as plt import matplotlib matplotlib.use('Agg') if st.session_state.get('results'): rewards = [r.get('reward', 0) for r in st.session_state.results] best_so_far = [max(rewards[:i+1]) for i in range(len(rewards))] fig, ax = plt.subplots(figsize=(8, 4)) ax.plot(range(1, len(rewards)+1), rewards, 'b-o', label='Iteration reward', linewidth=2) ax.plot(range(1, len(best_so_far)+1), best_so_far, 'g--', label='Best so far', linewidth=2) ax.axhline(y=50, color='orange', linestyle='--', label='Acceptable (50)', alpha=0.7) ax.axhline(y=80, color='red', linestyle='--', label='Optimal (80)', alpha=0.7) ax.xaxis.set_major_locator(plt.MaxNLocator(integer=True)) ax.set_xlabel('Iteration') ax.set_ylabel('Reward') ax.set_title('Reward Improvement') ax.legend() ax.grid(True, alpha=0.3) for i, (r_val, b_val) in enumerate(zip(rewards, best_so_far)): ax.annotate(str(int(r_val)), (i+1, r_val), textcoords="offset points", xytext=(0,10), ha='center', fontsize=10) st.pyplot(fig) plt.close() else: st.info("Run a scenario first to see the reward curve") # ═══════════════════════════════════════════════════════════════════════════ # TAB 3: Best Schedule # ═══════════════════════════════════════════════════════════════════════════ with tab_training: results = st.session_state.results if not results: st.info("Generate a schedule to view the best schedule.") else: best = max(results, key=lambda r: r.get("reward", 0.0)) st.markdown("### Best Schedule") st.caption(f"Iteration {best.get('iteration', 0)} - Highest scoring schedule") st.markdown(f"
{float(best.get('reward', 0.0)):.1f}
", unsafe_allow_html=True) plan = best.get("plan", {}) schedule = plan.get("schedule", {}) for date_str in sorted(schedule.keys()): st.markdown(f"

{date_str}

", unsafe_allow_html=True) for slot in schedule.get(date_str, []): badge = "study-badge" if slot.get("type") == "study" else ("break-badge" if slot.get("type") == "break" else "review-badge") st.markdown( f"
{slot.get('time', '?')}" f"{slot.get('type', '').upper()}: {slot.get('task', '?')}
", unsafe_allow_html=True, ) st.markdown("
", unsafe_allow_html=True)