Spaces:
Sleeping
Sleeping
Download main.py from stasaking/blood-brain-api: direct link, hf CLI and curl.
- Browser
- Download file 29.9 kB
-
https://huggingface.co/spaces/stasaking/blood-brain-api/resolve/b74042352b2100756b3efc82ce4a362c54208a00/main.py
- Command line
-
hf download hf://spaces/stasaking/blood-brain-api@b74042352b2100756b3efc82ce4a362c54208a00/main.py
-
curl -L -o main.py https://huggingface.co/spaces/stasaking/blood-brain-api/resolve/b74042352b2100756b3efc82ce4a362c54208a00/main.py
29.9 kB
| """ | |
| Blood-Brain Omics Benchmark API | |
| FastAPI + DuckDB backend serving benchmark results. | |
| Queries Parquet files remotely via DuckDB httpfs — no data downloaded to disk. | |
| DuckDB reads only the needed columns/row groups per query via HTTP range requests. | |
| Data source modes (in priority order): | |
| 1. Remote: HF Dataset repo via httpfs (default, zero RAM footprint) | |
| 2. Local: files in BENCHMARK_DATA_DIR (for local dev) | |
| Endpoints: | |
| GET /api/v1/registry - Full registry metadata | |
| GET /api/v1/maxn/heatmap - Blood x brain heatmap data | |
| GET /api/v1/maxn/detail - Per-module results for one combination | |
| GET /api/v1/maxn/panel_summary - One panel across all outcomes (mean/min/max) | |
| GET /api/v1/maxn/outcome_summary - One outcome across all panels (mean/min/max) | |
| GET /api/v1/maxn/target_comparison - Per-panel performance for a specific target | |
| GET /api/v1/h2h/blood - Blood H2H comparison | |
| GET /api/v1/h2h/blood/summary - H2H win counts | |
| GET /api/v1/h2h/brain - Brain H2H comparison | |
| GET /api/v1/h2h/models - Model H2H comparison (future) | |
| GET /api/v1/temporal - Temporal decay results | |
| GET /api/v1/features - Feature importance for a target | |
| GET /api/v1/features/cross - Cross-target feature importance | |
| GET /api/v1/health - Health check (DB row counts) | |
| GET /api/v1/ready - Readiness probe (SELECT 1) | |
| """ | |
| import asyncio | |
| import json | |
| import logging | |
| import os | |
| import time | |
| import traceback | |
| import urllib.request | |
| from typing import Optional | |
| import duckdb | |
| from fastapi import FastAPI, HTTPException, Query, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import JSONResponse, Response | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s %(levelname)s %(name)s: %(message)s", | |
| ) | |
| logger = logging.getLogger("blood_brain_api") | |
| # ============================================================================= | |
| # Configuration | |
| # ============================================================================= | |
| # HF Dataset repo for remote Parquet queries | |
| HF_DATASET = os.environ.get( | |
| "HF_DATASET", "stasaking/blood-brain-benchmark" | |
| ) | |
| HF_BASE_URL = f"https://huggingface.co/datasets/{HF_DATASET}/resolve/main" | |
| # Local fallback for dev | |
| DATA_DIR = os.environ.get( | |
| "BENCHMARK_DATA_DIR", | |
| os.path.join(os.path.dirname(__file__), "..", "data") | |
| ) | |
| # DuckDB resource settings — tuned for HF Spaces CPU Basic (2 GB RAM, 2 vCPU). | |
| # Override via environment variables when running on larger hardware. | |
| DUCKDB_MEMORY_LIMIT = os.environ.get("DUCKDB_MEMORY_LIMIT", "512MB") | |
| DUCKDB_THREADS = int(os.environ.get("DUCKDB_THREADS", "1")) | |
| DUCKDB_TEMP_DIR = os.environ.get("DUCKDB_TEMP_DIR", "/tmp/duckdb_tmp") | |
| # Concurrency cap — bound in-flight queries to avoid OOM/process crash on | |
| # CPU Basic. Excess requests get 429 and the frontend can retry. | |
| MAX_INFLIGHT = int(os.environ.get("API_MAX_INFLIGHT", "8")) | |
| # Response cache TTL (seconds) — absorbs repeat queries from rapid clicks. | |
| CACHE_TTL = float(os.environ.get("API_CACHE_TTL", "60")) | |
| CACHE_MAX_ENTRIES = int(os.environ.get("API_CACHE_MAX_ENTRIES", "256")) | |
| app = FastAPI( | |
| title="Blood-Brain Omics Benchmark API", | |
| version="1.0.0", | |
| description="Interactive exploration of blood omics → brain phenotype predictions", | |
| ) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # ============================================================================= | |
| # Concurrency cap + response cache | |
| # ============================================================================= | |
| # Bound the number of in-flight requests so that bursts cannot OOM/crash the | |
| # DuckDB process on CPU Basic. Excess requests fast-fail with 429 instead of | |
| # wedging the worker. | |
| _inflight_sem = asyncio.Semaphore(MAX_INFLIGHT) | |
| # Tiny TTL cache keyed on (path, query_string). Frontend rapid clicks | |
| # (e.g. switching panels) repeatedly hit the same URLs — caching them | |
| # absorbs the load before it ever reaches DuckDB. | |
| _cache: dict[str, tuple[float, int, bytes, str]] = {} | |
| _CACHEABLE_PATH_PREFIXES = ( | |
| "/api/v1/registry", | |
| "/api/v1/maxn/", | |
| "/api/v1/h2h/", | |
| "/api/v1/temporal", | |
| "/api/v1/features", | |
| ) | |
| def _cache_key(request: Request) -> Optional[str]: | |
| if request.method != "GET": | |
| return None | |
| path = request.url.path | |
| if not any(path.startswith(p) for p in _CACHEABLE_PATH_PREFIXES): | |
| return None | |
| return f"{path}?{request.url.query}" | |
| def _cache_get(key: str) -> Optional[tuple[int, bytes, str]]: | |
| entry = _cache.get(key) | |
| if entry is None: | |
| return None | |
| expires, status, body, ctype = entry | |
| if expires < time.monotonic(): | |
| _cache.pop(key, None) | |
| return None | |
| return status, body, ctype | |
| def _cache_put(key: str, status: int, body: bytes, ctype: str) -> None: | |
| if status != 200: | |
| return | |
| if len(_cache) >= CACHE_MAX_ENTRIES: | |
| # Cheap eviction: drop the oldest-expiring entry. | |
| oldest = min(_cache.items(), key=lambda kv: kv[1][0])[0] | |
| _cache.pop(oldest, None) | |
| _cache[key] = (time.monotonic() + CACHE_TTL, status, body, ctype) | |
| async def cache_and_throttle(request: Request, call_next): | |
| key = _cache_key(request) | |
| if key is not None: | |
| hit = _cache_get(key) | |
| if hit is not None: | |
| status, body, ctype = hit | |
| return Response( | |
| content=body, status_code=status, media_type=ctype, | |
| headers={"x-cache": "hit"}, | |
| ) | |
| # Bound concurrency. If saturated, fail fast with 429 rather than queue | |
| # indefinitely (queueing under sustained load is what crashes the worker). | |
| try: | |
| await asyncio.wait_for(_inflight_sem.acquire(), timeout=0.05) | |
| except asyncio.TimeoutError: | |
| return JSONResponse( | |
| status_code=429, | |
| content={"error": "busy", "detail": "Server busy, please retry."}, | |
| headers={"retry-after": "1"}, | |
| ) | |
| try: | |
| response = await call_next(request) | |
| finally: | |
| _inflight_sem.release() | |
| if key is not None and response.status_code == 200: | |
| # Buffer the streaming body so we can both cache and return it. | |
| body_chunks = [chunk async for chunk in response.body_iterator] | |
| body = b"".join(body_chunks) | |
| ctype = response.headers.get("content-type", "application/json") | |
| _cache_put(key, response.status_code, body, ctype) | |
| return Response( | |
| content=body, status_code=response.status_code, media_type=ctype, | |
| headers={"x-cache": "miss"}, | |
| ) | |
| return response | |
| # ============================================================================= | |
| # Middleware: request logging + unhandled-exception handler | |
| # ============================================================================= | |
| async def log_requests(request: Request, call_next): | |
| start = time.monotonic() | |
| try: | |
| response = await call_next(request) | |
| except Exception: | |
| elapsed = (time.monotonic() - start) * 1000 | |
| logger.error( | |
| "UNHANDLED %s %s — %.1f ms\n%s", | |
| request.method, request.url.path, elapsed, | |
| traceback.format_exc(), | |
| ) | |
| return JSONResponse( | |
| status_code=500, | |
| content={"error": "internal_server_error", "detail": "An unexpected error occurred."}, | |
| ) | |
| elapsed = (time.monotonic() - start) * 1000 | |
| logger.info( | |
| "%s %s %d %.1f ms", | |
| request.method, request.url.path, response.status_code, elapsed, | |
| ) | |
| return response | |
| # ============================================================================= | |
| # Startup: configure DuckDB with httpfs views or local tables | |
| # ============================================================================= | |
| db = None | |
| registry = None | |
| def _resolve_parquet(filename: str) -> str: | |
| """Return remote URL or local path for a Parquet file.""" | |
| local_path = os.path.join(DATA_DIR, filename) | |
| if os.path.exists(local_path): | |
| return local_path | |
| return f"{HF_BASE_URL}/{filename}" | |
| def startup(): | |
| global db, registry | |
| # Ensure DuckDB temp directory exists before connecting. | |
| os.makedirs(DUCKDB_TEMP_DIR, exist_ok=True) | |
| # Initialize DuckDB with resource limits suitable for CPU Basic. | |
| db = duckdb.connect(":memory:") | |
| db.execute("INSTALL httpfs; LOAD httpfs;") | |
| db.execute(f"SET memory_limit='{DUCKDB_MEMORY_LIMIT}';") | |
| db.execute(f"SET threads={DUCKDB_THREADS};") | |
| db.execute(f"SET temp_directory='{DUCKDB_TEMP_DIR}';") | |
| db.execute("SET preserve_insertion_order=false;") | |
| # Resolve data sources (remote or local) | |
| results_src = _resolve_parquet("benchmark_results.parquet") | |
| features_src = _resolve_parquet("feature_importance.parquet") | |
| is_remote = results_src.startswith("http") | |
| mode = "remote httpfs" if is_remote else "local files" | |
| logger.info( | |
| "DuckDB settings — memory_limit=%s threads=%d temp_dir=%s mode=%s", | |
| DUCKDB_MEMORY_LIMIT, DUCKDB_THREADS, DUCKDB_TEMP_DIR, mode, | |
| ) | |
| # Load both tables into memory for reliable, fast queries. | |
| # Total ~100MB in RAM — well within cpu-basic 2GB limit. | |
| try: | |
| db.execute(f""" | |
| CREATE TABLE results AS | |
| SELECT * FROM read_parquet('{results_src}') | |
| """) | |
| except Exception as exc: | |
| logger.error("STARTUP FAILED: could not load results table: %s", exc) | |
| raise RuntimeError(f"Failed to load results table: {exc}") from exc | |
| try: | |
| db.execute(f""" | |
| CREATE TABLE features AS | |
| SELECT * FROM read_parquet('{features_src}') | |
| """) | |
| except Exception as exc: | |
| logger.error("STARTUP FAILED: could not load features table: %s", exc) | |
| raise RuntimeError(f"Failed to load features table: {exc}") from exc | |
| n_results = db.execute("SELECT COUNT(*) FROM results").fetchone()[0] | |
| n_features = db.execute("SELECT COUNT(*) FROM features").fetchone()[0] | |
| logger.info("Tables loaded: %d results, %d features", n_results, n_features) | |
| # Load registry JSON (small file, always downloaded) | |
| registry_local = os.path.join(DATA_DIR, "benchmark_registry.json") | |
| if os.path.exists(registry_local): | |
| with open(registry_local) as f: | |
| registry = json.load(f) | |
| else: | |
| # Download from HF | |
| registry_url = f"{HF_BASE_URL}/benchmark_registry.json" | |
| try: | |
| with urllib.request.urlopen(registry_url) as resp: | |
| registry = json.loads(resp.read().decode()) | |
| logger.info("Downloaded registry from %s", registry_url) | |
| except Exception as e: | |
| logger.warning("Could not load registry: %s", e) | |
| registry = {"project": {}, "blood_platforms": {}, | |
| "brain_targets": {}, "phases": {}, "models": {}} | |
| def shutdown(): | |
| global db | |
| if db is not None: | |
| try: | |
| db.close() | |
| logger.info("DuckDB connection closed.") | |
| except Exception as exc: | |
| logger.warning("Error closing DuckDB connection: %s", exc) | |
| db = None | |
| # ============================================================================= | |
| # Helpers | |
| # ============================================================================= | |
| VALID_METRICS = {"r2", "pearson", "mse", "mae"} | |
| def q(sql: str, params=None): | |
| """Run a query on an isolated DuckDB cursor. | |
| DuckDB Connection objects are NOT safe for concurrent use across | |
| requests — sharing the global ``db`` connection between overlapping | |
| FastAPI requests causes corrupted result state and hung responses. | |
| Cursors created via ``db.cursor()`` are cheap, isolated, and safe | |
| for concurrent queries against the same in-memory database. | |
| """ | |
| cur = db.cursor() | |
| return cur.execute(sql, params) if params is not None else cur.execute(sql) | |
| def validate_metric(metric: str) -> str: | |
| if metric not in VALID_METRICS: | |
| raise HTTPException(400, f"Invalid metric: {metric}. " | |
| f"Must be one of {VALID_METRICS}") | |
| return metric | |
| def covariate_phase(base_phase: str, covariates: str) -> str: | |
| """Map covariates param to actual phase name.""" | |
| if covariates == "none": | |
| return base_phase | |
| elif covariates == "only": | |
| return f"{base_phase}_covonly" | |
| elif covariates == "included": | |
| return f"{base_phase}_withcov" | |
| return base_phase | |
| # ============================================================================= | |
| # Endpoints | |
| # ============================================================================= | |
| def get_registry(): | |
| """Full registry metadata.""" | |
| return registry | |
| def maxn_heatmap( | |
| metric: str = Query("r2", description="Metric to aggregate"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Blood x brain heatmap data. | |
| Returns median metric across modules for each combination. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT blood_platform, brain_target, | |
| MEDIAN({metric}) as value, | |
| MEDIAN(n_samples) as n_samples, | |
| COUNT(*) as n_targets | |
| FROM results | |
| WHERE phase = ? AND model = ? | |
| GROUP BY blood_platform, brain_target | |
| ORDER BY blood_platform, brain_target | |
| """, [phase, model]).fetchall() | |
| if not rows: | |
| return {"rows": [], "cols": [], "values": [], "n_samples": [], | |
| "metric": metric, "covariates": covariates} | |
| # Build matrix | |
| blood_set = sorted(set(r[0] for r in rows)) | |
| brain_set = sorted(set(r[1] for r in rows)) | |
| values = [[None] * len(brain_set) for _ in range(len(blood_set))] | |
| n_samples = [[None] * len(brain_set) for _ in range(len(blood_set))] | |
| blood_idx = {b: i for i, b in enumerate(blood_set)} | |
| brain_idx = {b: i for i, b in enumerate(brain_set)} | |
| for blood, brain, val, ns, nt in rows: | |
| i = blood_idx[blood] | |
| j = brain_idx[brain] | |
| values[i][j] = round(val, 4) if val is not None else None | |
| n_samples[i][j] = int(ns) if ns is not None else None | |
| return { | |
| "rows": blood_set, | |
| "cols": brain_set, | |
| "values": values, | |
| "n_samples": n_samples, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "model": model, | |
| } | |
| def maxn_detail( | |
| blood: str = Query(..., description="Blood platform name"), | |
| brain: str = Query(..., description="Brain target name"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Per-module results for one blood x brain combination.""" | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(""" | |
| SELECT target, n_samples, r2, pearson, mse, mae | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| AND model = ? | |
| ORDER BY r2 DESC | |
| """, [phase, blood, brain, model]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "covariates": covariates, | |
| "model": model, | |
| "targets": [ | |
| {"target": r[0], "n_samples": r[1], "r2": r[2], | |
| "pearson": r[3], "mse": r[4], "mae": r[5]} | |
| for r in rows | |
| ], | |
| } | |
| def h2h_blood( | |
| pair: str = Query(..., description="Platform pair, e.g. SomaScan_vs_TMT"), | |
| brain: str = Query(..., description="Brain target name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Per-module comparison for a blood platform pair on one brain target.""" | |
| validate_metric(metric) | |
| phase = "h2h_blood_withcov" if covariates == "included" else "h2h_blood" | |
| rows = q(f""" | |
| SELECT target, blood_platform, {metric} | |
| FROM results | |
| WHERE phase = ? AND h2h_pair = ? AND brain_target = ? AND model = ? | |
| ORDER BY target | |
| """, [phase, pair, brain, model]).fetchall() | |
| # Pivot: target -> {platform_a: val, platform_b: val} | |
| platforms = sorted(set(r[1] for r in rows)) | |
| targets = {} | |
| for target, platform, val in rows: | |
| if target not in targets: | |
| targets[target] = {} | |
| targets[target][platform] = round(val, 4) if val is not None else None | |
| return { | |
| "pair": pair, | |
| "brain": brain, | |
| "platforms": platforms, | |
| "metric": metric, | |
| "targets": [ | |
| {"target": t, **vals} | |
| for t, vals in sorted(targets.items()) | |
| ], | |
| } | |
| def h2h_blood_summary( | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Win counts across all blood H2H pairs.""" | |
| validate_metric(metric) | |
| phase = "h2h_blood_withcov" if covariates == "included" else "h2h_blood" | |
| rows = q(f""" | |
| SELECT h2h_pair, brain_target, target, blood_platform, {metric} | |
| FROM results | |
| WHERE phase = ? AND model = ? AND h2h_pair IS NOT NULL | |
| ORDER BY h2h_pair, brain_target, target | |
| """, [phase, model]).fetchall() | |
| # Count wins per pair | |
| pair_wins = {} | |
| current = None | |
| buffer = {} | |
| for pair, brain, target, platform, val in rows: | |
| key = (pair, brain, target) | |
| if key != current: | |
| if current and len(buffer) == 2: | |
| pair_key = current[0] | |
| if pair_key not in pair_wins: | |
| pair_wins[pair_key] = {} | |
| platforms = list(buffer.keys()) | |
| v0, v1 = buffer[platforms[0]], buffer[platforms[1]] | |
| if v0 is not None and v1 is not None: | |
| asc = metric in ("mse", "mae") | |
| winner = platforms[0] if (v0 < v1 if asc else v0 > v1) else platforms[1] | |
| pair_wins[pair_key][winner] = pair_wins[pair_key].get(winner, 0) + 1 | |
| current = key | |
| buffer = {} | |
| buffer[platform] = val | |
| # Process last group | |
| if current and len(buffer) == 2: | |
| pair_key = current[0] | |
| if pair_key not in pair_wins: | |
| pair_wins[pair_key] = {} | |
| platforms = list(buffer.keys()) | |
| v0, v1 = buffer[platforms[0]], buffer[platforms[1]] | |
| if v0 is not None and v1 is not None: | |
| asc = metric in ("mse", "mae") | |
| winner = platforms[0] if (v0 < v1 if asc else v0 > v1) else platforms[1] | |
| pair_wins[pair_key][winner] = pair_wins[pair_key].get(winner, 0) + 1 | |
| return { | |
| "metric": metric, | |
| "covariates": covariates, | |
| "pairs": pair_wins, | |
| } | |
| def h2h_brain( | |
| blood: str = Query(..., description="Blood platform name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Compare brain targets for one blood platform.""" | |
| validate_metric(metric) | |
| phase = "h2h_brain_withcov" if covariates == "included" else "h2h_brain" | |
| rows = q(f""" | |
| SELECT h2h_pair, brain_target, target, {metric} | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND model = ? | |
| ORDER BY h2h_pair, target | |
| """, [phase, blood, model]).fetchall() | |
| # Group by pair | |
| pairs = {} | |
| for pair, brain, target, val in rows: | |
| if pair not in pairs: | |
| pairs[pair] = {} | |
| if target not in pairs[pair]: | |
| pairs[pair][target] = {} | |
| pairs[pair][target][brain] = round(val, 4) if val is not None else None | |
| return { | |
| "blood": blood, | |
| "metric": metric, | |
| "pairs": pairs, | |
| } | |
| def h2h_models( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| ): | |
| """Compare models for one blood x brain combination (future).""" | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT target, model, {metric} | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| ORDER BY target, model | |
| """, [phase, blood, brain]).fetchall() | |
| models = sorted(set(r[1] for r in rows)) | |
| targets = {} | |
| for target, model, val in rows: | |
| if target not in targets: | |
| targets[target] = {} | |
| targets[target][model] = round(val, 4) if val is not None else None | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "models": models, | |
| "metric": metric, | |
| "targets": [ | |
| {"target": t, **vals} | |
| for t, vals in sorted(targets.items()) | |
| ], | |
| } | |
| def temporal( | |
| target: Optional[str] = Query(None, description="Specific target (e.g., gpath)"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """Temporal decay results: metric across time bins.""" | |
| validate_metric(metric) | |
| phase = "temporal_withcov" if covariates == "included" else "temporal" | |
| if target: | |
| rows = q(f""" | |
| SELECT temporal_bin, brain_target, target, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND model = ? AND target = ? | |
| ORDER BY temporal_bin | |
| """, [phase, model, target]).fetchall() | |
| else: | |
| rows = q(f""" | |
| SELECT temporal_bin, brain_target, target, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND model = ? | |
| ORDER BY temporal_bin, target | |
| """, [phase, model]).fetchall() | |
| results = [ | |
| {"bin": r[0], "brain_target": r[1], "target": r[2], | |
| "value": round(r[3], 4) if r[3] is not None else None, | |
| "n_samples": r[4]} | |
| for r in rows | |
| ] | |
| return { | |
| "metric": metric, | |
| "covariates": covariates, | |
| "results": results, | |
| } | |
| def feature_importance( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| target: str = Query(...), | |
| phase: str = Query("maxn"), | |
| model: str = Query("TabPFN"), | |
| limit: int = Query(30, ge=1, le=100), | |
| ): | |
| """Top features for a specific module/target.""" | |
| rows = q(""" | |
| SELECT feature_name, importance, rank | |
| FROM features | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? | |
| AND target = ? AND model = ? | |
| ORDER BY rank | |
| LIMIT ? | |
| """, [phase, blood, brain, target, model, limit]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "target": target, | |
| "phase": phase, | |
| "features": [ | |
| {"feature": r[0], "importance": round(r[1], 4), "rank": r[2]} | |
| for r in rows | |
| ], | |
| } | |
| def feature_cross_target( | |
| blood: str = Query(...), | |
| brain: str = Query(...), | |
| phase: str = Query("maxn"), | |
| model: str = Query("TabPFN"), | |
| limit: int = Query(30, ge=1, le=100), | |
| ): | |
| """Aggregated feature importance across all modules in a brain target.""" | |
| rows = q(""" | |
| SELECT feature_name, | |
| AVG(importance) as mean_importance, | |
| COUNT(DISTINCT target) as n_targets, | |
| MIN(rank) as best_rank | |
| FROM features | |
| WHERE phase = ? AND blood_platform = ? AND brain_target = ? AND model = ? | |
| GROUP BY feature_name | |
| ORDER BY mean_importance DESC | |
| LIMIT ? | |
| """, [phase, blood, brain, model, limit]).fetchall() | |
| return { | |
| "blood": blood, | |
| "brain": brain, | |
| "phase": phase, | |
| "features": [ | |
| {"feature": r[0], "mean_importance": round(r[1], 4), | |
| "n_targets": r[2], "best_rank": r[3]} | |
| for r in rows | |
| ], | |
| } | |
| def maxn_panel_summary( | |
| blood: str = Query(..., description="Blood platform name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Summary of one predictor panel across all outcome panels. | |
| Returns mean, min, max, n_targets, and median n_samples per outcome. | |
| Replaces N calls to /maxn/detail with a single query. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT brain_target, | |
| AVG({metric}) as mean_val, | |
| MIN({metric}) as min_val, | |
| MAX({metric}) as max_val, | |
| COUNT(*) as n_targets, | |
| MEDIAN(n_samples) as n_samples | |
| FROM results | |
| WHERE phase = ? AND blood_platform = ? AND model = ? | |
| GROUP BY brain_target | |
| ORDER BY mean_val DESC | |
| """, [phase, blood, model]).fetchall() | |
| return { | |
| "blood": blood, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "outcomes": [ | |
| { | |
| "brain_target": r[0], | |
| "mean": round(r[1], 4) if r[1] is not None else None, | |
| "min": round(r[2], 4) if r[2] is not None else None, | |
| "max": round(r[3], 4) if r[3] is not None else None, | |
| "n_targets": r[4], | |
| "n_samples": int(r[5]) if r[5] is not None else None, | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def maxn_outcome_summary( | |
| brain: str = Query(..., description="Brain target name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Summary of one outcome panel across all predictor panels. | |
| Returns mean, min, max, n_targets, and median n_samples per predictor. | |
| Replaces N calls to /maxn/detail with a single query. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT blood_platform, | |
| AVG({metric}) as mean_val, | |
| MIN({metric}) as min_val, | |
| MAX({metric}) as max_val, | |
| COUNT(*) as n_targets, | |
| MEDIAN(n_samples) as n_samples | |
| FROM results | |
| WHERE phase = ? AND brain_target = ? AND model = ? | |
| GROUP BY blood_platform | |
| ORDER BY mean_val DESC | |
| """, [phase, brain, model]).fetchall() | |
| return { | |
| "brain": brain, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "predictors": [ | |
| { | |
| "blood_platform": r[0], | |
| "mean": round(r[1], 4) if r[1] is not None else None, | |
| "min": round(r[2], 4) if r[2] is not None else None, | |
| "max": round(r[3], 4) if r[3] is not None else None, | |
| "n_targets": r[4], | |
| "n_samples": int(r[5]) if r[5] is not None else None, | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def maxn_target_comparison( | |
| brain: str = Query(..., description="Brain target name"), | |
| target: str = Query(..., description="Specific target/module name"), | |
| metric: str = Query("r2"), | |
| covariates: str = Query("none", enum=["none", "only", "included"]), | |
| model: str = Query("TabPFN"), | |
| ): | |
| """ | |
| Per-predictor performance for a specific target within an outcome panel. | |
| E.g., how each blood platform predicts module m4 of bulkrnaseq_dlpfc. | |
| Replaces N calls to /maxn/detail with a single query. | |
| """ | |
| validate_metric(metric) | |
| phase = covariate_phase("maxn", covariates) | |
| rows = q(f""" | |
| SELECT blood_platform, {metric}, n_samples | |
| FROM results | |
| WHERE phase = ? AND brain_target = ? AND target = ? AND model = ? | |
| ORDER BY {metric} DESC | |
| """, [phase, brain, target, model]).fetchall() | |
| return { | |
| "brain": brain, | |
| "target": target, | |
| "metric": metric, | |
| "covariates": covariates, | |
| "predictors": [ | |
| { | |
| "blood_platform": r[0], | |
| "value": round(r[1], 4) if r[1] is not None else None, | |
| "n_samples": r[2], | |
| } | |
| for r in rows | |
| ], | |
| } | |
| def health(): | |
| """Health check — returns row counts or degraded status if DB unavailable.""" | |
| if db is None: | |
| return JSONResponse( | |
| status_code=503, | |
| content={"status": "degraded", "detail": "Database not initialized"}, | |
| ) | |
| try: | |
| n_results = q("SELECT COUNT(*) FROM results").fetchone()[0] | |
| n_features = q("SELECT COUNT(*) FROM features").fetchone()[0] | |
| except Exception as exc: | |
| logger.error("Health check query failed: %s", exc) | |
| return JSONResponse( | |
| status_code=503, | |
| content={"status": "degraded", "detail": "Database query failed"}, | |
| ) | |
| return {"status": "ok", "n_results": n_results, "n_features": n_features} | |
| def ready(): | |
| """Readiness probe — verifies DB connectivity with a trivial query.""" | |
| if db is None: | |
| return JSONResponse( | |
| status_code=503, | |
| content={"ready": False, "detail": "Database not initialized"}, | |
| ) | |
| try: | |
| q("SELECT 1").fetchone() | |
| except Exception as exc: | |
| logger.error("Ready check query failed: %s", exc) | |
| return JSONResponse( | |
| status_code=503, | |
| content={"ready": False, "detail": "Database query failed"}, | |
| ) | |
| return {"ready": True} | |