main.py

Go to the documentation of this file.

"""FastAPI app: browse past Elodin runs, list sensors, plot, and export CSV.
 
Run (dev):   uvicorn backend.main:app --reload --port 8000
Prod-ish:    build the frontend to frontend/dist and this process serves it too.
"""
 
from __future__ import annotations
 
from pathlib import Path
 
from fastapi import Body, FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, PlainTextResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
 
from . import config, descriptions, export_cache, run_config, runs, series, summary, video
 
app = FastAPI(title="Elodin Past-Run Viewer")
 
# Dev convenience: the Vite dev server (5173) may call us cross-origin. In the
# single-process prod setup the frontend is same-origin so this is a no-op.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
 
 
def _parse_components(components: str | None) -> list[str]:
    if not components:
        return []
    return [c for c in components.split(",") if c]
 
 
def _time_source(value: str) -> str:
    """Which clock the x-axis is. Defaults to the sensor clock; see series.py."""
    if value not in series.TIME_SOURCES:
        raise HTTPException(400, f"time_source must be one of {series.TIME_SOURCES}")
    return value
 
 
@app.get("/api/runs")
def api_runs():
    return runs.list_runs()
 
 
@app.put("/api/runs/{run_id}/description")
def api_set_description(run_id: str, text: str = Body(..., embed=True, max_length=4000)):
    """Set (or clear, with empty text) this run's shared one-line description.
 
    Unowned on purpose: there is no login on this viewer, so anyone who can see a run can
    label it and everyone else sees that label. Last write wins.
    """
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    if not (config.ELODIN_DIR / run_id).is_dir():
        raise HTTPException(404, f"run not found: {run_id}")
    try:
        entry = descriptions.set_text(run_id, text)
    except OSError as e:
        # Almost always a read-only or missing mount. Say so, rather than accepting the
        # edit and quietly dropping it — the user would have no way to tell.
        raise HTTPException(503, f"could not save description: {e}")
    return {"run_id": run_id, "description": entry["text"], "updated_at": entry["updated_at"]}
 
 
@app.get("/api/runs/{run_id}/summary")
def api_summary(run_id: str):
    """What is knowable about a run without exporting it — size, component count and an
    approximate duration. Sub-second; safe to call on every run selection."""
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    if not (config.ELODIN_DIR / run_id).is_dir():
        raise HTTPException(404, f"run not found: {run_id}")
    return summary.summarize(run_id)
 
 
@app.get("/api/runs/{run_id}/components")
def api_components(run_id: str):
    """The component index. Only for an already-indexed run: selecting a run must not
    cost a multi-second export, so building one is POST /index, an explicit action."""
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    if not export_cache.is_cached(run_id):
        raise HTTPException(409, "run is not indexed yet")
    try:
        return export_cache.get_index(run_id)
    except FileNotFoundError as e:
        raise HTTPException(404, str(e))
    except RuntimeError as e:
        raise HTTPException(500, str(e))
 
 
@app.post("/api/runs/{run_id}/index")
def api_index(run_id: str):
    """Export the run to parquet and build its index — the expensive step, run only when
    asked for. Synchronous: tens of seconds on a multi-GB run, and the caller shows a
    busy state for the duration. Concurrent callers share one export (export_cache holds
    a per-run lock), so a double click costs nothing extra."""
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    try:
        return export_cache.get_index(run_id)
    except FileNotFoundError as e:
        raise HTTPException(404, str(e))
    except RuntimeError as e:
        raise HTTPException(500, str(e))
 
 
@app.get("/api/runs/{run_id}/config", response_class=PlainTextResponse)
def api_config(run_id: str):
    """The config snapshot taken beside this run's DB when the session started.
 
    Served verbatim — it is the record of what actually ran, so it is never reformatted
    or re-serialised. Runs recorded before the snapshot existed simply have no file.
    """
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    path = run_config.snapshot_path(run_id)
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        raise HTTPException(404, "no config snapshot for this run")
    return PlainTextResponse(
        text,
        media_type="text/plain; charset=utf-8",
        headers={"Content-Disposition": f'inline; filename="{run_id}.toml"'},
    )
 
 
@app.get("/api/runs/{run_id}/series")
def api_series(
    run_id: str,
    components: str = Query(...),
    start: float | None = None,
    end: float | None = None,
    max_points: int = 4000,
    time_source: str = "sensor",
):
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    comps = _parse_components(components)
    if not comps:
        raise HTTPException(400, "no components requested")
    src = _time_source(time_source)
    export_cache.ensure_exported(run_id)
    try:
        return series.series_json(
            run_id, comps, start, end, max(4, min(max_points, 50000)), src
        )
    except FileNotFoundError as e:
        raise HTTPException(404, str(e))
 
 
@app.get("/api/runs/{run_id}/download")
def api_download(
    run_id: str,
    components: str | None = None,
    start: float | None = None,
    end: float | None = None,
    time_source: str = "sensor",
):
    """CSV export.
 
    - With `components` (a plot selection): a **wide** CSV — one column per
      channel aligned on a shared time grid. Bounded by the selection, ideal for
      analysis of a few channels.
    - Without `components` (whole run): a **long/tidy** CSV (`time,component,
      value`) over every primary channel. Streams at constant memory; a wide
      whole-run frame would blow up to millions of rows × hundreds of columns.
    """
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    src = _time_source(time_source)
    index = export_cache.get_index(run_id)
    comps = _parse_components(components)
    if comps:
        rows = series.wide_csv_rows(run_id, comps, start, end, src)
        fname = f"{run_id}_selection.csv"
    else:
        comps = [c["name"] for c in index["components"] if c["primary"]]
        if not comps:
            raise HTTPException(404, "no exportable channels")
        rows = series.long_csv_rows(run_id, comps, start, end, src)
        fname = f"{run_id}.csv"
    # Rate-limit the stream so a large export can't saturate the host uplink.
    rows = series.throttle(rows, config.MAX_DOWNLOAD_BYTES_PER_SEC)
    return StreamingResponse(
        rows,
        media_type="text/csv",
        headers={"Content-Disposition": f'attachment; filename="{fname}"'},
    )
 
 
@app.get("/api/runs/{run_id}/video")
def api_run_video(run_id: str):
    """The video clips that overlap this run, on the epoch-seconds axis the plot uses.
 
    Matched against the run's whole extent on either clock, so switching the plot to
    DB write time never hides a clip. Needs the index (that is where the extent lives);
    an un-indexed run has no plot to put video on anyway. `available` is false when no
    recordings folder is mounted: the page then shows no video panel at all.
    """
    if not config.RUN_RE.match(run_id):
        raise HTTPException(400, "invalid run id")
    if not video.available():
        return {"available": False, "clips": []}
    if not export_cache.is_cached(run_id):
        raise HTTPException(409, "run is not indexed yet")
    index = export_cache.get_index(run_id)
    lows = [t for t in (index.get("t_min"), index.get("sensor_t_min")) if t is not None]
    highs = [t for t in (index.get("t_max"), index.get("sensor_t_max")) if t is not None]
    if not lows or not highs:
        return {"available": True, "clips": []}
    clips = video.clips_overlapping(min(lows), max(highs))
    return {"available": True, "clips": [c.as_json() for c in clips]}
 
 
# `private` keeps a shared cache in front of the viewer (Cloudflare, which caches .mp4 by
# extension) from storing the clip. On a cache miss Cloudflare answers a Range request by
# first pulling the whole object from byte 0, so a seek to the end of a fresh 190 MB Save
# waited ~25 s per camera for its first byte (measured 2026-11-10) and the panel looked
# dead. Uncached, the Range goes straight to this server and comes back at once. The
# browser may still keep it: a finished clip never changes.
VIDEO_CACHE_CONTROL = "private, max-age=86400"
 
 
@app.get("/api/video/{recording}/{stream}.mp4")
def api_video_file(recording: str, stream: str):
    """One clip's file, with HTTP Range (Starlette's FileResponse answers 206) so the
    browser can seek without downloading the whole recording."""
    path = video.clip_file(recording, stream)
    if path is None:
        raise HTTPException(404, "no such clip")
    return FileResponse(path, media_type="video/mp4", headers={"Cache-Control": VIDEO_CACHE_CONTROL})
 
 
# Serve the built frontend if present (single-process deployment). Mounted last
# so it never shadows /api routes.
_dist = Path(__file__).resolve().parent.parent / "frontend" / "dist"
if _dist.is_dir():
    app.mount("/", StaticFiles(directory=str(_dist), html=True), name="static")