Build a Visual Inspection Line

Build a containerized vision line that detects defects and logs rejected parts.

Introduction

30 Second Summary

A factory inspection line has seconds to spot a scratched part before it continues through production. Rejecting a good part wastes time while accepting a damaged one puts the batch at risk.

In this project, you will build a miniature visual inspection line for synthetic gear and washer images using a CNN classifier plus a PatchCore-style anomaly detector. The finished system runs in containers behind measurable quality gates.

What You'll Build

You will demo a local inspection line that returns ACCEPT or REJECT for each test image while its rejected-part counter rises.

By the end of this project, you'll have:

  • An auditable synthetic dataset where every gear or washer image belongs to one split. You can inject split leakage to watch the split guard stop an invalid experiment.
  • A tracked model comparison where MLflow records per-category AUROC for both detectors. The generated report ties the calibrated threshold to the false-reject rate. It also records defect recall. The export gate proves ONNX parity within 1e-4. The latency gate keeps CPU p95 latency below 200 ms.
  • A running inspection line that returns decisions through a REST endpoint built with FastAPI. Rejected parts persist in PostgreSQL, so the /events/count endpoint returns a growing count. GitHub Actions reruns the quality gates after each push.
  • Secret Mission: Introduce an uneven lighting shift and capture the resulting false-reject failure.

Do I need prior machine learning experience?

Basic Python syntax is enough because the guide supplies every project file. You will focus on tracing each quality decision through the inspection stack.

Before We Start

Before any hands-on work, commit to inspecting synthetic gear and washer images for defects. A 5% false-reject budget limits costly plant disruption caused when normal parts are rejected.

Set Up the Inspection Workspace

Your inspection line must produce repeatable evidence. Library drift can change model behavior between runs.

A pinned Python 3.12 virtual environment gives every local command the same dependency set. This isolation protects the model training results from packages installed by other projects.

VS Code holds the supplied repository. Git supports the later workflow in CI. Docker Desktop provides the container runtime for the finished inspection line.

In this step, get ready to:
  • Load the supplied inspection project into VS Code.
  • Create an isolated Python environment from the pinned dependency file.
  • Verify every required runtime before generating inspection data.
Load the supplied project files

The supplied files connect every later stage of the inspection line. Their folder structure must stay intact because the training commands import modules from src.

  • Create a folder named visual-inspection on your Desktop with File Explorer.
  • Open the empty visual-inspection folder in the next actions. You will create the complete project tree from the integrated PowerShell terminal.

File Explorer should now show the empty visual-inspection folder on your Desktop. This gives the project files a known location before you open the workspace.

  • Press the Windows key to display Windows search.

You should see the Windows search field ready for an application name.

  • Type Visual Studio Code into Windows search.
  • Press Enter to start VS Code.

You should see the VS Code window open.

  • Press Ctrl+K followed by Ctrl+O to open the folder picker.

You should see a folder picker where you can select the workspace.

  • Select the visual-inspection folder on your Desktop.
  • Click Select Folder.

The Explorer sidebar should now show the supplied project tree under visual-inspection.

  • Select Yes, I trust the authors if the Workspace Trust dialog appears.
  • Select View from the top menu.
  • Select Terminal.

The integrated terminal should open at the bottom of VS Code. Its prompt should end inside the visual-inspection workspace.

  • Choose PowerShell from the terminal profile menu if another shell is active.

Your editor now has the empty visual-inspection workspace plus a Windows PowerShell terminal in the correct folder. The next checkpoint gives you the complete project structure and every file you need.

✔️ I already have the project files

Continue when the Explorer sidebar shows requirements.txt, src, tests, Dockerfile, and compose.yaml.

ⓧ I need the complete project code

The original project bundle is not available through a download button. Use this copy-paste fallback to create the same workspace directly in VS Code.

  • Create the complete empty file structure from the integrated PowerShell terminal by running:
$files = @(
  "requirements.txt"
  ".gitignore"
  "Dockerfile"
  "compose.yaml"
  "src\__init__.py"
  "src\pipeline.py"
  "src\train.py"
  "src\export.py"
  "src\api.py"
  "src\controller.py"
  "tests\test_quality_gates.py"
  ".github\workflows\ci.yml"
)
$files | ForEach-Object {
  $parent = Split-Path $_ -Parent
  if ($parent) { New-Item -ItemType Directory -Force $parent | Out-Null }
  New-Item -ItemType File -Force $_ | Out-Null
}
Get-ChildItem -Recurse -File | Select-Object -ExpandProperty FullName

What should the project structure contain?

visual-inspection/
├── .github/workflows/ci.yml
├── src/
│   ├── __init__.py
│   ├── api.py
│   ├── controller.py
│   ├── export.py
│   ├── pipeline.py
│   └── train.py
├── tests/test_quality_gates.py
├── .gitignore
├── compose.yaml
├── Dockerfile
└── requirements.txt

The data, artifacts, and mlflow.db outputs appear later when you run the training workflow.

  • Open requirements.txt in the Explorer sidebar.
  • Paste this complete dependency list into the file:
torch==2.14.1
numpy==2.5.3
Pillow==13.0.0
scikit-learn==1.9.1
mlflow==3.17.2
onnx==1.23.2
onnxscript==0.7.2
onnxruntime==1.31.0
fastapi==0.156.8
uvicorn==0.54.0
python-multipart==0.0.32
psycopg[binary]==3.3.6
httpx==0.28.1
pytest==9.1.1
  • Save requirements.txt.
  • Open .gitignore and paste this content:
.venv/
__pycache__/
.pytest_cache/
data/
artifacts/
mlflow.db
mlruns/
*.pyc
  • Save .gitignore.
  • Leave src/__init__.py empty so Python treats src as the project package.
  • Open src/pipeline.py and paste this complete pipeline:
import csv
import json
import shutil
from collections import defaultdict
from pathlib import Path

import mlflow
import numpy as np
import torch
from PIL import Image
from sklearn.metrics import roc_auc_score
from sklearn.neighbors import NearestNeighbors
from torch import nn
from torch.utils.data import DataLoader, TensorDataset

# Keep every generated image and model input at the same size.
IMAGE_SIZE = 64
CATEGORIES = ("gear", "washer")
FALSE_REJECT_BUDGET = 0.05
ARTIFACT_DIR = Path("artifacts")


def render_part(category: str, defect: bool, seed: int, lighting_shift: float = 0.0) -> Image.Image:
    # Build one repeatable grayscale part from coordinate masks.
    rng = np.random.default_rng(seed)
    yy, xx = np.mgrid[:IMAGE_SIZE, :IMAGE_SIZE]
    cx = IMAGE_SIZE // 2 + int(rng.integers(-2, 3))
    cy = IMAGE_SIZE // 2 + int(rng.integers(-2, 3))
    radius = np.sqrt((xx - cx) ** 2 + (yy - cy) ** 2)
    canvas = np.full((IMAGE_SIZE, IMAGE_SIZE), 35.0 + lighting_shift * 255.0)

    if category == "gear":
        angle = np.arctan2(yy - cy, xx - cx)
        toothed_edge = 20.0 + 2.5 * np.cos(12.0 * angle)
        part_mask = (radius <= toothed_edge) & (radius >= 7.0)
    else:
        part_mask = (radius <= 21.0) & (radius >= 10.0)

    canvas[part_mask] = 190.0
    if defect:
        # Add a bright diagonal scratch across the part surface.
        scratch = np.abs((yy - cy) - 0.55 * (xx - cx)) < 1.2
        canvas[scratch & part_mask] = 250.0

    canvas += rng.normal(0.0, 5.0, canvas.shape)
    return Image.fromarray(np.clip(canvas, 0, 255).astype(np.uint8), mode="L")


def assert_disjoint(records: list[dict[str, str]]) -> None:
    # Reject any stable image ID assigned to more than one split.
    memberships: dict[str, set[str]] = defaultdict(set)
    for record in records:
        memberships[record["image_id"]].add(record["split"])
    leaked = sorted(image_id for image_id, splits in memberships.items() if len(splits) > 1)
    if leaked:
        raise ValueError(f"Split leakage detected: {leaked[:5]}")


def generate_dataset(inject_leak: bool = False, lighting_shift: float = 0.0) -> list[dict[str, str]]:
    # Rebuild the dataset so every run starts from a known state.
    root = Path("data")
    if root.exists():
        shutil.rmtree(root)
    records: list[dict[str, str]] = []
    samples_per_label = {"train": 60, "val": 20, "test": 20}
    seed = 1000

    for split, count in samples_per_label.items():
        for category in CATEGORIES:
            for label in (0, 1):
                kind = "defect" if label else "normal"
                target = root / split / category / kind
                target.mkdir(parents=True, exist_ok=True)
                for index in range(count):
                    image_id = f"{split}-{category}-{label}-{index:03d}"
                    path = target / f"{image_id}.png"
                    render_part(category, bool(label), seed, lighting_shift).save(path)
                    records.append({
                        "image_id": image_id,
                        "split": split,
                        "category": category,
                        "label": str(label),
                        "path": path.as_posix(),
                    })
                    seed += 1

    if inject_leak:
        # Copy one identity into a second split to challenge the guard.
        leaked_record = dict(records[0])
        leaked_record["split"] = "test"
        records.append(leaked_record)

    manifest = root / "manifest.csv"
    with manifest.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=["image_id", "split", "category", "label", "path"])
        writer.writeheader()
        writer.writerows(records)

    assert_disjoint(records)
    counts = {split: sum(record["split"] == split for record in records) for split in samples_per_label}
    print(f"split counts: {counts}")
    return records


def load_split(root: Path, split: str) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    # Load one manifest split into model-ready arrays.
    images: list[np.ndarray] = []
    labels: list[int] = []
    categories: list[str] = []
    with (root / "manifest.csv").open(newline="", encoding="utf-8") as handle:
        for record in csv.DictReader(handle):
            if record["split"] != split:
                continue
            with Image.open(record["path"]) as image:
                array = np.asarray(image.convert("L"), dtype=np.float32) / 255.0
            images.append(array[None, :, :])
            labels.append(int(record["label"]))
            categories.append(record["category"])
    return np.stack(images), np.asarray(labels), np.asarray(categories)


class TinyCNN(nn.Module):
    # Produce a defect logit while exposing frozen spatial features.
    def __init__(self) -> None:
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(1, 8, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(8, 16, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(16, 16, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
        )
        self.pool = nn.AdaptiveAvgPool2d(1)
        self.classifier = nn.Linear(16, 1)

    def forward(self, images: torch.Tensor) -> torch.Tensor:
        features = self.features(images)
        return self.classifier(self.pool(features).flatten(1))

    def patch_embeddings(self, images: torch.Tensor) -> np.ndarray:
        # Turn each spatial feature cell into one patch embedding.
        with torch.no_grad():
            features = self.features(images).permute(0, 2, 3, 1)
        return features.reshape(features.shape[0], -1, features.shape[-1]).numpy()


def train_cnn(images: np.ndarray, labels: np.ndarray, epochs: int = 18) -> TinyCNN:
    # Train the compact classifier on labeled synthetic parts.
    torch.manual_seed(7)
    model = TinyCNN()
    dataset = TensorDataset(torch.from_numpy(images), torch.from_numpy(labels.astype(np.float32))[:, None])
    loader = DataLoader(dataset, batch_size=64, shuffle=True, generator=torch.Generator().manual_seed(7))
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    loss_fn = nn.BCEWithLogitsLoss()
    model.train()
    for _ in range(epochs):
        for batch_images, batch_labels in loader:
            optimizer.zero_grad()
            loss = loss_fn(model(batch_images), batch_labels)
            loss.backward()
            optimizer.step()
    model.eval()
    return model


def cnn_scores(model: TinyCNN, images: np.ndarray) -> np.ndarray:
    # Convert classifier logits into defect probabilities.
    with torch.no_grad():
        return torch.sigmoid(model(torch.from_numpy(images))).numpy().reshape(-1)


def build_memory_banks(model: TinyCNN, images: np.ndarray, labels: np.ndarray, categories: np.ndarray) -> dict[str, np.ndarray]:
    # Store a compact sample of nominal patch embeddings per category.
    banks: dict[str, np.ndarray] = {}
    for category in CATEGORIES:
        mask = (categories == category) & (labels == 0)
        patches = model.patch_embeddings(torch.from_numpy(images[mask])).reshape(-1, 16)
        banks[category] = patches[::4]
    return banks


def patch_scores(model: TinyCNN, banks: dict[str, np.ndarray], images: np.ndarray, categories: np.ndarray) -> np.ndarray:
    # Score each image by its most unusual patch distance.
    indexes = {category: NearestNeighbors(n_neighbors=1).fit(bank) for category, bank in banks.items()}
    embeddings = model.patch_embeddings(torch.from_numpy(images))
    scores: list[float] = []
    for patches, category in zip(embeddings, categories, strict=True):
        distances, _ = indexes[str(category)].kneighbors(patches)
        scores.append(float(distances.max()))
    return np.asarray(scores)


def calibrate(scores: np.ndarray, labels: np.ndarray) -> float:
    # Set the operating point from normal validation scores only.
    return float(np.quantile(scores[labels == 0], 1.0 - FALSE_REJECT_BUDGET, method="higher"))


def evaluate(scores: np.ndarray, labels: np.ndarray, categories: np.ndarray, threshold: float) -> dict[str, object]:
    # Report ranking quality and fixed-threshold behavior.
    auroc = {
        category: float(roc_auc_score(labels[categories == category], scores[categories == category]))
        for category in CATEGORIES
    }
    normal = labels == 0
    defect = labels == 1
    return {
        "threshold": threshold,
        "auroc": auroc,
        "false_reject_rate": float(np.mean(scores[normal] > threshold)),
        "defect_recall": float(np.mean(scores[defect] > threshold)),
    }


def train_and_report() -> dict[str, object]:
    # Train both inspectors and persist their reproducible evidence.
    train_images, train_labels, train_categories = load_split(Path("data"), "train")
    val_images, val_labels, val_categories = load_split(Path("data"), "val")
    test_images, test_labels, test_categories = load_split(Path("data"), "test")
    model = train_cnn(train_images, train_labels)

    cnn_val = cnn_scores(model, val_images)
    cnn_test = cnn_scores(model, test_images)
    banks = build_memory_banks(model, train_images, train_labels, train_categories)
    patch_val = patch_scores(model, banks, val_images, val_categories)
    patch_test = patch_scores(model, banks, test_images, test_categories)

    cnn_threshold = calibrate(cnn_val, val_labels)
    patch_threshold = calibrate(patch_val, val_labels)
    report = {
        "false_reject_budget": FALSE_REJECT_BUDGET,
        "cnn": evaluate(cnn_test, test_labels, test_categories, cnn_threshold),
        "patchcore_style": evaluate(patch_test, test_labels, test_categories, patch_threshold),
    }

    ARTIFACT_DIR.mkdir(exist_ok=True)
    torch.save(model.state_dict(), ARTIFACT_DIR / "model.pt")
    np.savez(ARTIFACT_DIR / "memory_banks.npz", **banks)
    report_path = ARTIFACT_DIR / "report.json"
    report_path.write_text(json.dumps(report, indent=2), encoding="utf-8")

    # Record parameters, metrics, and key files in a local MLflow database.
    mlflow.set_tracking_uri("sqlite:///mlflow.db")
    mlflow.set_experiment("visual-inspection")
    with mlflow.start_run(run_name="cnn-vs-patchcore-style"):
        mlflow.log_params({"epochs": 18, "false_reject_budget": FALSE_REJECT_BUDGET})
        metrics: dict[str, float] = {}
        for name in ("cnn", "patchcore_style"):
            result = report[name]
            metrics[f"{name}_false_reject_rate"] = float(result["false_reject_rate"])
            metrics[f"{name}_defect_recall"] = float(result["defect_recall"])
            for category, value in result["auroc"].items():
                metrics[f"{name}_{category}_auroc"] = float(value)
        mlflow.log_metrics(metrics)
        mlflow.log_artifact(str(ARTIFACT_DIR / "model.pt"))
        mlflow.log_artifact(str(report_path))

    print(json.dumps(report, indent=2))
    return report
  • Save src/pipeline.py.
  • Open src/train.py and paste this command-line entry point:
import argparse

from src.pipeline import generate_dataset, train_and_report


def main() -> None:
    # Expose data generation and training through one repeatable command.
    parser = argparse.ArgumentParser(description="Train the visual inspection models.")
    parser.add_argument("--data-only", action="store_true")
    parser.add_argument("--inject-leak", action="store_true")
    parser.add_argument("--lighting-shift", type=float, default=0.0)
    args = parser.parse_args()

    generate_dataset(
        inject_leak=args.inject_leak,
        lighting_shift=args.lighting_shift,
    )
    if not args.data_only:
        train_and_report()


if __name__ == "__main__":
    main()
  • Save src/train.py.
  • Open src/export.py and paste this ONNX export gate:
import json
import math
import time
from pathlib import Path

import numpy as np
import onnx
import onnxruntime as ort
import torch

from src.pipeline import ARTIFACT_DIR, TinyCNN, load_split

# Fail export when numerical agreement or CPU latency misses the project limits.
MAX_ABS_ERROR = 1e-4
MAX_P95_MS = 200.0


def main() -> None:
    # Restore the trained classifier and export one-image inference.
    model = TinyCNN()
    model.load_state_dict(torch.load(ARTIFACT_DIR / "model.pt", weights_only=True))
    model.eval()
    images, _, _ = load_split(Path("data"), "test")
    sample = torch.from_numpy(images[:1])
    program = torch.onnx.export(
        model,
        (sample,),
        input_names=["image"],
        output_names=["logit"],
        dynamo=True,
    )
    onnx_path = ARTIFACT_DIR / "model.onnx"
    program.save(onnx_path)
    onnx.checker.check_model(onnx.load(onnx_path))

    # Compare PyTorch and ONNX Runtime on the same tensor.
    session = ort.InferenceSession(str(onnx_path), providers=["CPUExecutionProvider"])
    with torch.no_grad():
        torch_output = model(sample).numpy()
    ort_output = session.run(None, {"image": sample.numpy()})[0]
    max_abs_error = float(np.abs(torch_output - ort_output).max())

    # Warm the runtime before collecting one hundred latency measurements.
    for _ in range(10):
        session.run(None, {"image": sample.numpy()})
    timings = []
    for _ in range(100):
        start = time.perf_counter()
        session.run(None, {"image": sample.numpy()})
        timings.append((time.perf_counter() - start) * 1000.0)
    p95_ms = sorted(timings)[math.ceil(0.95 * len(timings)) - 1]

    result = {
        "max_abs_error": max_abs_error,
        "p95_ms": p95_ms,
        "max_allowed_error": MAX_ABS_ERROR,
        "max_allowed_p95_ms": MAX_P95_MS,
    }
    (ARTIFACT_DIR / "validation.json").write_text(json.dumps(result, indent=2), encoding="utf-8")
    print(json.dumps(result, indent=2))
    if max_abs_error > MAX_ABS_ERROR or p95_ms > MAX_P95_MS:
        raise SystemExit(1)


if __name__ == "__main__":
    main()
  • Save src/export.py.
  • Open src/api.py and paste this inspection service:
import json
import math
import os
from io import BytesIO
from pathlib import Path

import numpy as np
import onnxruntime as ort
import psycopg
from fastapi import FastAPI, UploadFile
from PIL import Image

# Load the calibrated threshold and ONNX graph once during service startup.
ARTIFACT_DIR = Path("artifacts")
DATABASE_URL = os.getenv("DATABASE_URL", "")
report = json.loads((ARTIFACT_DIR / "report.json").read_text(encoding="utf-8"))
threshold = float(report["cnn"]["threshold"])
session = ort.InferenceSession(str(ARTIFACT_DIR / "model.onnx"), providers=["CPUExecutionProvider"])
app = FastAPI(title="Visual Inspection Line")


def prepare_image(content: bytes) -> np.ndarray:
    # Match uploaded images to the model's grayscale input shape.
    with Image.open(BytesIO(content)) as image:
        array = np.asarray(image.convert("L").resize((64, 64)), dtype=np.float32) / 255.0
    return array[None, None, :, :]


def create_event_table(connection: psycopg.Connection) -> None:
    # Keep the event schema available before inserts or counts.
    connection.execute("""
        CREATE TABLE IF NOT EXISTS defect_events (
            id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            inspected_at timestamp with time zone DEFAULT CURRENT_TIMESTAMP,
            part_id text NOT NULL,
            category text NOT NULL,
            image_name text NOT NULL,
            score double precision NOT NULL,
            threshold double precision NOT NULL,
            rejected boolean NOT NULL
        )
    """)


def log_reject(part_id: str, category: str, image_name: str, score: float) -> None:
    # Persist only rejected parts when a database is configured.
    if not DATABASE_URL:
        return
    with psycopg.connect(DATABASE_URL) as connection:
        create_event_table(connection)
        connection.execute(
            "INSERT INTO defect_events (part_id, category, image_name, score, threshold, rejected) VALUES (%s, %s, %s, %s, %s, %s)",
            (part_id, category, image_name, score, threshold, True),
        )


@app.get("/health")
def health() -> dict[str, str]:
    # Give orchestrators a lightweight readiness response.
    return {"status": "ok"}


@app.post("/inspect")
async def inspect(file: UploadFile, part_id: str = "unknown", category: str = "unknown") -> dict[str, object]:
    # Convert one upload into a calibrated ACCEPT or REJECT decision.
    tensor = prepare_image(await file.read())
    logit = float(session.run(None, {"image": tensor})[0].reshape(-1)[0])
    score = 1.0 / (1.0 + math.exp(-logit))
    rejected = score > threshold
    if rejected:
        log_reject(part_id, category, file.filename or "uploaded-image", score)
    return {
        "part_id": part_id,
        "category": category,
        "score": score,
        "threshold": threshold,
        "decision": "REJECT" if rejected else "ACCEPT",
    }


@app.get("/events/count")
def event_count() -> dict[str, object]:
    # Report how many rejected-part events reached PostgreSQL.
    if not DATABASE_URL:
        return {"count": 0, "database": "disabled"}
    with psycopg.connect(DATABASE_URL) as connection:
        create_event_table(connection)
        row = connection.execute("SELECT count(*) FROM defect_events").fetchone()
    return {"count": int(row[0]), "database": "postgresql"}
  • Save src/api.py.
  • Open src/controller.py and paste this mock line controller:
import math
import time
from pathlib import Path

import httpx

# Send generated test parts to the local inspection service.
API_URL = "http://127.0.0.1:8000"
MAX_P95_MS = 200.0


def main() -> None:
    # Measure each request while printing the returned decision.
    paths = sorted(Path("data/test").glob("*/*/*.png"))
    timings: list[float] = []
    with httpx.Client(timeout=30.0) as client:
        client.get(f"{API_URL}/health").raise_for_status()
        for index, path in enumerate(paths):
            part_id = f"LINE-{index:04d}"
            category = path.parts[-3]
            start = time.perf_counter()
            with path.open("rb") as image:
                response = client.post(
                    f"{API_URL}/inspect",
                    params={"part_id": part_id, "category": category},
                    files={"file": (path.name, image, "image/png")},
                )
            timings.append((time.perf_counter() - start) * 1000.0)
            response.raise_for_status()
            result = response.json()
            print(part_id, result["decision"], f"score={result['score']:.4f}")

        # Read the durable reject count after every image has passed the line.
        count_response = client.get(f"{API_URL}/events/count")
        count_response.raise_for_status()
        logged_rejects = int(count_response.json()["count"])

    p95_ms = sorted(timings)[math.ceil(0.95 * len(timings)) - 1]
    print({"p95_ms": p95_ms, "logged_rejects": logged_rejects})
    if p95_ms > MAX_P95_MS or logged_rejects < 1:
        raise SystemExit(1)


if __name__ == "__main__":
    main()
  • Save src/controller.py.
  • Open tests/test_quality_gates.py and paste these automated gates:
import csv
import json
from collections import defaultdict
from pathlib import Path

from src.export import MAX_ABS_ERROR, MAX_P95_MS
from src.pipeline import CATEGORIES


def test_manifest_splits_are_disjoint() -> None:
    # Confirm every stable image identity belongs to one split.
    memberships: dict[str, set[str]] = defaultdict(set)
    with Path("data/manifest.csv").open(newline="", encoding="utf-8") as handle:
        for record in csv.DictReader(handle):
            memberships[record["image_id"]].add(record["split"])
    assert all(len(splits) == 1 for splits in memberships.values())


def test_report_contains_per_category_auroc() -> None:
    # Require both inspectors to report every supported part category.
    report = json.loads(Path("artifacts/report.json").read_text(encoding="utf-8"))
    for inspector in ("cnn", "patchcore_style"):
        assert set(report[inspector]["auroc"]) == set(CATEGORIES)
        assert 0.0 <= report[inspector]["false_reject_rate"] <= 1.0
        assert 0.0 <= report[inspector]["defect_recall"] <= 1.0


def test_onnx_parity_and_latency() -> None:
    # Block models that miss either serving-quality limit.
    result = json.loads(Path("artifacts/validation.json").read_text(encoding="utf-8"))
    assert result["max_abs_error"] <= MAX_ABS_ERROR
    assert result["p95_ms"] <= MAX_P95_MS
  • Save tests/test_quality_gates.py.
  • Open Dockerfile and paste this image definition:
FROM python:3.12.15-slim

WORKDIR /code
COPY requirements.txt /code/requirements.txt
RUN pip install --no-cache-dir -r /code/requirements.txt
COPY . /code
RUN python -m src.train && python -m src.export

CMD ["uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "8000"]
  • Save Dockerfile.
  • Open compose.yaml and paste this two-service configuration:
services:
  db:
    image: postgres:18.6-alpine
    environment:
      POSTGRES_USER: inspection
      POSTGRES_PASSWORD: inspection
      POSTGRES_DB: inspection
    volumes:
      - inspection_postgres:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U inspection -d inspection"]
      interval: 5s
      timeout: 5s
      retries: 5

  api:
    build: .
    environment:
      DATABASE_URL: postgresql://inspection:inspection@db:5432/inspection
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  inspection_postgres:
  • Save compose.yaml.
  • Open .github/workflows/ci.yml and paste this workflow:
name: inspection-quality-gates

on:
  push:
  pull_request:

permissions:
  contents: read

jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v7
        with:
          python-version: "3.12"
          cache: pip
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Build model artifacts
        run: |
          python -m src.train
          python -m src.export
      - name: Run quality gates
        run: pytest
  • Save .github/workflows/ci.yml.
  • Confirm every required file exists by running:
Get-ChildItem -Recurse -File | ForEach-Object {
  $_.FullName.Replace((Get-Location).Path + "\", "")
}

You should see all twelve project files in the terminal output. Your workspace now contains the complete code needed by every later step.

Create the pinned environment

The virtual environment must come from Python 3.12. Checking the interpreter first prevents an incompatible runtime from creating .venv.

  • Check the Python version available to PowerShell by running this command:
python --version

What does this check prove?

This command reports the interpreter PowerShell reaches through python. That interpreter becomes the foundation of the project environment.

✔️ I see Python 3.12

Your local interpreter matches the project runtime. Continue with the environment creation below.

ⓧ I see another version

A different Python version can produce incompatible package results. Install the Windows build for Python 3.12.10 before creating the environment.

  • Open the official Python release page in your browser.
  • Download the Windows installer (64-bit).
  • Run the downloaded installer.
  • Enable the installer option that makes Python available from your terminal.
  • Restart VS Code after the installation completes.
  • Return to the visual-inspection workspace.
  • Open a new integrated PowerShell terminal.
  • Repeat the version check by running:
python --version

What should the new terminal use?

The restarted editor receives the updated runtime path. Continue when the command reports Python 3.12.

ⓧ Command not found

PowerShell cannot reach a Python installation yet. Install Python 3.12.10 before creating the project environment.

  • Open the official Python release page in your browser.
  • Download the Windows installer (64-bit).
  • Run the downloaded installer.
  • Enable the installer option that makes Python available from your terminal.
  • Restart VS Code after the installation completes.
  • Return to the visual-inspection workspace.
  • Open a new integrated PowerShell terminal.
  • Confirm the installation by running:
python --version

What confirms the installation?

PowerShell can now reach Python when the command prints a 3.12 version. That result clears the way for the project environment.

A virtual environment stores this project's interpreter links plus packages inside .venv. The repository can now use pinned dependencies without changing your system packages.

  • Create the isolated environment inside visual-inspection by running this command:
python -m venv .venv

What does this command create?

The venv module creates a project-specific Python environment inside .venv.

Its package directory begins empty. The pinned requirements fill it in the next action.

  • Check the Explorer sidebar for the new .venv folder.

You should see .venv at the top level of visual-inspection.

Missing the .venv folder?

Check that the terminal prompt points to visual-inspection. Running the command from another folder creates the environment there.

If the folder remains missing, help me locate the virtual environment created by PowerShell.

  • Activate the project environment by running this command:
.venv\Scripts\Activate.ps1

What does activation change?

Activation points the current PowerShell session at the interpreter inside .venv.

Package installation commands now target this project environment.

The PowerShell prompt should now show (.venv) at the front. That prefix confirms the environment is active.

Activation blocked?

Confirm the terminal is inside visual-inspection before checking the activation path. PowerShell may also block local activation scripts under your machine's script policy.

Use Python's Windows virtual environment guidance for the documented activation pattern. You can also help me diagnose why Activate.ps1 is blocked.

The dependency installation is the slowest setup action. Allow several minutes because the pinned machine-learning packages are larger than a typical Python library.

  • Install every pinned dependency from requirements.txt by running this command:
pip install -r requirements.txt

What gets installed?

  • PyTorch provides the compact CNN plus its frozen feature extractor.
  • NumPy plus Pillow handle image arrays and synthetic image files.
  • scikit-learn provides nearest-neighbor search plus AUROC calculation.
  • MLflow records the experiment parameters and metrics.
  • ONNX defines the exported model format. ONNX Runtime executes that model on the CPU.
  • FastAPI exposes inspection endpoints. Uvicorn runs the API server.
  • Psycopg writes rejected-part events to PostgreSQL.
  • HTTPX sends images from the mock controller to the inspection API.
  • pytest runs the automated quality gates.

PowerShell should display package download and installation activity before returning to the active (.venv) prompt.

Dependency installation failed?

Confirm that (.venv) appears at the front of the prompt. Also confirm that requirements.txt is visible in the top level of the Explorer sidebar.

First, help me diagnose the pinned dependency installation failure. Share only non-sensitive error output with the NextWork community if the issue depends on your machine.

Verify every required tool

The final check covers the interpreter, repository tooling, container tooling, and core imports. Together these results prove the workspace can support the later training and serving stages.

Before you run these checks, do you expect all four commands to succeed from the activated environment?

  • Run the complete environment check in PowerShell with these commands:
python --version
docker --version
git --version
python -c "import torch, fastapi, onnxruntime; print('inspection environment ready')"

What does each check prove?

  • The Python check confirms that the active environment uses the required runtime.
  • The Docker check confirms that the container command is available.
  • The Git check confirms that repository tooling is available.
  • The import check loads PyTorch, FastAPI, and ONNX Runtime in one Python process.

You should see a Python 3.12 version. Docker and Git should each print version details.

The final command should print inspection environment ready.

That is the setup locked down. Your editor, virtual environment, dependencies, repository tools, and container runtime are ready for the same inspection workflow.

One of the checks failed?

If an import fails, confirm that (.venv) still appears at the front of the prompt. Repeat the dependency installation if PowerShell is using another environment.

If Docker or Git fails, confirm the relevant desktop tool is installed before starting a fresh PowerShell terminal. You can also help me identify which environment check is failing.

Your inspection workspace is ready. Next, you will generate synthetic parts and watch the split-integrity guard stop a leaked experiment.

Generate Parts and Guard the Splits

Your pinned PowerShell workspace is ready. The isolated environment gives every later quality check a stable baseline.

A vision model can look excellent when test images leak into training. In this step, you will generate synthetic parts across clean splits. You will then make split leakage visible before training begins.

In this step, get ready to:
  • Generate 64 by 64 grayscale gear and washer images.
  • Audit each image ID through the dataset manifest.
  • Observe the split guard stop an invalid experiment.
Generate the synthetic parts

The supplied generator creates normal parts plus scratched parts with repeatable seeds. It places each image into a training, validation, or test split.

  • Generate the clean dataset by running this command:
python -m src.train --data-only

What does this command create?

The --data-only flag stops the command after dataset generation.

  • The generate_dataset() function removes any older data folder before rebuilding it.
  • The render_part() function creates 64 by 64 grayscale PNG images for gears and washers.
  • The generator stores normal images under normal. It stores scratched images under defect.
  • The command writes data/manifest.csv after assembling every image record.

These images make the integration workflow reproducible on a local CPU. They provide a pipeline prototype instead of a production accuracy benchmark.

You should see split counts: {'train': 240, 'val': 80, 'test': 80}. The command should finish without an exception.

Good start. Your inspection line now has repeatable image data across three clean splits.

Dataset generation stopped?

  • Confirm that your activated environment still shows .venv at the start of the PowerShell prompt.
  • Confirm that PowerShell is still inside the visual-inspection project folder containing src/train.py.
  • Check that src/pipeline.py remains available beside the training CLI.

Help me diagnose the dataset generation failure.

  • Switch back to VS Code.
  • Expand the data folder in the Explorer sidebar.
  • Expand data/train to confirm that gear and washer folders exist.
  • Expand data/val to confirm that both part categories exist.
  • Expand data/test to confirm that both part categories exist.

Each category contains normal and defect folders. The PNG filenames carry their split, category, label, and sequence number.

Audit the manifest

The manifest is a CSV audit trail for every generated image. A stable image identity makes split membership testable before a model sees the data.

  • Select data/manifest.csv from the Explorer sidebar.
  • Confirm that the header contains image_id, split, category, label, and path.
  • Press Cmd+F on macOS or Ctrl+F on Windows to search for train-gear-0-000.

You should find one row. Its split value is train. Its category value is gear. Its label value is 0.

  • Search for test-washer-1-019 using the same Find control.

You should find one row with test membership. The row points to a defect washer PNG under data/test.

How does the manifest support an audit?

  • The image_id field gives each generated image a stable identity.
  • The split field records its training, validation, or test membership.
  • The category field distinguishes gears from washers.
  • The label field uses 0 for normal parts. It uses 1 for scratched parts.
  • The path field connects the record to its PNG file.

The two spot checks confirm the manifest format. The split guard checks every row programmatically.

Trigger the split guard

A quality gate earns trust when it blocks invalid input. You will challenge this gate by assigning one existing image ID to a second split.

Before you run this command, predict whether the duplicate membership can pass the guard.

  • Inject the duplicate membership by running this command:
python -m src.train --data-only --inject-leak

What happened inside the guard?

  • The --inject-leak flag copies one training record.
  • The copied record keeps its original image_id. Its split value becomes test.
  • The assert_disjoint() function maps each image ID to its recorded split membership.
  • An image ID associated with multiple splits raises a ValueError before train_and_report() can begin.

PowerShell ends the run with a non-zero exit. You will see Split leakage detected near the end of the traceback.

That failure proves the gate works. Invalid evaluation data cannot reach model training.

Did the leaked run succeed?

  • Confirm that the command includes the exact --inject-leak flag.
  • Confirm that src/train.py passes args.inject_leak into generate_dataset().
  • Confirm that generate_dataset() calls assert_disjoint() after writing the manifest.

Help me find why the split guard allowed the leaked image ID.

The failed experiment leaves the duplicated row in data/manifest.csv because generation finished before the guard raised the exception. Regenerating without the injection flag replaces that manifest with clean membership.

Before you rerun the clean command, predict whether the duplicate row will remain in the regenerated manifest.

  • Restore the clean dataset by running this command:
python -m src.train --data-only

How is the clean state restored?

The generate_dataset() function deletes the existing data folder at the start of each run. The clean command rebuilds every PNG plus the manifest without adding the duplicate row.

You should see split counts: {'train': 240, 'val': 80, 'test': 80} again. The command should complete with a successful exit.

You have now proved both sides of the gate. Clean membership passes while duplicate membership stops the pipeline.

Clean regeneration still failing?

  • Confirm that the restored command omits the --inject-leak flag.
  • Confirm that no editor has locked a PNG or data/manifest.csv while the generator replaces the data folder.

Help me restore the clean generated dataset.

Your clean dataset is ready for a valid experiment. Next up, you will train the CNN plus the PatchCore-style detector against the same disjoint splits.

Train Inspectors and Calibrate Rejects

Your disjoint dataset now gives every result a trustworthy foundation. The next question is how a labeled classifier compares with a nominal patch-memory detector on the same parts.

In this step, you will train TinyCNN with PyTorch. You will build category-specific patch memory with scikit-learn.

You will compare per-category AUROC for the CNN score stream with the PatchCore-style detector. You will calibrate each reject threshold from normal validation images before MLflow records the experiment.

In this step, get ready to:
  • Train TinyCNN on labeled gear and washer images.
  • Build normal-only patch memory banks for each category.
  • Track thresholds, AUROC, false-reject rate, and defect recall in MLflow.
Train both inspectors

The classifier learns from labeled normal and scratched images. The anomaly detector stores nominal patch embeddings for each part category.

This is the heaviest local compute step in the project. Expect the terminal to stay busy for a few minutes while the CPU trains the model and evaluates both score streams.

  • Before you start the training run, predict whether both score streams will rank defects equally.
  • Train both inspectors by running this command:
python -m src.train

What Does the Training Command Do?

  • The command refreshes the disjoint synthetic dataset through generate_dataset().
  • The train_cnn() function trains TinyCNN for 18 epochs.
  • The frozen convolutional features produce patch embeddings for normal training images.
  • The pipeline builds one nearest-neighbor memory bank for gear images.
  • The pipeline builds a separate nearest-neighbor memory bank for washer images.
  • Normal validation scores set the reject threshold for each detector.
  • The test split produces per-category AUROC plus operating metrics.

When the run finishes, you will see a JSON report containing cnn and patchcore_style results. Each result contains a threshold, per-category AUROC, false-reject rate, and defect recall.

That run has turned your clean dataset into two evaluated inspection methods. The model files and experiment history are now stored in the workspace.

Training Run Stopped?

If the run stops while importing a package, confirm that your PowerShell prompt still begins with .venv.

If the run cannot write an artifact, confirm that your project workspace allows new files and folders.

Help me diagnose the failed training run.

Interpret the reject calibration

AUROC measures how well scores rank normal parts against defective parts across many thresholds. A production decision still needs one fixed operating point for ACCEPT or REJECT.

How Does Threshold Calibration Work?

The pipeline sorts scores from normal validation images. It selects a cutoff that targets FALSE_REJECT_BUDGET = 0.05.

The CNN receives its own threshold. The PatchCore-style score stream receives a separate threshold.

The test labels evaluate the fixed thresholds after calibration. This keeps test outcomes out of the threshold-setting decision.

  • Find false_reject_budget in the terminal report.
  • Confirm its value is 0.05.
  • Compare the gear and washer values inside cnn.auroc.
  • Compare the gear and washer values inside patchcore_style.auroc.
  • Locate threshold for each model.
  • Locate false_reject_rate for each model.
  • Locate defect_recall for each model.
  • Confirm the CNN false_reject_rate is at or below 0.05.

You should see numeric gear and washer AUROC values for both score streams. The CNN false-reject rate should stay within the project budget.

Confirm the saved experiment

The training pipeline saves the model state, patch memories, and evaluation report under artifacts. It also writes the local MLflow tracking database at the top level of the project workspace.

  • Before you refresh the file list, predict which four outputs the training run created.
  • Switch back to the VS Code workspace from setup.
  • Refresh the project file list.
  • Expand the artifacts folder.
  • Confirm model.pt appears inside artifacts.
  • Confirm memory_banks.npz appears inside artifacts.
  • Confirm report.json appears inside artifacts.
  • Confirm mlflow.db appears at the top level of the project workspace.

You should see model.pt inside artifacts. You should also see memory_banks.npz.

The same folder should contain report.json. The project workspace should contain mlflow.db.

What Did MLflow Capture?

  • The tracking URI is sqlite:///mlflow.db.
  • The experiment is named visual-inspection.
  • The run is named cnn-vs-patchcore-style.
  • The parameters record 18 epochs.
  • The parameters record a false-reject budget of 0.05.
  • The metrics record AUROC, false-reject rate, and defect recall for both inspectors.
  • The run stores model.pt as an artifact.
  • The run stores report.json as an artifact.

You now have a tracked comparison with a plant-style reject budget. Next, you will export the CNN to ONNX and test whether its production runtime preserves the same outputs.

Export and Gate the ONNX Service

Your CNN and PatchCore-style detector now have measured performance for gears and washers. The CNN also has a reject threshold calibrated to the plant-style false-reject budget.

A training checkpoint cannot power the inspection line on its own. The exported runtime needs numerical evidence that it agrees with PyTorch.

You will export the classifier to ONNX for ONNX Runtime. You will gate its numerical parity against an explicit tolerance.

You will also gate warmed CPU latency. The validated model then becomes an inspection service through FastAPI.

In this step, get ready to:
  • Export the trained CNN as a checked ONNX graph.
  • Measure numerical parity and warmed CPU p95 latency.
  • Run the REST inspection service and verify its endpoints.
Export and validate the ONNX graph

The supplied src/export.py turns artifacts/model.pt into a portable inference graph. It also checks that the graph is structurally valid before measuring its behavior.

Before you run the export, do you expect conversion alone to prove that the ONNX model behaves like the PyTorch model?

  • Export the model and run its quality gates by running:
python -m src.export

What does the export command do?

  • The script reloads the trained TinyCNN state from artifacts/model.pt.
  • The dynamo exporter creates artifacts/model.onnx with an image input.
  • The graph produces a logit output for the defect classifier.
  • The ONNX checker validates the saved graph before runtime testing begins.

You should see a JSON object containing max_abs_error and p95_ms. The command also creates artifacts/validation.json.

That clears the first production hurdle. Your trained checkpoint now has a checked ONNX artifact beside it.

Export stopped before the JSON result?

  • Confirm artifacts/model.pt exists in the current project workspace.
  • Confirm data/manifest.csv exists in the data folder.
  • Confirm your activated .venv contains the pinned packages from requirements.txt.

Help me diagnose why the ONNX export stops before producing its validation result.

Interpret the parity and latency gates

The exporter runs the same test tensors through both runtimes. It then measures the largest absolute difference between their logits.

  • Find max_abs_error in the JSON output.
  • Confirm its value is at or below 0.0001.
  • Find p95_ms in the JSON output.
  • Confirm its value is at or below 200.0.
  • Use your editor's project file tree to locate artifacts/validation.json.

What do these gates prove?

The max_abs_error gate checks whether ONNX Runtime preserves the PyTorch logits within 1e-4. A passing value provides numerical parity evidence for the tested tensors.

The session warms with 10 untimed runs. It then records 100 CPU inference timings.

The p95_ms value marks the latency that covers 95 percent of those timed runs. The 200.0 millisecond limit is this project's serving requirement.

✔️ Awesome, I've got everything!

The supplied export and API files match the gated serving path used in this step.

ⓧ I'd like to double check the full code

  • Compare your supplied src/export.py with this complete reference file:
import json
import math
import time
from pathlib import Path

import numpy as np
import onnx
import onnxruntime as ort
import torch

from src.pipeline import ARTIFACT_DIR, TinyCNN, load_split

MAX_ABS_ERROR = 1e-4
MAX_P95_MS = 200.0


def main() -> None:
    model = TinyCNN()
    model.load_state_dict(torch.load(ARTIFACT_DIR / "model.pt", weights_only=True))
    model.eval()
    images, _, _ = load_split(Path("data"), "test")
    sample = torch.from_numpy(images[:16])
    example = torch.from_numpy(images[:1])
    program = torch.onnx.export(model, (example,), input_names=["image"], output_names=["logit"], dynamo=True)
    onnx_path = ARTIFACT_DIR / "model.onnx"
    program.save(onnx_path)
    onnx.checker.check_model(onnx.load(onnx_path))
    session = ort.InferenceSession(str(onnx_path), providers=["CPUExecutionProvider"])
    with torch.no_grad():
        torch_output = model(sample).numpy()
    ort_output = session.run(None, {"image": sample.numpy()})[0]
    max_abs_error = float(np.abs(torch_output - ort_output).max())
    for _ in range(10):
        session.run(None, {"image": example.numpy()})
    timings = []
    for _ in range(100):
        start = time.perf_counter()
        session.run(None, {"image": example.numpy()})
        timings.append((time.perf_counter() - start) * 1000.0)
    p95_ms = sorted(timings)[math.ceil(0.95 * len(timings)) - 1]
    result = {"max_abs_error": max_abs_error, "p95_ms": p95_ms, "max_allowed_error": MAX_ABS_ERROR, "max_allowed_p95_ms": MAX_P95_MS}
    (ARTIFACT_DIR / "validation.json").write_text(json.dumps(result, indent=2), encoding="utf-8")
    print(json.dumps(result, indent=2))
    if max_abs_error > MAX_ABS_ERROR or p95_ms > MAX_P95_MS:
        raise SystemExit(1)


if __name__ == "__main__":
    main()

How does the export gate work?

  • The first section restores TinyCNN from the trained state dictionary.
  • The export section saves the dynamo-produced graph as artifacts/model.onnx.
  • The validation section compares logits from PyTorch with logits from ONNX Runtime.
  • The benchmark section exits with a non-zero status when either quality gate fails.
  • Compare your supplied src/api.py with this complete reference file:
import json
import math
import os
from io import BytesIO
from pathlib import Path

import numpy as np
import onnxruntime as ort
import psycopg
from fastapi import FastAPI, UploadFile
from PIL import Image

ARTIFACT_DIR = Path("artifacts")
DATABASE_URL = os.getenv("DATABASE_URL", "")
report = json.loads((ARTIFACT_DIR / "report.json").read_text(encoding="utf-8"))
threshold = float(report["cnn"]["threshold"])
session = ort.InferenceSession(str(ARTIFACT_DIR / "model.onnx"), providers=["CPUExecutionProvider"])
app = FastAPI(title="Visual Inspection Line")


def prepare_image(content: bytes) -> np.ndarray:
    with Image.open(BytesIO(content)) as image:
        array = np.asarray(image.convert("L").resize((64, 64)), dtype=np.float32) / 255.0
    return array[None, None, :, :]


def log_reject(part_id: str, category: str, image_name: str, score: float) -> None:
    if not DATABASE_URL:
        return
    with psycopg.connect(DATABASE_URL) as connection:
        connection.execute("""
            CREATE TABLE IF NOT EXISTS defect_events (
                id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                inspected_at timestamp with time zone DEFAULT CURRENT_TIMESTAMP,
                part_id text NOT NULL,
                category text NOT NULL,
                image_name text NOT NULL,
                score double precision NOT NULL,
                threshold double precision NOT NULL,
                rejected boolean NOT NULL
            )
        """)
        connection.execute(
            "INSERT INTO defect_events (part_id, category, image_name, score, threshold, rejected) VALUES (%s, %s, %s, %s, %s, %s)",
            (part_id, category, image_name, score, threshold, True),
        )


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/inspect")
async def inspect(file: UploadFile, part_id: str = "unknown", category: str = "unknown") -> dict[str, object]:
    tensor = prepare_image(await file.read())
    logit = float(session.run(None, {"image": tensor})[0].reshape(-1)[0])
    score = 1.0 / (1.0 + math.exp(-logit))
    rejected = score > threshold
    if rejected:
        log_reject(part_id, category, file.filename or "uploaded-image", score)
    return {"part_id": part_id, "category": category, "score": score, "threshold": threshold, "decision": "REJECT" if rejected else "ACCEPT"}


@app.get("/events/count")
def event_count() -> dict[str, object]:
    if not DATABASE_URL:
        return {"count": 0, "database": "disabled"}
    with psycopg.connect(DATABASE_URL) as connection:
        connection.execute("CREATE TABLE IF NOT EXISTS defect_events (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, inspected_at timestamp with time zone DEFAULT CURRENT_TIMESTAMP, part_id text NOT NULL, category text NOT NULL, image_name text NOT NULL, score double precision NOT NULL, threshold double precision NOT NULL, rejected boolean NOT NULL)")
        count = connection.execute("SELECT count(*) FROM defect_events").fetchone()[0]
    return {"count": count, "database": "postgresql"}

How does the service stay ready?

  • Module startup reads the calibrated CNN threshold from artifacts/report.json.
  • Module startup creates one ONNX Runtime session with CPUExecutionProvider.
  • The /inspect route converts an uploaded image into the model's input shape.
  • The /events/count route reports the number of persisted reject events when the database is enabled.
Start and inspect the FastAPI service

The supplied API imports the exported graph once during startup. Each inspection request reuses that CPUExecutionProvider session.

Uvicorn imports app from src/api.py and serves it on the local machine. The server keeps this PowerShell window busy until you stop it.

  • Start the inspection service by running:
uvicorn src.api:app --host 127.0.0.1 --port 8000

What does the server command do?

  • The import string loads the FastAPI application named app from src.api.
  • The host value keeps the service available on your local machine.
  • Port 8000 becomes the local entry point for inspection requests.

You should see startup messages while PowerShell remains attached to the running service.

Before you load the documentation, which two operational routes do you expect the service to show?

  • Enter http://127.0.0.1:8000/docs in your browser's address bar.
  • Confirm POST /inspect is listed for image uploads.
  • Confirm GET /events/count is listed for the operational log.

You will see interactive API documentation for both endpoints. This proves the validated ONNX artifact loads successfully inside the service.

Documentation page not loading?

  • Confirm the PowerShell window still shows the running Uvicorn process.
  • Confirm artifacts/model.onnx exists after the export.
  • Confirm artifacts/report.json still contains the calibrated CNN threshold.

Help me debug why the FastAPI documentation page does not load for my local ONNX inspection service.

  • Return to the PowerShell window running Uvicorn.
  • Press Ctrl+C to stop the local service.

Your screenshot confirms that the gated ONNX model is available through /inspect. It also confirms that the operational event count has its own endpoint.

Your model now has parity evidence, a CPU latency result, and a working API surface. Next up, you will run the complete line with Docker Compose and persist rejected parts in PostgreSQL.

Run the Containerized Inspection Line

Your ONNX model now matches PyTorch within the parity gate. The FastAPI service can already turn an uploaded part image into an inspection decision.

A production-style line also needs a controller that can reach the service. Rejected parts need durable records in PostgreSQL.

In this step, Docker Compose runs the API beside its database. GitHub Actions then repeats the training, export, and test gates whenever code changes.

In this step, get ready to:
  • Build the API image and start the containerized inspection services.
  • Run the mock controller against the test images.
  • Confirm the repository quality gates pass in GitHub Actions.
Build and start the line

The supplied compose.yaml coordinates two services. The database must pass its health check before the API starts.

The first image build can take several minutes while Docker downloads the base images and installs the pinned dependencies. A long stream of build logs is expected during this first run.

  • Press the Windows key to open Windows Search.
  • Type Docker Desktop into the search field.
  • Press Enter to start Docker Desktop.
  • Wait for Docker Desktop to indicate that its engine is ready.
  • Build the API image from the supplied Dockerfile by running this command:
docker compose build

Why Docker Compose here?

The build creates one reproducible environment for the API. The Dockerfile installs the pinned packages before training and exporting the model inside that environment.

This keeps the serving artifact tied to the same code that produced it. The line can be rebuilt from the repository instead of depending on one workstation.

The terminal shows the dependency installation followed by model training and export logs. The build finishes with a completed API image.

Image build did not finish?

  • Check that Docker Desktop still indicates its engine is ready.
  • Check that PowerShell remains inside the project workspace containing Dockerfile and compose.yaml.
  • Retry the build after restoring your internet connection if a dependency download stopped.

Ask for help with the build logs: help me diagnose why docker compose build failed in this visual inspection project

  • Start the API and database services by running this command:
docker compose up -d

What starts in the background?

  • The db service runs postgres:18.6-alpine with the inspection database.
  • The health check confirms that PostgreSQL accepts connections before the API starts.
  • The api service receives DATABASE_URL and exposes the inspection endpoints on host port 8000.
  • The inspection_postgres volume keeps database rows between container restarts.

Compose starts the database in the background. It waits for the configured health check before starting the API service.

Services did not start?

  • Check Docker Desktop for a stopped engine before retrying the command.
  • Stop any local process already using port 8000 before starting the API service again.
  • Review the Compose output for the service name that failed to start.

Ask for help with the service startup: help me diagnose why my Docker Compose API or PostgreSQL service did not start

Run the mock controller

The controller acts like equipment on the inspection line. It uploads test images to /inspect and measures each request.

The API writes a row to defect_events only when the decision is REJECT. The final count proves those rejected-part records reached PostgreSQL.

Before you run the controller, predict whether both decision types will appear in its output.

  • Send the generated test images through the containerized line by running this command:
python -m src.controller

What does the controller prove?

  • Each request carries a LINE part ID plus its gear or washer category.
  • The API returns its score plus the calibrated CNN threshold.
  • The controller calculates request-level p95_ms across the inspected images.
  • The final request to /events/count reads the persisted rejected-part count.

You will see part IDs such as LINE-0000 followed by ACCEPT or REJECT decisions. The final line reports p95_ms at or below 200.0.

The same final line reports logged_rejects greater than zero. That is the inspection loop working end to end from image upload to a durable rejection record.

Controller cannot complete the line run?

  • Check that the Compose services are still running before retrying the controller.
  • Check that the generated images still exist under data/test/.
  • Check the controller output for the request that failed before investigating the matching service.

Ask for help with the controller result: help me debug why the visual inspection controller cannot get decisions or persisted event counts

Push the quality gates to GitHub

The supplied workflow turns the local checks into CI gates. Every push or pull request rebuilds the model artifacts before running the test suite.

Use this checkpoint to compare the container and workflow files before publishing them.

✔️ The supplied files match

Great. Keep the supplied Dockerfile, compose.yaml, and .github/workflows/ci.yml unchanged before pushing.

ⓧ I'd like to double check the full code

Compare your complete Dockerfile with this reference.

FROM python:3.12.15-slim

WORKDIR /code
COPY requirements.txt /code/requirements.txt
RUN pip install --no-cache-dir -r /code/requirements.txt
COPY . /code
RUN python -m src.train && python -m src.export

CMD ["uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "8000"]

This image installs the pinned dependencies. It trains and exports the inspection model before starting Uvicorn.

Compare your complete compose.yaml with this reference.

services:
  db:
    image: postgres:18.6-alpine
    environment:
      POSTGRES_USER: inspection
      POSTGRES_PASSWORD: inspection
      POSTGRES_DB: inspection
    volumes:
      - inspection_postgres:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U inspection -d inspection"]
      interval: 5s
      timeout: 5s
      retries: 5

  api:
    build: .
    environment:
      DATABASE_URL: postgresql://inspection:inspection@db:5432/inspection
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  inspection_postgres:

This definition keeps the database in a persistent volume. It starts the API after the database becomes healthy.

Compare your complete .github/workflows/ci.yml with this reference.

name: inspection-quality-gates

on:
  push:
  pull_request:

permissions:
  contents: read

jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v7
        with:
          python-version: "3.12"
          cache: pip
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Build model artifacts
        run: |
          python -m src.train
          python -m src.export
      - name: Run quality gates
        run: pytest

This workflow installs the dependencies before rebuilding the model artifacts. It finishes by running the supplied pytest gates.

Git may request authentication during the push if this machine has not connected to the repository before.

  • Complete the browser sign-in if Git requests authentication.
  • Publish the supplied workflow to the current repository by running these commands:
git add .github/workflows/ci.yml
git commit -m "Add inspection quality gates"
git push

What does this publish?

The first command stages .github/workflows/ci.yml. The second command records that staged workflow in the repository history.

The push sends the current branch to its configured remote. The workflow responds to that push by starting the quality job.

PowerShell shows the commit result followed by the completed push. GitHub now has the workflow file under .github/workflows.

Workflow did not reach GitHub?

  • Check that the current repository has an upstream remote before retrying the push.
  • Complete any pending GitHub authentication request in your browser.
  • Confirm that .github/workflows/ci.yml was included in the commit.

Ask for help with the push: help me diagnose why my Git commit or push did not publish the GitHub Actions workflow

The first workflow run can take a few minutes while the runner installs every pinned package. The changing job status shows that the run is still active.

Before you inspect the latest run, predict whether the same training, export, and test gates will pass on the hosted runner.

  • Open the GitHub repository connected to this project in your browser.
  • Click the Actions tab under the repository name.
  • Select inspection-quality-gates in the workflow list.
  • Open the newest workflow run.
  • Select the quality job.
  • Confirm that Install dependencies, Build model artifacts, and Run quality gates each show a successful status.

You will see the workflow complete after training, ONNX export, and pytest finish successfully. Your repository now blocks regressions in split integrity, per-category reporting, ONNX parity, and CPU latency.

That is the full production-style loop complete. Images reach a containerized API while rejected parts become persistent database events and every code change faces the same quality gates.

Secret mission

Break the Line with New Lighting

A reject threshold only reflects the conditions used during calibration. Apply a bright diagonal lighting shift to normal parts, test the unchanged production model, and capture the resulting false-reject failure as a reproducible robustness artifact.

Clean Up Your Resources

Clean Up Your Resources

Everything generated by the inspection line lives on your Windows machine. The software path is free for personal educational use.

Choose whether to keep the line available for demonstrations, pause its containers for later, or delete the generated experiment resources.

Cost warning

Docker Desktop is free for personal use and education. Larger commercial organizations require a paid subscription under Docker's published terms.

Your organization's Docker plan determines whether a subscription applies to this workspace.

Resources you used:

  • The running Docker Compose stack backed by the persistent inspection_postgres PostgreSQL volume.
  • The baseline synthetic images and split manifest in data/.
  • The lighting-shift images and manifest in data_shift/.
  • The artifacts/ directory containing the PyTorch checkpoint, ONNX model, memory banks, quality reports, and lighting-shift result.
  • The MLflow experiment history stored in mlflow.db.

Keep everything running

No action needed. Choose this if you want the inspection line ready for demonstrations or further experiments.

  • The API and database containers remain available for mock controller runs.
  • Rejected-part events remain available in the inspection_postgres volume.
  • The baseline and lighting-shift datasets remain available in data/ and data_shift/.
  • Your model evidence remains available in artifacts/ and mlflow.db.

Pause - I'll come back to this later

Pausing frees the memory used by the API and database containers. Your generated files and persistent database volume stay ready for your next session.

  • Stop the Compose services by running this command:
docker compose down

This stops the inspection API and PostgreSQL containers. The inspection_postgres volume keeps your rejected-part events.

  • Keep data/, data_shift/, artifacts/, and mlflow.db for future training and robustness experiments.
  • Use the Compose start command from Step 5 when you return to the inspection line.

Delete - I don't want to use this again

Deleting these generated results is permanent. Your supplied source files, tests, Compose definition, and CI workflow stay in the project workspace.

  • Stop the Compose services and delete the persistent database volume by running this command:
docker compose down -v

This removes the API and database containers from the Compose stack. It also deletes the inspection_postgres volume containing every persisted rejected-part event.

  • Delete both generated datasets, every model artifact, and the MLflow database by running these commands:
Remove-Item -Recurse -Force data, data_shift, artifacts
Remove-Item -Force mlflow.db

These commands remove the baseline images, lighting-shift images, model outputs, quality reports, robustness result, and local experiment history.

  • List the remaining project files by running this command:
dir

The command lists what remains in the project workspace. You should no longer see data, data_shift, artifacts, or mlflow.db.

Nice Work!

Nice Work!

You did it! You built a production-style visual inspection line that turns synthetic gear and washer images into measurable ACCEPT or REJECT decisions.

You've learned how to:

  • Protected model evaluation with a split leakage guard. Compared a PyTorch CNN classifier with a PatchCore-style nominal memory detector through per-category AUROC. Calibrated the CNN threshold to a 5% false-reject budget.
  • Exported the production model to ONNX with numerical parity within 1e-4. Measured warmed CPU p95 latency below 200 ms. Served REST inference through FastAPI backed by ONNX Runtime.
  • Containerized the API with Docker Compose. Persisted rejected-part events in PostgreSQL. Used a mock controller to prove both line decisions. Automated the quality gates with GitHub Actions.
  • Completed the optional Secret Mission by reproducing a false-reject failure under a lighting distribution shift. Captured the bright-band robustness result in artifacts/lighting_shift.json.

Ready to quiz yourself?