app.py

"""
FastAPI entrypoint for the PDF-plots Hugging Face Space.

This file only wires HTTP endpoints to the logic living in `services/`.
Update the plotting/storage/color logic there without ever touching
this file (and vice versa).

Every endpoint follows the same pattern:
  1. Hash the request parameters -> deterministic plot_id.
  2. If use_cache and a matching plot already exists -> return it.
  3. Otherwise: load + validate the PDF set's error type, build the
     figure, upload it under its conventional filename, update the
     grid's index.json, and return the new entry.

Only PDF sets with ErrorType in {"hessian", "symmhessian", "replicas"}
are supported right now (see services/error_types.py) — anything else
returns HTTP 400 with an explanation, rather than silently computing
wrong uncertainties.
"""

import logging
import os
import threading
import time

import numpy as np
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse

from services.config import (
    COMPUTE_SEMAPHORE_LIMIT,
    FLUSH_INTERVAL_SECONDS,
    TMP_DIR,
    correlation_plot_filename,
    main_plot_filename,
    ratio_plot_filename,
    standard_plot_filename,
)
from services.error_types import UnsupportedErrorTypeError
from services.models import (
    CorrelationPlotRequest,
    MainPlotRequest,
    PlotResponse,
    RatioPlotRequest,
    StandardPlotRequest,
)
from services.pdf_utils import COMPUTE_SEMAPHORE, LHAPDF_LOCK, get_pdf_members, invalidate_cache, load_validated_pdf_set
from services.plotting import (
    build_correlation_plot,
    build_main_plot,
    build_standard_plot,
    build_standard_ratio_plot,
)
from services.storage import (
    find_cached_plot,
    find_pending_plot,
    flush_pending,
    make_plot_id,
    pending_status,
    queue_plot,
)

logger = logging.getLogger(__name__)

app = FastAPI(title="LHAPDF Plotly Plots Space")


def _background_flush_loop():
    """
    Runs for the lifetime of the container: every FLUSH_INTERVAL_SECONDS,
    batches every plot queued since the last flush into one HF commit per
    dataset_repo (see services/storage.py::flush_pending). Errors are
    swallowed and retried next cycle rather than crashing the Space —
    a missed flush just means plots stay queued (and locally servable)
    a bit longer.
    """
    while True:
        time.sleep(FLUSH_INTERVAL_SECONDS)
        token = os.environ.get("HF_TOKEN")
        if not token:
            logger.warning("[flush loop] HF_TOKEN not set, skipping this cycle")
            continue
        try:
            logger.info("[flush loop] running scheduled flush...")
            result = flush_pending(token)
            logger.info("[flush loop] scheduled flush result: %s", result)
        except Exception as e:
            logger.exception("[flush loop] flush_pending failed: %s", e)


@app.on_event("startup")
def _start_background_flush():
    threading.Thread(target=_background_flush_loop, daemon=True).start()

# This Space is called directly from the CTEQ-TEA website's client-side JS
# (a different origin), which the browser will block without CORS headers.
# The API is read-mostly (it only ever writes to one known dataset repo
# using the server-side token below), so allowing any origin is acceptable;
# tighten allow_origins to the site's domain if that changes.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/health")
def health():
    logger.info("GET /health (likely a wake-up ping from the website)")
    return {"status": "ok"}


@app.get("/plots/{folder}/{filename}")
def get_plot_file(folder: str, filename: str):
    """
    Serves a generated plot's JSON straight off local disk. This is the
    URL returned as `plotly_json_url` for any plot queued but not yet
    committed to the HF dataset (see services/storage.py::queue_plot), so
    the requesting page can render it immediately without waiting for the
    next batch flush. Files stay here even after a flush, so this keeps
    working afterwards too — it only goes stale if the container restarts
    (ephemeral disk), at which point a fresh request just regenerates it.
    """
    path = os.path.join(TMP_DIR, folder, filename)
    if not os.path.isfile(path):
        raise HTTPException(
            status_code=404,
            detail="Plot not found locally — it may not have been generated yet, or this Space container restarted since it was.",
        )
    return FileResponse(path, media_type="application/json")


@app.get("/pending-status")
def get_pending_status():
    """Number of plots queued and waiting for the next batched commit, per dataset_repo|grid_name."""
    return pending_status()


@app.post("/flush-cache")
def force_flush():
    """Manually trigger a batch commit now instead of waiting for the next scheduled flush."""
    token = _server_hf_token()
    return flush_pending(token)


def _server_hf_token() -> str:
    """
    The HF token with write access to dataset_repo, read from this Space's
    own HF_TOKEN secret. Never accepted from the caller (see
    services.models.BasePlotRequest) so a public webpage can call these
    endpoints directly without ever handling a write-capable credential.
    """
    token = os.environ.get("HF_TOKEN")
    if not token:
        raise HTTPException(
            status_code=500,
            detail="Server misconfigured: HF_TOKEN secret is not set on this Space.",
        )
    return token


def _get_pdf_set_or_400(dataset_repo: str, grid_name: str, hf_token: str):
    """Shared error-type gate for every endpoint below."""
    try:
        return load_validated_pdf_set(dataset_repo, grid_name, hf_token)
    except UnsupportedErrorTypeError as e:
        logger.warning("grid=%s dataset_repo=%s rejected: %s", grid_name, dataset_repo, e)
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("grid=%s dataset_repo=%s failed to load PDFSet: %s", grid_name, dataset_repo, e)
        raise HTTPException(status_code=500, detail=str(e))


def _get_pdf_members_or_400(dataset_repo: str, grid_name: str, hf_token: str):
    """
    Shared member-loading gate, mirroring _get_pdf_set_or_400. Returns
    `pdf_set.mkPDFs()`, cached per (dataset_repo, grid_name) so requests
    for the same PDF set (any Q/flavor/plot type) reuse the same loaded
    members instead of re-parsing every member's grid file from disk.
    """
    try:
        return get_pdf_members(dataset_repo, grid_name, hf_token)
    except UnsupportedErrorTypeError as e:
        logger.warning("grid=%s dataset_repo=%s rejected: %s", grid_name, dataset_repo, e)
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("grid=%s dataset_repo=%s failed to load PDF members: %s", grid_name, dataset_repo, e)
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/compute-standard-plot", response_model=PlotResponse)
def compute_standard_plot(request: StandardPlotRequest):
    hf_token = _server_hf_token()
    parameters = {
        "q_scale": request.q_scale,
        "parton_id": request.parton_id,
        "plot_color": request.plot_color,
        "custom_title": request.custom_title,
        "x_min": request.x_min,
        "x_max": request.x_max,
        "n_points": request.n_points,
    }
    plot_id = make_plot_id("standard_plot", request.grid_name, parameters)
    logger.info("POST /compute-standard-plot: grid=%s dataset_repo=%s plot_id=%s params=%s", request.grid_name, request.dataset_repo, plot_id, parameters)
    t0 = time.monotonic()

    if request.use_cache:
        pending = find_pending_plot(plot_id)
        if pending:
            logger.info("plot_id=%s served from pending cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already generated this session, awaiting next batch upload.",
                plot_id=plot_id,
                plot_url=pending["plot_url"],
                plotly_json_url=pending["plotly_json_url"],
            )
        cached = find_cached_plot(request.dataset_repo, request.grid_name, plot_id, hf_token)
        if cached:
            logger.info("plot_id=%s served from committed HF cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already existed, returning cached version.",
                plot_id=plot_id,
                plot_url=cached["plot_url"],
                plotly_json_url=cached["plotly_json_url"],
            )

    logger.info("plot_id=%s not cached, generating...", plot_id)
    logger.info("plot_id=%s waiting for compute slot (max %d concurrent)...", plot_id, COMPUTE_SEMAPHORE_LIMIT)
    with COMPUTE_SEMAPHORE:
        logger.info("plot_id=%s acquired compute slot", plot_id)
        pdf_set = _get_pdf_set_or_400(request.dataset_repo, request.grid_name, hf_token)
        pdfs = _get_pdf_members_or_400(request.dataset_repo, request.grid_name, hf_token)

        try:
            x_vals = np.logspace(np.log10(request.x_min), np.log10(request.x_max), request.n_points)

            # See services/pdf_utils.py::LHAPDF_LOCK — pdf.xfxQ()/pdf_set.uncertainty()
            # below are native lhapdf calls and are not safe to run concurrently
            # with any other request's lhapdf calls.
            logger.info("plot_id=%s waiting for LHAPDF_LOCK...", plot_id)
            with LHAPDF_LOCK:
                logger.info("plot_id=%s acquired LHAPDF_LOCK, building figure...", plot_id)
                fig = build_standard_plot(
                    pdf_set=pdf_set,
                    pdfs=pdfs,
                    grid_name=request.grid_name,
                    q_scale=request.q_scale,
                    parton_id=request.parton_id,
                    plot_color=request.plot_color,
                    x_vals=x_vals,
                    custom_title=request.custom_title,
                )
            logger.info("plot_id=%s figure built, released LHAPDF_LOCK", plot_id)

            json_filename = standard_plot_filename(request.grid_name, request.parton_id, request.q_scale)

            entry = queue_plot(
                dataset_repo=request.dataset_repo,
                grid_name=request.grid_name,
                plot_id=plot_id,
                plot_type="standard_plot",
                parameters=parameters,
                fig=fig,
                json_filename=json_filename,
            )
            logger.info("plot_id=%s done in %.2fs total", plot_id, time.monotonic() - t0)

            return PlotResponse(
                status="success",
                cached=False,
                message="Plot generated; queued for the next batched upload to " + request.dataset_repo + ".",
                plot_id=plot_id,
                plot_url=entry["plot_url"],
                plotly_json_url=entry["plotly_json_url"],
            )

        except HTTPException:
            raise
        except Exception as e:
            logger.exception("plot_id=%s compute-standard-plot failed: %s", plot_id, e)
            invalidate_cache(request.dataset_repo, request.grid_name)
            raise HTTPException(status_code=500, detail=str(e))


@app.post("/compute-ratio-plot", response_model=PlotResponse)
def compute_ratio_plot(request: RatioPlotRequest):
    hf_token = _server_hf_token()
    parameters = {
        "q_scale": request.q_scale,
        "parton_id": request.parton_id,
        "plot_color": request.plot_color,
        "custom_title": request.custom_title,
        "x_min": request.x_min,
        "x_max": request.x_max,
        "n_points": request.n_points,
    }
    plot_id = make_plot_id("ratio_plot", request.grid_name, parameters)
    logger.info("POST /compute-ratio-plot: grid=%s dataset_repo=%s plot_id=%s params=%s", request.grid_name, request.dataset_repo, plot_id, parameters)
    t0 = time.monotonic()

    if request.use_cache:
        pending = find_pending_plot(plot_id)
        if pending:
            logger.info("plot_id=%s served from pending cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already generated this session, awaiting next batch upload.",
                plot_id=plot_id,
                plot_url=pending["plot_url"],
                plotly_json_url=pending["plotly_json_url"],
            )
        cached = find_cached_plot(request.dataset_repo, request.grid_name, plot_id, hf_token)
        if cached:
            logger.info("plot_id=%s served from committed HF cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already existed, returning cached version.",
                plot_id=plot_id,
                plot_url=cached["plot_url"],
                plotly_json_url=cached["plotly_json_url"],
            )

    logger.info("plot_id=%s not cached, generating...", plot_id)
    logger.info("plot_id=%s waiting for compute slot (max %d concurrent)...", plot_id, COMPUTE_SEMAPHORE_LIMIT)
    with COMPUTE_SEMAPHORE:
        logger.info("plot_id=%s acquired compute slot", plot_id)
        pdf_set = _get_pdf_set_or_400(request.dataset_repo, request.grid_name, hf_token)
        pdfs = _get_pdf_members_or_400(request.dataset_repo, request.grid_name, hf_token)

        try:
            x_vals = np.logspace(np.log10(request.x_min), np.log10(request.x_max), request.n_points)

            # See services/pdf_utils.py::LHAPDF_LOCK.
            logger.info("plot_id=%s waiting for LHAPDF_LOCK...", plot_id)
            with LHAPDF_LOCK:
                logger.info("plot_id=%s acquired LHAPDF_LOCK, building figure...", plot_id)
                fig = build_standard_ratio_plot(
                    pdf_set=pdf_set,
                    pdfs=pdfs,
                    grid_name=request.grid_name,
                    q_scale=request.q_scale,
                    parton_id=request.parton_id,
                    plot_color=request.plot_color,
                    x_vals=x_vals,
                    custom_title=request.custom_title,
                )
            logger.info("plot_id=%s figure built, released LHAPDF_LOCK", plot_id)

            json_filename = ratio_plot_filename(request.grid_name, request.parton_id, request.q_scale)

            entry = queue_plot(
                dataset_repo=request.dataset_repo,
                grid_name=request.grid_name,
                plot_id=plot_id,
                plot_type="ratio_plot",
                parameters=parameters,
                fig=fig,
                json_filename=json_filename,
            )
            logger.info("plot_id=%s done in %.2fs total", plot_id, time.monotonic() - t0)

            return PlotResponse(
                status="success",
                cached=False,
                message="Plot generated; queued for the next batched upload to " + request.dataset_repo + ".",
                plot_id=plot_id,
                plot_url=entry["plot_url"],
                plotly_json_url=entry["plotly_json_url"],
            )

        except HTTPException:
            raise
        except Exception as e:
            logger.exception("plot_id=%s compute-ratio-plot failed: %s", plot_id, e)
            invalidate_cache(request.dataset_repo, request.grid_name)
            raise HTTPException(status_code=500, detail=str(e))


@app.post("/compute-correlation-plot", response_model=PlotResponse)
def compute_correlation_plot(request: CorrelationPlotRequest):
    hf_token = _server_hf_token()
    parameters = {
        "q_scale": request.q_scale,
        "parton_id_1": request.parton_id_1,
        "parton_id_2": request.parton_id_2,
        "color_correlated": request.color_correlated,
        "color_anti_correlated": request.color_anti_correlated,
        "x_min": request.x_min,
        "x_max": request.x_max,
        "n_points": request.n_points,
    }
    plot_id = make_plot_id("correlation_plot", request.grid_name, parameters)
    logger.info("POST /compute-correlation-plot: grid=%s dataset_repo=%s plot_id=%s params=%s", request.grid_name, request.dataset_repo, plot_id, parameters)
    t0 = time.monotonic()

    if request.use_cache:
        pending = find_pending_plot(plot_id)
        if pending:
            logger.info("plot_id=%s served from pending cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already generated this session, awaiting next batch upload.",
                plot_id=plot_id,
                plot_url=pending["plot_url"],
                plotly_json_url=pending["plotly_json_url"],
            )
        cached = find_cached_plot(request.dataset_repo, request.grid_name, plot_id, hf_token)
        if cached:
            logger.info("plot_id=%s served from committed HF cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already existed, returning cached version.",
                plot_id=plot_id,
                plot_url=cached["plot_url"],
                plotly_json_url=cached["plotly_json_url"],
            )

    logger.info("plot_id=%s not cached, generating...", plot_id)
    logger.info("plot_id=%s waiting for compute slot (max %d concurrent)...", plot_id, COMPUTE_SEMAPHORE_LIMIT)
    with COMPUTE_SEMAPHORE:
        logger.info("plot_id=%s acquired compute slot", plot_id)
        pdfs = _get_pdf_members_or_400(request.dataset_repo, request.grid_name, hf_token)

        try:
            x_vals = np.logspace(np.log10(request.x_min), np.log10(request.x_max), request.n_points)

            # See services/pdf_utils.py::LHAPDF_LOCK.
            logger.info("plot_id=%s waiting for LHAPDF_LOCK...", plot_id)
            with LHAPDF_LOCK:
                logger.info("plot_id=%s acquired LHAPDF_LOCK, building figure...", plot_id)
                fig = build_correlation_plot(
                    pdfs=pdfs,
                    q_scale=request.q_scale,
                    parton_id_1=request.parton_id_1,
                    parton_id_2=request.parton_id_2,
                    x_vals=x_vals,
                    color_correlated=request.color_correlated,
                    color_anti_correlated=request.color_anti_correlated,
                )
            logger.info("plot_id=%s figure built, released LHAPDF_LOCK", plot_id)

            json_filename = correlation_plot_filename(
                request.grid_name, request.parton_id_1, request.parton_id_2, request.q_scale
            )

            entry = queue_plot(
                dataset_repo=request.dataset_repo,
                grid_name=request.grid_name,
                plot_id=plot_id,
                plot_type="correlation_plot",
                parameters=parameters,
                fig=fig,
                json_filename=json_filename,
            )
            logger.info("plot_id=%s done in %.2fs total", plot_id, time.monotonic() - t0)

            return PlotResponse(
                status="success",
                cached=False,
                message="Plot generated; queued for the next batched upload to " + request.dataset_repo + ".",
                plot_id=plot_id,
                plot_url=entry["plot_url"],
                plotly_json_url=entry["plotly_json_url"],
            )

        except HTTPException:
            raise
        except Exception as e:
            logger.exception("plot_id=%s compute-correlation-plot failed: %s", plot_id, e)
            invalidate_cache(request.dataset_repo, request.grid_name)
            raise HTTPException(status_code=500, detail=str(e))


@app.post("/compute-main-plot", response_model=PlotResponse)
def compute_main_plot(request: MainPlotRequest):
    hf_token = _server_hf_token()
    parameters = {
        "q_scale": request.q_scale,
        "parton_ids": request.parton_ids,
        "custom_title": request.custom_title,
        "x_min": request.x_min,
        "x_max": request.x_max,
        "n_points": request.n_points,
    }
    plot_id = make_plot_id("main_plot", request.grid_name, parameters)
    logger.info("POST /compute-main-plot: grid=%s dataset_repo=%s plot_id=%s params=%s", request.grid_name, request.dataset_repo, plot_id, parameters)
    t0 = time.monotonic()

    if request.use_cache:
        pending = find_pending_plot(plot_id)
        if pending:
            logger.info("plot_id=%s served from pending cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already generated this session, awaiting next batch upload.",
                plot_id=plot_id,
                plot_url=pending["plot_url"],
                plotly_json_url=pending["plotly_json_url"],
            )
        cached = find_cached_plot(request.dataset_repo, request.grid_name, plot_id, hf_token)
        if cached:
            logger.info("plot_id=%s served from committed HF cache in %.3fs", plot_id, time.monotonic() - t0)
            return PlotResponse(
                status="success",
                cached=True,
                message="Plot already existed, returning cached version.",
                plot_id=plot_id,
                plot_url=cached["plot_url"],
                plotly_json_url=cached["plotly_json_url"],
            )

    logger.info("plot_id=%s not cached, generating...", plot_id)
    logger.info("plot_id=%s waiting for compute slot (max %d concurrent)...", plot_id, COMPUTE_SEMAPHORE_LIMIT)
    with COMPUTE_SEMAPHORE:
        logger.info("plot_id=%s acquired compute slot", plot_id)
        pdf_set = _get_pdf_set_or_400(request.dataset_repo, request.grid_name, hf_token)
        pdfs = _get_pdf_members_or_400(request.dataset_repo, request.grid_name, hf_token)

        try:
            x_vals = np.logspace(np.log10(request.x_min), np.log10(request.x_max), request.n_points)

            # See services/pdf_utils.py::LHAPDF_LOCK.
            logger.info("plot_id=%s waiting for LHAPDF_LOCK...", plot_id)
            with LHAPDF_LOCK:
                logger.info("plot_id=%s acquired LHAPDF_LOCK, building figure...", plot_id)
                fig = build_main_plot(
                    pdf_set=pdf_set,
                    pdfs=pdfs,
                    grid_name=request.grid_name,
                    q_scale=request.q_scale,
                    parton_ids=request.parton_ids,
                    x_vals=x_vals,
                    custom_title=request.custom_title,
                )
            logger.info("plot_id=%s figure built, released LHAPDF_LOCK", plot_id)

            json_filename = main_plot_filename(request.grid_name, request.parton_ids, request.q_scale)

            entry = queue_plot(
                dataset_repo=request.dataset_repo,
                grid_name=request.grid_name,
                plot_id=plot_id,
                plot_type="main_plot",
                parameters=parameters,
                fig=fig,
                json_filename=json_filename,
            )
            logger.info("plot_id=%s done in %.2fs total", plot_id, time.monotonic() - t0)

            return PlotResponse(
                status="success",
                cached=False,
                message="Plot generated; queued for the next batched upload to " + request.dataset_repo + ".",
                plot_id=plot_id,
                plot_url=entry["plot_url"],
                plotly_json_url=entry["plotly_json_url"],
            )

        except HTTPException:
            raise
        except Exception as e:
            logger.exception("plot_id=%s compute-main-plot failed: %s", plot_id, e)
            invalidate_cache(request.dataset_repo, request.grid_name)
            raise HTTPException(status_code=500, detail=str(e))