test: Wired POST /engine/trace endpoint with preprocess + vectorize pip…
- "engine/api/routes.py" - "engine/tests/test_api.py" - "engine/main.py" GSD-Task: S01/T05
This commit is contained in:
parent
b33e883a6b
commit
a91c99dd6c
3 changed files with 287 additions and 0 deletions
105
engine/api/routes.py
Normal file
105
engine/api/routes.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
"""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,
|
||||
}
|
||||
|
|
@ -2,12 +2,16 @@
|
|||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from api.routes import router
|
||||
|
||||
app = FastAPI(
|
||||
title="Kerf Engine",
|
||||
description="Raster-to-vector conversion pipeline with Potrace and VTracer modes",
|
||||
version="0.1.0",
|
||||
)
|
||||
|
||||
app.include_router(router)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
|
|
|
|||
178
engine/tests/test_api.py
Normal file
178
engine/tests/test_api.py
Normal file
|
|
@ -0,0 +1,178 @@
|
|||
"""Integration tests for the /engine/trace endpoint."""
|
||||
|
||||
import json
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from main import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def _make_test_png(width: int = 100, height: int = 100) -> bytes:
|
||||
"""Create a simple test PNG with a white rectangle on black background."""
|
||||
img = np.zeros((height, width, 3), dtype=np.uint8)
|
||||
cv2.rectangle(img, (20, 20), (80, 80), (255, 255, 255), -1)
|
||||
ok, buf = cv2.imencode(".png", img)
|
||||
assert ok
|
||||
return buf.tobytes()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_png() -> bytes:
|
||||
return _make_test_png()
|
||||
|
||||
|
||||
class TestTraceEndpointPotrace:
|
||||
"""Tests for /engine/trace with mode=potrace."""
|
||||
|
||||
def test_basic_trace(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["format"] == "svg"
|
||||
assert "<svg" in body["output"]
|
||||
assert "metadata" in body
|
||||
|
||||
def test_metadata_shape(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace"},
|
||||
)
|
||||
body = resp.json()
|
||||
meta = body["metadata"]
|
||||
assert "path_count" in meta
|
||||
assert "node_count_total" in meta
|
||||
assert "open_paths" in meta
|
||||
assert "warnings" in meta
|
||||
assert "processing_ms" in meta
|
||||
assert isinstance(meta["path_count"], int)
|
||||
assert isinstance(meta["processing_ms"], float)
|
||||
|
||||
def test_svg_has_paths(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace"},
|
||||
)
|
||||
body = resp.json()
|
||||
assert body["metadata"]["path_count"] >= 1
|
||||
|
||||
def test_custom_params(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={
|
||||
"mode": "potrace",
|
||||
"params": json.dumps({"turdsize": 5, "alphamax": 0.5}),
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert "<svg" in resp.json()["output"]
|
||||
|
||||
|
||||
class TestTraceEndpointVtracer:
|
||||
"""Tests for /engine/trace with mode=vtracer."""
|
||||
|
||||
def test_basic_trace(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "vtracer"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["format"] == "svg"
|
||||
assert "<svg" in body["output"].lower() or "svg" in body["output"]
|
||||
|
||||
def test_metadata_present(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "vtracer"},
|
||||
)
|
||||
body = resp.json()
|
||||
meta = body["metadata"]
|
||||
assert isinstance(meta["path_count"], int)
|
||||
assert isinstance(meta["processing_ms"], float)
|
||||
|
||||
def test_custom_params(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={
|
||||
"mode": "vtracer",
|
||||
"params": json.dumps({"filter_speckle": 10, "corner_threshold": 90}),
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
class TestTraceEndpointValidation:
|
||||
"""Tests for input validation on /engine/trace."""
|
||||
|
||||
def test_invalid_mode(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "invalid"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_unsupported_output_format(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace", "output_format": "pdf"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_invalid_params_json(self, test_png):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace", "params": "not-json"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_empty_file(self):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("empty.png", b"", "image/png")},
|
||||
data={"mode": "potrace"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_corrupt_image(self):
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("bad.png", b"not an image", "image/png")},
|
||||
data={"mode": "potrace"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_defaults_to_potrace(self, test_png):
|
||||
"""Mode defaults to potrace when not specified."""
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert "<svg" in resp.json()["output"]
|
||||
|
||||
def test_preset_ignored(self, test_png):
|
||||
"""Preset param is accepted but ignored for now."""
|
||||
resp = client.post(
|
||||
"/engine/trace",
|
||||
files={"file": ("test.png", test_png, "image/png")},
|
||||
data={"mode": "potrace", "preset": "logo"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
Loading…
Add table
Reference in a new issue