146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import get_db
|
|
from app.schemas import DetectionQaRequest, DetectionRunRequest
|
|
from app.services.detection_service import DetectionService
|
|
from app.services.model_registry_service import ModelRegistryService
|
|
from app.utils.response import envelope
|
|
|
|
router = APIRouter(prefix="/detection", tags=["detection"])
|
|
|
|
|
|
@router.get("/models", response_model=dict)
|
|
def list_detection_models() -> dict:
|
|
return envelope({"models": [model.model_dump() for model in ModelRegistryService.list_model_capabilities()]})
|
|
|
|
|
|
@router.post("/run", response_model=dict)
|
|
def run_detection(payload: DetectionRunRequest, db: Session = Depends(get_db)) -> dict:
|
|
result = DetectionService.run_detection(
|
|
db=db,
|
|
project_id=payload.project_id,
|
|
dataset_id=payload.dataset_id,
|
|
model_id=payload.model_id,
|
|
confidence_threshold=payload.confidence_threshold,
|
|
class_filter=payload.class_filter,
|
|
tile_manifest_path=payload.tile_manifest_path,
|
|
parameters_json=payload.parameters_json,
|
|
)
|
|
return envelope(result.model_dump())
|
|
|
|
|
|
@router.get("/runs", response_model=dict)
|
|
def list_detection_runs(
|
|
project_id: UUID | None = None,
|
|
dataset_id: UUID | None = None,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(DetectionService.list_runs(db, project_id=project_id, dataset_id=dataset_id).model_dump())
|
|
|
|
|
|
@router.get("/runs/{analysis_run_id}", response_model=dict)
|
|
def get_detection_run(analysis_run_id: UUID, db: Session = Depends(get_db)) -> dict:
|
|
return envelope(DetectionService.get_run(db, analysis_run_id).model_dump())
|
|
|
|
|
|
@router.get("/runs/{analysis_run_id}/detections", response_model=dict)
|
|
def list_detection_run_detections(
|
|
analysis_run_id: UUID,
|
|
dataset_id: UUID | None = None,
|
|
class_name: str | None = None,
|
|
min_confidence: float | None = None,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(
|
|
DetectionService.list_detections(
|
|
db,
|
|
analysis_run_id=analysis_run_id,
|
|
dataset_id=dataset_id,
|
|
class_name=class_name,
|
|
min_confidence=min_confidence,
|
|
).model_dump()
|
|
)
|
|
|
|
|
|
@router.get("/datasets/{dataset_id}/detections", response_model=dict)
|
|
def list_dataset_detections(
|
|
dataset_id: UUID,
|
|
analysis_run_id: UUID | None = None,
|
|
class_name: str | None = None,
|
|
min_confidence: float | None = None,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(
|
|
DetectionService.list_detections(
|
|
db,
|
|
analysis_run_id=analysis_run_id,
|
|
dataset_id=dataset_id,
|
|
class_name=class_name,
|
|
min_confidence=min_confidence,
|
|
).model_dump()
|
|
)
|
|
|
|
|
|
@router.get("/detections/{detection_id}", response_model=dict)
|
|
def get_detection(detection_id: UUID, db: Session = Depends(get_db)) -> dict:
|
|
return envelope(DetectionService.get_detection(db, detection_id).model_dump())
|
|
|
|
|
|
@router.get("/runs/{analysis_run_id}/geojson", response_model=dict)
|
|
def get_detection_run_geojson(
|
|
analysis_run_id: UUID,
|
|
class_name: str | None = None,
|
|
min_confidence: float | None = None,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(
|
|
DetectionService.detections_to_geojson(
|
|
db,
|
|
analysis_run_id=analysis_run_id,
|
|
class_name=class_name,
|
|
min_confidence=min_confidence,
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/datasets/{dataset_id}/geojson", response_model=dict)
|
|
def get_dataset_detection_geojson(
|
|
dataset_id: UUID,
|
|
analysis_run_id: UUID | None = None,
|
|
class_name: str | None = None,
|
|
min_confidence: float | None = None,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(
|
|
DetectionService.detections_to_geojson(
|
|
db,
|
|
analysis_run_id=analysis_run_id,
|
|
dataset_id=dataset_id,
|
|
class_name=class_name,
|
|
min_confidence=min_confidence,
|
|
)
|
|
)
|
|
|
|
|
|
@router.post("/runs/{analysis_run_id}/qa/reference", response_model=dict)
|
|
def compare_detection_run_with_reference(
|
|
analysis_run_id: UUID,
|
|
payload: DetectionQaRequest,
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return envelope(
|
|
DetectionService.compare_detections_with_reference(
|
|
db=db,
|
|
analysis_run_id=analysis_run_id,
|
|
reference_dataset_id=payload.reference_dataset_id,
|
|
iou_threshold=payload.iou_threshold,
|
|
class_name=payload.class_name,
|
|
min_confidence=payload.min_confidence,
|
|
)
|
|
)
|