develop

src/api/__init__.py

@@ -4,8 +4,10 @@ REST API для сервиса оценки качества
 
 from .endpoints import router as endpoints
 from .root import router as root
+from .annotation import router as annotation
 
 __all__ = [
     "endpoints",
-    "root"
+    "root",
+    "annotation"
 ]

src/api/annotation.py

@@ -0,0 +1,224 @@
+# src/api/annotation_endpoints.py
+from fastapi import APIRouter, File, UploadFile, HTTPException, Body
+from fastapi.responses import JSONResponse, FileResponse
+from pydantic import BaseModel
+from typing import List, Dict, Any, Optional
+import json
+from pathlib import Path
+import shutil
+import uuid
+from datetime import datetime
+import base64
+from PIL import Image
+import io
+
+router = APIRouter(prefix="/api/annotation", tags=["annotation"])
+
+# Конфигурация
+ANNOTATION_DIR = Path("data/annotations")
+ANNOTATION_DIR.mkdir(parents=True, exist_ok=True)
+
+# Хранилище сессий разметки
+sessions = {}
+
+
+class AnnotationPoint(BaseModel):
+    x: int
+    y: int
+
+
+class AnnotationPolygon(BaseModel):
+    points: List[AnnotationPoint]
+    label: str = "bone"
+    class_id: int = 1
+
+
+class AnnotationData(BaseModel):
+    image_id: str
+    polygons: List[AnnotationPolygon]
+    classification: Optional[Dict[str, Any]] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class AnnotationSession:
+    """Сессия разметки для одного пользователя"""
+
+    def __init__(self, session_id: str):
+        self.session_id = session_id
+        self.current_index = 0
+        self.images = []
+        self.annotations = {}
+        self.created_at = datetime.now()
+
+    def load_images(self, image_paths: List[str]):
+        self.images = image_paths
+
+    def get_current_image(self):
+        if self.current_index < len(self.images):
+            return self.images[self.current_index]
+        return None
+
+    def next_image(self):
+        self.current_index += 1
+        return self.get_current_image()
+
+    def prev_image(self):
+        self.current_index -= 1
+        if self.current_index < 0:
+            self.current_index = 0
+        return self.get_current_image()
+
+    def save_annotation(self, image_path: str, annotation: AnnotationData):
+        self.annotations[image_path] = annotation
+
+    def get_annotation(self, image_path: str):
+        return self.annotations.get(image_path)
+
+
+# Эндпоинты
+
+@router.post("/session/create")
+async def create_session():
+    """Создание новой сессии разметки"""
+    session_id = str(uuid.uuid4())[:8]
+    sessions[session_id] = AnnotationSession(session_id)
+    return {"session_id": session_id}
+
+
+@router.post("/session/{session_id}/load")
+async def load_images(session_id: str, image_paths: List[str] = Body(...)):
+    """Загрузка списка изображений в сессию"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    session.load_images(image_paths)
+    return {"status": "ok", "total": len(image_paths)}
+
+
+@router.get("/session/{session_id}/current")
+async def get_current_image(session_id: str):
+    """Получение текущего изображения"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    image_path = session.get_current_image()
+
+    if not image_path:
+        return {"status": "complete", "message": "All images annotated"}
+
+    # Читаем изображение
+    if Path(image_path).exists():
+        image = Image.open(image_path)
+        buffered = io.BytesIO()
+        image.save(buffered, format="PNG")
+        img_base64 = base64.b64encode(buffered.getvalue()).decode()
+
+        # Получаем существующую аннотацию
+        annotation = session.get_annotation(image_path)
+
+        return {
+            "status": "ok",
+            "image": {
+                "path": image_path,
+                "data": f"data:image/png;base64,{img_base64}",
+                "index": session.current_index,
+                "total": len(session.images)
+            },
+            "annotation": annotation.dict() if annotation else None
+        }
+
+    return {"status": "error", "message": "Image not found"}
+
+
+@router.post("/session/{session_id}/save")
+async def save_annotation(session_id: str, annotation: AnnotationData):
+    """Сохранение разметки для текущего изображения"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    image_path = session.get_current_image()
+
+    if not image_path:
+        raise HTTPException(status_code=400, detail="No current image")
+
+    session.save_annotation(image_path, annotation)
+
+    # Сохраняем в файл
+    save_path = ANNOTATION_DIR / f"{Path(image_path).stem}_annotation.json"
+    with open(save_path, 'w') as f:
+        json.dump(annotation.dict(), f, indent=2)
+
+    return {"status": "ok", "saved": str(save_path)}
+
+
+@router.post("/session/{session_id}/next")
+async def next_image(session_id: str):
+    """Переход к следующему изображению"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    session.next_image()
+
+    # Получаем следующее изображение
+    return await get_current_image(session_id)
+
+
+@router.post("/session/{session_id}/prev")
+async def prev_image(session_id: str):
+    """Переход к предыдущему изображению"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    session.prev_image()
+
+    return await get_current_image(session_id)
+
+
+@router.post("/session/{session_id}/export")
+async def export_dataset(session_id: str, format: str = "coco"):
+    """Экспорт датасета в нужном формате"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+
+    if format == "coco":
+        return await export_coco(session)
+    elif format == "yolo":
+        return await export_yolo(session)
+    else:
+        raise HTTPException(status_code=400, detail="Unsupported format")
+
+
+async def export_coco(session: AnnotationSession):
+    """Экспорт в COCO формат"""
+    # TODO: Реализовать экспорт в COCO
+    return {"status": "ok", "format": "coco"}
+
+
+async def export_yolo(session: AnnotationSession):
+    """Экспорт в YOLO формат"""
+    # TODO: Реализовать экспорт в YOLO
+    return {"status": "ok", "format": "yolo"}
+
+
+@router.get("/session/{session_id}/progress")
+async def get_progress(session_id: str):
+    """Получение прогресса разметки"""
+    if session_id not in sessions:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    session = sessions[session_id]
+    annotated = len(session.annotations)
+    total = len(session.images)
+
+    return {
+        "annotated": annotated,
+        "total": total,
+        "progress": (annotated / total * 100) if total > 0 else 0
+    }

src/main.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 from starlette.responses import JSONResponse
 
-from src.api.endpoints import router
+from src.api import endpoints, root, annotation
 
 # Создаем приложение
 app = FastAPI(
@@ -24,7 +24,9 @@ if static_dir.exists():
 else:
     print(f"⚠️ Папка static не найдена: {static_dir}")
 
-app.include_router(router)
+app.include_router(endpoints)
+app.include_router(root)
+app.include_router(annotation)
 
 # Глобальный обработчик 404
 @app.exception_handler(404)