- "engine/api/routes.py" - "engine/tests/test_api.py" - "engine/main.py" GSD-Task: S01/T05
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""API routes for the Kerf Engine trace endpoint."""
|
|
|
|
import json
|
|
import re
|
|
import time
|
|
|
|
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
|
|
|
from pipeline.preprocessing import preprocess
|
|
from pipeline.vectorize import potrace_trace, vtracer_trace
|
|
|
|
router = APIRouter()
|
|
|
|
VALID_MODES = {"potrace", "vtracer"}
|
|
|
|
|
|
def _extract_svg_metadata(svg: str) -> dict:
|
|
"""Extract basic metadata from an SVG string."""
|
|
path_matches = re.findall(r'<path\b[^>]*\bd="([^"]*)"', svg)
|
|
path_count = len(path_matches)
|
|
|
|
node_count_total = 0
|
|
open_paths = 0
|
|
for d_attr in path_matches:
|
|
# Count SVG path commands (M, L, C, Q, A, Z, etc.)
|
|
commands = re.findall(r"[MLHVCSQTAZ]", d_attr, re.IGNORECASE)
|
|
node_count_total += len(commands)
|
|
# A path is "open" if it doesn't end with Z
|
|
if not d_attr.rstrip().upper().endswith("Z"):
|
|
open_paths += 1
|
|
|
|
return {
|
|
"path_count": path_count,
|
|
"node_count_total": node_count_total,
|
|
"open_paths": open_paths,
|
|
}
|
|
|
|
|
|
@router.post("/engine/trace")
|
|
async def trace(
|
|
file: UploadFile = File(...),
|
|
mode: str = Form("potrace"),
|
|
output_format: str = Form("svg"),
|
|
preset: str = Form("default"),
|
|
params: str = Form("{}"),
|
|
):
|
|
"""Convert a raster image to SVG via the preprocessing + vectorization pipeline."""
|
|
if mode not in VALID_MODES:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Invalid mode '{mode}'. Must be one of: {', '.join(sorted(VALID_MODES))}",
|
|
)
|
|
|
|
if output_format != "svg":
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Unsupported output_format '{output_format}'. Only 'svg' is supported.",
|
|
)
|
|
|
|
try:
|
|
user_params = json.loads(params)
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(status_code=422, detail=f"Invalid params JSON: {exc}")
|
|
|
|
raw_bytes = await file.read()
|
|
if not raw_bytes:
|
|
raise HTTPException(status_code=422, detail="Uploaded file is empty")
|
|
|
|
warnings: list[str] = []
|
|
start = time.perf_counter()
|
|
|
|
try:
|
|
preprocessed = preprocess(raw_bytes, params=user_params)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=f"Preprocessing failed: {exc}")
|
|
|
|
try:
|
|
if mode == "potrace":
|
|
svg_output = potrace_trace(preprocessed, **{
|
|
k: v for k, v in user_params.items()
|
|
if k in ("turdsize", "alphamax", "opticurve", "opttolerance")
|
|
})
|
|
else:
|
|
svg_output = vtracer_trace(preprocessed, **{
|
|
k: v for k, v in user_params.items()
|
|
if k in (
|
|
"colormode", "hierarchical", "filter_speckle", "color_precision",
|
|
"layer_difference", "corner_threshold", "length_threshold",
|
|
"splice_threshold", "mode", "path_precision", "max_iterations",
|
|
)
|
|
})
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"Vectorization failed: {exc}")
|
|
|
|
processing_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
|
|
metadata = _extract_svg_metadata(svg_output)
|
|
metadata["warnings"] = warnings
|
|
metadata["processing_ms"] = processing_ms
|
|
|
|
return {
|
|
"output": svg_output,
|
|
"format": "svg",
|
|
"metadata": metadata,
|
|
}
|