comic_panel_extractor/annorator_server.py

from fastapi import APIRouter, HTTPException, UploadFile, File
from fastapi.responses import FileResponse
from pydantic import BaseModel, field_validator
from typing import List
from PIL import Image
import os
import base64
from io import BytesIO
import shutil
from .config import Config
from typing import List, Optional, Union, Dict, Any
from . import utils
import copy
import traceback

app = APIRouter()

# === Configuration ===
IMAGE_ROOT = os.path.join(Config.current_path, "dataset/images")
LABEL_ROOT = os.path.join(Config.current_path, "dataset/labels")
IMAGE_LABEL_ROOT = os.path.join(Config.current_path, "image_labels")

CLASS_ID = 0

# === Pydantic Models ===
class Point(BaseModel):
    x: float
    y: float

class Box(BaseModel):
    type: str = "bbox"  # "bbox" or "segmentation"
    # For bbox
    left: Optional[int] = None
    top: Optional[int] = None
    width: Optional[int] = None
    height: Optional[int] = None
    # For segmentation
    points: Optional[List[Point]] = None
    # Common fields
    classId: int = CLASS_ID
    stroke: str = "#00ff00"
    strokeWidth: int = 3
    fill: str = "rgba(0, 255, 0, 0.2)"
    saved: bool = True

    @field_validator("left", "top", "width", "height", mode="before")
    def round_floats(cls, v):
        return round(v) if v is not None else None

class SaveAnnotationsRequest(BaseModel):
    annotations: List[Box]  # Changed from 'boxes' to 'annotations'
    image_name: str
    original_width: int
    original_height: int

class ImageInfo(BaseModel):
    name: str  # Relative path like train/image1.jpg
    width: int
    height: int
    has_annotations: bool

# === Helpers ===
def get_image_path(image_name: str) -> str:
    return os.path.join(IMAGE_ROOT, image_name)

def get_label_path(image_name: str) -> str:
    return os.path.join(LABEL_ROOT, os.path.splitext(image_name)[0] + ".txt")

# === Core Functions ===
def load_yolo_annotations(image_path: str, label_path: str, detect: bool = False):
    """Load both bbox and segmentation annotations from YOLO format"""
    try:
        img = Image.open(image_path)
        w, h = img.size
        annotations = []

        # Auto-detect if needed
        normalise = False
        if detect and not os.path.exists(label_path):
            from .yolo_manager import YOLOManager
            with YOLOManager() as yolo_manager:
                weights_path = Config.yolo_trained_model_path
                yolo_manager.load_model(weights_path)
                yolo_manager.annotate_images(
                    image_paths=[image_path],
                    output_dir=IMAGE_LABEL_ROOT,
                    save_image=False,
                    label_path=label_path
                )
                normalise = True

        if os.path.exists(label_path):
            with open(label_path, "r") as f:
                for line in f:
                    parts = list(map(float, line.strip().split()))
                    if len(parts) < 5:
                        continue

                    class_id = int(parts[0])

                    if len(parts) == 5:  # Bounding box format
                        _, xc, yc, bw, bh = parts
                        left = int((xc - bw / 2) * w)
                        top = int((yc - bh / 2) * h)
                        width = int(bw * w)
                        height = int(bh * h)

                        annotations.append({
                            "type": "bbox",
                            "left": left,
                            "top": top,
                            "width": width,
                            "height": height,
                            "classId": class_id,
                            "stroke": "#00ff00",
                            "strokeWidth": 3,
                            "fill": "rgba(0, 255, 0, 0.2)",
                            "saved": True
                        })

                    elif len(parts) > 5 and len(parts) % 2 == 1:  # Segmentation format
                        # Skip class_id, then pairs of x,y coordinates
                        coords = parts[1:]
                        if len(coords) >= 6:  # At least 3 points
                            points = []
                            for i in range(0, len(coords), 2):
                                if i + 1 < len(coords):
                                    x = coords[i] * w
                                    y = coords[i + 1] * h
                                    points.append({"x": x, "y": y})

                            annotations.append({
                                "type": "segmentation",
                                "points": points,
                                "classId": class_id,
                                "stroke": "#00ff00",
                                "strokeWidth": 3,
                                "fill": "rgba(0, 255, 0, 0.2)",
                                "saved": True
                            })
            if normalise:
                annotations = utils.normalize_segmentation(annotations)
                save_yolo_annotations(
                    copy.deepcopy(annotations),
                    (w, h),
                    label_path
                )
        return annotations, (w, h)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error loading annotations: {str(e)} {traceback.format_exc()}")

def normalize_annotations(annotations: List[Union[Box, dict]]) -> List[Box]:
    """Convert all annotations to Box objects."""
    normalized = []
    for ann in annotations:
        if isinstance(ann, Box):
            normalized.append(ann)
        elif isinstance(ann, dict):
            normalized.append(Box(**ann))
        else:
            raise TypeError(f"Unsupported annotation type: {type(ann)}")
    return normalized

def save_yolo_annotations(annotations: List[Box], original_size: tuple, label_path: str):
    """Save annotations in YOLO format (both bbox and segmentation)"""
    annotations = normalize_annotations(annotations)
    os.makedirs(os.path.dirname(label_path), exist_ok=True)
    w, h = original_size

    try:
        with open(label_path, "w") as f:
            # Generate YOLO format from annotations
            for annotation in annotations:
                if annotation.type == "bbox":
                    left, top, width, height = annotation.left, annotation.top, annotation.width, annotation.height
                    xc = (left + width / 2) / w
                    yc = (top + height / 2) / h
                    bw = width / w
                    bh = height / h
                    f.write(f"{annotation.classId} {xc:.6f} {yc:.6f} {bw:.6f} {bh:.6f}\n")

                elif annotation.type == "segmentation" and annotation.points:
                    # Convert points to normalized coordinates
                    normalized_points = []
                    for point in annotation.points:
                        normalized_points.extend([point.x / w, point.y / h])

                    coords_str = " ".join(f"{coord:.6f}" for coord in normalized_points)
                    f.write(f"{annotation.classId} {coords_str}\n")

        # Copy to image_labels directory
        shutil.copy2(label_path, f"{IMAGE_LABEL_ROOT}/{os.path.basename(label_path)}")
        return True
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error saving annotations: {str(e)} {traceback.format_exc()}")

def parse_yolo_line(line: str, image_width: int, image_height: int) -> Dict[str, Any]:
    """Parse a single YOLO format line and return annotation dict"""
    parts = list(map(float, line.strip().split()))
    if len(parts) < 5:
        return None

    class_id = int(parts[0])

    if len(parts) == 5:  # Bounding box
        _, xc, yc, bw, bh = parts
        left = int((xc - bw / 2) * image_width)
        top = int((yc - bh / 2) * image_height)
        width = int(bw * image_width)
        height = int(bh * image_height)

        return {
            "type": "bbox",
            "left": left,
            "top": top,
            "width": width,
            "height": height,
            "classId": class_id,
            "stroke": "#00ff00",
            "strokeWidth": 3,
            "fill": "rgba(0, 255, 0, 0.2)",
            "saved": True
        }

    elif len(parts) > 5 and len(parts) % 2 == 1:  # Segmentation
        coords = parts[1:]
        if len(coords) >= 6:  # At least 3 points
            points = []
            for i in range(0, len(coords), 2):
                if i + 1 < len(coords):
                    x = coords[i] * image_width
                    y = coords[i + 1] * image_height
                    points.append({"x": x, "y": y})

            return {
                "type": "segmentation",
                "points": points,
                "classId": class_id,
                "stroke": "#00ff00",
                "strokeWidth": 3,
                "fill": "rgba(0, 255, 0, 0.2)",
                "saved": True
            }

    return None

# === API Routes ===

@app.get("/api/annotate/images", response_model=List[ImageInfo])
async def list_all_images():
    image_info_list = []
    for root, _, files in os.walk(IMAGE_ROOT):
        for file in sorted(files):
            if file.lower().endswith((".jpg", ".jpeg", ".png")):
                image_path = os.path.join(root, file)
                rel_path = os.path.relpath(image_path, IMAGE_ROOT)
                label_path = get_label_path(rel_path)

                img = Image.open(image_path)
                width, height = img.size

                image_info_list.append(ImageInfo(
                    name=rel_path.replace("\\", "/"),
                    width=width,
                    height=height,
                    has_annotations=os.path.exists(label_path)
                ))
    return image_info_list

@app.get("/api/annotate/image/{image_name:path}")
async def get_image(image_name: str):
    image_path = get_image_path(image_name)
    if not os.path.exists(image_path):
        raise HTTPException(status_code=404, detail="Image not found")

    with Image.open(image_path) as img:
        if img.mode != "RGB":
            img = img.convert("RGB")
        buffer = BytesIO()
        img.save(buffer, format="JPEG")
        img_data = base64.b64encode(buffer.getvalue()).decode()
        return {
            "image_data": f"data:image/jpeg;base64,{img_data}",
            "width": img.width,
            "height": img.height
        }

@app.get("/api/annotate/annotations/{image_name:path}")
async def get_annotations(image_name: str):
    image_path = get_image_path(image_name)
    label_path = get_label_path(image_name)

    if not os.path.exists(image_path):
        raise HTTPException(status_code=404, detail="Image not found")

    annotations, (width, height) = load_yolo_annotations(image_path, label_path)

    return {
        "annotations": annotations,
        "original_width": width,
        "original_height": height
    }

@app.get("/api/annotate/detect_annotations/{image_name:path}")
async def get_detected_annotations(image_name: str):
    image_path = get_image_path(image_name)
    label_path = get_label_path(image_name)

    if not os.path.exists(image_path):
        raise HTTPException(status_code=404, detail="Image not found")

    annotations, (width, height) = load_yolo_annotations(image_path, label_path, True)
    return {
        "annotations": annotations,
        "original_width": width,
        "original_height": height
    }

@app.post("/api/annotate/annotations")
async def save_annotations(request: SaveAnnotationsRequest):
    label_path = get_label_path(request.image_name)
    success = save_yolo_annotations(
        request.annotations,
        (request.original_width, request.original_height),
        label_path
    )
    return {"message": f"Saved {len(request.annotations)} annotations successfully"}

@app.delete("/api/annotate/annotations/{image_name:path}")
async def delete_annotations(image_name: str):
    label_path = get_label_path(image_name)
    if os.path.exists(label_path):
        os.remove(label_path)
        return {"message": "Annotations deleted"}
    return {"message": "No annotations to delete"}

@app.get("/api/annotate/annotations/{image_name:path}/download")
async def download_annotations(image_name: str):
    label_path = get_label_path(image_name)
    if not os.path.exists(label_path):
        raise HTTPException(status_code=404, detail="Annotations not found")
    return FileResponse(
        label_path,
        media_type="text/plain",
        filename=os.path.basename(label_path)
    )

@app.post("/api/annotate/upload")
async def upload_image(file: UploadFile = File(...)):
    if not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")

    file_path = os.path.join(IMAGE_ROOT, "train", file.filename)
    with open(file_path, "wb") as f:
        f.write(await file.read())
    return {"message": f"Uploaded {file.filename} to train set"}