import os
import gradio as gr
import json
import base64
from datetime import datetime
from io import BytesIO
from pathlib import Path
from theme import create_theme, CUSTOM_CSS
from data import (
ensure_dirs,
load_decisions,
create_decision_record,
get_open_decisions,
get_decision_by_id,
resolve_decision as resolve_decision_data,
export_decisions,
CARDS_DIR,
)
from prompts import (
FOLLOW_UP_SYSTEM,
FOLLOW_UP_NEXT,
CATEGORIZE_PROMPT,
PREDICT_PROMPT,
MOMENT_CARD_PROMPT,
PATTERN_ANALYSIS_PROMPT,
IMAGE_DESCRIBE_PROMPT,
)
from models import (
generate_text,
transcribe_audio,
describe_image,
generate_moment_card,
)
ensure_dirs()
LOADING_MESSAGES = [
"Compiling life choices…",
"Running static analysis on your decisions…",
"Checking for dependency conflicts…",
"Generating memory snapshot…",
]
# ── helpers ──────────────────────────────────────────────────────────────────
def _format_qa(qa_list: list[dict]) -> str:
if not qa_list:
return "(no follow-up yet)"
return "\n".join(f"Q: {q['question']}\nA: {q['answer']}" for q in qa_list)
def _parse_json(text: str, fallback):
clean = text.strip()
if clean.startswith("```"):
clean = clean.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
try:
return json.loads(clean)
except (json.JSONDecodeError, ValueError):
return fallback
def _card_thumbnail_b64(path: str | None, size: int = 96) -> str:
if not path or not Path(path).exists():
return ""
from PIL import Image as PILImage
img = PILImage.open(path)
img.thumbnail((size, size))
buf = BytesIO()
img.save(buf, format="PNG")
encoded = base64.b64encode(buf.getvalue()).decode()
return (
f'
'
)
def _initial_state() -> dict:
return {
"raw_input": "",
"input_type": "text",
"qa": [],
"step": 0,
"image_description": None,
"category_data": None,
"predictions": None,
"card_prompt": None,
"card_path": None,
}
# ── analysis pipeline ────────────────────────────────────────────────────────
def _run_categorize(decision: str, qa_ctx: str) -> dict:
prompt = CATEGORIZE_PROMPT.format(decision=decision, qa_context=qa_ctx)
raw = generate_text([{"role": "user", "content": prompt}], max_tokens=200)
return _parse_json(raw, {
"category": "lifestyle",
"subcategory": "general",
"severity": 5,
"status_emoji": "🔧",
})
def _run_predict(decision: str, cat: str, sev: int, qa_ctx: str) -> list:
prompt = PREDICT_PROMPT.format(
decision=decision, category=cat, severity=sev, qa_context=qa_ctx,
)
raw = generate_text([{"role": "user", "content": prompt}], max_tokens=500)
return _parse_json(raw, [
{"outcome": "Uncertain outcome", "probability": "medium",
"valence": "neutral", "timeframe": "months"},
])
def _run_card_prompt(decision: str, category: str, tone: str) -> str:
prompt = MOMENT_CARD_PROMPT.format(
decision=decision, category=category, tone=tone,
)
return generate_text([{"role": "user", "content": prompt}], max_tokens=150)
# ── HTML builders ────────────────────────────────────────────────────────────
def _build_analysis_html(cat_data: dict, predictions: list) -> str:
category = cat_data.get("category", "unknown")
severity = cat_data.get("severity", 5)
emoji = cat_data.get("status_emoji", "🔧")
sev_class = "low" if severity <= 3 else "medium" if severity <= 6 else "high"
preds = ""
for p in predictions:
v = p.get("valence", "neutral")
icon = {"negative": "🐛", "positive": "✨"}.get(v, "🔧")
cls = {"negative": "bug", "positive": "feature"}.get(v, "neutral")
preds += (
f'
'
f'{icon} {p["outcome"]} '
f'({p.get("probability","")}, '
f'{p.get("timeframe","")})
'
)
return f"""
{emoji} Debug Report Complete
[{category.upper()}]
{cat_data.get("subcategory","")}
Predicted Consequences
{preds}
"""
def _build_timeline_html(decisions: list) -> str:
if not decisions:
return (
''
'
📭
'
"
No commits yet. Head to New Commit to log your first "
"life decision.
"
)
entries = []
for d in reversed(decisions):
meta = d.get("debug_metadata", {})
h = meta.get("commit_hash", "0000000")
branch = meta.get("branch", "unknown")
emoji = meta.get("status_emoji", "🔧")
status = d.get("status", "open")
resolved_cls = " resolved" if status == "resolved" else ""
try:
dt = datetime.fromisoformat(d["timestamp"])
date_str = dt.strftime("%b %d, %Y %H:%M")
except Exception:
date_str = d.get("timestamp", "")
msg = d.get("raw_input", "")[:100]
if len(d.get("raw_input", "")) > 100:
msg += "…"
preds = ""
for p in d.get("consequence_predictions", [])[:3]:
v = p.get("valence", "neutral")
icon = {"negative": "🐛", "positive": "✨"}.get(v, "🔧")
cls = {"negative": "bug", "positive": "feature"}.get(v, "neutral")
preds += (
f''
f"{icon} {p.get('outcome','')}
"
)
thumb = _card_thumbnail_b64(d.get("moment_card_path"))
if status == "resolved" and d.get("outcome"):
ov = d["outcome"].get("actual_valence", "mixed")
outcome_html = (
f''
f'RESOLVED: {d["outcome"].get("description","")[:60]}
'
)
else:
outcome_html = (
''
"⏳ Outcome pending
"
)
entries.append(f"""
{preds}
{thumb}
{outcome_html}
""")
return f'{"".join(entries)}
'
# ── chart helpers ────────────────────────────────────────────────────────────
def _category_chart(decisions: list):
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
cats: dict[str, int] = {}
for d in decisions:
c = d.get("category", "unknown")
cats[c] = cats.get(c, 0) + 1
fig, ax = plt.subplots(figsize=(5, 4))
fig.patch.set_facecolor("#161b22")
ax.set_facecolor("#161b22")
if not cats:
ax.text(
0.5, 0.5, "No data yet",
ha="center", va="center", color="#484f58", fontsize=14,
)
ax.axis("off")
return fig
color_map = {
"career": "#58a6ff", "financial": "#f0883e", "health": "#3fb950",
"relationship": "#db61a2", "education": "#a06ef6",
"housing": "#79c0ff", "lifestyle": "#d29922", "creative": "#ff7b72",
}
labels = list(cats.keys())
sizes = list(cats.values())
colors = [color_map.get(l, "#8b949e") for l in labels]
ax.pie(
sizes,
labels=[l.capitalize() for l in labels],
colors=colors,
autopct="%1.0f%%",
textprops={"color": "#e6edf3", "fontsize": 10},
)
return fig
def _compute_stats(decisions: list) -> tuple:
total = len(decisions)
open_n = sum(1 for d in decisions if d.get("status") == "open")
resolved = total - open_n
accs = [
d["outcome"]["prediction_accuracy"]
for d in decisions
if d.get("outcome") and d["outcome"].get("prediction_accuracy") is not None
]
avg_acc = f"{sum(accs)/len(accs)*100:.0f}%" if accs else "--"
return total, open_n, resolved, avg_acc
def _stat_html(value, label):
return (
f''
)
# ── build app ────────────────────────────────────────────────────────────────
theme = create_theme()
with gr.Blocks(title="LifeLog", theme=theme, css=CUSTOM_CSS) as app:
# Header
gr.HTML("""
""")
state = gr.State(_initial_state())
with gr.Tabs() as tabs:
# ════════════════════════════════════════════════════════════════════
# TAB 1 — NEW COMMIT
# ════════════════════════════════════════════════════════════════════
with gr.Tab("📝 New Commit", id="tab-commit"):
gr.HTML('')
with gr.Row(equal_height=True):
with gr.Column(scale=1):
text_input = gr.Textbox(
label="📝 Type It",
placeholder=(
"What decision did you make? "
"What crossroads are you at?"
),
lines=4,
)
text_btn = gr.Button("Log Decision", variant="primary")
with gr.Column(scale=1):
audio_input = gr.Audio(
label="🎙️ Speak It",
sources=["microphone"],
type="filepath",
)
audio_btn = gr.Button(
"Transcribe & Log", variant="primary",
)
with gr.Column(scale=1):
image_input = gr.Image(
label="📷 Upload It", type="filepath",
)
image_ctx = gr.Textbox(
label="Context",
placeholder="e.g. 'This is the job offer I received'",
lines=2,
)
image_btn = gr.Button(
"Analyze & Log", variant="primary",
)
# Follow-up conversation
chat_col = gr.Column(visible=False)
with chat_col:
gr.HTML(
'"
)
chatbot = gr.Chatbot(
label="Follow-up Questions",
height=320,
type="messages",
)
with gr.Row():
user_resp = gr.Textbox(
placeholder="Your answer…",
label="",
scale=4,
show_label=False,
)
submit_btn = gr.Button("Submit", variant="primary", scale=1)
# Results (no visibility wrapper — content presence is the cue)
analysis_html = gr.HTML()
with gr.Row():
moment_img = gr.Image(
label="🎨 Your Moment Card", height=400,
visible=False,
)
save_btn = gr.Button(
"💾 Save to Timeline", variant="primary", size="lg",
visible=False,
)
save_status = gr.HTML()
# ── event handlers ──
def _start_session(text: str, st: dict, input_type: str = "text"):
if not text or not text.strip():
gr.Warning("Please enter a decision first.")
return None, st, gr.skip()
st = _initial_state()
st["raw_input"] = text.strip()
st["input_type"] = input_type
msgs = [
{"role": "system", "content": FOLLOW_UP_SYSTEM},
{"role": "user", "content": (
f"Decision logged: {text.strip()}\n\n"
"Ask follow-up question #1 of 3 (ROOT CAUSE)."
)},
]
question = generate_text(msgs)
st["current_question"] = question
chat = [
{"role": "assistant", "content": (
f"**Debugging session started for commit:** "
f"`{text.strip()[:80]}`\n\n{question}"
)},
]
return chat, st, gr.Column(visible=True)
def start_text(text, st):
return _start_session(text, st, "text")
def start_voice(audio, st):
if not audio:
gr.Warning("Please record audio first.")
return gr.skip(), None, st, gr.skip()
transcript = transcribe_audio(audio)
chat, st, cv = _start_session(transcript, st, "voice")
return transcript, chat, st, cv
def start_image(img, ctx, st):
if not img:
gr.Warning("Please upload an image first.")
return gr.skip(), None, st, gr.skip()
desc = describe_image(img, IMAGE_DESCRIBE_PROMPT)
st_new = _initial_state()
st_new["image_description"] = desc
combined = (
f"{ctx.strip()}\n\n[Image analysis: {desc}]"
if ctx and ctx.strip() else desc
)
chat, st_new, cv = _start_session(
combined, st_new, "image",
)
return combined, chat, st_new, cv
def handle_followup(user_msg, chat_history, st):
if not user_msg or not user_msg.strip():
return (
"", chat_history, st,
gr.skip(), gr.skip(), gr.skip(), gr.skip(),
)
st["qa"].append({
"question": st.get("current_question", ""),
"answer": user_msg.strip(),
})
st["step"] += 1
chat_history = list(chat_history)
chat_history.append({"role": "user", "content": user_msg.strip()})
if st["step"] < 3:
qa_ctx = _format_qa(st["qa"])
prompt = FOLLOW_UP_NEXT.format(
decision=st["raw_input"],
qa_context=qa_ctx,
question_number=st["step"] + 1,
)
next_q = generate_text([
{"role": "system", "content": FOLLOW_UP_SYSTEM},
{"role": "user", "content": prompt},
])
st["current_question"] = next_q
chat_history.append(
{"role": "assistant", "content": next_q},
)
return (
"", chat_history, st,
gr.skip(), gr.skip(), gr.skip(), gr.skip(),
)
# ── all 3 questions done → run analysis ──
chat_history.append({
"role": "assistant",
"content": "✅ All questions answered. Compiling debug report…",
})
qa_ctx = _format_qa(st["qa"])
cat_data = _run_categorize(st["raw_input"], qa_ctx)
category = cat_data.get("category", "lifestyle")
severity = cat_data.get("severity", 5)
predictions = _run_predict(
st["raw_input"], category, severity, qa_ctx,
)
pos = sum(1 for p in predictions if p.get("valence") == "positive")
neg = sum(1 for p in predictions if p.get("valence") == "negative")
tone = (
"hopeful and optimistic" if pos > neg
else "tense and uncertain" if neg > pos
else "contemplative and balanced"
)
card_prompt = _run_card_prompt(st["raw_input"], category, tone)
card_image = generate_moment_card(card_prompt)
card_name = f"card_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
card_path = str(CARDS_DIR / card_name)
card_image.save(card_path)
st["category_data"] = cat_data
st["predictions"] = predictions
st["card_prompt"] = card_prompt
st["card_path"] = card_path
section_hdr = (
'"
)
html = section_hdr + _build_analysis_html(cat_data, predictions)
return (
"", chat_history, st,
html, gr.Image(value=card_image, visible=True),
gr.Button(visible=True), gr.skip(),
)
def save_to_timeline(st):
if not st.get("category_data"):
return (
'No data to save.
',
st,
)
cd = st["category_data"]
record = create_decision_record(
raw_input=st["raw_input"],
input_type=st["input_type"],
follow_up_qa=st["qa"],
category=cd.get("category", "lifestyle"),
subcategory=cd.get("subcategory", "general"),
severity=cd.get("severity", 5),
status_emoji=cd.get("status_emoji", "🔧"),
consequence_predictions=st.get("predictions", []),
moment_card_prompt=st.get("card_prompt", ""),
moment_card_path=st.get("card_path"),
image_description=st.get("image_description"),
)
h = record["debug_metadata"]["commit_hash"]
return (
f'✅ Commit '
f'{h} saved to timeline.'
f"
",
_initial_state(),
)
# Wire events
text_btn.click(
start_text,
inputs=[text_input, state],
outputs=[chatbot, state, chat_col],
)
text_input.submit(
start_text,
inputs=[text_input, state],
outputs=[chatbot, state, chat_col],
)
audio_btn.click(
start_voice,
inputs=[audio_input, state],
outputs=[text_input, chatbot, state, chat_col],
)
image_btn.click(
start_image,
inputs=[image_input, image_ctx, state],
outputs=[text_input, chatbot, state, chat_col],
)
submit_btn.click(
handle_followup,
inputs=[user_resp, chatbot, state],
outputs=[
user_resp, chatbot, state,
analysis_html, moment_img,
save_btn, save_status,
],
)
user_resp.submit(
handle_followup,
inputs=[user_resp, chatbot, state],
outputs=[
user_resp, chatbot, state,
analysis_html, moment_img,
save_btn, save_status,
],
)
save_btn.click(
save_to_timeline,
inputs=[state],
outputs=[save_status, state],
)
# ════════════════════════════════════════════════════════════════════
# TAB 2 — GIT LOG
# ════════════════════════════════════════════════════════════════════
with gr.Tab("📜 Git Log", id="tab-log") as tab_log:
gr.HTML(
'"
)
with gr.Row():
cat_filter = gr.Dropdown(
choices=[
"All", "career", "financial", "health",
"relationship", "education", "housing",
"lifestyle", "creative",
],
value="All",
label="Filter by branch",
scale=2,
)
refresh_log_btn = gr.Button("🔄 Refresh", scale=1)
timeline_html = gr.HTML()
card_gallery = gr.Gallery(
label="🎨 Moment Cards", columns=4, height=280,
)
def refresh_log(cat):
decisions = load_decisions()
if cat != "All":
decisions = [
d for d in decisions if d.get("category") == cat
]
html = _build_timeline_html(decisions)
cards = [
(d["moment_card_path"],
d.get("debug_metadata", {}).get("commit_hash", ""))
for d in reversed(decisions)
if d.get("moment_card_path")
and Path(d["moment_card_path"]).exists()
]
return html, cards
refresh_log_btn.click(
refresh_log, inputs=[cat_filter],
outputs=[timeline_html, card_gallery],
)
cat_filter.change(
refresh_log, inputs=[cat_filter],
outputs=[timeline_html, card_gallery],
)
tab_log.select(
refresh_log, inputs=[cat_filter],
outputs=[timeline_html, card_gallery],
)
# ════════════════════════════════════════════════════════════════════
# TAB 3 — DEBUG DASHBOARD
# ════════════════════════════════════════════════════════════════════
with gr.Tab("📊 Debug Dashboard", id="tab-dash") as tab_dash:
gr.HTML(
'"
)
with gr.Row():
s_total = gr.HTML(_stat_html(0, "Total Commits"))
s_open = gr.HTML(_stat_html(0, "Open"))
s_resolved = gr.HTML(_stat_html(0, "Resolved"))
s_accuracy = gr.HTML(_stat_html("--", "Prediction Accuracy"))
chart = gr.Plot(label="Category Distribution")
analyze_btn = gr.Button(
"🔍 Run Pattern Analysis", variant="primary", size="lg",
)
pattern_md = gr.Markdown(
"*Click 'Run Pattern Analysis' to generate your debug "
"report…*"
)
export_btn = gr.Button("📥 Export All Data (JSON)", size="sm")
export_file = gr.File(label="Download", visible=False)
def refresh_dash():
decisions = load_decisions()
total, open_n, resolved, acc = _compute_stats(decisions)
fig = _category_chart(decisions)
return (
_stat_html(total, "Total Commits"),
_stat_html(open_n, "Open"),
_stat_html(resolved, "Resolved"),
_stat_html(acc, "Prediction Accuracy"),
fig,
)
def run_patterns():
decisions = load_decisions()
if len(decisions) < 2:
return (
"**Need at least 2 decisions to analyze patterns.** "
"Log more decisions in the New Commit tab."
)
summary = [
{
"decision": d["raw_input"][:200],
"category": d.get("category"),
"severity": d.get("severity"),
"predictions": d.get("consequence_predictions", []),
"status": d.get("status"),
"outcome": d.get("outcome"),
"timestamp": d.get("timestamp"),
}
for d in decisions
]
prompt = PATTERN_ANALYSIS_PROMPT.format(
decisions_json=json.dumps(summary, indent=2),
)
return generate_text(
[{"role": "user", "content": prompt}], max_tokens=1000,
)
def do_export():
path = Path("data") / "lifelog_export.json"
path.write_text(export_decisions(), encoding="utf-8")
return gr.File(value=str(path), visible=True)
analyze_btn.click(
refresh_dash,
outputs=[s_total, s_open, s_resolved, s_accuracy, chart],
).then(run_patterns, outputs=[pattern_md])
tab_dash.select(
refresh_dash,
outputs=[s_total, s_open, s_resolved, s_accuracy, chart],
)
export_btn.click(do_export, outputs=[export_file])
# ════════════════════════════════════════════════════════════════════
# TAB 4 — RESOLVE
# ════════════════════════════════════════════════════════════════════
with gr.Tab("✅ Resolve", id="tab-resolve") as tab_resolve:
gr.HTML(
'"
)
dec_dropdown = gr.Dropdown(
label="Select a decision to resolve",
choices=[],
interactive=True,
)
refresh_dec_btn = gr.Button("🔄 Refresh List", size="sm")
dec_detail = gr.HTML()
outcome_text = gr.Textbox(
label="What actually happened?",
placeholder="Describe the outcome…",
lines=3,
)
outcome_valence = gr.Radio(
choices=["positive", "negative", "mixed"],
label="How did it turn out?",
value="mixed",
)
resolve_btn = gr.Button(
"🔧 Close This Bug/Feature", variant="primary",
)
resolve_status = gr.HTML()
def refresh_open():
items = get_open_decisions()
choices = [
(
f"{d['debug_metadata']['commit_hash']} "
f"[{d.get('category','?').upper()}] "
f"{d['raw_input'][:55]}",
d["id"],
)
for d in items
]
return gr.Dropdown(choices=choices, value=None)
def show_detail(did):
if not did:
return ""
d = get_decision_by_id(did)
if not d:
return ""
qa = "".join(
f"Q: {q['question']}
"
f"A: {q['answer']}
"
for q in d.get("follow_up_qa", [])
)
preds = "".join(
f"{'🐛' if p.get('valence')=='negative' else '✨' if p.get('valence')=='positive' else '🔧'} "
f"{p.get('outcome','')} ({p.get('probability','')}, "
f"{p.get('timeframe','')})
"
for p in d.get("consequence_predictions", [])
)
return f"""
{d['debug_metadata']['status_emoji']} Commit
{d['debug_metadata']['commit_hash']}
Decision: {d['raw_input']}
Category:
[{d.get('category','?').upper()}] ·
Severity: {d.get('severity','?')}/10
Follow-up Discussion
{qa}
Predicted Consequences
{preds}
"""
def do_resolve(did, desc, valence):
if not did:
return 'Select a decision.
'
if not desc or not desc.strip():
return 'Describe the outcome.
'
resolve_decision_data(did, desc.strip(), valence)
return (
f'✅ Decision resolved as '
f"{valence}. Timeline updated.
"
)
refresh_dec_btn.click(refresh_open, outputs=[dec_dropdown])
tab_resolve.select(refresh_open, outputs=[dec_dropdown])
dec_dropdown.change(
show_detail, inputs=[dec_dropdown], outputs=[dec_detail],
)
resolve_btn.click(
do_resolve,
inputs=[dec_dropdown, outcome_text, outcome_valence],
outputs=[resolve_status],
)
if __name__ == "__main__":
app.launch(
allowed_paths=["data/cards"],
)