Spaces:
Sleeping
Sleeping
File size: 4,684 Bytes
f8a3ca2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """Data import / sample dataset endpoints."""
from __future__ import annotations
import csv
import io
import json
from pathlib import Path
from fastapi import APIRouter, File, HTTPException, UploadFile
from app.api.schemas import SampleDatasetModel, SampleInfoModel
SAMPLES_DIR = Path(__file__).resolve().parents[1] / "data_sources" / "samples"
router = APIRouter(prefix="/api/data", tags=["data"])
SAMPLE_REGISTRY = {
"gdp": {"file": "sample-gdp.json", "icon": "📊"},
"climate": {"file": "sample-climate.json", "icon": "🌡️"},
"population": {"file": "sample-population.json", "icon": "👥"},
}
def _load(name: str) -> dict:
path = SAMPLES_DIR / SAMPLE_REGISTRY[name]["file"]
if not path.exists():
raise HTTPException(status_code=404, detail=f"sample {name} not found")
return json.loads(path.read_text(encoding="utf-8"))
@router.get("/samples", response_model=list[SampleInfoModel])
async def list_samples() -> list[SampleInfoModel]:
out: list[SampleInfoModel] = []
for sid, meta in SAMPLE_REGISTRY.items():
if sid not in SAMPLE_REGISTRY:
continue
data = _load(sid)
out.append(
SampleInfoModel(
id=sid,
name=data.get("name", sid),
name_en=data.get("name_en"),
source=data.get("source", ""),
unit=data.get("unit", ""),
period=data.get("period", ""),
description=data.get("description", ""),
n=len(data.get("values", [])),
icon=meta["icon"],
)
)
return out
@router.get("/samples/{sid}", response_model=SampleDatasetModel)
async def get_sample(sid: str) -> SampleDatasetModel:
if sid not in SAMPLE_REGISTRY:
raise HTTPException(status_code=404, detail=f"unknown sample id {sid}")
data = _load(sid)
return SampleDatasetModel(
id=sid,
name=data.get("name", sid),
name_en=data.get("name_en"),
source=data.get("source", ""),
unit=data.get("unit", ""),
period=data.get("period", ""),
description=data.get("description", ""),
n=len(data.get("values", [])),
icon=SAMPLE_REGISTRY[sid]["icon"],
values=list(data.get("values", [])),
)
@router.post("/upload")
async def upload_csv(file: UploadFile = File(...)) -> dict:
"""Accept a CSV upload, parse the first numeric column, return values + metadata.
Designed for the data-import workflow; the frontend renders a preview and
feeds the values back into /bayesian/compute.
"""
if file.content_type and "csv" not in file.content_type and not file.filename.endswith(".csv"):
raise HTTPException(status_code=415, detail="please upload a CSV file")
raw = await file.read()
try:
text = raw.decode("utf-8-sig")
except UnicodeDecodeError:
text = raw.decode("latin-1")
reader = csv.reader(io.StringIO(text))
rows = list(reader)
if not rows:
raise HTTPException(status_code=400, detail="empty CSV")
# Detect header
header = rows[0]
has_header = any(not _is_number(c) for c in header)
data_rows = rows[1:] if has_header else rows
# Find first column with at least 50% numeric values
n_cols = max((len(r) for r in data_rows), default=0)
chosen_col = 0
best_score = -1.0
for col in range(n_cols):
nums = sum(1 for r in data_rows if col < len(r) and _is_number(r[col]))
score = nums / max(len(data_rows), 1)
if score > best_score:
best_score = score
chosen_col = col
values: list[float] = []
skipped = 0
for r in data_rows:
if chosen_col >= len(r):
skipped += 1
continue
cell = r[chosen_col].strip()
if not cell:
skipped += 1
continue
try:
values.append(float(cell))
except ValueError:
skipped += 1
if len(values) < 2:
raise HTTPException(
status_code=400,
detail=f"could not parse at least 2 numeric values (got {len(values)})",
)
return {
"filename": file.filename,
"values": values,
"n_total_rows": len(data_rows),
"n_used": len(values),
"n_skipped": skipped,
"chosen_column_index": chosen_col,
"chosen_column_label": header[chosen_col]
if has_header and chosen_col < len(header)
else f"column_{chosen_col}",
}
def _is_number(s: str) -> bool:
try:
float(s.strip())
return True
except (ValueError, AttributeError):
return False
|