Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import json | |
| import os | |
| import sys | |
| import tempfile | |
| import time | |
| import html | |
| from collections.abc import Callable | |
| from datetime import datetime, timedelta | |
| from pathlib import Path | |
| from typing import Any | |
| ROOT = Path(__file__).resolve().parent | |
| SRC = ROOT / "src" | |
| if str(SRC) not in sys.path: | |
| sys.path.insert(0, str(SRC)) | |
| import gradio as gr | |
| from unstuck import embeddings, recall | |
| from unstuck.calibration import calibrate, multiplier | |
| from unstuck.prompts import format_exemplar | |
| from unstuck.service import BreakdownView, Unstuck | |
| from unstuck.store import Store | |
| THEME = gr.themes.Base( | |
| primary_hue="indigo", | |
| neutral_hue="stone", | |
| font=[gr.themes.GoogleFont("Inter"), "system-ui", "sans-serif"], | |
| ).set( | |
| body_background_fill="#f7f5f1", | |
| block_background_fill="#ffffff", | |
| block_border_width="0px", | |
| block_shadow="0 1px 3px rgba(40, 35, 60, 0.06)", | |
| button_primary_background_fill="linear-gradient(135deg, #4f46e5, #6366f1)", | |
| button_primary_background_fill_hover="linear-gradient(135deg, #4338ca, #4f46e5)", | |
| button_primary_text_color="#ffffff", | |
| button_primary_shadow="0 2px 10px rgba(79, 70, 229, 0.28)", | |
| button_secondary_background_fill="#ffffff", | |
| button_secondary_border_color="#e7e2d9", | |
| block_radius="16px", | |
| button_large_radius="12px", | |
| input_background_fill="#ffffff", | |
| input_border_color="#e7e2d9", | |
| input_border_color_focus="#4f46e5", | |
| ) | |
| CSS = """ | |
| @import url('https://fonts.googleapis.com/css2?family=Fraunces:opsz,wght@9..144,500;9..144,600;9..144,700&display=swap'); | |
| /* off-brand polish — a distinct identity, well past the default Gradio look */ | |
| .gradio-container { max-width: 760px !important; margin: 0 auto !important; | |
| background: radial-gradient(1100px 380px at 50% -120px, #ecebff 0%, rgba(236,235,255,0) 70%), #f7f5f1 !important; } | |
| button { transition: transform .12s ease, box-shadow .12s ease, filter .12s ease !important; } | |
| button:hover { transform: translateY(-1px); } | |
| textarea:focus, input:focus { outline: none !important; border-color: #4f46e5 !important; | |
| box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.15) !important; } | |
| footer, .built-with, a[href*="gradio.app"] { display: none !important; } | |
| #hero { text-align: center; padding: 1.6rem 0 0.2rem; } | |
| #hero h1 { font-family: 'Fraunces', Georgia, serif; font-size: 2.5rem; font-weight: 600; margin: 0; letter-spacing: -0.01em; | |
| background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 55%, #9333ea 100%); | |
| -webkit-background-clip: text; background-clip: text; color: transparent; } | |
| #hero p { color: #78716c; margin: 0.45rem 0 0; font-size: 1.04rem; } | |
| #steps-list { display: flex; flex-direction: column; gap: 10px; } | |
| .step-card { display: flex; align-items: center; gap: 14px; min-width: 0; background: #fff; | |
| border: 1px solid #eee9e2; border-radius: 14px; padding: 12px 16px; | |
| box-shadow: 0 1px 2px rgba(40, 35, 60, 0.05); | |
| transition: box-shadow .15s ease, transform .15s ease, border-color .15s ease; } | |
| .step-card:hover { box-shadow: 0 4px 14px rgba(40, 35, 60, 0.09); } | |
| .step-next { border-color:#4f46e5; box-shadow:0 6px 20px rgba(79,70,229,0.22); | |
| background: linear-gradient(180deg, #fbfaff 0%, #ffffff 60%); } | |
| .step-next .step-num { background:#4f46e5; color:#fff; } | |
| .step-later { opacity:0.55; } | |
| .step-skipped { opacity:0.45; } | |
| .step-skipped .step-text { text-decoration:line-through; } | |
| .step-num { flex: none; width: 30px; height: 30px; border-radius: 50%; | |
| background: #eef2ff; color: #4f46e5; font-weight: 600; display: flex; | |
| align-items: center; justify-content: center; font-size: 0.9rem; } | |
| .step-text { flex: 1; min-width: 0; overflow-wrap: anywhere; color: #292524; font-size: 1rem; line-height: 1.4; } | |
| .chip { flex: none; border-radius: 999px; padding: 3px 11px; font-size: 0.82rem; | |
| white-space: nowrap; } | |
| .chip-raw { background: #f5f5f4; color: #78716c; } | |
| .chip-you { background: #eef2ff; color: #4338ca; font-weight: 600; } | |
| .chip-done { background: #ecfdf5; color: #047857; font-weight: 600; } | |
| .chip-timer { background: #fef3c7; color: #b45309; font-weight: 600; } | |
| .chip-skip { background:#f5f5f4; color:#a8a29e; } | |
| .step-row { align-items: center !important; gap: 8px !important; margin-bottom: 10px; flex-wrap: wrap !important; } | |
| .step-row .step-html { flex: 1 1 320px !important; min-width: 0 !important; } | |
| .step-row .step-card { flex: 1; } | |
| .took-input input { border-radius: 10px !important; } | |
| .readout { background: #eef2ff; color: #4338ca; border-radius: 12px; padding: 12px 18px; | |
| font-size: 0.95rem; text-align: center; margin-top: 6px; } | |
| .completion { background:#ecfdf5; color:#047857; border-radius:12px; | |
| padding:14px 18px; font-size:1rem; text-align:center; margin-top:10px; font-weight:600; } | |
| .summary { background: #f5f5f4; color: #57534e; border-radius: 12px; padding: 10px 16px; | |
| font-size: 0.94rem; text-align: center; margin-top: 8px; font-weight: 600; } | |
| .restored-banner { background:#eff6ff; color:#1d4ed8; border-radius:12px; | |
| padding:10px 16px; font-size:0.92rem; text-align:center; margin-top:8px; font-weight:600; } | |
| .recall-banner { font-size: 13px; color: #4f46e5; background: rgba(79,70,229,0.08); | |
| border-radius: 10px; padding: 8px 12px; margin: 8px 0; } | |
| .patterns { display:flex; flex-direction:column; gap:8px; } | |
| .pattern-row { display:flex; align-items:flex-end; gap:10px; background:#fff; | |
| border:1px solid #eee9e2; border-radius:10px; padding:8px 14px; font-size:0.88rem; } | |
| .pattern-cat { font-weight:600; color:#292524; min-width:90px; } | |
| .pattern-mult { color:#4338ca; flex:1; } | |
| .bar { display:inline-block; width:7px; background:#c7d2fe; border-radius:3px 3px 0 0; | |
| margin-right:2px; } | |
| .pattern-n { color:#a8a29e; font-size:0.8rem; } | |
| .explainer { color: #a8a29e; font-size: 0.86rem; text-align: center; margin-top: 10px; } | |
| footer { display: none !important; } | |
| @media (max-width: 640px) { | |
| .gradio-container { padding: 0 10px !important; } | |
| #hero h1 { font-size: 1.6rem; } | |
| #hero p { font-size: 0.94rem; } | |
| .step-row { flex-wrap: wrap !important; } | |
| .step-row .step-html { flex: 1 1 100% !important; } | |
| .step-row .step-card { flex: 1 1 100%; } | |
| .step-card { flex-wrap: wrap; gap: 8px; padding: 10px 12px; } | |
| .step-text { flex: 1 1 calc(100% - 40px); font-size: 0.95rem; } | |
| .chip { font-size: 0.78rem; padding: 2px 9px; } | |
| .took-input input { font-size: 16px !important; } /* prevents iOS zoom-on-focus */ | |
| .pattern-row { flex-wrap: wrap; } | |
| .pattern-cat { min-width: 70px; } | |
| } | |
| """ | |
| def finish_minutes( | |
| manual: float | None, started_at: float | None, now: float | |
| ) -> int | None: | |
| """Return manually entered minutes or compute elapsed timer minutes.""" | |
| if manual is not None and manual > 0: | |
| return int(round(manual)) | |
| if started_at is not None: | |
| return max(1, int(round((now - started_at) / 60.0))) | |
| return None | |
| def next_step_id(rows: list[dict]) -> int | None: | |
| """Return the first unresolved step id in visible row order.""" | |
| for row in rows: | |
| if not row["logged"] and not row.get("skipped"): | |
| return int(row["step_id"]) | |
| return None | |
| def make_record( | |
| category: str, est_minutes: int, actual_minutes: int, now: float | |
| ) -> dict[str, Any]: | |
| """Return one browser-stored calibration record.""" | |
| return { | |
| "category": category, | |
| "est_minutes": est_minutes, | |
| "actual_minutes": actual_minutes, | |
| "completed_at": now, | |
| } | |
| def make_history_entry( | |
| text: str, embedding: list[float], breakdown: list[dict[str, Any]] | |
| ) -> dict[str, Any]: | |
| """Build a recall-history entry for one completed/created task.""" | |
| return { | |
| "text": text, | |
| "embedding": list(embedding), | |
| "breakdown": [ | |
| { | |
| "text": str(step["text"]), | |
| "category": str(step["category"]), | |
| "est_minutes": int(step["est_minutes"]), | |
| } | |
| for step in breakdown | |
| ], | |
| "durations": [], | |
| "dismissals": 0, | |
| } | |
| def _history_from_data(data: dict | None) -> list[dict[str, Any]]: | |
| """Return the recall history from BrowserState, or empty if absent/malformed.""" | |
| if not isinstance(data, dict): | |
| return [] | |
| history = data.get("history") | |
| return history if isinstance(history, list) else [] | |
| def with_history(data: dict, history: list[dict[str, Any]]) -> dict: | |
| """Return a copy of BrowserState data with the recall history replaced.""" | |
| return {**data, "history": history} | |
| def bump_dismissal( | |
| history: list[dict[str, Any]], index: int | |
| ) -> list[dict[str, Any]]: | |
| """Return a copy of history with entry[index]'s dismissals incremented by one.""" | |
| updated = [dict(entry) for entry in history] | |
| if 0 <= index < len(updated): | |
| updated[index]["dismissals"] = int(updated[index].get("dismissals", 0)) + 1 | |
| return updated | |
| def apply_dismissal(data: dict, pointer: dict | None) -> dict: | |
| """Return BrowserState data with the recalled entry's dismissal bumped, if any. | |
| Two dismissals push an entry's count to MAX_DISMISSALS, after which | |
| recall.select skips it — the 'shake off a sticky recall' behaviour.""" | |
| dismiss_index = (pointer or {}).get("dismiss_index") | |
| if dismiss_index is None: | |
| return data | |
| history = bump_dismissal(_history_from_data(data), int(dismiss_index)) | |
| return with_history(data, history) | |
| def record_duration_in_history( | |
| history: list[dict[str, Any]], index: int, category: str, actual_minutes: int | |
| ) -> list[dict[str, Any]]: | |
| """Return a copy of history with one real duration appended to entry[index].""" | |
| updated = [dict(entry) for entry in history] | |
| if 0 <= index < len(updated): | |
| durations = list(updated[index].get("durations", [])) | |
| durations.append({"category": category, "actual_minutes": int(actual_minutes)}) | |
| updated[index]["durations"] = durations | |
| return updated | |
| def undo_row(rows: list[dict], step_id: int) -> list[dict]: | |
| """Return rows with one resolved step restored to unresolved state.""" | |
| undone = [] | |
| for row in rows: | |
| if row["step_id"] != step_id: | |
| undone.append(row) | |
| continue | |
| updated = { | |
| **row, | |
| "logged": False, | |
| "skipped": False, | |
| "actual_minutes": None, | |
| "started_at": None, | |
| } | |
| updated.pop("record_at", None) | |
| undone.append(updated) | |
| return undone | |
| def edit_row_text(rows: list[dict], step_id: int, new_text: str) -> list[dict]: | |
| """Return rows with one step's text replaced by non-empty user wording.""" | |
| text = new_text.strip() | |
| if not text: | |
| return rows | |
| edited = [] | |
| found = False | |
| for row in rows: | |
| if row["step_id"] == step_id: | |
| edited.append({**row, "text": text}) | |
| found = True | |
| else: | |
| edited.append(row) | |
| return edited if found else rows | |
| def add_manual_row( | |
| rows: list[dict], text: str, minutes: float | None, records: list[dict] | |
| ) -> list[dict]: | |
| """Return rows with one user-authored admin step appended.""" | |
| clean_text = text.strip() | |
| if not clean_text: | |
| return rows | |
| est = int(round(minutes)) if minutes and minutes > 0 else 10 | |
| next_id = max((int(row["step_id"]) for row in rows), default=0) + 1 | |
| row = { | |
| "step_id": next_id, | |
| "text": clean_text, | |
| "category": "admin", | |
| "raw_minutes": est, | |
| "calibrated_minutes": calibrate(est, multiplier("admin", records)), | |
| "logged": False, | |
| "skipped": False, | |
| "actual_minutes": None, | |
| "started_at": None, | |
| } | |
| return [*rows, row] | |
| def remove_record( | |
| records: list[dict], | |
| category: str, | |
| est_minutes: int, | |
| actual_minutes: int, | |
| completed_at: float | None, | |
| ) -> list[dict]: | |
| """Return records with one matching calibration row removed.""" | |
| remove_index = None | |
| for index, record in enumerate(records): | |
| fields_match = ( | |
| record.get("category") == category | |
| and record.get("est_minutes") == est_minutes | |
| and record.get("actual_minutes") == actual_minutes | |
| ) | |
| if not fields_match: | |
| continue | |
| if completed_at is None: | |
| remove_index = index | |
| elif record.get("completed_at") == completed_at: | |
| remove_index = index | |
| break | |
| if remove_index is None: | |
| return list(records) | |
| return [record for index, record in enumerate(records) if index != remove_index] | |
| def _record_key(record: dict[str, Any]) -> tuple[Any, Any, Any, Any]: | |
| return ( | |
| record["category"], | |
| record["est_minutes"], | |
| record["actual_minutes"], | |
| record["completed_at"], | |
| ) | |
| def _validate_record_row(row: Any, index: int) -> dict[str, Any]: | |
| if not isinstance(row, dict): | |
| raise ValueError(f"record {index} must be an object") | |
| category = row.get("category") | |
| est_minutes = row.get("est_minutes") | |
| actual_minutes = row.get("actual_minutes") | |
| completed_at = row.get("completed_at") | |
| if not isinstance(category, str): | |
| raise ValueError(f"record {index} category must be a string") | |
| if type(est_minutes) is not int or est_minutes <= 0: | |
| raise ValueError(f"record {index} est_minutes must be positive") | |
| if type(actual_minutes) is not int or actual_minutes <= 0: | |
| raise ValueError(f"record {index} actual_minutes must be positive") | |
| if type(completed_at) not in (int, float): | |
| raise ValueError(f"record {index} completed_at must be numeric") | |
| return make_record(category, est_minutes, actual_minutes, float(completed_at)) | |
| def merge_records(existing: list[dict], payload: str) -> tuple[list[dict], int, int]: | |
| """Merge valid exported records without mutating existing browser state.""" | |
| try: | |
| data = json.loads(payload) | |
| except json.JSONDecodeError as exc: | |
| raise ValueError("invalid json") from exc | |
| if not isinstance(data, dict): | |
| raise ValueError("payload must be an object") | |
| records = data.get("records") | |
| if not isinstance(records, list): | |
| raise ValueError("records must be a list") | |
| rows = [ | |
| _validate_record_row(row, index) | |
| for index, row in enumerate(records, start=1) | |
| ] | |
| merged = list(existing) | |
| seen = {_record_key(record) for record in merged} | |
| imported = 0 | |
| skipped = 0 | |
| for row in rows: | |
| key = _record_key(row) | |
| if key in seen: | |
| skipped += 1 | |
| continue | |
| merged.append(row) | |
| seen.add(key) | |
| imported += 1 | |
| return merged, imported, skipped | |
| def export_payload(records: list[dict]) -> str: | |
| """Return a Store.export_json-compatible payload for browser records.""" | |
| return json.dumps({"tasks": [], "steps": [], "records": records}) | |
| def with_plan(data: dict, task: str, rows: list) -> dict: | |
| """Return browser state with the visible plan snapshot replaced.""" | |
| records = _records_from_data(data) | |
| return { | |
| "records": list(records), | |
| "plan": {"task": task, "rows": list(rows)}, | |
| } | |
| def with_records(data: dict, records: list) -> dict: | |
| """Return browser state with calibration records replaced.""" | |
| plan = data.get("plan") if isinstance(data, dict) else None | |
| return {"records": list(records), "plan": plan} | |
| def snapshot(data: dict, task: str, rows: list[dict]) -> dict: | |
| """Return browser state with the visible plan rows saved.""" | |
| return with_plan(data, task, rows) | |
| def summary_html(rows: list[dict[str, Any]]) -> str: | |
| """Return total plan timing for the current UI rows.""" | |
| if not rows: | |
| return "" | |
| active_rows = [row for row in rows if not row.get("skipped")] | |
| total_for_you = sum( | |
| int(row["actual_minutes"]) | |
| if row["logged"] | |
| else int(row["calibrated_minutes"]) | |
| for row in active_rows | |
| ) | |
| total_raw = sum(int(row["raw_minutes"]) for row in active_rows) | |
| n_done = sum(1 for row in rows if row["logged"]) | |
| n_skipped = sum(1 for row in rows if row.get("skipped")) | |
| text = f"For you: ~{total_for_you} min total · AI estimate: {total_raw} min" | |
| if n_done: | |
| text += f" · {n_done}/{len(rows)} done" | |
| if n_skipped: | |
| text += f" · {n_skipped} skipped" | |
| return f'<div class="summary">{text}</div>' | |
| def completion_html(rows: list[dict]) -> str: | |
| """Return the done-state banner once a plan has resolved.""" | |
| if not rows: | |
| return "" | |
| if not all(row["logged"] or row.get("skipped") for row in rows): | |
| return "" | |
| logged_rows = [row for row in rows if row["logged"]] | |
| if not logged_rows: | |
| return "" | |
| n_done = len(logged_rows) | |
| n_skipped = sum(1 for row in rows if row.get("skipped")) | |
| took = sum(int(row["actual_minutes"]) for row in logged_rows) | |
| est = sum(int(row["raw_minutes"]) for row in logged_rows) | |
| line1 = f"🎉 Done — {n_done} steps in {took} min." | |
| if n_skipped > 0: | |
| line1 += f" ({n_skipped} skipped)" | |
| ratio = took / est | |
| if ratio > 1.05: | |
| line2 = ( | |
| f"The AI guessed {est} min — you took {ratio:.1f}× longer, " | |
| "and Unstuck now knows that." | |
| ) | |
| elif ratio < 0.95: | |
| line2 = ( | |
| f"The AI guessed {est} min — you beat it, " | |
| f"finishing in {ratio:.1f}× the time." | |
| ) | |
| else: | |
| line2 = f"The AI guessed {est} min — almost exactly right." | |
| return f'<div class="completion">{line1}<br>{line2}</div>' | |
| def patterns_html(records: list[dict]) -> str: | |
| """Return per-category calibration history as compact HTML.""" | |
| if not records: | |
| return "" | |
| blocks = [] | |
| for category in sorted({str(record["category"]) for record in records}): | |
| category_records = [ | |
| (index, record) | |
| for index, record in enumerate(records) | |
| if str(record.get("category")) == category | |
| ] | |
| ordered = sorted( | |
| category_records, | |
| key=lambda item: (float(item[1].get("completed_at", item[0])), item[0]), | |
| ) | |
| recent = [record for _index, record in ordered[-5:]] | |
| mult = multiplier(category, records) | |
| if mult > 1.05: | |
| verdict = "you underestimate these" | |
| elif mult < 0.95: | |
| verdict = "you overestimate these" | |
| else: | |
| verdict = "your gut is right" | |
| bars = [] | |
| for record in recent: | |
| est = int(record["est_minutes"]) | |
| actual = int(record["actual_minutes"]) | |
| ratio = actual / est | |
| height = max(4, min(36, int(round(ratio * 14)))) | |
| title = html.escape(f"est {est} → took {actual} min", quote=True) | |
| bars.append( | |
| f'<span class="bar" style="height:{height}px" ' | |
| f'title="{title}"></span>' | |
| ) | |
| cat = html.escape(category) | |
| blocks.append( | |
| '<div class="pattern-row">' | |
| f'<span class="pattern-cat">{cat}</span>' | |
| f'<span class="pattern-mult">~{mult:.1f}× — {verdict}</span>' | |
| + "".join(bars) | |
| + f'<span class="pattern-n">{len(category_records)} logged</span>' | |
| "</div>" | |
| ) | |
| return '<div class="patterns">' + "".join(blocks) + "</div>" | |
| def _records_from_data(data: dict | None) -> list[dict[str, Any]]: | |
| if not isinstance(data, dict): | |
| return [] | |
| records = data.get("records") | |
| return records if isinstance(records, list) else [] | |
| def new_plan( | |
| data: dict, patterns: Callable[[list[dict[str, Any]]], str] | |
| ) -> tuple[list[dict[str, Any]], str, str, str, Any, dict]: | |
| """Clear the visible plan while keeping browser-stored records.""" | |
| records = _records_from_data(data) | |
| updated = {"records": list(records), "plan": None} | |
| return [], "", "", patterns(records), gr.update(value=""), updated | |
| def restored_banner_html(rows: list[dict]) -> str: | |
| """Transient 'welcome back' cue, shown only when a saved plan is restored on load. | |
| Any subsequent action recomputes the summary without it, so it self-clears | |
| after the first interaction — orienting a returning user (or a judge who | |
| reloaded mid-demo) without lingering. | |
| """ | |
| if not rows: | |
| return "" | |
| left = sum(1 for r in rows if not r.get("logged") and not r.get("skipped")) | |
| if left == 0: | |
| return "" | |
| steps = "step" if left == 1 else "steps" | |
| return ( | |
| '<div class="restored-banner">↩ Restored your plan from earlier ' | |
| f"— {left} {steps} left</div>" | |
| ) | |
| def recall_banner_html(matched_task: str) -> str: | |
| """Banner shown when a breakdown was shaped by a recalled similar task.""" | |
| safe = html.escape(matched_task) | |
| return ( | |
| '<div class="recall-banner">Shaped by a similar task you did before: ' | |
| f'<em>{safe}</em></div>' | |
| ) | |
| def restore_snapshot( | |
| data: dict, readout: Callable[[list[dict[str, Any]]], str] | |
| ) -> tuple[list[dict[str, Any]], str, str, str, Any]: | |
| """Return browser plan state for Gradio load, or empty state on bad data.""" | |
| records = _records_from_data(data) | |
| patterns = patterns_html(records) | |
| plan = data.get("plan") if isinstance(data, dict) else None | |
| if not isinstance(plan, dict): | |
| return [], readout(records), "", patterns, gr.update() | |
| saved_task = plan.get("task") | |
| rows = plan.get("rows") | |
| if not isinstance(saved_task, str) or not isinstance(rows, list): | |
| return [], readout(records), "", patterns, gr.update() | |
| summary = restored_banner_html(rows) + completion_html(rows) + summary_html(rows) | |
| return rows, readout(records), summary, patterns, gr.update(value=saved_task) | |
| def plan_markdown(task: str, rows: list[dict]) -> str: | |
| """Return the current plan as a portable markdown checklist.""" | |
| if not rows: | |
| return "" | |
| title = task.strip() or "My plan" | |
| lines = [f"## {title}", ""] | |
| total_for_you = 0 | |
| for row in rows: | |
| text = str(row["text"]) | |
| if row.get("skipped"): | |
| lines.append(f"- [-] {text} (skipped)") | |
| elif row["logged"]: | |
| actual = int(row["actual_minutes"]) | |
| total_for_you += actual | |
| lines.append(f"- [x] {text} (took {actual} min)") | |
| else: | |
| calibrated = int(row["calibrated_minutes"]) | |
| total_for_you += calibrated | |
| lines.append(f"- [ ] {text} (~{calibrated} min)") | |
| lines.extend(["", f"Total for you: ~{total_for_you} min"]) | |
| return "\n".join(lines) | |
| def share_text(task: str, rows: list[dict]) -> str: | |
| """Return a short, paste-anywhere progress update for the current plan.""" | |
| if not rows: | |
| return "" | |
| title = task.strip() or "my task" | |
| n_done = sum(1 for row in rows if row["logged"]) | |
| n_skipped = sum(1 for row in rows if row.get("skipped")) | |
| active = [row for row in rows if not row.get("skipped")] | |
| complete = all(row["logged"] or row.get("skipped") for row in rows) | |
| if complete: | |
| took = sum(int(row["actual_minutes"]) for row in active if row["logged"]) | |
| guessed = sum(int(row["raw_minutes"]) for row in active) | |
| line = ( | |
| f'Got unstuck: "{title}" — {n_done} steps in {took} min' | |
| f" (the AI guessed {guessed})." | |
| ) | |
| else: | |
| remaining = sum( | |
| int(row["calibrated_minutes"]) for row in active if not row["logged"] | |
| ) | |
| line = ( | |
| f'Getting unstuck: "{title}" — {n_done} of {len(rows)} steps done,' | |
| f" ~{remaining} min to go." | |
| ) | |
| return f"{line} Made with Unstuck: https://build-small-hackathon-unstuck.hf.space" | |
| def _ics_escape(text: str) -> str: | |
| """Escape a value for an iCalendar text field (RFC 5545 §3.3.11).""" | |
| return ( | |
| str(text) | |
| .replace("\\", "\\\\") | |
| .replace(";", "\\;") | |
| .replace(",", "\\,") | |
| .replace("\n", "\\n") | |
| ) | |
| def plan_ics(task: str, rows: list[dict], start: datetime) -> str: | |
| """Return the remaining steps as an iCalendar (.ics), back-to-back from ``start``. | |
| Only unlogged, non-skipped steps become events (the work still ahead), each | |
| ``calibrated_minutes`` long. Times are *floating* (no TZID / no trailing Z) so | |
| the blocks land in the importing calendar's own timezone — "block my next | |
| hour" imports exactly as authored, wherever it's opened. | |
| """ | |
| remaining = [r for r in rows if not r.get("logged") and not r.get("skipped")] | |
| if not remaining: | |
| return "" | |
| title = task.strip() or "My plan" | |
| stamp = start.strftime("%Y%m%dT%H%M%S") | |
| lines = [ | |
| "BEGIN:VCALENDAR", | |
| "VERSION:2.0", | |
| "PRODID:-//Unstuck//Plan//EN", | |
| "CALSCALE:GREGORIAN", | |
| ] | |
| cursor = start | |
| for row in remaining: | |
| minutes = int(row["calibrated_minutes"]) | |
| lines += [ | |
| "BEGIN:VEVENT", | |
| f"UID:unstuck-{row.get('step_id', 0)}-{stamp}@unstuck.app", | |
| f"DTSTAMP:{stamp}", | |
| f"DTSTART:{cursor.strftime('%Y%m%dT%H%M%S')}", | |
| f"DURATION:PT{minutes}M", | |
| f"SUMMARY:{_ics_escape(row['text'])}", | |
| f"DESCRIPTION:{_ics_escape(title)} — via Unstuck", | |
| "END:VEVENT", | |
| ] | |
| cursor = cursor + timedelta(minutes=minutes) | |
| lines.append("END:VCALENDAR") | |
| return "\r\n".join(lines) + "\r\n" | |
| def splice_rows( | |
| rows: list[dict[str, Any]], step_id: int, new_rows: list[dict[str, Any]] | |
| ) -> list[dict[str, Any]]: | |
| """Replace one visible step row with its freshly broken-down sub-steps.""" | |
| spliced: list[dict[str, Any]] = [] | |
| found = False | |
| for row in rows: | |
| if row["step_id"] == step_id: | |
| spliced.extend(new_rows) | |
| found = True | |
| else: | |
| spliced.append(row) | |
| return spliced if found else rows | |
| def build_ui( | |
| service: Unstuck, | |
| embed: Callable[[str], list[float] | None] | None = None, | |
| ) -> gr.Blocks: | |
| """Build the Gradio UI around an injected Unstuck service.""" | |
| embed_fn = embed if embed is not None else embeddings.embed | |
| def view_rows(view: BreakdownView) -> list[dict[str, Any]]: | |
| return [ | |
| { | |
| "step_id": row.step_id, | |
| "text": row.text, | |
| "category": row.category, | |
| "raw_minutes": row.raw_minutes, | |
| "calibrated_minutes": row.calibrated_minutes, | |
| "logged": False, | |
| "skipped": False, | |
| "actual_minutes": None, | |
| "started_at": None, | |
| } | |
| for row in view.rows | |
| ] | |
| def recalibrated( | |
| rows: list[dict[str, Any]], records: list[dict[str, Any]] | |
| ) -> list[dict[str, Any]]: | |
| return [ | |
| row | |
| if row["logged"] or row.get("skipped") | |
| else { | |
| **row, | |
| "calibrated_minutes": calibrate( | |
| row["raw_minutes"], multiplier(row["category"], records) | |
| ), | |
| } | |
| for row in rows | |
| ] | |
| def readout(records: list[dict[str, Any]]) -> str: | |
| if not records: | |
| return "" | |
| lines = [] | |
| for category in sorted({r["category"] for r in records}): | |
| mult = multiplier(category, records) | |
| n = sum(1 for r in records if r["category"] == category) | |
| if mult > 1.05: | |
| verdict = f"run ~{mult:.1f}× long" | |
| elif mult < 0.95: | |
| verdict = f"run short (~{mult:.1f}×)" | |
| else: | |
| verdict = "are about right" | |
| lines.append( | |
| f"Your <em>{category}</em> estimates {verdict} " | |
| f"({n} logged) — adjusting for that." | |
| ) | |
| return '<div class="readout">' + "<br>".join(lines) + "</div>" | |
| def patterns(records: list[dict[str, Any]]) -> str: | |
| return patterns_html(records) | |
| def persist(data: dict, task: str, rows: list[dict[str, Any]]) -> dict: | |
| return snapshot(data, task, rows) | |
| def restore( | |
| data: dict, | |
| ) -> tuple[list[dict[str, Any]], str, str, str, Any, dict, None]: | |
| rows, readout_html, summary, patterns_html_, task_update = restore_snapshot( | |
| data, readout | |
| ) | |
| return rows, readout_html, summary, patterns_html_, task_update, data, None | |
| def break_down( | |
| task: str, data: dict, granularity: str | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict, None, str, dict | None]: | |
| clean_task = task.strip() | |
| records = _records_from_data(data) | |
| if not clean_task: | |
| updated = persist(data, clean_task, []) | |
| return ( | |
| [], | |
| '<div class="explainer">Enter a task to break down.</div>', | |
| "", | |
| patterns(records), | |
| updated, | |
| None, | |
| "", | |
| None, | |
| ) | |
| vector = embed_fn(clean_task) | |
| match = recall.select(vector, _history_from_data(data)) | |
| exemplar = ( | |
| format_exemplar(match.entry["text"], match.entry["breakdown"]) | |
| if match is not None | |
| else None | |
| ) | |
| try: | |
| view = service.breakdown(clean_task, granularity, exemplar=exemplar) | |
| rows = recalibrated(view_rows(view), records) | |
| except Exception: | |
| gr.Warning("The model backend is busy. Try again in a minute.") | |
| updated = persist(data, clean_task, []) | |
| return ( | |
| [], | |
| '<div class="explainer">The model backend is busy or out of GPU quota. ' | |
| "Try again in a minute. Logging in to Hugging Face raises the free " | |
| "ZeroGPU quota.</div>", | |
| "", | |
| patterns(records), | |
| updated, | |
| None, | |
| "", | |
| None, | |
| ) | |
| banner = "" | |
| recall_pointer = None | |
| if match is not None: | |
| rows = recall.seed_estimates(rows, match.entry) | |
| banner = recall_banner_html(match.entry["text"]) | |
| recall_pointer = match.index | |
| history = _history_from_data(data) | |
| new_index = len(history) | |
| history = history + [ | |
| make_history_entry( | |
| clean_task, | |
| vector or [], | |
| [ | |
| { | |
| "text": row["text"], | |
| "category": row["category"], | |
| "est_minutes": row["raw_minutes"], | |
| } | |
| for row in rows | |
| ], | |
| ) | |
| ] | |
| updated = with_history(persist(data, clean_task, rows), history) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| None, | |
| banner, | |
| {"history_index": new_index, "dismiss_index": recall_pointer, | |
| "task": clean_task, "granularity": granularity}, | |
| ) | |
| def start_fresh( | |
| task: str, data: dict, granularity: str, pointer: dict | None | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict, None, str, None]: | |
| clean_task = task.strip() | |
| records = _records_from_data(data) | |
| data = apply_dismissal(data, pointer) | |
| history = _history_from_data(data) | |
| try: | |
| view = service.breakdown(clean_task, granularity, exemplar=None) | |
| rows = recalibrated(view_rows(view), records) | |
| except Exception: | |
| gr.Warning("The model backend is busy. Try again in a minute.") | |
| return ( | |
| [], | |
| readout(records), | |
| "", | |
| patterns(records), | |
| with_history(persist(data, clean_task, []), history), | |
| None, | |
| "", | |
| None, | |
| ) | |
| updated = with_history(persist(data, clean_task, rows), history) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| None, | |
| "", | |
| None, | |
| ) | |
| def new_plan_ui( | |
| data: dict, | |
| ) -> tuple[list[dict[str, Any]], str, str, str, Any, dict, None]: | |
| rows, readout_html, summary, patterns_html_, task_update, updated = new_plan( | |
| data, patterns | |
| ) | |
| return rows, readout_html, summary, patterns_html_, task_update, updated, None | |
| def log_step( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| minutes: float | None, | |
| task: str, | |
| rows: list[dict[str, Any]], | |
| data: dict, | |
| recall_state: dict[str, Any] | None, | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| records = _records_from_data(data) | |
| row = next((row for row in rows if row["step_id"] == step_id), None) | |
| if row is None: | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| now = time.time() | |
| actual = finish_minutes(minutes, row.get("started_at"), now) | |
| if actual is None: | |
| gr.Warning("Press Start first or enter minutes.") | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| category = str(row["category"]) | |
| records = records + [ | |
| make_record( | |
| category, | |
| int(row["raw_minutes"]), | |
| actual, | |
| now, | |
| ) | |
| ] | |
| rows = [ | |
| { | |
| **row, | |
| "logged": True, | |
| "actual_minutes": actual, | |
| "started_at": None, | |
| "record_at": now, | |
| } | |
| if row["step_id"] == step_id | |
| else row | |
| for row in rows | |
| ] | |
| rows = recalibrated(rows, records) | |
| updated = persist(with_records(data, records), task, rows) | |
| pointer = recall_state if isinstance(recall_state, dict) else {} | |
| history_index = pointer.get("history_index") | |
| if history_index is not None: | |
| history = record_duration_in_history( | |
| _history_from_data(updated), | |
| int(history_index), | |
| category, | |
| actual, | |
| ) | |
| updated = with_history(updated, history) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| return handler | |
| def undo_step( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| task: str, rows: list[dict[str, Any]], data: dict | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| records = _records_from_data(data) | |
| row = next((row for row in rows if row["step_id"] == step_id), None) | |
| if row is None: | |
| updated = persist(with_records(data, records), task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| if row.get("skipped"): | |
| rows = undo_row(rows, step_id) | |
| elif row["logged"]: | |
| records = remove_record( | |
| records, | |
| str(row["category"]), | |
| int(row["raw_minutes"]), | |
| int(row["actual_minutes"]), | |
| row.get("record_at"), | |
| ) | |
| rows = undo_row(rows, step_id) | |
| rows = recalibrated(rows, records) | |
| updated = persist(with_records(data, records), task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| return handler | |
| def start_step( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| task: str, rows: list[dict[str, Any]], data: dict | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| records = _records_from_data(data) | |
| rows = [ | |
| {**row, "started_at": time.time()} | |
| if row["step_id"] == step_id | |
| else row | |
| for row in rows | |
| ] | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| return handler | |
| def skip_step( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| task: str, rows: list[dict[str, Any]], data: dict | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| records = _records_from_data(data) | |
| rows = [ | |
| {**row, "skipped": True, "started_at": None} | |
| if row["step_id"] == step_id | |
| else row | |
| for row in rows | |
| ] | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| return handler | |
| def break_down_step( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| task: str, rows: list[dict[str, Any]], data: dict | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| records = _records_from_data(data) | |
| if len(rows) >= 16: | |
| gr.Warning("That's plenty of steps — try starting the first tiny one") | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| step_text = next( | |
| (str(row["text"]) for row in rows if row["step_id"] == step_id), | |
| None, | |
| ) | |
| if step_text is None: | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| try: | |
| new_rows = recalibrated( | |
| view_rows(service.breakdown(step_text, "tiny")), records | |
| ) | |
| except Exception: | |
| gr.Warning( | |
| "The model backend is busy or out of GPU quota. " | |
| "Try again in a minute." | |
| ) | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| spliced = splice_rows(rows, step_id, new_rows) | |
| updated = persist(data, task, spliced) | |
| return ( | |
| spliced, | |
| readout(records), | |
| completion_html(spliced) + summary_html(spliced), | |
| patterns(records), | |
| updated, | |
| ) | |
| return handler | |
| def edit_step(step_id: int) -> Callable[[], int]: | |
| def handler() -> int: | |
| return step_id | |
| return handler | |
| def save_step_text( | |
| step_id: int, | |
| ) -> Any: | |
| def handler( | |
| new_text: str, | |
| task: str, | |
| rows: list[dict[str, Any]], | |
| data: dict, | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict, None]: | |
| records = _records_from_data(data) | |
| rows = edit_row_text(rows, step_id, new_text) | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| None, | |
| ) | |
| return handler | |
| def add_step( | |
| text: str, | |
| minutes: float | None, | |
| task: str, | |
| rows: list[dict[str, Any]], | |
| data: dict, | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict, Any]: | |
| records = _records_from_data(data) | |
| rows = add_manual_row(rows, text, minutes, records) | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| gr.update(value=""), | |
| ) | |
| def export_data(data: dict) -> tuple[str, dict]: | |
| records = _records_from_data(data) | |
| handle = tempfile.NamedTemporaryFile( | |
| mode="w", | |
| encoding="utf-8", | |
| suffix=".json", | |
| prefix="unstuck-", | |
| delete=False, | |
| ) | |
| with handle: | |
| handle.write(export_payload(records)) | |
| return gr.update(value=handle.name, visible=True), data | |
| def import_data( | |
| file: Any, task: str, rows: list[dict[str, Any]], data: dict | |
| ) -> tuple[list[dict[str, Any]], str, str, str, dict]: | |
| file_path = getattr(file, "name", file) | |
| records = _records_from_data(data) | |
| try: | |
| payload = Path(file_path).read_text(encoding="utf-8") | |
| records, imported, skipped = merge_records(records, payload) | |
| status = f"Imported {imported} records ({skipped} duplicates skipped)" | |
| except (OSError, TypeError, ValueError): | |
| gr.Warning("That file doesn't look like an Unstuck export.") | |
| updated = persist(data, task, rows) | |
| return ( | |
| rows, | |
| readout(records), | |
| completion_html(rows) + summary_html(rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| updated_rows = recalibrated(rows, records) | |
| status_html = f'<div class="summary">{status}</div>' | |
| updated = persist(with_records(data, records), task, updated_rows) | |
| return ( | |
| updated_rows, | |
| readout(records) + status_html, | |
| completion_html(updated_rows) + summary_html(updated_rows), | |
| patterns(records), | |
| updated, | |
| ) | |
| def copy_checklist(task: str, rows: list[dict[str, Any]]) -> Any: | |
| markdown = plan_markdown(task, rows) | |
| return gr.update(value=markdown, visible=bool(markdown)) | |
| def copy_share(task: str, rows: list[dict[str, Any]]) -> Any: | |
| text = share_text(task, rows) | |
| return gr.update(value=text, visible=bool(text)) | |
| def export_ics(task: str, rows: list[dict[str, Any]]) -> Any: | |
| text = plan_ics(task, rows, datetime.now()) | |
| if not text: | |
| gr.Warning("No remaining steps to put on the calendar.") | |
| return gr.update(visible=False) | |
| handle = tempfile.NamedTemporaryFile( | |
| mode="w", encoding="utf-8", suffix=".ics", prefix="unstuck-plan-", delete=False | |
| ) | |
| with handle: | |
| handle.write(text) | |
| return gr.update(value=handle.name, visible=True) | |
| with gr.Blocks(title="Unstuck") as ui: | |
| gr.HTML( | |
| '<div id="hero"><h1>Unstuck</h1>' | |
| "<p>One overwhelming task → tiny timed steps, honest about how long " | |
| "<em>you</em> actually take.</p></div>" | |
| ) | |
| rows_state = gr.State([]) | |
| user_data = gr.BrowserState( | |
| {"records": [], "plan": None, "history": []}, storage_key="unstuck-v1" | |
| ) | |
| task = gr.Textbox( | |
| label="Task", | |
| placeholder="Paste the overwhelming thing here", | |
| lines=3, | |
| ) | |
| editing_step_id = gr.State(None) | |
| granularity = gr.Radio( | |
| choices=["chunky", "regular", "tiny"], | |
| value="regular", | |
| label="Step size", | |
| info="How small should the pieces be?", | |
| ) | |
| with gr.Row(): | |
| break_button = gr.Button("Break it down", variant="primary") | |
| new_plan_button = gr.Button("New plan", variant="secondary") | |
| gr.Examples( | |
| examples=[ | |
| ["Clean up my inbox and reply to the important emails"], | |
| ["Prepare a small birthday dinner for four"], | |
| ["Unpack and organise my desk after moving"], | |
| ], | |
| inputs=task, | |
| cache_examples=False, | |
| ) | |
| readout_output = gr.HTML() | |
| recall_banner_output = gr.HTML() | |
| recall_state = gr.State(None) | |
| summary_output = gr.HTML() | |
| with gr.Accordion("Your patterns", open=False): | |
| patterns_output = gr.HTML() | |
| def render_recall_actions(pointer: dict | None) -> None: | |
| if not pointer or pointer.get("dismiss_index") is None: | |
| return | |
| fresh = gr.Button("Start fresh", size="sm", variant="secondary") | |
| fresh.click( | |
| start_fresh, | |
| inputs=[task, user_data, granularity, recall_state], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| editing_step_id, | |
| recall_banner_output, | |
| recall_state, | |
| ], | |
| api_name="start_fresh", | |
| ) | |
| def render_rows(rows: list[dict[str, Any]], editing_id: int | None) -> None: | |
| import html as html_lib | |
| if not rows: | |
| return | |
| spotlight = next_step_id(rows) | |
| for index, row in enumerate(rows, start=1): | |
| text = html_lib.escape(str(row["text"])) | |
| is_spotlight = row["step_id"] == spotlight | |
| if row.get("skipped"): | |
| card_class = "step-card step-skipped" | |
| step_num = "–" | |
| chips = '<div class="chip chip-skip">skipped</div>' | |
| elif row["logged"]: | |
| card_class = "step-card" | |
| step_num = "✓" | |
| chips = ( | |
| f'<div class="chip chip-raw">AI: {row["raw_minutes"]} min</div>' | |
| f'<div class="chip chip-done">took {row["actual_minutes"]} min</div>' | |
| ) | |
| elif is_spotlight: | |
| card_class = "step-card step-next" | |
| step_num = str(index) | |
| chips = ( | |
| f'<div class="chip chip-raw">AI: {row["raw_minutes"]} min</div>' | |
| f'<div class="chip chip-you">For you: ' | |
| f'{row["calibrated_minutes"]} min</div>' | |
| ) | |
| else: | |
| card_class = "step-card step-later" | |
| step_num = str(index) | |
| chips = ( | |
| f'<div class="chip chip-raw">AI: {row["raw_minutes"]} min</div>' | |
| f'<div class="chip chip-you">For you: ' | |
| f'{row["calibrated_minutes"]} min</div>' | |
| ) | |
| timer_chip = ( | |
| '<div class="chip chip-timer">⏱ timing</div>' | |
| if row.get("started_at") is not None | |
| and not row["logged"] | |
| and not row.get("skipped") | |
| else "" | |
| ) | |
| with gr.Row(elem_classes="step-row"): | |
| gr.HTML( | |
| f'<div class="{card_class}">' | |
| f'<div class="step-num">{step_num}</div>' | |
| f'<div class="step-text">{text}</div>' | |
| + chips | |
| + timer_chip | |
| + "</div>", | |
| padding=False, | |
| elem_classes="step-html", | |
| ) | |
| if row["logged"] or row.get("skipped"): | |
| undo = gr.Button( | |
| "Undo", | |
| size="sm", | |
| scale=0, | |
| min_width=70, | |
| variant="secondary", | |
| ) | |
| undo.click( | |
| undo_step(int(row["step_id"])), | |
| inputs=[task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| api_visibility="private", | |
| ) | |
| elif is_spotlight: | |
| start = gr.Button( | |
| "Restart" if row.get("started_at") is not None else "Start", | |
| size="sm", | |
| scale=0, | |
| min_width=80, | |
| variant="secondary", | |
| ) | |
| start.click( | |
| start_step(int(row["step_id"])), | |
| inputs=[task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| api_visibility="private", | |
| ) | |
| minutes = gr.Number( | |
| show_label=False, | |
| container=False, | |
| minimum=1, | |
| precision=0, | |
| placeholder="took (min)", | |
| scale=0, | |
| min_width=110, | |
| elem_classes="took-input", | |
| ) | |
| done = gr.Button( | |
| "Done", size="sm", scale=0, min_width=70, variant="secondary" | |
| ) | |
| done.click( | |
| log_step(int(row["step_id"])), | |
| inputs=[minutes, task, rows_state, user_data, recall_state], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| api_visibility="private", | |
| ) | |
| still_stuck = gr.Button( | |
| "Still stuck?", | |
| size="sm", | |
| scale=0, | |
| min_width=110, | |
| variant="secondary", | |
| ) | |
| still_stuck.click( | |
| break_down_step(int(row["step_id"])), | |
| inputs=[task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| api_visibility="private", | |
| ) | |
| skip = gr.Button( | |
| "Skip", | |
| size="sm", | |
| scale=0, | |
| min_width=70, | |
| variant="secondary", | |
| ) | |
| skip.click( | |
| skip_step(int(row["step_id"])), | |
| inputs=[task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| api_visibility="private", | |
| ) | |
| edit = gr.Button( | |
| "Edit", | |
| size="sm", | |
| scale=0, | |
| min_width=70, | |
| variant="secondary", | |
| ) | |
| edit.click( | |
| edit_step(int(row["step_id"])), | |
| outputs=editing_step_id, | |
| api_visibility="private", | |
| ) | |
| if is_spotlight and editing_id == row["step_id"]: | |
| with gr.Row(elem_classes="step-row"): | |
| edit_text = gr.Textbox( | |
| value=str(row["text"]), | |
| show_label=False, | |
| container=False, | |
| scale=1, | |
| ) | |
| save = gr.Button( | |
| "Save", | |
| size="sm", | |
| scale=0, | |
| min_width=70, | |
| variant="primary", | |
| ) | |
| save.click( | |
| save_step_text(int(row["step_id"])), | |
| inputs=[edit_text, task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| editing_step_id, | |
| ], | |
| api_visibility="private", | |
| ) | |
| gr.HTML( | |
| '<div class="explainer">"For you" recalibrates the AI estimate from the ' | |
| "actual times you log — your personal time-blindness, learned per " | |
| "category. Log a step with <em>took (min) → Done</em> and watch the " | |
| "remaining estimates adjust.</div>" | |
| ) | |
| with gr.Row(): | |
| manual_text = gr.Textbox( | |
| label="Your own step", | |
| placeholder="Add your own step", | |
| show_label=False, | |
| scale=1, | |
| ) | |
| manual_minutes = gr.Number( | |
| label="Minutes", | |
| value=5, | |
| placeholder="min", | |
| show_label=False, | |
| minimum=1, | |
| precision=0, | |
| scale=0, | |
| min_width=90, | |
| ) | |
| add_button = gr.Button("Add step", variant="secondary", scale=0, min_width=90) | |
| with gr.Row(): | |
| export_button = gr.Button("Export my data (JSON)") | |
| import_button = gr.UploadButton( | |
| "Import my data (JSON)", | |
| file_types=[".json"], | |
| ) | |
| copy_button = gr.Button("Copy as checklist") | |
| share_button = gr.Button("Copy share update") | |
| ics_button = gr.Button("Add remaining steps to calendar (.ics)") | |
| export_file = gr.File(label="Download", interactive=False, visible=False) | |
| ics_file = gr.File(label="Calendar (.ics)", interactive=False, visible=False) | |
| checklist_output = gr.Textbox( | |
| label="Checklist", lines=8, visible=False, buttons=["copy"] | |
| ) | |
| share_output = gr.Textbox( | |
| label="Share update", lines=2, visible=False, buttons=["copy"] | |
| ) | |
| break_button.click( | |
| break_down, | |
| inputs=[task, user_data, granularity], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| editing_step_id, | |
| recall_banner_output, | |
| recall_state, | |
| ], | |
| ) | |
| new_plan_button.click( | |
| new_plan_ui, | |
| inputs=user_data, | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| task, | |
| user_data, | |
| editing_step_id, | |
| ], | |
| ) | |
| task.submit( | |
| break_down, | |
| inputs=[task, user_data, granularity], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| editing_step_id, | |
| recall_banner_output, | |
| recall_state, | |
| ], | |
| ) | |
| add_button.click( | |
| add_step, | |
| inputs=[manual_text, manual_minutes, task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| manual_text, | |
| ], | |
| api_visibility="private", | |
| ) | |
| export_button.click(export_data, inputs=user_data, outputs=[export_file, user_data]) | |
| import_button.upload( | |
| import_data, | |
| inputs=[import_button, task, rows_state, user_data], | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| user_data, | |
| ], | |
| ) | |
| copy_button.click( | |
| copy_checklist, | |
| inputs=[task, rows_state], | |
| outputs=checklist_output, | |
| ) | |
| share_button.click( | |
| copy_share, | |
| inputs=[task, rows_state], | |
| outputs=share_output, | |
| ) | |
| ics_button.click( | |
| export_ics, | |
| inputs=[task, rows_state], | |
| outputs=ics_file, | |
| ) | |
| ui.load( | |
| restore, | |
| inputs=user_data, | |
| outputs=[ | |
| rows_state, | |
| readout_output, | |
| summary_output, | |
| patterns_output, | |
| task, | |
| user_data, | |
| editing_step_id, | |
| ], | |
| ) | |
| return ui | |
| def main() -> None: # pragma: no cover | |
| data_dir = Path(os.environ.get("UNSTUCK_DATA_DIR", "data")) | |
| data_dir.mkdir(parents=True, exist_ok=True) | |
| from unstuck.backend import generate | |
| service = Unstuck(generate=generate, store=Store(str(data_dir / "unstuck.sqlite3"))) | |
| build_ui(service).launch(theme=THEME, css=CSS) | |
| if __name__ == "__main__": | |
| main() | |