BIDS Neuroimaging Viewer

from fastapi import FastAPI from fastapi.responses import HTMLResponse from datasets import load_dataset import base64 import asyncio from contextlib import asynccontextmanager ds_iter = None first_sample = None nifti_base_url = "data:application/octet-stream;base64,{}" initial_metadata = {} num_iterations = 0 async def load_dataset_async(): global ds_iter dataset = load_dataset("TobiasPitters/ds004884-mini", streaming=True) ds_iter = iter(dataset["train"]) @asynccontextmanager async def lifespan(app: FastAPI): global first_sample, initial_metadata asyncio.create_task(load_dataset_async()) yield app = FastAPI(title="BIDS Neuroimaging Viewer", lifespan=lifespan) @app.get("/", response_class=HTMLResponse) async def root(): """Main page with NiiVue viewer""" html_content = f""" BIDS Neuroimaging Viewer

Loading NiiVue...

Controls:

Slice views (3 panels): Click to move crosshair, drag to pan, scroll to zoom
3D view (bottom-right): Left-click drag to rotate, scroll to zoom

Sample Info

""" return html_content @app.get("/initial") async def get_initial(): print("Loading dataset...") # dataset = load_dataset("TobiasPitters/ds004884-mini", streaming=True) global ds_iter dataset = load_dataset("parquet", data_dir="./data/" ,streaming=True) ds_iter = iter(dataset["train"]) first_sample = next(ds_iter) if isinstance(first_sample['nifti'], dict): nifti_bytes = first_sample['nifti']['bytes'] else: nifti_bytes = first_sample['nifti'].to_bytes() nifti_b64 = base64.b64encode(nifti_bytes).decode("utf-8") data_url = nifti_base_url.format(nifti_b64) initial_metadata = { 'subject': first_sample.get('subject', 'N/A'), 'session': first_sample.get('session', 'N/A'), 'datatype': first_sample.get('datatype', 'N/A'), 'suffix': first_sample.get('suffix', 'N/A'), 'task': first_sample.get('task', 'N/A'), 'run': first_sample.get('run', 'N/A'), 'path': first_sample.get('path', 'N/A') } return { "status": "success", "data_url": data_url, "metadata": initial_metadata } async def load_dataset_iterator(): global num_iterations dataset = load_dataset("TobiasPitters/ds004884-mini", streaming=True) ds_iter_new = iter(dataset["train"]) for _ in range(num_iterations): next(ds_iter_new) global ds_iter ds_iter = ds_iter_new @app.get("/next") async def next_sample(): """Load next sample from dataset""" global ds_iter global num_iterations try: ex = next(ds_iter) if isinstance(ex['nifti'], dict): nifti_bytes = ex['nifti']['bytes'] else: nifti_bytes = ex['nifti'].to_bytes() nifti_b64 = base64.b64encode(nifti_bytes).decode("utf-8") new_data_url = nifti_base_url.format(nifti_b64) metadata = { 'subject': ex.get('subject', 'N/A'), 'session': ex.get('session', 'N/A'), 'datatype': ex.get('datatype', 'N/A'), 'suffix': ex.get('suffix', 'N/A'), 'task': ex.get('task', 'N/A'), 'run': ex.get('run', 'N/A'), 'path': ex.get('path', 'N/A') } num_iterations += 1 return { "status": "success", "data_url": new_data_url, "metadata": metadata } except StopIteration: return {"status": "error", "message": "No more samples in dataset"} @app.get("/health") async def health(): """Health check endpoint""" return {"status": "healthy", "dataset_loaded": True}

🧠 BIDS Neuroimaging Viewer

Controls:

Sample Info