import os import uuid import json from pathlib import Path import gradio as gr try: import spaces # only available on Hugging Face Spaces runtime except ImportError: # Provide a dummy so @spaces.GPU(duration=...) is a no-op locally # while keeping the decorator statically visible in the source code. class _spaces: @staticmethod def GPU(duration=300): return lambda fn: fn spaces = _spaces from app.services.retrieval import load_games_dataset, normalize_game_record, retrieve_examples from app.services.generator import ( generate_game, generate_game_with_model, build_generation_prompt, unload_nemotron, ) from app.services.validator import validate_game, repair_game from app.services.schema_validator import create_minimal_game_template from app.services.tracing import log_event, log_generation_trace, load_events from app.services.journal import ( create_journal_entry, save_journal_entry, summarize_journal, load_journal_entries, detect_mood, assess_story_value, ) from app.services.scoring import compute_scores from app.services.story import build_story_packet, generate_story from app.services.image_gen import generate_poster_sync # ── Load dataset once on startup ────────────────────────────────────────────── DATASET_PATH = Path(__file__).resolve().parent / "app/data/games_dataset.json" DATA_RECORDS = [] try: raw = load_games_dataset(str(DATASET_PATH)) DATA_RECORDS = [normalize_game_record(r) for r in raw] print(f"[OK] Loaded {len(DATA_RECORDS)} game records for retrieval") except FileNotFoundError: print(f"[!] Dataset not found at {DATASET_PATH}, retrieval will be empty") # ── In-memory session store (resets on restart — fine for hackathon) ─────── SESSION_STORE: dict[str, dict] = {} # session_id -> {config: {...}, game: {...}, events: [...], journals: [...]} # ── HF Spaces Zero GPU generation ────────────────────────────────────────── # This function is statically decorated so HF Spaces detects the GPU # requirement at startup. The 2.84 GB GGUF model is lazily downloaded # on first call inside the GPU context. @spaces.GPU(duration=300) def _generate_with_gpu(config: dict, retrieved: list[dict]): """Generate game using GPU (Nemotron 3 Nano 4B via llama.cpp). Runs inside @spaces.GPU so CUDA is available. The model download and llama.cpp initialisation happen lazily on first call. """ import json prompt = build_generation_prompt(config, retrieved) json_str = generate_game_with_model(prompt, model_name="nemotron") if json_str: try: game = json.loads(json_str) if all(field in game for field in ["game_id", "title", "setup", "tasks", "safety"]): return game except json.JSONDecodeError: pass return None def _generate_game(config: dict, retrieved: list[dict]): """Try GPU generation; fall back to CPU/mock if unavailable.""" if spaces is not None: game = _generate_with_gpu(config, retrieved) if game is not None: return game print("[app] GPU generation returned None, falling back to CPU/mock") return generate_game(config, retrieved) # ── Pipeline entry point ────────────────────────────────────────────────────── def run_pipeline( game_type: str, city: str, area: str, location_type: str, duration_minutes: int, num_players: int, difficulty: str, age_group: str, energy_level: str, ): """Run the full AI generation pipeline end-to-end.""" session_id = str(uuid.uuid4()) config = { "game_type": game_type, "city": city or "Paris", "area": area or "Downtown", "location_type": location_type, "duration_minutes": int(duration_minutes), "num_players": int(num_players), "difficulty": difficulty, "age_group": age_group, "energy_level": energy_level, "photo_enabled": True, } state = {} # 1 ── Retrieval state["num_retrieved"] = 0 if DATA_RECORDS: retrieved = retrieve_examples(config, DATA_RECORDS, k=3) state["num_retrieved"] = len(retrieved) state["retrieved_ids"] = [r["id"] for r in retrieved] else: retrieved = [] # 2 ── Generation (GPU if available, CPU/mock fallback otherwise) game = _generate_game(config, retrieved) state["game_id"] = game["game_id"] state["game_title"] = game["title"] # Free GPU memory — Nemotron is no longer needed after generation. # This makes room for FLUX poster / Cohere ASR later. unload_nemotron() # 3 ── Validation is_valid, failures = validate_game(game, config) state["validation_passed"] = is_valid state["validation_failures"] = failures # 4 ── Repair (if needed) repaired = None if not is_valid: repaired = repair_game(game, failures, config) state["repair_applied"] = True is_valid2, failures2 = validate_game(repaired, config) state["repair_valid"] = is_valid2 state["remaining_failures"] = failures2 else: state["repair_applied"] = False final_game = repaired if repaired is not None else game # 5 ── Log generation trace log_generation_trace( session_id=session_id, config=config, retrieved_examples=retrieved, game=final_game, validation_passed=is_valid or (repaired is not None and state.get("repair_valid", False)), validation_failures=failures, repaired_game=repaired, ) # 6 ── Reveal all tasks as events for task in final_game.get("tasks", []): log_event(session_id, "task_revealed", { "task_id": task["task_id"], "title": task["title"], "points": task["points"], }) # 7 ── Store session SESSION_STORE[session_id] = { "config": config, "game": final_game, "events": [], "journals": [], } # 8 ── Build summary text summary = build_summary(final_game, state, session_id) return summary, session_id # ── Phase 3: Gameplay helpers ───────────────────────────────────────────────── def complete_task(session_id: str, task_id: str, team_id: str = "team-a"): """Log a task completion event and return updated scoreboard.""" if session_id not in SESSION_STORE: return "⚠ Unknown session" ev = log_event(session_id, "task_completed", { "task_id": task_id, "summary": f"Team {team_id} completed {task_id}", }, team_id=team_id) SESSION_STORE[session_id]["events"].append(ev) return f"✅ Task {task_id} completed!" def skip_task(session_id: str, task_id: str, team_id: str = "team-a"): """Log a task skip event.""" if session_id not in SESSION_STORE: return "⚠ Unknown session" ev = log_event(session_id, "task_skipped", { "task_id": task_id, "summary": f"Team {team_id} skipped {task_id}", }, team_id=team_id) SESSION_STORE[session_id]["events"].append(ev) return f"⏭️ Task {task_id} skipped." def use_hint(session_id: str, task_id: str, team_id: str = "team-a"): """Log a hint usage event.""" if session_id not in SESSION_STORE: return "⚠ Unknown session" ev = log_event(session_id, "hint_used", { "task_id": task_id, "summary": f"Team {team_id} used a hint for {task_id}", }, team_id=team_id) SESSION_STORE[session_id]["events"].append(ev) return f"💡 Hint used for {task_id} (−5 pts)" @spaces.GPU(duration=120) def record_journal( session_id: str, transcript: str = "", task_id: str = "", location_note: str = "", team_id: str = "team-a", audio_path: str = "", language: str = "en", ): """Record a journal entry, summarize it, and return the result. Decorated with ``@spaces.GPU`` so the Cohere ASR model can run on the GPU. On HF ZeroGPU, ``torch.cuda.is_available()`` is ``False`` outside a ``@spaces.GPU`` function, so without this the ASR model would load on CPU and the voice-journal button would appear to hang. Two input paths are supported: * **Voice path** — pass ``audio_path`` (e.g. a path returned by ``gr.Audio(type="filepath")``). The audio is transcribed with the Cohere ASR service (``CohereLabs/cohere-transcribe-03-2026``) and the transcript is stored with ASR metadata. * **Typed path** — pass ``transcript`` directly. The user can also edit an ASR transcript in the UI before submitting; if both ``audio_path`` and ``transcript`` are present, the typed transcript wins and is treated as a manual correction (``transcript_source == "hybrid"``). """ if session_id not in SESSION_STORE: return "[!] Unknown session" asr_metadata: dict | None = None audio_ref: str | None = None transcript_source = "typed" # ── 1. Voice path — transcribe audio if provided ─────────────────── if audio_path: audio_ref = audio_path try: from app.services.asr import transcribe as _asr_transcribe asr_result = _asr_transcribe(audio_path, language=language) asr_metadata = { "model": asr_result.get("model"), "language": asr_result.get("language"), "status": asr_result.get("status"), "error": asr_result.get("error"), } asr_text = (asr_result.get("transcript") or "").strip() if asr_text and not transcript.strip(): transcript = asr_text transcript_source = "asr" elif asr_text and transcript.strip() and asr_text != transcript.strip(): transcript_source = "hybrid" except Exception as exc: print(f"[app] ASR transcription failed: {type(exc).__name__}: {exc}") asr_metadata = { "model": None, "language": language, "status": "error", "error": f"{type(exc).__name__}: {exc}", } if not transcript or not transcript.strip(): # Surface the actual error instead of a generic message hint = "" if asr_metadata: err = asr_metadata.get("error") or "" status = asr_metadata.get("status") or "" if status == "error" and err: hint = f"\n\n**ASR error:** {err}" elif status == "skipped": hint = ( "\n\n**ASR was skipped** — set `CITYQUEST_SKIP_MODEL=0` " "and ensure `HF_TOKEN` is configured." ) return ( "⚠️ No transcript available — record audio or type a note first." + hint ) # ── 2. Build journal entry ──────────────────────────────────────── entry = create_journal_entry( transcript=transcript, session_id=session_id, team_id=team_id, task_id=task_id or None, location_note=location_note, audio_ref=audio_ref, asr_metadata=asr_metadata, transcript_source=transcript_source, ) # Summarize summary = summarize_journal(transcript, task_id=task_id or None, location_note=location_note) entry["moment_summary"] = summary["moment_summary"] entry["tags"] = summary["tags"] entry["story_value"] = summary["story_value"] # Persist save_journal_entry(entry) # Log event ev = log_event(session_id, "journal_recorded", { "journal_id": entry["journal_id"], "mood": entry["mood"], "story_value": summary["story_value"], "summary": summary["moment_summary"], "transcript_source": transcript_source, "asr_status": asr_metadata.get("status") if asr_metadata else None, "asr_model": asr_metadata.get("model") if asr_metadata else None, }, team_id=team_id) SESSION_STORE[session_id]["events"].append(ev) SESSION_STORE[session_id]["journals"].append(entry) # Build display text source_label = { "asr": "🎙️ (transcribed)", "hybrid": "🎙️✏️ (transcribed, edited)", "typed": "⌨️ (typed)", }.get(transcript_source, transcript_source) asr_line = "" if asr_metadata and asr_metadata.get("status") != "ok": asr_line = f"- ASR status: **{asr_metadata.get('status')}** — {asr_metadata.get('error') or ''}\n" display = ( f"🎙️ **Journal recorded!** {source_label}\n" f"- Mood: *{entry['mood']}*\n" f"- Story value: **{summary['story_value']}**\n" f"- Tags: {', '.join(summary['tags'])}\n" f"{asr_line}" f"- Summary: {summary['moment_summary']}" ) return display def upload_photo( session_id: str, photo_file, caption: str = "", task_id: str = "", team_id: str = "team-a", ): """Log a photo upload event and return a confirmation.""" if session_id not in SESSION_STORE: return "[!] Unknown session" photo_name = "" if photo_file is not None: # Gradio 4+ returns a filepath string or a PIL image if isinstance(photo_file, str): photo_name = photo_file.split("/")[-1].split("\\")[-1] else: photo_name = getattr(photo_file, "name", "photo") photo_id = f"photo-{uuid.uuid4().hex[:8]}" payload = { "photo_id": photo_id, "photo_name": photo_name, "caption": caption, "summary": f"Team {team_id} uploaded photo for {task_id or 'general'}", } if task_id: payload["task_id"] = task_id ev = log_event(session_id, "photo_uploaded", payload, team_id=team_id) SESSION_STORE[session_id]["events"].append(ev) # Track photo in session store for recap if "photos" not in SESSION_STORE[session_id]: SESSION_STORE[session_id]["photos"] = [] SESSION_STORE[session_id]["photos"].append({ "photo_id": photo_id, "photo_name": photo_name, "photo_path": photo_file if isinstance(photo_file, str) else "", "caption": caption, "task_id": task_id, }) # Build gallery list photos = SESSION_STORE[session_id]["photos"] gallery_lines = [f"📸 **{p['photo_id']}** — {p['caption'] or '(no caption)'} [{p['task_id'] or 'general'}]" for p in photos] display = ( f"📸 **Photo uploaded!**\n" f"- ID: `{photo_id}`\n" f"- Caption: {caption or '(none)'}\n" f"- Related task: {task_id or 'general'}\n\n" f"**All photos ({len(photos)}):**\n" + "\n".join(gallery_lines) ) return display def end_game(session_id: str, team_id: str = "team-a"): """End the game, compute scores, and return a scoreboard.""" if session_id not in SESSION_STORE: return "[!] Unknown session" session = SESSION_STORE[session_id] game = session["game"] events = load_events(session_id=session_id) ev = log_event(session_id, "game_finished", { "summary": f"Game finished — session {session_id}", }, team_id=team_id) events.append(ev) scores = compute_scores(events, game) session["scores"] = scores lines = ["# Final Scoreboard\n"] for ts in scores.get("team_scores", []): marker = " [WINNER]" if ts["team_id"] == scores.get("winner") else "" lines.append(f"### Team: {ts['team_id']}{marker}") lines.append(f"- **Total points:** {ts['points']}") lines.append(f"- Tasks completed: {ts['completed_tasks']}/{ts['total_tasks']}") lines.append(f"- Hints used: {ts['hints_used']}") if ts.get("bonuses"): lines.append(f"- Bonuses: {', '.join(ts['bonuses'])}") lines.append("") lines.append("**Breakdown:**") for b in ts.get("scoring_breakdown", []): lines.append(f" * {b}") lines.append("") if scores.get("winner"): lines.append(f"**Winner: {scores['winner']}**") return "\n".join(lines) def generate_recap(session_id: str): """Generate the final story recap from all collected session data.""" if session_id not in SESSION_STORE: return "[!] Unknown session", {} session = SESSION_STORE[session_id] game = session["game"] events = load_events(session_id=session_id) journals = load_journal_entries(session_id=session_id) scores = session.get("scores", compute_scores(events, game)) photos = session.get("photos", []) # Build story packet packet = build_story_packet( game=game, events=events, scores=scores, journal_entries=journals, photo_captions=photos, ) # Generate story result = generate_story(packet, session_id=session_id) # Format for display (poster goes in dedicated section below) lines = [ "# Episode Recap\n", result["short_recap"], "", "---\n", result["long_summary"], "", "---\n", ] lines.append(f"**Poster prompt:** _{result['poster_prompt']}_\n") lines.append(f"*Click **Generate Poster** below to create an image from this prompt.*\n") return "\n".join(lines), result @spaces.GPU(duration=300) def generate_poster_from_session(session_id: str): """Generate a poster image from an existing session's recap data.""" if session_id not in SESSION_STORE: return None, "[!] Unknown session - generate a recap first" session = SESSION_STORE[session_id] # Check if we have a recap result or can build one if "poster_url" in session: return session["poster_url"], "Poster ready from previous generation" # Build story packet from stored data game = session.get("game") if not game: return None, "[!] No game data in session" events = load_events(session_id=session_id) journals = load_journal_entries(session_id=session_id) scores = session.get("scores", compute_scores(events, game)) photos = session.get("photos", []) packet = build_story_packet( game=game, events=events, scores=scores, journal_entries=journals, photo_captions=photos, ) result = generate_story(packet, session_id=session_id) poster_url, status = generate_poster_sync( session_id, result["poster_prompt"], photo_paths=[p.get("photo_path", "") for p in photos if p.get("photo_path")], ) session["poster_url"] = poster_url return poster_url, status # ── Build summary text ──────────────────────────────────────────────────────── def build_summary(game: dict, state: dict, session_id: str = "") -> str: """Format a human-readable game summary for the UI.""" lines = [] lines.append(f"# 🎮 {game.get('title', 'Untitled')}") lines.append("") if session_id: lines.append(f"> Session `{session_id[:12]}…` — paste this in the **Play** tab to log progress.") lines.append("") lines.append("## 📋 Setup") setup = game.get("setup", {}) lines.append(f"- **Location:** {setup.get('city', '?')} — {setup.get('area', '?')}") lines.append(f"- **Meeting point:** {setup.get('meeting_point', '?')}") lines.append(f"- **Duration:** {setup.get('duration_minutes', '?')} min") lines.append(f"- **Players:** {setup.get('num_players', '?')}") lines.append("") lines.append("## 📜 Rules") for i, rule in enumerate(game.get("rules", []), 1): lines.append(f" {i}. {rule}") lines.append("") lines.append("## 🎯 Tasks") for t in game.get("tasks", []): time_str = f"{t.get('time_limit_minutes', '∞')} min" if t.get("time_limit_minutes") else "No time limit" lines.append(f" **{t.get('task_id', '?')}:** {t.get('title', '?')}") lines.append(f" - *{t.get('description', '')[:80]}*") lines.append(f" - 🏆 {t.get('points', 0)} pts | ⏱ {time_str} | 📸 {t.get('proof_type', '?')}") lines.append(f" - 💡 {t.get('hint', '')[:70]}") lines.append(f" - 🛡 {t.get('safety_note', '')[:70]}") lines.append("") lines.append("## 💡 Global Hints") for h in game.get("global_hints", []): lines.append(f" • {h}") lines.append("") lines.append("## 🔒 Safety") safety = game.get("safety", {}) lines.append(f"- **Zone:** {safety.get('allowed_zone', '?')}") lines.append(f"- **Supervision required:** {'Yes' if safety.get('adult_supervision') else 'No'}") lines.append(f"- **Forbidden:** {', '.join(safety.get('forbidden_behaviors', []))}") lines.append("") lines.append("## 📊 Scoring") for s in game.get("score_rules", []): lines.append(f" • {s}") lines.append(f"- **Tie-breaker:** {game.get('tie_breaker', '?')}") lines.append("") lines.append("## 📖 Story Seed") seed = game.get("story_seed", {}) lines.append(f"- **Tone:** {seed.get('tone', '?')}") lines.append(f"- **Motifs:** {', '.join(seed.get('motifs', []))}") lines.append(f"- **Recap style:** {seed.get('recap_style', '?')}") lines.append("") # Pipeline trace lines.append("---") lines.append("### 🔍 Pipeline trace") lines.append(f"- Retrieval: {state.get('num_retrieved', 0)} examples found") if state.get("retrieved_ids"): lines.append(f"- Retrieved IDs: {', '.join(state['retrieved_ids'])}") lines.append(f"- Game ID: {state.get('game_id', '?')}") if state.get("validation_passed"): lines.append(f"- ✅ Validation passed") else: lines.append(f"- ❌ Validation failed ({len(state.get('validation_failures', []))} issues)") if state.get("repair_applied"): lines.append(f"- 🔧 Repair applied → {'✅ Passed' if state.get('repair_valid') else '❌ Still has issues'}") for f in state.get("validation_failures", [])[:5]: lines.append(f" - {f}") leftover = state.get("remaining_failures", []) if leftover: for f in leftover[:5]: lines.append(f" - ⚠ {f}") return "\n".join(lines) # ── Multiplayer layer (additive — no existing backend functions modified) ──── import json as _json import os as _os import secrets as _secrets ADVENTURE_CODES: dict[str, str] = {} # code -> session_id PERSIST_PATH = "app/data/sessions_store.json" _CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789" # no 0/O/1/I ambiguity def make_adventure_code() -> str: while True: code = "".join(_secrets.choice(_CODE_ALPHABET) for _ in range(6)) if code not in ADVENTURE_CODES: return code def save_state(): """Persist sessions + codes so a process restart doesn't kill live games. Events live in the existing trace logs; we only persist session metadata.""" try: payload = {"codes": ADVENTURE_CODES, "sessions": {}} for sid, s in SESSION_STORE.items(): payload["sessions"][sid] = { "config": s.get("config"), "game": s.get("game"), "adventure_code": s.get("adventure_code"), "teams": s.get("teams", []), "players": s.get("players", []), "photos": s.get("photos", []), } _os.makedirs(_os.path.dirname(PERSIST_PATH), exist_ok=True) with open(PERSIST_PATH, "w") as f: _json.dump(payload, f) except Exception as e: print(f"[persist] save failed: {e}") def load_state(): """Restore sessions + codes on startup (events reload lazily via load_events).""" try: if not _os.path.exists(PERSIST_PATH): return with open(PERSIST_PATH) as f: payload = _json.load(f) ADVENTURE_CODES.update(payload.get("codes", {})) for sid, s in payload.get("sessions", {}).items(): SESSION_STORE.setdefault(sid, { "config": s.get("config"), "game": s.get("game"), "adventure_code": s.get("adventure_code"), "teams": s.get("teams", []), "players": s.get("players", []), "photos": s.get("photos", []), "events": [], "journals": [], }) print(f"✓ Restored {len(payload.get('sessions', {}))} session(s) from disk") except Exception as e: print(f"[persist] load failed: {e}") load_state() def join_adventure(code: str, player_name: str, team_id: str): """Join an existing adventure by code. Returns (session_id_or_None, message, game, teams).""" code = (code or "").strip().upper() player_name = (player_name or "").strip() if not code: return None, "⚠ Enter an adventure code.", None, [] if not player_name: return None, "⚠ Enter your name.", None, [] session_id = ADVENTURE_CODES.get(code) if session_id is None or session_id not in SESSION_STORE: return None, f"⚠ No adventure found for code **{code}**. Check with your game host.", None, [] session = SESSION_STORE[session_id] teams = session.get("teams", ["team-a", "team-b"]) if team_id not in teams: team_id = teams[0] session.setdefault("players", []).append({"name": player_name, "team_id": team_id}) save_state() title = session.get("game", {}).get("title", "Adventure") return session_id, f"✅ Joined **{title}** as **{player_name}** ({team_id})", session.get("game"), teams def get_live_feed(session_id: str) -> str: """Render recent activity for all players in the adventure (polled by gr.Timer).""" if not session_id or session_id not in SESSION_STORE: return "" try: events = load_events(session_id=session_id) except Exception: events = SESSION_STORE[session_id].get("events", []) session = SESSION_STORE[session_id] lines = [] players = session.get("players", []) if players: roster = ", ".join(f"{p['name']} ({p['team_id']})" for p in players[-8:]) lines.append(f"**👥 Players:** {roster}") lines.append("") if events: lines.append("**📡 Recent activity:**") for ev in events[-10:]: etype = ev.get("type") or ev.get("event_type") or "event" team = ev.get("team_id", "") payload = ev.get("payload") or ev.get("data") or {} summary = ( payload.get("summary") or ev.get("summary") or etype.replace("_", " ") ) icon = { "task_completed": "✅", "task_skipped": "⏭️", "hint_used": "💡", "journal_recorded": "🎙️", "photo_uploaded": "📸", "player_joined": "👋", "game_finished": "🏁", "task_revealed": "🎯", }.get(etype, "•") team_tag = f" `[{team}]`" if team else "" lines.append(f"{icon} {summary}{team_tag}") else: lines.append("*No activity yet — waiting for players…*") return "\n".join(lines) # ── CSS ─────────────────────────────────────────────────────────────────────── CSS = """ @import url('https://fonts.googleapis.com/css2?family=Syne:wght@400;600;700;800&family=DM+Sans:ital,wght@0,300;0,400;0,500;1,300&display=swap'); :root { --bg: #0a0c10; --surface: #12151c; --surface2: #1a1e28; --border: #252a38; --accent: #f0c040; --accent2: #3b82f6; --accent3: #10b981; --danger: #ef4444; --text: #e8eaf0; --muted: #6b7280; --radius: 12px; } body, .gradio-container { background: var(--bg) !important; font-family: 'DM Sans', sans-serif !important; color: var(--text) !important; } footer { display: none !important; } /* ── Config card ── */ .config-card { background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 24px; } /* ── Section labels ── */ .section-label { font-family: 'Syne', sans-serif !important; font-size: 11px !important; font-weight: 700 !important; letter-spacing: 0.12em !important; text-transform: uppercase !important; color: var(--muted) !important; margin-bottom: 16px !important; display: flex; align-items: center; gap: 10px; } .section-label::after { content: ''; flex: 1; height: 1px; background: var(--border); } /* ── Form overrides ── */ label span, .label-wrap span { font-family: 'Syne', sans-serif !important; font-size: 11px !important; font-weight: 700 !important; letter-spacing: 0.1em !important; text-transform: uppercase !important; color: var(--muted) !important; } input[type="text"], textarea, .gr-input { background: var(--surface2) !important; border: 1px solid var(--border) !important; border-radius: 8px !important; color: var(--text) !important; font-family: 'DM Sans', sans-serif !important; } input[type="text"]:focus, textarea:focus { border-color: var(--accent) !important; outline: none !important; box-shadow: 0 0 0 3px rgba(240,192,64,0.1) !important; } /* ── Buttons ── */ .gr-button { font-family: 'Syne', sans-serif !important; font-weight: 700 !important; letter-spacing: 0.04em !important; border-radius: 8px !important; transition: all 0.15s ease !important; } .gr-button.primary { background: var(--accent) !important; color: #0a0c10 !important; border: none !important; } .gr-button.primary:hover { background: #f5ce5a !important; transform: translateY(-1px) !important; box-shadow: 0 6px 24px rgba(240,192,64,0.25) !important; } .gr-button.secondary { background: transparent !important; border: 1.5px solid var(--border) !important; color: var(--text) !important; } .gr-button.secondary:hover { border-color: var(--accent) !important; color: var(--accent) !important; } /* ── Named button overrides ── */ #complete-btn { background: var(--accent3) !important; color: #fff !important; border: none !important; } #skip-btn { background: transparent !important; border: 1.5px solid var(--border) !important; color: var(--text) !important; } #hint-btn { background: rgba(59,130,246,0.12) !important; border: 1.5px solid var(--accent2) !important; color: var(--accent2) !important; } #end-btn { background: var(--danger) !important; color: #fff !important; border: none !important; width: 100% !important; height: 52px !important; font-size: 1rem !important; } #generate-btn { margin-top: 28px !important; width: 100% !important; height: 52px !important; font-size: 1rem !important; } #proceed-btn { width: 100% !important; margin-top: 12px !important; } #recap-btn { width: 100% !important; margin-bottom: 16px !important; } #poster-btn { width: 100% !important; } #new-game-btn { height: 44px !important; } /* ── Markdown output ── */ .gr-markdown { color: var(--text) !important; line-height: 1.7 !important; } .gr-markdown h1 { font-family: 'Syne', sans-serif !important; font-size: 1.6rem !important; font-weight: 800 !important; color: var(--accent) !important; border-bottom: 1px solid var(--border) !important; padding-bottom: 8px !important; margin-bottom: 16px !important; } .gr-markdown h2 { font-family: 'Syne', sans-serif !important; font-size: 1rem !important; font-weight: 700 !important; letter-spacing: 0.08em !important; text-transform: uppercase !important; color: var(--muted) !important; margin-top: 24px !important; } .gr-markdown h3 { font-family: 'Syne', sans-serif !important; color: var(--text) !important; } .gr-markdown code { background: var(--surface2) !important; color: var(--accent) !important; border-radius: 4px !important; padding: 2px 6px !important; font-size: 0.85em !important; } .gr-markdown pre { background: var(--surface2) !important; border: 1px solid var(--border) !important; border-radius: 8px !important; padding: 16px !important; } /* ── Scrollable panels ── */ .output-panel { background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 24px; max-height: 580px; overflow-y: auto; } .output-panel::-webkit-scrollbar { width: 4px; } .output-panel::-webkit-scrollbar-thumb { background: var(--border); border-radius: 99px; } .game-summary-panel { background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 20px; max-height: 520px; overflow-y: auto; } """ JS_INIT = """ function() { document.addEventListener('DOMContentLoaded', () => {}); } """ # ── Gradio UI ───────────────────────────────────────────────────────────────── with gr.Blocks(title="CityQuest-AI") as demo: # ── Per-browser-tab state (each player carries their own) ───────────── current_session = gr.State("") current_team = gr.State("team-a") current_player = gr.State("") is_host = gr.State(False) # ── Nav header ──────────────────────────────────────────────────────── gr.HTML("""
🌍 CityQuest·AI
""") # ════════════════════════════════════════════════════════════════════ # SCREEN 0 — Landing: Create or Join # ════════════════════════════════════════════════════════════════════ with gr.Group(visible=True) as landing_screen: gr.HTML("""
Your city is the
game board
AI-generated real-world adventures. Create one and invite your friends,
or join with a code.
""") with gr.Row(): with gr.Column(scale=1): gr.HTML("""
🗺️
New Adventure
Configure a game, get a share code,
and host your crew.
""") create_adventure_btn = gr.Button("🗺️ Create New Adventure", variant="primary", size="lg", elem_id="create-adv-btn") with gr.Column(scale=1): gr.HTML("""
🎟️
Join Adventure
Got a code from a friend?
Jump straight into their game.
""") join_adventure_btn = gr.Button("🎟️ Join with a Code", variant="secondary", size="lg", elem_id="join-adv-btn") # ════════════════════════════════════════════════════════════════════ # SCREEN 1a — Create (host configures the game) # ════════════════════════════════════════════════════════════════════ with gr.Group(visible=False) as create_screen: gr.HTML("""
Design your
real-world game
Configure once. Share the code. Play together.
""") with gr.Row(equal_height=False): # Left: config with gr.Column(scale=1, elem_classes=["config-card"]): gr.HTML('
Host
') host_name = gr.Textbox(label="Your Name", placeholder="Alex") gr.HTML('
Game Type
') game_type = gr.Dropdown( label="", choices=["scavenger_hunt", "hide_and_seek", "tag"], value="scavenger_hunt", container=False, ) gr.HTML('
Location
') with gr.Row(): city = gr.Textbox(label="City", value="Paris") area = gr.Textbox(label="Area / District", value="Le Marais") location_type = gr.Radio( label="Location Type", choices=["park", "street", "landmark", "mixed"], value="mixed", ) gr.HTML('
Players & Teams
') with gr.Row(): num_players = gr.Slider(label="Players", minimum=2, maximum=10, value=4, step=1) num_teams = gr.Slider(label="Teams", minimum=2, maximum=4, value=2, step=1) duration_minutes = gr.Slider(label="Duration (min)", minimum=15, maximum=120, value=60, step=5) gr.HTML('
Preferences
') with gr.Row(): difficulty = gr.Dropdown(label="Difficulty", choices=["easy", "medium", "hard"], value="medium") age_group = gr.Dropdown(label="Age Group", choices=["kids", "teens", "adults", "mixed"], value="adults") energy_level = gr.Radio(label="Energy Level", choices=["low", "medium", "high"], value="medium") generate_btn = gr.Button("🚀 Generate Adventure", variant="primary", size="lg", elem_id="generate-btn") back_from_create_btn = gr.Button("← Back", variant="secondary") # Right: output (revealed after generation) with gr.Column(scale=2, visible=False) as gen_preview_col: # Adventure code banner adventure_code_html = gr.HTML(value="") gr.HTML('
Generated Game
') with gr.Column(elem_classes=["output-panel"]): output_md = gr.Markdown(value="") session_id_box = gr.Textbox(label="Session ID", interactive=False) proceed_btn = gr.Button("▶ Start Playing →", variant="primary", size="lg", elem_id="proceed-btn") # ════════════════════════════════════════════════════════════════════ # SCREEN 1b — Join (player enters a code) # ════════════════════════════════════════════════════════════════════ with gr.Group(visible=False) as join_screen: gr.HTML("""
Join an adventure
Enter the 6-character code your host shared with you.
""") with gr.Column(elem_classes=["config-card"]): join_code = gr.Textbox(label="Adventure Code", placeholder="XK4P7M", max_lines=1) join_name = gr.Textbox(label="Your Name", placeholder="Sam") join_team = gr.Dropdown(label="Team", choices=["team-a", "team-b"], value="team-a") join_submit_btn = gr.Button("🎟️ Join Adventure", variant="primary", size="lg", elem_id="join-submit-btn") join_feedback = gr.Markdown(value="") back_from_join_btn = gr.Button("← Back", variant="secondary") # ════════════════════════════════════════════════════════════════════ # SCREEN 2 — Play # ════════════════════════════════════════════════════════════════════ with gr.Group(visible=False) as play_screen: gr.HTML("""
Game in progress ● Live
Log task completions, hints, and journal entries as you play.
""") with gr.Row(equal_height=False): # Left: game reference + live feed with gr.Column(scale=1): play_code_html = gr.HTML(value="") gr.HTML('
Live Feed
') with gr.Column(elem_classes=["game-summary-panel"]): live_feed_md = gr.Markdown(value="*Waiting for activity…*") gr.HTML('
Game Reference
') with gr.Column(elem_classes=["game-summary-panel"]): play_game_md = gr.Markdown(value="") # Right: actions with gr.Column(scale=1): play_identity_md = gr.Markdown(value="") play_team_dd = gr.Dropdown(label="Your Team", choices=["team-a", "team-b"], value="team-a") gr.HTML('
Task Actions
') play_task_id = gr.Textbox(label="Task ID", placeholder="t1, t2 …") with gr.Row(): complete_btn = gr.Button("✅ Complete", variant="primary", elem_id="complete-btn") skip_btn = gr.Button("⏭️ Skip", elem_id="skip-btn") hint_btn = gr.Button("💡 Hint", elem_id="hint-btn") task_feedback = gr.Markdown(value="") gr.HTML('
Voice Journal
') journal_audio = gr.Audio( label="🎙️ Record voice (auto-transcribed via Cohere ASR)", sources=["microphone", "upload"], type="filepath", ) journal_transcript = gr.Textbox( label="Transcript (edit the ASR result or type your own)", lines=3, placeholder="We just found the mural near the canal — it was incredible!", ) with gr.Row(): journal_task_id = gr.Textbox(label="Task ID (opt.)", placeholder="t1") journal_location = gr.Textbox(label="Location note", placeholder="Rue de Rivoli") journal_language = gr.Dropdown( label="Language", value="en", choices=["en", "fr", "de", "it", "es", "pt", "el", "nl", "pl", "zh", "ja", "ko", "vi", "ar"], ) journal_btn = gr.Button("🎙️ Record Journal", variant="secondary") journal_output = gr.Markdown(value="") gr.HTML('
Photo Upload
') photo_file = gr.Image(label="Upload photo", type="filepath", height=160) photo_caption = gr.Textbox(label="Caption", placeholder="The mural we discovered!") photo_task_id = gr.Textbox(label="Task ID (opt.)", placeholder="t1") photo_btn = gr.Button("📸 Upload Photo", variant="secondary") photo_output = gr.Markdown(value="") gr.HTML('
') end_btn = gr.Button("🏁 End Game & See Scores →", variant="primary", size="lg", elem_id="end-btn") # Poll shared session state every 3s so all players see each other's progress sync_timer = gr.Timer(value=3) # ════════════════════════════════════════════════════════════════════ # SCREEN 3 — Recap # ════════════════════════════════════════════════════════════════════ with gr.Group(visible=False) as recap_screen: gr.HTML("""
Game over — your story awaits
Read the final scoreboard, generate your AI recap, and create a poster.
""") with gr.Row(equal_height=False): # Left: scoreboard with gr.Column(scale=1): gr.HTML('
Final Scoreboard
') with gr.Column(elem_classes=["output-panel"]): scoreboard_md = gr.Markdown(value="Scores will appear here after ending the game.") # Right: recap + poster with gr.Column(scale=1): gr.HTML('
Episode Recap
') recap_btn = gr.Button("📖 Generate AI Recap", variant="primary", size="lg", elem_id="recap-btn") with gr.Column(elem_classes=["output-panel"]): recap_md = gr.Markdown(value="Click **Generate AI Recap** to narrate your game.") recap_json = gr.JSON(label="Recap data", visible=False) gr.HTML('
Poster
') poster_btn = gr.Button("🎨 Generate Poster", variant="secondary", elem_id="poster-btn") poster_image = gr.Image(label="Poster", height=400, visible=True) poster_status = gr.Markdown(value="") with gr.Row(): new_game_btn = gr.Button("← Start a New Adventure", variant="secondary", elem_id="new-game-btn") # ── Footer ──────────────────────────────────────────────────────────── gr.HTML("""
CityQuest·AI · Built for the NVIDIA / HF Hackathon · Nemotron 3 Nano 4B · llama.cpp
""") # ════════════════════════════════════════════════════════════════════ # Helpers (UI layer) # ════════════════════════════════════════════════════════════════════ def _teams_for(n: int) -> list[str]: return [f"team-{c}" for c in "abcd"[:int(n)]] def _code_banner(code: str) -> str: return f"""
Adventure Code — share with your friends
{code}
""" # ════════════════════════════════════════════════════════════════════ # Wire: Landing # ════════════════════════════════════════════════════════════════════ create_adventure_btn.click( fn=lambda: (gr.update(visible=False), gr.update(visible=True), gr.update(visible=False)), inputs=[], outputs=[landing_screen, create_screen, join_screen], ) join_adventure_btn.click( fn=lambda: (gr.update(visible=False), gr.update(visible=False), gr.update(visible=True)), inputs=[], outputs=[landing_screen, create_screen, join_screen], ) back_from_create_btn.click( fn=lambda: (gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)), inputs=[], outputs=[landing_screen, create_screen, join_screen], ) back_from_join_btn.click( fn=lambda: (gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)), inputs=[], outputs=[landing_screen, create_screen, join_screen], ) # ════════════════════════════════════════════════════════════════════ # Wire: Create / Generate (host) # ════════════════════════════════════════════════════════════════════ def on_generate(host_name_v, game_type, city, area, location_type, duration_minutes, num_players, num_teams_v, difficulty, age_group, energy_level): summary, session_id = run_pipeline( game_type, city, area, location_type, duration_minutes, num_players, difficulty, age_group, energy_level, ) # Multiplayer registration (UI layer — backend untouched) teams = _teams_for(num_teams_v) code = make_adventure_code() ADVENTURE_CODES[code] = session_id session = SESSION_STORE[session_id] session["adventure_code"] = code session["teams"] = teams host = (host_name_v or "Host").strip() or "Host" session["players"] = [{"name": host, "team_id": teams[0]}] save_state() return ( summary, # output_md session_id, # session_id_box session_id, # current_session teams[0], # current_team host, # current_player True, # is_host gr.update(visible=True), # gen_preview_col _code_banner(code), # adventure_code_html summary, # play_game_md _code_banner(code), # play_code_html f"Playing as **{host}** · Host", # play_identity_md gr.update(choices=teams, value=teams[0]), # play_team_dd ) generate_btn.click( fn=on_generate, inputs=[host_name, game_type, city, area, location_type, duration_minutes, num_players, num_teams, difficulty, age_group, energy_level], outputs=[output_md, session_id_box, current_session, current_team, current_player, is_host, gen_preview_col, adventure_code_html, play_game_md, play_code_html, play_identity_md, play_team_dd], ) proceed_btn.click( fn=lambda: (gr.update(visible=False), gr.update(visible=True)), inputs=[], outputs=[create_screen, play_screen], ) # ════════════════════════════════════════════════════════════════════ # Wire: Join (player) # ════════════════════════════════════════════════════════════════════ def on_join(code, name, team): session_id, msg, game, teams = join_adventure(code, name, team) if session_id is None: # Stay on join screen, show error return ( msg, "", "team-a", "", gr.update(visible=True), # join_screen stays gr.update(visible=False), # play_screen "", "", "", gr.update(), ) session = SESSION_STORE[session_id] actual_team = team if team in teams else teams[0] summary = build_summary(game, {}, session_id) if game else "*Game data unavailable.*" code_html = _code_banner(session.get("adventure_code", "")) return ( msg, # join_feedback session_id, # current_session actual_team, # current_team name, # current_player gr.update(visible=False), # join_screen gr.update(visible=True), # play_screen summary, # play_game_md code_html, # play_code_html f"Playing as **{name}** · {actual_team}", # play_identity_md gr.update(choices=teams, value=actual_team), # play_team_dd ) join_submit_btn.click( fn=on_join, inputs=[join_code, join_name, join_team], outputs=[join_feedback, current_session, current_team, current_player, join_screen, play_screen, play_game_md, play_code_html, play_identity_md, play_team_dd], ) # Keep current_team in sync if the player switches teams mid-game play_team_dd.change(fn=lambda t: t, inputs=[play_team_dd], outputs=[current_team]) # ════════════════════════════════════════════════════════════════════ # Wire: Play actions (session/team from State — read-only boxes don't pass) # ════════════════════════════════════════════════════════════════════ complete_btn.click(fn=complete_task, inputs=[current_session, play_task_id, current_team], outputs=[task_feedback]) skip_btn.click( fn=skip_task, inputs=[current_session, play_task_id, current_team], outputs=[task_feedback]) hint_btn.click( fn=use_hint, inputs=[current_session, play_task_id, current_team], outputs=[task_feedback]) journal_btn.click( fn=record_journal, inputs=[current_session, journal_transcript, journal_task_id, journal_location, current_team, journal_audio, journal_language], outputs=[journal_output], ) photo_btn.click( fn=upload_photo, inputs=[current_session, photo_file, photo_caption, photo_task_id, current_team], outputs=[photo_output], ) # Live feed polling — every player sees the shared adventure state sync_timer.tick( fn=get_live_feed, inputs=[current_session], outputs=[live_feed_md], ) # End game → scoreboard → recap screen def on_end_game(session_id, team_id): scoreboard_text = end_game(session_id, team_id) return ( scoreboard_text, gr.update(visible=False), gr.update(visible=True), ) end_btn.click( fn=on_end_game, inputs=[current_session, current_team], outputs=[scoreboard_md, play_screen, recap_screen], ) # ════════════════════════════════════════════════════════════════════ # Wire: Recap # ════════════════════════════════════════════════════════════════════ recap_btn.click( fn=generate_recap, inputs=[current_session], outputs=[recap_md, recap_json], ) poster_btn.click( fn=generate_poster_from_session, inputs=[current_session], outputs=[poster_image, poster_status], ) # New adventure → back to landing new_game_btn.click( fn=lambda: (gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)), inputs=[], outputs=[landing_screen, create_screen, join_screen, play_screen, recap_screen], ) # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": demo.launch(theme=gr.themes.Base(), css=CSS, js=JS_INIT)