Anomaly Detection for ManufacturingFrom MVTec benchmarks to production inspection lines

# PatchCore inference with anomalib (v2)
from anomalib.models import Patchcore
from anomalib.engine import Engine
from anomalib.data import MVTec

# 1. Configure model
model = Patchcore(
    backbone="wide_resnet50_2",
    layers_to_extract=["layer2", "layer3"],
    coreset_sampling_ratio=0.1,  # 10% of patches → memory bank
    num_neighbors=9,
)

# 2. Load MVTec AD dataset (or your custom dataset)
datamodule = MVTec(
    root="./datasets/MVTec",
    category="bottle",
    image_size=(256, 256),
    train_batch_size=32,
    eval_batch_size=32,
)

# 3. Train (builds memory bank from normal images)
engine = Engine(max_epochs=1)  # PatchCore needs only 1 epoch
engine.fit(model=model, datamodule=datamodule)

# 4. Evaluate
results = engine.test(model=model, datamodule=datamodule)
# Returns: image_AUROC, pixel_AUROC, F1, PRO score

# Export trained model to ONNX for edge deployment
import torch
from anomalib.deploy import ExportMode

# Export to ONNX (quantized INT8 for Jetson)
engine.export(
    model=model,
    export_mode=ExportMode.ONNX,
    export_root="./exported_models/bottle_patchcore",
)

# ── Inference on edge device ──
import onnxruntime as ort
import numpy as np
from PIL import Image

session = ort.InferenceSession(
    "bottle_patchcore/model.onnx",
    providers=["TensorrtExecutionProvider", "CUDAExecutionProvider"],
)

def inspect(image_path: str, threshold: float = 0.5):
    img = Image.open(image_path).resize((256, 256))
    input_array = np.array(img).astype(np.float32) / 255.0
    input_array = np.transpose(input_array, (2, 0, 1))  # HWC → CHW
    input_array = np.expand_dims(input_array, axis=0)

    outputs = session.run(None, {"input": input_array})
    anomaly_score = outputs[0][0]    # Scalar score
    anomaly_map = outputs[1][0]      # Pixel-level heatmap

    return {
        "is_defective": float(anomaly_score) > threshold,
        "confidence": float(anomaly_score),
        "heatmap": anomaly_map,
    }

# Production API wrapper for anomaly detection
from fastapi import FastAPI, UploadFile
from contextlib import asynccontextmanager
import onnxruntime as ort
import numpy as np
from PIL import Image
import io, time

models: dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load models at startup
    models["bottle"] = ort.InferenceSession("models/bottle.onnx")
    models["capsule"] = ort.InferenceSession("models/capsule.onnx")
    models["metal_nut"] = ort.InferenceSession("models/metal_nut.onnx")
    yield
    models.clear()

app = FastAPI(title="Manufacturing AD API", lifespan=lifespan)

@app.post("/inspect/{product_type}")
async def inspect(product_type: str, file: UploadFile):
    if product_type not in models:
        return {"error": f"No model for {product_type}"}

    t0 = time.perf_counter()
    img = Image.open(io.BytesIO(await file.read())).resize((256, 256))
    input_arr = np.expand_dims(
        np.transpose(np.array(img, dtype=np.float32) / 255.0, (2, 0, 1)),
        axis=0,
    )

    outputs = models[product_type].run(None, {"input": input_arr})
    latency_ms = (time.perf_counter() - t0) * 1000

    return {
        "product_type": product_type,
        "anomaly_score": float(outputs[0][0]),
        "is_defective": float(outputs[0][0]) > 0.5,
        "latency_ms": round(latency_ms, 1),
    }