api ¶

@router.get("/ping")
async def ping(request: Request) -> dict[str, bool | float | None]:
    """
    Cheap liveness check the frontend polls to show a connected/disconnected indicator.

    Also reports the `.db` file's mtime, so the client can detect that someone regenerated it
    (e.g. re-ran `trendify generate`/`run` while this server is still up) and prompt a refresh.
    The underlying `RecordStore` connection itself stays valid across a regeneration (the
    generate pipeline writes into the same file/inode via WAL, it doesn't replace it); only
    this process's *response cache* goes stale, so an mtime change clears it here.
    """
    db_path: Path = request.app.state.db_path
    try:
        mtime = db_path.stat().st_mtime
    except OSError:
        mtime = None

    if mtime is not None and mtime != request.app.state.db_mtime:
        request.app.state.db_mtime = mtime
        request.app.state.response_cache.clear()

    return {"ok": True, "db_updated_at": mtime}

@router.get("/table", response_model=TableResponse)
async def get_table(tag: str, view: TableView, request: Request) -> TableResponse:
    """
    Table data for the table viewer's Melted/Pivot/Statistics tabs, as DataTables-ready
    `{available, columns, rows}` JSON.

    `tag` is the tag's `encode_tag` (JSON) form, the same string the sidebar's `tojson`-
    rendered `tag-selected` payload already produces client-side, since a tuple tag has no
    single unambiguous plain-string/path encoding to put directly in a URL path segment.

    `pivot`/`stats` report `available: false` (not a 500) rather than raising when the pivot
    fails (a repeated `(row, col)` pair) or there are no numeric pivoted columns: both are
    normal, expected shapes for some tags' data, not error conditions.
    """
    decoded_tag = _decode_tag_or_422(tag)
    is_hydration = _is_hydration_request(request)
    if is_hydration:
        logger.debug(
            f"Hydrating tag {decoded_tag!r} in the background (table, view={view})"
        )

    def build(store: RecordStore) -> TableResponse:
        melted = store.get_table_entries(decoded_tag)
        if view == "melted":
            return TableResponse(
                available=melted.height > 0,
                columns=melted.columns,
                rows=melted.to_dicts(),
            )

        pivot = TableBuilder.pivot_table(melted) if melted.height > 0 else None
        if view == "pivot":
            if pivot is None:
                return TableResponse(available=False, columns=[], rows=[])
            return TableResponse(
                available=True, columns=pivot.columns, rows=pivot.to_dicts()
            )

        stats = TableBuilder.get_stats_table(pivot) if pivot is not None else None
        if stats is None:
            return TableResponse(available=False, columns=[], rows=[])
        return TableResponse(
            available=True, columns=stats.columns, rows=stats.to_dicts()
        )

    return await _cached(request, ("table", view, tag), lambda: _run(request, build))

@router.get("/plot", response_model=PlotResponse)
async def get_plot(
    tag: str,
    request: Request,
    line_mode: LineMode = LineMode.LINES_AND_MARKERS,
    interp: InterpMode = InterpMode.LINEAR,
    hover: HoverMode = HoverMode.CLOSEST,
    show_spike: bool = False,
    max_points: int | None = Query(None, gt=0, le=_MAX_POINTS_LIMIT),
) -> PlotResponse:
    """
    Plotly figure JSON (`{available, data, layout}`) for the plot viewer, built the same way as
    the static Plotly/matplotlib renderers (`generator.render._render_tag_assets`): every
    `PlottableData2D` record sharing `tag` is added to one shared `PlotlyFigure`, then that
    tag's `Format2D` (if any) is applied.

    `line_mode`/`interp`/`hover`/`show_spike`/`max_points` are the dashboard's view-only
    `PlotConfig` settings (`viewer.plot_config`), applied as a post-process pass over the
    already-built figure JSON rather than baked into `PlotlyFigure`/the record classes
    themselves -- they're display overrides the *viewer* controls, independent of whatever
    style a record's own `Pen`/`Marker` authored. `AxLine` renders as layout shapes
    (`add_hline`/`add_vline`), not `data` traces, so it's naturally unaffected by the
    trace-level overrides below.
    """
    decoded_tag = _decode_tag_or_422(tag)
    is_hydration = _is_hydration_request(request)
    if is_hydration:
        logger.debug(f"Hydrating tag {decoded_tag!r} in the background (plot)")

    def build(store: RecordStore) -> PlotResponse:
        format2d_records = store.get_records_of_type(Format2D, tag=decoded_tag)
        format2d = format2d_records[0] if format2d_records else None

        points = store.get_records_of_type(Point2D, tag=decoded_tag)
        traces = store.get_records_of_type(Trace2D, tag=decoded_tag)
        scatters = store.get_records_of_type(Scatter2D, tag=decoded_tag)
        axlines = store.get_records_of_type(AxLine, tag=decoded_tag)
        histogram_entries = store.get_records_of_type(HistogramEntry, tag=decoded_tag)

        figure = PlotlyFigure.new(decoded_tag)
        for record in (*points, *traces, *scatters, *axlines, *histogram_entries):
            figure.add_record(record)

        if not figure.fig.data:
            return PlotResponse(available=False, data=[], layout={})

        if format2d is not None:
            figure.apply_format(format2d)

        raw = figure.fig.to_plotly_json()
        traces_json = cast(list[dict[str, Any]], raw["data"])
        layout_json = cast(dict[str, Any], raw["layout"])

        for trace in traces_json:
            if trace.get("type") != "scatter":
                continue
            trace["mode"] = line_mode.value
            trace.setdefault("line", {})["shape"] = interp.value
            trace["x"] = _plain_list(trace.get("x"))
            trace["y"] = _plain_list(trace.get("y"))
            if max_points is not None and len(trace["x"]) > max_points:
                trace["x"], trace["y"] = _downsample_xy(
                    trace["x"], trace["y"], max_points
                )

        layout_json["hovermode"] = False if hover == HoverMode.NONE else hover.value
        if show_spike:
            for axis_key in ("xaxis", "yaxis"):
                axis = layout_json.setdefault(axis_key, {})
                axis["showspikes"] = True
                axis["spikemode"] = "across"
                axis["spikedash"] = "dot"
                # Plotly's spike default (a heavy black/white line) reads harshly against either
                # theme -- rose-500 (this app's one accent color, see DASHBOARD.md's "Color
                # Scheme") stays legible and consistent in both, and a thinner line is less
                # visually loud than the default.
                axis["spikecolor"] = "#f43f5e"
                axis["spikethickness"] = 1

        return PlotResponse(available=True, data=traces_json, layout=layout_json)

    return await _cached(
        request,
        ("plot", tag, line_mode, interp, hover, show_spike, max_points),
        lambda: _run(request, build),
    )