app.py

"""CoastWise Gradio application."""

from __future__ import annotations

import html
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone

import gradio as gr

from coastwise.ai import (
    build_hosted_interpreter,
    build_local_interpreter,
    build_lookup_workflow_trace,
    explain_safety_result,
    explanation_to_dict,
    workflow_trace_to_dict,
)
from coastwise.data import get_rule_card
from coastwise.photo import candidate_from_photo, photo_result_to_dict
from coastwise.qna import answer_question, answer_to_dict
from coastwise.rules import compose_safety_result, default_context, result_to_dict
from coastwise.schemas import CoastalRegion, UserIntent
from coastwise.search import search_species
from coastwise.source_cache import DEFAULT_SEED_CACHE_DIR, SourceCache, load_seed_cache
from coastwise.source_refresh import fetch_official_source, refresh_official_sources
from coastwise.source_registry import load_official_sources
from coastwise.submission import default_model_inventory
from coastwise.webui import (
    APP_CSS,
    EMPTY_STATE_HTML,
    FOOTER_NOTE,
    build_header_html,
    build_theme,
    human_retrieved,
    render_source_backed_answer_payload,
    render_source_bar_html,
)


APP_TITLE = "CoastWise"

SOURCE_BACKED_CONTROL_LABELS = (
    "Question",
    "Ask",
    "Refresh official sources",
    "Source status",
)

INTRO = """
California CDFW/CDPH Q&A.
"""

EXAMPLE_QUESTIONS = (
    "what's the min size of lingcod?",
    "can I eat mussels from Pacifica today?",
    "Dungeness crab rules",
)

# Ocean & Trust theme + brand CSS now live in coastwise.webui so the server app
# and the in-browser offline (Gradio-Lite) build render identically.
THEME = build_theme()

LIMITS = """
Conservative decision-support only.

Does not certify species ID or legality.

Does not authorize harvest or food safety.

Verify CDFW/CDPH before harvest.
"""

SOURCE_STATUS_DEFAULT = (
    "**Source status** — seed cache loaded and offline-ready. "
    "Refresh re-checks the official CDFW/CDPH sources."
)

HEADER_HTML = build_header_html()

INTENT_LABELS = {
    "Just curious": UserIntent.JUST_CURIOUS,
    "Catch and release": UserIntent.CATCH_AND_RELEASE,
    "Thinking of harvest": UserIntent.THINKING_OF_HARVEST,
}

REGION_LABELS = {
    "Point Arena to Pigeon Point": CoastalRegion.POINT_ARENA_TO_PIGEON_POINT,
    "San Francisco Bay": CoastalRegion.SAN_FRANCISCO_BAY,
    "Unknown": CoastalRegion.UNKNOWN,
}


SOURCE_BACKED_INTERPRETER = build_local_interpreter() or build_hosted_interpreter()


# Holds the most recent refreshed cache for the running session, so the
# "Refresh official sources" button actually changes what subsequent answers
# read. None means "use the bundled reviewed seed".
_SESSION_CACHE: SourceCache | None = None


def _active_cache():
    return _SESSION_CACHE if _SESSION_CACHE is not None else load_seed_cache(DEFAULT_SEED_CACHE_DIR)


def ask_source_backed_question(question: str, beach_or_coastal_area: str = "") -> dict[str, object]:
    answer = answer_question(
        question,
        beach_or_coastal_area,
        _active_cache(),
        datetime.now(timezone.utc),
        model=SOURCE_BACKED_INTERPRETER,
    )
    return answer_to_dict(answer)


def render_source_backed_answer_from_cache(
    question: str,
    beach_or_coastal_area: str,
    cache,
    now: datetime,
) -> str:
    answer = answer_question(question, beach_or_coastal_area, cache, now, model=SOURCE_BACKED_INTERPRETER)
    return render_source_backed_answer_payload(answer_to_dict(answer))


def render_source_backed_answer(question: str, beach_or_coastal_area: str = "") -> str:
    result = ask_source_backed_question(question, beach_or_coastal_area)
    return render_source_backed_answer_payload(result)


def _answer_from_example(example: str) -> tuple[str, str]:
    """Populate the question box with an example and render its answer in one click."""
    return example, render_source_backed_answer(example, "")


def _bounded_live_fetcher(sources, per_source_timeout: float = 2.5, overall_deadline: float = 3.0):
    """Fetch sources concurrently within a tight overall deadline so the live Refresh
    never hangs the UI on slow or unreachable networks (offline-first). Returns a
    fetcher that serves pre-fetched text; misses raise so refresh_official_sources
    keeps the last-good cache for them."""
    fetched: dict[str, str] = {}
    executor = ThreadPoolExecutor(max_workers=min(8, max(1, len(sources))))
    futures = {
        executor.submit(fetch_official_source, source, timeout=per_source_timeout): source
        for source in sources
    }
    try:
        for future in as_completed(futures, timeout=overall_deadline):
            try:
                fetched[futures[future].id] = future.result()
            except Exception:
                pass
    except Exception:
        pass  # overall deadline reached — proceed with whatever completed
    executor.shutdown(wait=False, cancel_futures=True)

    def fetcher(source):
        if source.id in fetched:
            return fetched[source.id]
        raise RuntimeError("source unavailable within refresh deadline")

    return fetcher


def refresh_official_sources_status(fetcher=None) -> dict[str, object]:
    global _SESSION_CACHE
    sources = load_official_sources()
    if fetcher is None:
        fetcher = _bounded_live_fetcher(sources)
    base = _active_cache()
    result, refreshed = refresh_official_sources(
        sources,
        base,
        datetime.now(timezone.utc),
        fetcher,
    )
    # Apply the re-fetch to the live session as updated snapshot freshness, but
    # keep the reviewed chunks/facts: structured regulation values stay curated
    # and are never auto-derived from freshly fetched HTML.
    _SESSION_CACHE = SourceCache(
        sources=refreshed.sources,
        snapshots=refreshed.snapshots,
        chunks=base.chunks,
        facts=base.facts,
        advisory_facts=base.advisory_facts,
    )
    return {
        "message": result.user_message,
        "updated": result.updated_source_ids,
        "unchanged": result.unchanged_source_ids,
        "failed": result.failed_source_ids,
        "cache_preserved": result.cache_preserved_source_ids,
    }


def render_source_bar() -> str:
    """Top-of-app source freshness for the running session cache."""
    return render_source_bar_html(_active_cache(), datetime.now(timezone.utc))


def _refresh_pending() -> str:
    return (
        '<span class="cw-srcbar"><span class="cw-dot cw-dot--ok cw-dot--pulse"></span>'
        '<span class="cw-srcbar__txt">Checking official sources…</span></span>'
    )


def render_refresh_status() -> str:
    status = refresh_official_sources_status()
    cache = _active_cache()
    retrieved = max((snapshot.retrieved_at for snapshot in cache.snapshots), default=None)
    when = human_retrieved(retrieved) if retrieved else "just now"
    reachable = len(status["updated"]) + len(status["unchanged"])
    if reachable == 0 and status["failed"]:
        # Offline / unreachable: the last-good cache still answers (offline-first).
        variant = "warn"
        text = f"Couldn't reach official sources just now — using last-good cache from {when}."
    else:
        variant = "ok"
        text = (
            f"Checked just now — {len(status['updated'])} updated, "
            f"{len(status['unchanged'])} unchanged. Last good: {when}."
        )
    return (
        f'<span class="cw-srcbar"><span class="cw-dot cw-dot--{variant}"></span>'
        f'<span class="cw-srcbar__txt">{html.escape(text)}</span></span>'
    )


def lookup_species(
    query: str,
    intent_label: str,
    region_label: str,
    county_or_beach: str,
) -> dict[str, object]:
    matches = search_species(query)
    card = get_rule_card(matches[0].rule_card_key) if matches else get_rule_card("unknown")
    context = default_context(
        intent=INTENT_LABELS.get(intent_label, UserIntent.JUST_CURIOUS),
        region=REGION_LABELS.get(region_label, CoastalRegion.UNKNOWN),
        county_or_beach=county_or_beach.strip() or None,
    )
    result = compose_safety_result(card, context)
    explanation = explain_safety_result(result, default_model_inventory(), generator=None)
    trace = build_lookup_workflow_trace(
        "name_search",
        candidate_steps=(
            f"Normalized query: {query.strip()}",
            f"Candidate matches: {', '.join(match.display_name for match in matches) or 'none'}",
        ),
        rule_steps=(
            f"Selected rule card: {card.key}",
            f"Deterministic status: {result.status.value}",
            "Generated explanation was created after deterministic safety composition.",
        ),
        result=result,
        explanation=explanation,
    )
    payload = result_to_dict(result)
    payload["matches"] = tuple(match.display_name for match in matches)
    payload["generated_explanation"] = explanation_to_dict(explanation)
    payload["workflow_trace"] = workflow_trace_to_dict(trace)
    return payload


def render_lookup_result(
    query: str,
    intent_label: str,
    region_label: str,
    county_or_beach: str,
) -> str:
    result = lookup_species(query, intent_label, region_label, county_or_beach)
    source_lines = "\n".join(f"- {url}" for url in result["source_links"]) or "- No source-backed card shown."
    cache_lines = "\n".join(f"- Cache date: {cache_date}" for cache_date in result["cache_dates"])
    notes = "\n".join(f"- {note}" for note in result["rule_notes"]) or "- Search by known name or verify with official sources."
    warnings = "\n".join(f"- {warning}" for warning in result["advisory_warnings"]) or "- No cached advisory overlay shown."
    explanation = result["generated_explanation"]["text"]
    return f"""
## {result["display_name"]}

**Status:** `{result["status"]}`

{result["summary"]}

**Advisory and location warnings**
{warnings}

**Rule notes**
{notes}

**Sources**
{source_lines}

**Source freshness**
{cache_lines}

**Next safe action:** {result["next_safe_action"]}

**Safety limits**
CoastWise does not certify species identification, legality, harvest authorization, or food safety. Verify current official guidance before harvest or consumption.

**Generated explanation**
{explanation}
"""


def lookup_photo_candidate(image) -> dict[str, object]:
    if image is None:
        return {
            "message": "Observe only: upload a demo photo to compare with curated references.",
            "fallback_status": "observe_only",
            "reference_set_keys": (),
            "image_retained": False,
            "candidates": (),
        }
    return photo_result_to_dict(candidate_from_photo(image))


def render_photo_candidate(image) -> str:
    result = lookup_photo_candidate(image)
    if not result["candidates"]:
        return f"**Photo candidate:** {result['message']}"

    rows = []
    for candidate in result["candidates"]:
        rows.append(
            f"- {candidate['display_name']} ({candidate['confidence']}): "
            f"{candidate['limitations']}"
        )
    return (
        "**Photo candidates are visually similar possibilities, not confirmed ID.**\n\n"
        + "\n".join(rows)
    )


def build_app() -> gr.Blocks:
    # Gradio 6 applies theme/css at launch() (see __main__), not on Blocks.
    with gr.Blocks(title=APP_TITLE) as demo:
        gr.Markdown(HEADER_HTML, sanitize_html=False, elem_classes=["cw-header-wrap"])

        with gr.Row(elem_classes=["cw-source-row"]):
            source_status = gr.Markdown(
                render_source_bar(),
                sanitize_html=False,
                elem_classes=["cw-source-status"],
                scale=5,
            )
            refresh = gr.Button(
                "Refresh",
                variant="secondary",
                elem_classes=["cw-refresh-top"],
                scale=1,
                min_width=132,
            )

        with gr.Group(elem_classes=["cw-search"]):
            query = gr.Textbox(
                label="Question",
                show_label=False,
                lines=1,
                max_lines=1,
                placeholder="Ask about a species or rule — e.g. what's the min size of lingcod?",
                elem_classes=["cw-question"],
            )
            lookup = gr.Button("Ask", variant="primary", elem_classes=["cw-ask"])

        with gr.Row(elem_classes=["cw-chips"]):
            chips = [
                gr.Button(example, variant="secondary", size="sm")
                for example in EXAMPLE_QUESTIONS
            ]

        output = gr.Markdown(EMPTY_STATE_HTML, sanitize_html=False, elem_classes=["cw-answer"])

        gr.Markdown(FOOTER_NOTE, elem_classes=["cw-footer"])

        lookup.click(render_source_backed_answer, inputs=[query], outputs=output)
        query.submit(render_source_backed_answer, inputs=[query], outputs=output)
        for chip, example in zip(chips, EXAMPLE_QUESTIONS, strict=False):
            chip.click(
                lambda example=example: _answer_from_example(example),
                outputs=[query, output],
            )
        refresh.click(
            _refresh_pending, outputs=source_status, show_progress="hidden"
        ).then(
            render_refresh_status, outputs=source_status, show_progress="hidden"
        )
    return demo


demo = build_app()


def _warm_up_local_model() -> None:
    """Preload the in-process MiniCPM model off the request path.

    The interpreter loads its GGUF lazily on first use; warming it here in a
    daemon thread keeps Space startup non-blocking while making the first real
    question fast. No-ops in deterministic mode (no model configured) and never
    raises.
    """
    if SOURCE_BACKED_INTERPRETER is None:
        return
    try:
        SOURCE_BACKED_INTERPRETER("what is the minimum size for lingcod?")
    except Exception:
        pass


threading.Thread(
    target=_warm_up_local_model, name="coastwise-model-warmup", daemon=True
).start()


if __name__ == "__main__":
    demo.launch(theme=THEME, css=APP_CSS)