mosaic ¶

def get_network_ip() -> str:
    """Detects the primary local network IP of the host machine."""
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # Doesn't actually have to be reachable; just triggers IP selection
        s.connect(("8.8.8.8", 1))
        ip = s.getsockname()[0]
    except Exception:
        ip = "localhost"
    finally:
        s.close()
    return ip

@router.get("/", response_class=HTMLResponse)
async def get_mosaic(request: Request) -> HTMLResponse:
    """Serves the initial mosiac frame."""
    return shared.templates.TemplateResponse(
        request=request, name="mosaic.html", context={"request": request}
    )

@router.get("/api/trials")
async def get_valid_trials() -> dict[str, list[str]]:
    """Scans the workdir for folders containing telemetry data."""
    job = shared.CURRENT_JOB
    from mujoco_mojo.runtime.signal_manager import SignalManager

    if job is None:
        logger.warning("Mosaic accessed but CURRENT_JOB is None.")
        return {"trials": []}

    valid_trials = []

    for tn in job.trial_nums:
        trial_dir = job.trial_num_to_path(tn)

        # if there is a db and the trial is actually done
        if (
            trial_dir / SignalManager.default_output_name()
        ).exists() and tn in job._cache:
            valid_trials.append(trial_dir.name)

    valid_trials.sort()

    return {"trials": valid_trials}

@router.get("/{trial_id}", response_class=HTMLResponse)
async def get_trial_viewer(request: Request, trial_id: str) -> HTMLResponse:
    """Land here when clicking a trial."""
    job = shared.CURRENT_JOB

    if not job:
        raise HTTPException(status_code=404, detail="No active job found.")

    server_ip = get_network_ip()
    port = request.url.port or 8000

    # 1. Get the sorted list of all trials that actually have data
    from mujoco_mojo.runtime.signal_manager import SignalManager

    valid_ids = []
    for tn in job.trial_nums:
        path = job.trial_num_to_path(tn)
        if (path / SignalManager.default_output_name()).exists():
            valid_ids.append(path.name)

    if trial_id not in valid_ids:
        raise HTTPException(
            status_code=404,
            detail=f"Trial '{trial_id}' does not exist or has no telemetry data.",
        )

    # 2. Find the neighbors of the current trial_id
    idx = valid_ids.index(trial_id)
    prev_id = valid_ids[idx - 1] if idx > 0 else None
    next_id = valid_ids[idx + 1] if idx < len(valid_ids) - 1 else None

    return shared.templates.TemplateResponse(
        request=request,
        name="trial_viewer.html",
        context={
            "request": request,
            "trial_id": trial_id,
            "prev_id": prev_id,
            "next_id": next_id,
            "external_url": f"http://{server_ip}:{port}",
            "show_quick_filters": MujocoMojoSettings().dojo.show_quick_filters,
            "column_metadata_fields": _column_metadata_fields(),
        },
    )

@router.get("/api/filter-schema")
async def get_filter_schema() -> list[dict]:
    """
    Returns metadata for all available filter types, derived from Pydantic models.

    Filter classes are auto-discovered from AnyFilter's union; no changes needed here
    when a new filter is added to filters.py.
    """
    from pydantic_core import PydanticUndefined

    def _infer_type(prop: dict) -> str:
        if prop.get("ui_type") == "col":
            return "col"
        if prop.get("ui_type") == "quat_col":
            return "quat_col"
        if prop.get("ui_type") == "vec_col":
            return "vec_col"
        if prop.get("ui_type") == "select":
            return "select"
        if "anyOf" in prop:
            non_null = [s for s in prop["anyOf"] if s.get("type") != "null"]
            prop = non_null[0] if non_null else {}
        t = prop.get("type", "")
        if t == "integer":
            return "int"
        if t == "number":
            return "float"
        if t == "boolean":
            return "bool"
        return "string"

    result = []
    for cls in _FILTER_CLASSES:
        schema = cls.model_json_schema()
        props = schema.get("properties", {})
        type_val = str(cls.model_fields["type"].default)

        params = []
        for name, field_info in cls.model_fields.items():
            if name in ("type", "enabled"):
                continue
            # model_json_schema() keys properties by alias when serialize_by_alias=True.
            # alias_generator may be a plain callable (to_camel) or an AliasGenerator
            # object; only call it when it's actually callable.
            ag = cls.model_config.get("alias_generator")
            alias: str = field_info.alias or (ag(name) if callable(ag) else name)
            prop = props.get(alias, props.get(name, {}))
            if "anyOf" in prop:
                non_null = [s for s in prop["anyOf"] if s.get("type") != "null"]
                prop_clean = {**prop, **(non_null[0] if non_null else {})}
            else:
                prop_clean = prop

            default = field_info.default
            if default is PydanticUndefined:
                default = None
            elif isinstance(default, float):
                default = round(float(default), 8)

            p: dict = {"name": alias, "type": _infer_type(prop), "default": default}
            if p["type"] == "select" and "enum" in prop:
                p["options"] = prop["enum"]
            if "minimum" in prop_clean:
                p["min"] = prop_clean["minimum"]
            if "maximum" in prop_clean:
                p["max"] = prop_clean["maximum"]
            if "exclusiveMinimum" in prop_clean:
                p["exclusive_min"] = prop_clean["exclusiveMinimum"]
            if "exclusiveMaximum" in prop_clean:
                p["exclusive_max"] = prop_clean["exclusiveMaximum"]
            params.append(p)

        description = (cls.__doc__ or "").strip()
        description = description.split("\n")[0].strip()

        entry: dict = {
            "type": type_val,
            "label": type_val.replace("_", " ").title(),
            "description": description,
            "category": cls.category,
            "params": params,
        }
        if type_val == "unit":
            entry["unit_groups"] = [
                {
                    "label": label,
                    "units": units,
                    "dimension": _unit_group_dimension(units),
                }
                for label, units in _UNIT_GROUPS
            ]
        result.append(entry)

    return result

@router.get("/api/plot-config-schema")
async def get_plot_config_schema() -> dict[str, Any]:
    """
    Returns PlotConfig's raw JSON schema, descriptions included - single
    source of truth for the Plot Editor's and JSON editor's hover tooltips
    (trial-viewer.ts), so a Field(description=...) change in plot_config.py
    shows up in both without any manual sync. Properties are keyed by their
    camelCase alias (e.g. "lineMode"), matching `config.*` in the frontend
    and the JSON editor's own document, since PlotConfig's model_config
    serializes by alias.

    Uses GenerateJsonSchemaWithDefaults (settings.py - the same generator
    the Dojo settings panel's own schema endpoint uses, not redefined here)
    for two extras beyond plain model_json_schema(): each field's
    description gets its default value appended, and each enum type in
    $defs gets an `x-enum-descriptions` map of value -> attribute-docstring
    (e.g. GridMode.ALL's own docstring), since standard JSON Schema's
    `enum` keyword has no room for per-value metadata - that's what lets a
    dropdown show a tooltip on "Major + Minor" itself, not just on the
    Gridlines field as a whole.
    """
    return _PlotConfig.model_json_schema(
        schema_generator=GenerateJsonSchemaWithDefaults
    )

@router.get("/api/profiles")
async def list_profiles() -> list[dict[str, Any]]:
    """List all saved profiles, including those in sub-folders."""
    d = _get_profiles_dir()
    profiles = [
        {
            "name": f.relative_to(d).with_suffix("").as_posix(),
            "modified": int(f.stat().st_mtime * 1000),
        }
        for f in sorted(
            d.rglob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True
        )
    ]
    return profiles

@router.get("/api/profiles/{name:path}")
async def get_profile(name: str) -> dict[str, Any]:
    """Return the saved profile (tabs + active index), validated against the schema."""
    from pydantic import ValidationError

    path = _resolve_profile_path(name)
    if not path.exists():
        raise HTTPException(status_code=404, detail="Profile not found")
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        profile = _parse_profile(data, name)
    except (json.JSONDecodeError, ValidationError) as exc:
        raise HTTPException(
            status_code=422,
            detail=f"Profile '{name}' failed validation and cannot be loaded: {exc}",
        ) from exc
    return profile.model_dump()

@router.post("/api/profiles/{name:path}")
async def save_profile(name: str, request: Request, body: dict) -> dict[str, str]:
    """
    Save the current set of plot tabs as a named profile.

    `body` is a plain dict (not `_PlotProfile` directly) so the legacy-format
    discriminator in `_parse_profile` can run before Pydantic's own request
    validation would otherwise reject an old-shaped payload with a 422 - in
    practice the frontend only ever POSTs the current (tabs-based) format,
    but accepting a legacy body here too keeps this endpoint's behavior
    symmetric with `get_profile`. The Content-Length header is checked first
    as a lightweight size guard. Sub-folder paths (e.g. 'project/baseline')
    are supported; directories are created automatically.
    """
    from pydantic import ValidationError

    cl = request.headers.get("content-length")
    if cl and int(cl) > _PROFILE_MAX_BYTES:
        raise HTTPException(status_code=413, detail="Profile payload too large")
    try:
        profile = _parse_profile(body, name)
    except ValidationError as exc:
        raise HTTPException(
            status_code=422,
            detail=f"Invalid profile payload: {exc}",
        ) from exc
    path = _resolve_profile_path(name)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(profile.model_dump_json(), encoding="utf-8")
    d = _get_profiles_dir()
    return {"name": path.relative_to(d).with_suffix("").as_posix()}

@router.delete("/api/profiles/{name:path}")
async def delete_profile(name: str) -> dict[str, str]:
    """Delete a saved profile."""
    path = _resolve_profile_path(name)
    if not path.exists():
        raise HTTPException(status_code=404, detail="Profile not found")
    path.unlink()
    d = _get_profiles_dir()
    # Remove empty parent directories up to (but not including) the profiles root.
    parent = path.parent
    while parent != d and parent.is_dir() and not any(parent.iterdir()):
        parent.rmdir()
        parent = parent.parent
    return {"deleted": path.relative_to(d).with_suffix("").as_posix()}

@router.get("/api/lab")
async def list_labs() -> list[dict]:
    """List all saved lab graphs with their input column requirements and output labels."""
    d = _get_lab_dir()
    return [
        _lab_meta(f, d)
        for f in sorted(
            d.rglob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True
        )
    ]

@router.get("/api/lab/{name:path}")
async def get_lab(name: str) -> Any:
    """Return the raw LiteGraph JSON for a saved lab."""
    path = _resolve_lab_path(name)
    if not path.exists():
        raise HTTPException(status_code=404, detail="Lab not found")
    return json.loads(path.read_text(encoding="utf-8"))

@router.post("/api/lab/{name:path}")
async def save_lab(name: str, request: Request) -> dict[str, str]:
    """Save a LiteGraph graph JSON as a named lab."""
    cl = request.headers.get("content-length")
    if cl and int(cl) > _LAB_MAX_BYTES:
        raise HTTPException(status_code=413, detail="Lab payload too large")
    body = await request.json()
    path = _resolve_lab_path(name)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(body), encoding="utf-8")
    d = _get_lab_dir()
    return {"name": path.relative_to(d).with_suffix("").as_posix()}

@router.delete("/api/lab/{name:path}")
async def delete_lab(name: str) -> dict[str, str]:
    """Delete a saved lab."""
    path = _resolve_lab_path(name)
    if not path.exists():
        raise HTTPException(status_code=404, detail="Lab not found")
    path.unlink()
    d = _get_lab_dir()
    # Remove empty parent directories up to (but not including) the lab root.
    parent = path.parent
    while parent != d and parent.is_dir() and not any(parent.iterdir()):
        parent.rmdir()
        parent = parent.parent
    return {"deleted": path.relative_to(d).with_suffix("").as_posix()}

@router.get("/{trial_id}/data", response_model=None)
async def get_trial_data(
    trial_id: str,
    cols: str = Query(None),
    filters: str = Query(None),
    display_unit_system: str | None = Query(None),
    max_points: int | None = Query(None, gt=0),
) -> dict[str, Any]:
    """
    Loops over the columns in the trial_id provided and returns their data.

    Args:
        trial_id (str): Trial to search (e.g. `"trial_001"`).
        cols (str, optional): Comma separated list of column names to return data for (e.g. `"/Bodies/body1/xpos:x,/Bodies/body2/xpos:m"`). Defaults to Query(None).
        filters (str, optional): String representation of filters to be applied sequentially. Defaults to Query(None).
        display_unit_system (str, optional): Named unit system to convert data into before returning (e.g. `"si"`, `"ips"`). Only columns whose metadata carries a resolvable unit or dimension are converted. Defaults to Query(None).
        max_points (int, optional): Maximum number of data points per trace. When the raw data exceeds this limit the response is downsampled using uniform time-domain buckets. Defaults to Query(None).

    Raises:
        HTTPException: Raised if no shared.CURRENT_JOB was set.
        HTTPException: Raised if the database path for the trial was found.
        HTTPException: Raised if an error occured while extracting data from the database.

    Returns:
        dict: Dictionary containing split columns and their associated data.

    """
    from mujoco_mojo.runtime.signal_manager import SignalManager

    job = shared.CURRENT_JOB
    if not job:
        raise HTTPException(status_code=404, detail="No job active")

    db_path = (
        job.workdir / "trials" / trial_id / SignalManager.default_output_name()
    ).resolve()

    if not db_path.exists():
        raise HTTPException(
            status_code=404, detail=f"Database not found for {trial_id}"
        )

    # use the files last modified time as a cache breaker
    mtime = db_path.stat().st_mtime

    # Parquet-only manifest (cached), then augment with valid lab virtual columns.
    parquet_manifest = _get_column_manifest(db_path, mtime)
    lab_mtime = _lab_dir_mtime()
    lab_extra = _valid_lab_columns_cached(frozenset(parquet_manifest["all"]), lab_mtime)
    if lab_extra:
        raw_manifest = _get_mojo_df(db_path, mtime).mojo.get_manifest(
            extra_columns=list(lab_extra),
            column_metadata={
                **_get_column_metadata(db_path, mtime),
                **_lab_column_metadata_cached(lab_mtime),
            },
        )
        column_manifest: ColumnManifest = {
            **raw_manifest,
            "column_metadata": {
                col: _augment_col_meta(meta)
                for col, meta in raw_manifest["column_metadata"].items()
            },
        }
    else:
        column_manifest = parquet_manifest

    try:
        requested = cols.split(",") if cols else []
        # available_cols covers only parquet columns for actual fetch logic
        available_cols = set(parquet_manifest["all"])

        # determine columns to request
        fetch_targets = [c for c in requested if c in available_cols]

        # ── Collect lab SI columns before building the df ─────────────────────
        # Lab virtual columns are computed from the graph, not read from parquet.
        # We load executors now so we can add their parquet SI deps to fetch_targets.
        lab_cols = [c for c in requested if c.startswith(f"{_LAB_PREFIX}/")]
        from collections import defaultdict as _dd

        by_lab: dict[str, list[tuple[str, str]]] = _dd(list)
        lab_executors: dict = {}

        if lab_cols:
            from mujoco_mojo.utils.layers.dojo.lab_executor import LabExecutor

            for col in lab_cols:
                # Lab names may themselves contain '/' (nested folders), so split
                # from the right - the output label is always the final segment.
                rest = col[len(_LAB_PREFIX) + 1 :]
                if "/" in rest:
                    lab_name, output_label = rest.rsplit("/", 1)
                    by_lab[lab_name].append((col, output_label))

            # Load executors transitively so chained labs have their deps available.
            to_load = list(by_lab.keys())
            while to_load:
                name = to_load.pop()
                if name in lab_executors:
                    continue
                lab_path = _resolve_lab_path(name)
                if not lab_path.exists():
                    continue
                try:
                    g = json.loads(lab_path.read_text(encoding="utf-8"))
                    lab_executors[name] = LabExecutor(g)
                    for si_col in lab_executors[name].signal_in_columns:
                        if si_col.startswith(f"{_LAB_PREFIX}/"):
                            dep_rest = si_col[len(_LAB_PREFIX) + 1 :]
                            if "/" in dep_rest:
                                dep_lab_name = dep_rest.rsplit("/", 1)[0]
                                if dep_lab_name not in lab_executors:
                                    to_load.append(dep_lab_name)
                except Exception:
                    pass

            # Add parquet SI columns and the time column to fetch_targets.
            extra_si: set[str] = set()
            for executor in lab_executors.values():
                for si_col in executor.signal_in_columns:
                    if si_col in available_cols:
                        extra_si.add(si_col)
                # Sibling vector components and quaternion/origin columns needed by
                # any in-graph Rotation or Translation node - without these,
                # apply_with_context() can't find the columns it needs and the
                # transform is a no-op.
                for dep_col in executor.frame_dependencies:
                    if dep_col in available_cols:
                        extra_si.add(dep_col)
            if _TIME_COLUMN_NAME in available_cols:
                extra_si.add(_TIME_COLUMN_NAME)
            existing_targets = set(fetch_targets)
            fetch_targets.extend(c for c in extra_si if c not in existing_targets)

        # ── Pre-flight: ensure Rotation/Translation filter dependencies are fetched ──
        # Parse filters early (errors are tolerated; full parse happens again below).
        if filters:
            try:
                _preflight = _filter_adapter.validate_python(json.loads(filters))
                _existing = set(fetch_targets)
                for _col, _flist in _preflight.items():
                    for _f in _flist:
                        if not isinstance(_f, (_RotationFilter, _TranslationFilter)):
                            continue
                        # Sibling vector components for the column being transformed -
                        # fetched unconditionally (cheap if unused) since either
                        # quat_col or origin_col alone is enough to need them, and a
                        # RotationFilter with neither set is just a no-op.
                        if (
                            _col.endswith(":x")
                            or _col.endswith(":y")
                            or _col.endswith(":z")
                        ):
                            _base = _col.rsplit(":", 1)[0]
                            for _comp in ("x", "y", "z"):
                                _sib = f"{_base}:{_comp}"
                                if _sib in available_cols and _sib not in _existing:
                                    fetch_targets.append(_sib)
                                    _existing.add(_sib)
                        # Quaternion components (RotationFilter only, when set)
                        if isinstance(_f, _RotationFilter) and _f.quat_col:
                            for _comp in ("w", "x", "y", "z"):
                                _qc = f"{_f.quat_col}:{_comp}"
                                if _qc in available_cols and _qc not in _existing:
                                    fetch_targets.append(_qc)
                                    _existing.add(_qc)
                        # Origin (frame position) components, when set
                        if _f.origin_col:
                            for _comp in ("x", "y", "z"):
                                _oc = f"{_f.origin_col}:{_comp}"
                                if _oc in available_cols and _oc not in _existing:
                                    fetch_targets.append(_oc)
                                    _existing.add(_oc)
            except Exception:
                pass

        # early exit for no found columns
        if not fetch_targets:
            return {"columns": column_manifest, "data": {}}

        # assemble dataframe
        raw_data = {
            col: _get_atomic_column(db_path, col, mtime) for col in fetch_targets
        }
        df = MojoDataFrame.from_dict(raw_data)

        # apply display unit system conversion before filters so filter params operate in display units
        if display_unit_system and display_unit_system in _UNIT_SYSTEMS:
            target_us = _UNIT_SYSTEMS[display_unit_system]
            df = df.mojo.with_unit_system(
                target_us, column_metadata=_get_column_metadata(db_path, mtime)
            )
            # update manifest column_metadata with resolved units and group_unit for the UI
            display_col_meta = {
                col: _augment_col_meta(_resolve_dim_meta(meta, target_us))
                for col, meta in column_manifest["column_metadata"].items()
            }
            column_manifest = {
                "all": column_manifest["all"],
                "rotatable_vectors": column_manifest["rotatable_vectors"],
                "available_quats": column_manifest["available_quats"],
                "column_metadata": display_col_meta,
            }

        # parse validated filter stacks (col_name -> list[AnyFilter])
        col_filters: dict = {}
        filter_errors: list[str] = []
        if filters:
            try:
                col_filters = _filter_adapter.validate_python(json.loads(filters))
            except Exception as e:
                logger.warning(f"Could not parse filters for {trial_id}: {e}")
                filter_errors.append(_format_filter_error(e))

        data: dict = {}

        # ── Execute lab virtual columns (multi-pass for chained labs) ──────────
        # exec_df accumulates lab outputs alongside the raw fetched columns, so
        # that stacked filters on Lab/... columns (e.g. RotationFilter) can find
        # sibling Lab/... vector components via apply_with_context.
        exec_df = df
        if lab_cols and lab_executors:
            lab_cols_set = set(lab_cols)

            # For transitive deps not directly requested, expose all their outputs
            # so downstream labs can use them as inputs via exec_df.
            all_by_lab: dict[str, list[tuple[str, str]]] = dict(by_lab)
            for lab_name, executor in lab_executors.items():
                if lab_name not in all_by_lab:
                    all_by_lab[lab_name] = [
                        (f"{_LAB_PREFIX}/{lab_name}/{out}", out)
                        for out in executor.output_labels
                    ]

            remaining = dict(all_by_lab)
            for _ in range(len(remaining) + 1):
                if not remaining:
                    break
                progress = False
                for lab_name in list(remaining.keys()):
                    if lab_name not in lab_executors:
                        del remaining[lab_name]
                        progress = True
                        continue
                    executor = lab_executors[lab_name]
                    # Wait until all lab-virtual SI deps are present in exec_df.
                    if any(
                        c not in exec_df.columns
                        for c in executor.signal_in_columns
                        if c.startswith(f"{_LAB_PREFIX}/")
                    ):
                        continue
                    try:
                        outputs = executor.execute(exec_df)
                        new_series: list[pl.Series] = []
                        for full_col, output_label in remaining[lab_name]:
                            if output_label in outputs:
                                s = outputs[output_label].rename(full_col)
                                new_series.append(s)
                                if full_col in lab_cols_set:
                                    data[full_col] = _json_safe_list(s)
                        if new_series:
                            exec_df = MojoDataFrame.from_pl(exec_df.hstack(new_series))
                        del remaining[lab_name]
                        progress = True
                    except Exception as exc:
                        logger.warning(f"Lab '{lab_name}' execution failed: {exc}")
                        del remaining[lab_name]
                        progress = True
                if not progress:
                    break

        column_meta = column_manifest["column_metadata"]

        def _transform_type_of(col: str) -> TransformType | None:
            meta = column_meta.get(col)
            return meta.transform_type if meta else None

        # ── build response data, applying per-column filters where present ────
        for col in requested:
            if col.startswith(f"{_LAB_PREFIX}/"):
                # Lab output already in data; apply any stacked filters on top.
                filter_list = col_filters.get(col)
                if filter_list and col in data:
                    series = pl.Series(name=col, values=data[col], dtype=pl.Float64)
                    for f in filter_list:
                        f = _origin_for_positions_only(f, _transform_type_of(col))
                        ctx = f.apply_with_context(series, exec_df)
                        if ctx is not None:
                            series = ctx
                        else:
                            tmp = pl.DataFrame({col: series})
                            tmp = tmp.with_columns(f.apply(pl.col(col)).alias(col))
                            series = tmp[col]
                    data[col] = _json_safe_list(series)
                continue
            if col not in df.columns:
                continue
            series = df[col]
            filter_list = col_filters.get(col)
            if filter_list:
                if series.dtype != pl.Float64:
                    series = series.cast(pl.Float64)
                for f in filter_list:
                    f = _origin_for_positions_only(f, _transform_type_of(col))
                    # context-aware filters (e.g. derivative/integral wrt another col)
                    ctx = f.apply_with_context(series, df)
                    if ctx is not None:
                        series = ctx
                    else:
                        tmp = pl.DataFrame({col: series})
                        tmp = tmp.with_columns(f.apply(pl.col(col)).alias(col))
                        series = tmp[col]
            data[col] = _json_safe_list(series)

        if max_points is not None:
            data = _downsample(data, max_points)

        return {
            "columns": column_manifest,
            "data": data,
            "filter_errors": filter_errors,
        }

    except Exception as e:
        logger.error(f"Data retrieval failed for {trial_id}: {e}")
        raise HTTPException(status_code=500, detail="Internal Server Error")

@router.get("/{trial_id}/media")
async def list_trial_media(trial_id: str) -> dict:
    """Lists media files in the trial directory with their fps metadata."""
    trial_dir = _resolve_trial_dir(trial_id)
    if not trial_dir.is_dir():
        return {"files": []}
    loop = asyncio.get_running_loop()
    paths = sorted(
        p
        for p in trial_dir.iterdir()
        if p.suffix.lower() in _MEDIA_EXTENSIONS and not p.stem.endswith(".mojo_webm")
    )
    files = []
    for p in paths:
        fps = await loop.run_in_executor(None, _read_file_fps, p)
        files.append({"name": p.name, "fps": fps, "mtime": p.stat().st_mtime})
    return {"files": files}

@router.get("/{trial_id}/media/{filename}/as-webm")
async def get_gif_as_webm(trial_id: str, filename: str) -> FileResponse:
    """Converts a GIF to WebM on-demand, cached in the system temp directory."""
    trial_dir = _resolve_trial_dir(trial_id)
    gif_path = (trial_dir / filename).resolve()
    if not gif_path.is_relative_to(trial_dir):
        raise HTTPException(status_code=400, detail="Invalid filename")
    if not gif_path.is_file() or gif_path.suffix.lower() != ".gif":
        raise HTTPException(status_code=404, detail="File not found or not a gif")
    webm_path = _gif_webm_cache_path(gif_path)
    if not webm_path.exists():
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, _convert_gif_to_webm_sync, gif_path, webm_path)
    return FileResponse(webm_path, media_type="video/webm")

@router.get("/{trial_id}/media/{filename}")
async def get_trial_media_file(trial_id: str, filename: str) -> FileResponse:
    """Serves a single media file from the trial directory."""
    trial_dir = _resolve_trial_dir(trial_id)
    file_path = (trial_dir / filename).resolve()
    if not file_path.is_relative_to(trial_dir):
        raise HTTPException(status_code=400, detail="Invalid filename")
    if not file_path.is_file() or file_path.suffix.lower() not in _MEDIA_EXTENSIONS:
        raise HTTPException(status_code=404, detail="File not found")
    _mime: dict[str, str] = {
        ".mp4": "video/mp4",
        ".webm": "video/webm",
        ".gif": "image/gif",
    }
    return FileResponse(
        file_path,
        media_type=_mime.get(file_path.suffix.lower(), "application/octet-stream"),
    )