Blame for server.py

"""FastAPI server — DiffuMT project page + 2AFC game + DDIM generation."""
from __future__ import annotations

import asyncio
import base64
import io
import json
import os
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

import numpy as np
import torch
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from fastapi.staticfiles import StaticFiles
from PIL import Image
from pydantic import BaseModel

DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
IMAGE_SIZE = 256
IMAGES_DIR = Path(__file__).parent / "images"
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "")

def _resolve_data_file() -> Path:
    """
    Prefer /data/results.json (HF persistent storage) if /data is a writable mount.
    Fall back to /app/results.json (committed to repo, survives restarts within the
    same build, resets on the next Docker rebuild).
    Seeds /data from the repo copy on first boot if persistent storage is now attached.
    """
    explicit = os.environ.get("DATA_PATH")
    if explicit:
        return Path(explicit)
    persistent = Path("/data/results.json")
    bundled    = Path(__file__).parent / "results.json"
    # /data is available if the directory exists and is writable
    if persistent.parent.exists() and os.access(persistent.parent, os.W_OK):
        if not persistent.exists() and bundled.exists():
            import shutil
            shutil.copy(bundled, persistent)
        return persistent
    return bundled

DATA_FILE = _resolve_data_file()
MODEL_CACHE = Path("/data/model") if Path("/data").exists() and os.access("/data", os.W_OK) \
              else Path(__file__).parent / ".model_cache"

model_state: dict = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("lifespan: importing diffusers/huggingface_hub", flush=True)
    from diffusers import DDIMScheduler, UNet2DModel
    from huggingface_hub import snapshot_download
    print("lifespan: imports done, downloading model snapshot", flush=True)

    MODEL_CACHE.mkdir(parents=True, exist_ok=True)
    model_dir = snapshot_download(
        "HTW-KI-Werkstatt/DiffuMT",
        repo_type="model",
        local_dir=str(MODEL_CACHE),
    )
    print(f"lifespan: snapshot at {model_dir}, loading UNet", flush=True)
    unet = (
        UNet2DModel.from_pretrained(f"{model_dir}/unet", use_safetensors=True)
        .to(DEVICE)
        .eval()
    )
    print("lifespan: UNet loaded, loading scheduler", flush=True)
    scheduler = DDIMScheduler.from_pretrained(f"{model_dir}/scheduler")
    model_state["unet"] = unet
    model_state["scheduler"] = scheduler
    print(f"Model loaded on {DEVICE}", flush=True)
    yield


# ── Data store ───────────────────────────────────────────────────────────────

DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
if not DATA_FILE.exists():
    DATA_FILE.write_text(json.dumps({"sessions": [], "trials": []}, indent=2))


def _read() -> dict:
    return json.loads(DATA_FILE.read_text())


def _write(data: dict) -> None:
    DATA_FILE.write_text(json.dumps(data, indent=2))


# ── App ──────────────────────────────────────────────────────────────────────

app = FastAPI(lifespan=lifespan)

IMAGE_EXT = {".jpg", ".jpeg", ".png", ".tiff", ".tif", ".webp", ".bmp", ".gif"}


# ── Game API ─────────────────────────────────────────────────────────────────

@app.get("/api/images")
def list_images():
    def ls(subdir: str):
        d = IMAGES_DIR / subdir
        if not d.exists():
            return []
        return [
            {"id": f.name, "src": f"/images/{subdir}/{f.name}", "type": subdir}
            for f in sorted(d.iterdir())
            if f.suffix.lower() in IMAGE_EXT
        ]
    return {"real": ls("real"), "fake": ls("fake")}


class TrialPayload(BaseModel):
    sessionId: str
    trial: dict


@app.post("/api/trial")
def save_trial(payload: TrialPayload):
    data = _read()
    data["trials"].append({"session_id": payload.sessionId, "player_name": None, **payload.trial})
    _write(data)
    return {"ok": True}


class SubmitPayload(BaseModel):
    sessionId: str
    playerName: str


@app.post("/api/submit")
def submit_score(payload: SubmitPayload):
    data = _read()
    trials = [t for t in data["trials"] if t["session_id"] == payload.sessionId]
    if not trials:
        raise HTTPException(status_code=404, detail="Session not found")
    data["trials"] = [
        {**t, "player_name": payload.playerName} if t["session_id"] == payload.sessionId else t
        for t in data["trials"]
    ]
    score = sum(1 for t in trials if t.get("correct"))
    total = len(trials)
    data["sessions"].append({
        "session_id": payload.sessionId,
        "player_name": payload.playerName,
        "timestamp": trials[0].get("timestamp", ""),
        "score": score,
        "total": total,
        "percentage": round((score / total) * 100) if total else 0,
    })
    _write(data)
    return {"sessionId": payload.sessionId, "score": score, "total": total}


@app.get("/api/leaderboard")
def leaderboard():
    sessions = _read()["sessions"]
    return sorted(sessions, key=lambda s: (-s["percentage"], s["timestamp"]))[:50]


def _check_admin(x_admin_token: str, authorization: str):
    if not ADMIN_TOKEN:
        raise HTTPException(status_code=503, detail="Admin access disabled: ADMIN_TOKEN not set")
    bearer = authorization.removeprefix("Bearer ").strip()
    if bearer != ADMIN_TOKEN and x_admin_token != ADMIN_TOKEN:
        raise HTTPException(status_code=401, detail="Unauthorized")


@app.get("/api/admin/sessions")
def admin_sessions():
    sessions = _read()["sessions"]
    return sorted(sessions, key=lambda s: s["timestamp"], reverse=True)


@app.get("/api/export")
def export_data(x_admin_token: str = Header(default=""), authorization: str = Header(default="")):
    _check_admin(x_admin_token, authorization)
    return Response(
        content=DATA_FILE.read_text(),
        media_type="application/json",
        headers={"Content-Disposition": 'attachment; filename="2afc_data.json"'},
    )


@app.post("/api/admin/clear")
def clear_all(x_admin_token: str = Header(default=""), authorization: str = Header(default="")):
    _check_admin(x_admin_token, authorization)
    _write({"sessions": [], "trials": []})
    return {"ok": True}


class ClearUserPayload(BaseModel):
    playerName: str


@app.post("/api/admin/clear-user")
def clear_user(payload: ClearUserPayload, x_admin_token: str = Header(default=""), authorization: str = Header(default="")):
    _check_admin(x_admin_token, authorization)
    data = _read()
    bs, bt = len(data["sessions"]), len(data["trials"])
    data["sessions"] = [s for s in data["sessions"] if s["player_name"] != payload.playerName]
    data["trials"]   = [t for t in data["trials"]   if t.get("player_name") != payload.playerName]
    _write(data)
    return {"ok": True, "removedSessions": bs - len(data["sessions"]), "removedTrials": bt - len(data["trials"])}


class ClearSessionPayload(BaseModel):
    sessionId: str
    deleteTrials: bool = True


@app.post("/api/admin/clear-session")
def clear_session(payload: ClearSessionPayload, x_admin_token: str = Header(default=""), authorization: str = Header(default="")):
    _check_admin(x_admin_token, authorization)
    data = _read()
    bs, bt = len(data["sessions"]), len(data["trials"])
    data["sessions"] = [s for s in data["sessions"] if s["session_id"] != payload.sessionId]
    if payload.deleteTrials:
        data["trials"] = [t for t in data["trials"] if t.get("session_id") != payload.sessionId]
    _write(data)
    return {"ok": True, "removedSessions": bs - len(data["sessions"]), "removedTrials": bt - len(data["trials"])}


@app.post("/api/admin/import")
async def import_data(request: Request, x_admin_token: str = Header(default=""), authorization: str = Header(default="")):
    _check_admin(x_admin_token, authorization)
    body = await request.json()
    if "sessions" not in body or "trials" not in body:
        raise HTTPException(status_code=400, detail="Body must have 'sessions' and 'trials' keys")
    _write(body)
    return {"ok": True, "sessions": len(body["sessions"]), "trials": len(body["trials"])}


# ── DDIM generation (SSE) ─────────────────────────────────────────────────────

class GeneratePayload(BaseModel):
    mask: str       # base64 PNG from canvas.toDataURL()
    steps: int = 30
    seed: int = -1


def _prep_mask(b64: str) -> torch.Tensor:
    raw = base64.b64decode(b64)
    img = Image.open(io.BytesIO(raw)).convert("L")
    img = img.resize((IMAGE_SIZE, IMAGE_SIZE), Image.NEAREST)
    arr = (np.asarray(img) > 0).astype(np.float32) / 255.0
    return torch.from_numpy(arr)[None, None].to(DEVICE)


def _tensor_to_b64(x: torch.Tensor) -> str:
    arr = (x / 2 + 0.5).clamp(0, 1)
    arr = (arr[0].cpu().permute(1, 2, 0).numpy() * 255).astype(np.uint8)
    buf = io.BytesIO()
    Image.fromarray(arr).save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode()


@app.post("/api/generate")
async def generate(payload: GeneratePayload):
    unet = model_state.get("unet")
    scheduler = model_state.get("scheduler")
    if unet is None:
        raise HTTPException(status_code=503, detail="Model not loaded yet — please wait a moment")

    async def stream():
        try:
            mask = _prep_mask(payload.mask)
            if mask.sum() == 0:
                yield f"data: {json.dumps({'error': 'Empty mask — draw some strokes first'})}\n\n"
                return

            gen = None
            if payload.seed >= 0:
                gen = torch.Generator(device=DEVICE).manual_seed(payload.seed)

            x = torch.randn((1, 3, IMAGE_SIZE, IMAGE_SIZE), generator=gen, device=DEVICE)
            scheduler.set_timesteps(payload.steps)
            timesteps = scheduler.timesteps
            n = len(timesteps)
            snap = set(np.linspace(0, n - 1, min(8, n)).round().astype(int).tolist())

            with torch.no_grad():
                for i, t in enumerate(timesteps):
                    x_in = torch.cat([x, mask], dim=1)
                    eps = unet(x_in, t).sample
                    x = scheduler.step(eps, t, x, generator=gen).prev_sample

                    if i in snap or i == n - 1:
                        msg = json.dumps({
                            "step": i + 1,
                            "total": n,
                            "image": _tensor_to_b64(x),
                            "done": i == n - 1,
                        })
                        yield f"data: {msg}\n\n"
                        await asyncio.sleep(0)

        except Exception as exc:
            yield f"data: {json.dumps({'error': str(exc)})}\n\n"

    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ── Static files (must be last) ───────────────────────────────────────────────

app.mount("/images", StaticFiles(directory=str(IMAGES_DIR)), name="images")
app.mount("/", StaticFiles(directory=str(Path(__file__).parent / "public"), html=True), name="static")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)))