Spaces:
Sleeping
Sleeping
File size: 4,420 Bytes
536551b | 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 | 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")
|