Pathora / backend /tile_server /app /api /routes.py
malavikapradeep2001's picture
Deploy Pathora Viewer: tile server, viewer components, and root app.py (#3)
536551b
raw
history blame
4.42 kB
import os
import shutil
from pathlib import Path
from fastapi import APIRouter, HTTPException, Query, UploadFile, File
from fastapi.responses import Response
import openslide
from app.models.schemas import SlideLoadRequest, SlideMetadata
from app.services.slide_cache import get_slide, load_slide
from app.services.tile_service import get_tile_jpeg, get_thumbnail_jpeg
router = APIRouter()
UPLOAD_DIR = Path(
os.getenv(
"TILE_SERVER_UPLOAD_DIR",
Path(__file__).resolve().parents[2] / "data" / "uploads",
)
)
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
@router.post("/slides/{slide_id}/load")
def load(slide_id: str, payload: SlideLoadRequest):
try:
load_slide(slide_id, payload.path)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"status": "loaded", "slide_id": slide_id}
@router.post("/slides/{slide_id}/upload")
def upload(slide_id: str, file: UploadFile = File(...)):
if not slide_id:
raise HTTPException(status_code=400, detail="Slide id is required")
filename = Path(file.filename or "")
suffix = filename.suffix.lower()
if suffix not in {".svs", ".tif", ".tiff"}:
suffix = ".svs"
dest_path = UPLOAD_DIR / f"{slide_id}{suffix}"
try:
with dest_path.open("wb") as buffer:
shutil.copyfileobj(file.file, buffer)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Upload failed: {exc}")
finally:
try:
file.file.close()
except Exception:
pass
try:
load_slide(slide_id, str(dest_path))
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"status": "uploaded", "slide_id": slide_id}
@router.post("/slides/{slide_id}/reload")
def reload_uploaded(slide_id: str):
if not slide_id:
raise HTTPException(status_code=400, detail="Slide id is required")
matches = list(UPLOAD_DIR.glob(f"{slide_id}.*"))
if not matches:
raise HTTPException(status_code=404, detail="Uploaded slide not found")
try:
load_slide(slide_id, str(matches[0]))
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"status": "reloaded", "slide_id": slide_id}
@router.get("/slides/{slide_id}/metadata", response_model=SlideMetadata)
def metadata(slide_id: str):
slide = get_slide(slide_id)
if slide is None:
raise HTTPException(status_code=404, detail="Slide not loaded")
mpp_x = slide.properties.get(openslide.PROPERTY_NAME_MPP_X)
mpp_y = slide.properties.get(openslide.PROPERTY_NAME_MPP_Y)
def _to_float(value):
try:
return float(value)
except (TypeError, ValueError):
return None
return SlideMetadata(
width=slide.dimensions[0],
height=slide.dimensions[1],
level_count=slide.level_count,
level_dimensions=[[int(w), int(h)] for w, h in slide.level_dimensions],
level_downsamples=[float(d) for d in slide.level_downsamples],
mpp_x=_to_float(mpp_x),
mpp_y=_to_float(mpp_y),
)
@router.get("/tiles/{slide_id}/{level}/{x}/{y}.jpg")
def tile(
slide_id: str,
level: int,
x: int,
y: int,
tile_size: int = Query(256, ge=64, le=2048),
channel: str = Query("original"),
):
slide = get_slide(slide_id)
if slide is None:
raise HTTPException(status_code=404, detail="Slide not loaded")
try:
jpeg_bytes = get_tile_jpeg(slide, level, x, y, tile_size, channel)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
return Response(content=jpeg_bytes, media_type="image/jpeg")
@router.get("/slides/{slide_id}/thumbnail")
def thumbnail(
slide_id: str,
size: int = Query(256, ge=64, le=1024),
channel: str = Query("original"),
):
slide = get_slide(slide_id)
if slide is None:
raise HTTPException(status_code=404, detail="Slide not loaded")
try:
jpeg_bytes = get_thumbnail_jpeg(slide, size, channel)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
return Response(content=jpeg_bytes, media_type="image/jpeg")