Spaces:
Sleeping
Sleeping
File size: 6,790 Bytes
66006d5 ac7ea7c 66006d5 ac7ea7c 66006d5 214c544 66006d5 ac7ea7c 66006d5 ac7ea7c 66006d5 bec7397 66006d5 669f7f8 66006d5 bec7397 66006d5 bec7397 66006d5 bec7397 66006d5 ac7ea7c 66006d5 bec7397 66006d5 | 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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | """Async detection job API (FR-04)."""
from __future__ import annotations
import json
import logging
from typing import Optional
from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request
from sqlalchemy.orm import Session
from ..database import get_db
from ..models import DetectionRun, User
from .dda_auth import current_dda_user
from .job_runner import (
create_local_folder_job,
enqueue_detection_job,
is_job_runner_busy,
job_to_dict,
)
from .local_routes import safe_resolve
from .models import DetectionJob
logger = logging.getLogger(__name__)
router = APIRouter()
def _require_dda():
from .config import IS_DDA_MODE
if not IS_DDA_MODE:
raise HTTPException(status_code=404, detail="DDA mode is not enabled")
@router.post("/jobs")
async def create_job(
base_path: str = Form(...),
comparison_path: str = Form(...),
method: str = Form("AI-Based Deep Learning"),
title: str = Form(""),
zone: str = Form(""),
village: str = Form(""),
enable_registration: bool = Form(True),
enable_normalization: bool = Form(True),
detection_sensitivity: float = Form(0.45),
min_region_area: Optional[int] = Form(150),
notify_email: Optional[str] = Form(None),
db: Session = Depends(get_db),
user: User = Depends(current_dda_user),
):
"""Queue async detection from local library paths. Returns immediately with jobId."""
_require_dda()
base_norm = base_path.replace("\\", "/").strip()
comp_norm = comparison_path.replace("\\", "/").strip()
if not base_norm or not comp_norm:
raise HTTPException(status_code=400, detail="base_path and comparison_path are required")
if base_norm == comp_norm:
raise HTTPException(status_code=400, detail="Base and comparison images must be different")
try:
safe_resolve(base_norm)
safe_resolve(comp_norm)
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=400, detail=f"Invalid library path: {exc}") from exc
if is_job_runner_busy():
raise HTTPException(
status_code=409,
detail="Another detection job is already running. Wait for it to finish, then try again.",
)
if not title.strip():
from pathlib import Path
title = f"{Path(base_norm).name} vs {Path(comp_norm).name}"
job = create_local_folder_job(
db,
base_path=base_norm,
comparison_path=comp_norm,
method=method,
title=title,
zone=zone,
village=village,
enable_registration=enable_registration,
enable_normalization=enable_normalization,
detection_sensitivity=detection_sensitivity,
min_region_area=min_region_area,
notify_email=notify_email or "",
created_by=user.id,
)
if not enqueue_detection_job(job.id):
job.status = "failed"
job.error_message = "Could not start background worker"
db.commit()
raise HTTPException(status_code=503, detail="Job queue is busy")
return {"jobId": job.id, "status": "queued", "message": "Detection job queued. Poll GET /api/dda/jobs/{id} for status."}
@router.get("/jobs/{job_id}")
def get_job(job_id: int, db: Session = Depends(get_db), user: User = Depends(current_dda_user)):
_require_dda()
job = db.query(DetectionJob).filter(DetectionJob.id == job_id).first()
if not job:
raise HTTPException(status_code=404, detail="Job not found")
if job.created_by and job.created_by != user.id:
raise HTTPException(status_code=403, detail="Not allowed to view this job")
run = None
if job.run_id:
run = db.query(DetectionRun).filter(DetectionRun.id == job.run_id).first()
data = job_to_dict(job, run=run)
if job.status == "completed" and run:
try:
data["result"] = _run_detail(db, run, user.id)
except HTTPException:
raise
except Exception as exc:
logger.warning("Could not load full run for job %s: %s", job_id, exc)
data["resultError"] = str(exc)[:500]
return data
def _run_detail(db: Session, run: DetectionRun, user_id: int) -> dict:
import base64
from ..database import DATA_DIR
if run.user_id != user_id:
raise HTTPException(status_code=403, detail="Not allowed")
regions = json.loads(run.regions_json or "[]")
from .review_service import merge_reviews
regions = merge_reviews(db, run.id, regions)
overlay_b64 = ""
if run.overlay_path:
overlay_file = DATA_DIR / run.overlay_path
if overlay_file.exists():
overlay_b64 = base64.b64encode(overlay_file.read_bytes()).decode("utf-8")
from .config import get_detection_max_side
from .detect_service import _isoformat_ist
return {
"id": run.id,
"title": run.title,
"method": run.method,
"zone": run.zone or "",
"village": run.village or "",
"statistics": {
"totalPixels": run.total_pixels,
"changedPixels": run.changed_pixels,
"unchangedPixels": run.total_pixels - run.changed_pixels,
"changePercentage": run.change_percentage,
},
"regions": regions,
"overlayBase64Png": overlay_b64,
"overlayUrl": f"/api/overlay/{run.overlay_path}" if run.overlay_path else None,
"beforeFullUrl": f"/api/overlay/{run.before_full_path}" if run.before_full_path else None,
"beforeThumbUrl": f"/api/overlay/{run.before_thumb_path}" if run.before_thumb_path else None,
"afterThumbUrl": f"/api/overlay/{run.after_thumb_path}" if run.after_thumb_path else None,
"afterFullUrl": f"/api/overlay/{run.after_full_path}" if getattr(run, "after_full_path", None) else None,
"createdAt": _isoformat_ist(run.created_at),
"detectionMaxSide": get_detection_max_side(),
}
@router.get("/jobs")
def list_jobs(
status: Optional[str] = Query(None),
limit: int = Query(20, ge=1, le=100),
db: Session = Depends(get_db),
user: User = Depends(current_dda_user),
):
"""Recent detection jobs for in-app notifications / reports feed (FR-05 partial)."""
_require_dda()
from .job_runner import reconcile_stale_jobs
reconcile_stale_jobs(db)
q = db.query(DetectionJob).filter(DetectionJob.created_by == user.id)
if status:
q = q.filter(DetectionJob.status == status)
jobs = q.order_by(DetectionJob.created_at.desc()).limit(limit).all()
out = []
for job in jobs:
run = db.query(DetectionRun).filter(DetectionRun.id == job.run_id).first() if job.run_id else None
out.append(job_to_dict(job, run=run))
return {"jobs": out, "runnerBusy": is_job_runner_busy()}
|