README.md

---
license: apache-2.0
---
# VisA Anomaly Classifier: Complete Runnable Setup

A Mixture of Experts (MoE) image classification system that identifies industrial product categories and detects anomalies using a product classification router and 12 specialized ResNet-18 experts.

**Hugging Face Model:** https://huggingface.co/bipulsardar/Anomaly_Classifier

## 1. Project Structure

Create the following project structure:

```text
visa-anomaly-classifier/
├── app.py
├── requirements.txt
└── README.md
```

Model weights are downloaded automatically from Hugging Face when the application starts.

## 2. Requirements

- Python 3.10 or later
- Internet access for downloading model weights
- CPU or CUDA-compatible GPU
- Approximately 600 MB for model weights, plus additional runtime memory

## 3. Install Dependencies

Create a virtual environment and install the dependencies.

### Windows

```powershell
python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
```

### Linux / macOS / Google Colab

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

Create `requirements.txt`:

```text
torch
torchvision
huggingface_hub
pillow
numpy
pandas
scikit-learn
tqdm
```

For GPU acceleration, install the PyTorch build compatible with your CUDA environment using the official selector: https://pytorch.org/get-started/locally/

## 4. Complete Application

Create `app.py` with the following code.

```python
import argparse
import json
import time
from pathlib import Path

import numpy as np
import pandas as pd
import torch
import torch.nn as nn

from PIL import Image
from torchvision import models, transforms
from huggingface_hub import snapshot_download
from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score,
    roc_auc_score,
    classification_report,
    confusion_matrix
)

CATEGORIES = [
    "candle", "capsules", "cashew", "chewinggum",
    "fryum", "macaroni1", "macaroni2", "pcb1",
    "pcb2", "pcb3", "pcb4", "pipe_fryum"
]

IMAGE_EXTENSIONS = {
    ".jpg", ".jpeg", ".png", ".bmp", ".tif", ".tiff"
}

REPOSITORY = "bipulsardar/Anomaly_Classifier"
IMAGE_SIZE = 224
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")

TRANSFORM = transforms.Compose([
    transforms.Resize((IMAGE_SIZE, IMAGE_SIZE)),
    transforms.ToTensor(),
    transforms.Normalize(
        mean=[0.485, 0.456, 0.406],
        std=[0.229, 0.224, 0.225]
    )
])

def create_model(num_classes):
    model = models.resnet18(weights=None)
    model.fc = nn.Linear(model.fc.in_features, num_classes)
    return model

def load_model(path, num_classes):
    checkpoint = torch.load(
        path,
        map_location=DEVICE,
        weights_only=True
    )

    state_dict = checkpoint.get("model_state_dict", checkpoint)

    model = create_model(num_classes)
    model.load_state_dict(state_dict)
    model.to(DEVICE)
    model.eval()

    return model

def load_models(model_dir=None, token=None):
    if model_dir:
        directory = Path(model_dir)
    else:
        directory = Path(snapshot_download(
            repo_id=REPOSITORY,
            repo_type="model",
            token=token
        ))

    router = load_model(directory / "router.pth", len(CATEGORIES))

    experts = {
        category: load_model(
            directory / f"expert_{category}.pth",
            2
        )
        for category in CATEGORIES
    }

    return router, experts

def predict_image(image_path, router, experts):
    with Image.open(image_path) as image:
        tensor = TRANSFORM(image.convert("RGB")).unsqueeze(0).to(DEVICE)

    if DEVICE.type == "cuda":
        torch.cuda.synchronize()

    start = time.perf_counter()

    with torch.inference_mode():
        router_probabilities = torch.softmax(router(tensor), dim=1)

        category_id = int(router_probabilities.argmax(dim=1).item())
        category = CATEGORIES[category_id]

        expert_probabilities = torch.softmax(
            experts[category](tensor),
            dim=1
        )

        anomaly_score = float(expert_probabilities[0, 1].item())
        prediction = int(expert_probabilities.argmax(dim=1).item())

    if DEVICE.type == "cuda":
        torch.cuda.synchronize()

    elapsed_ms = (time.perf_counter() - start) * 1000

    return {
        "image_path": str(image_path),
        "category": category,
        "category_confidence": float(
            router_probabilities[0, category_id].item()
        ),
        "prediction": "anomaly" if prediction == 1 else "normal",
        "anomaly_score": anomaly_score,
        "latency_ms": elapsed_ms
    }

def discover_dataset(dataset_dir):
    root = Path(dataset_dir)
    records = []

    for category in CATEGORIES:
        image_root = root / category / "Data" / "Images"

        for folder, label in [("Normal", 0), ("Anomaly", 1)]:
            directory = image_root / folder

            if not directory.exists():
                continue

            for path in directory.rglob("*"):
                if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS:
                    records.append({
                        "image_path": str(path),
                        "true_category": category,
                        "true_label": label
                    })

    dataframe = pd.DataFrame(records)

    if dataframe.empty:
        raise RuntimeError(
            f"No VisA images found under {root}. "
            "Expected category/Data/Images/Normal and Anomaly folders."
        )

    return dataframe.drop_duplicates("image_path").reset_index(drop=True)

def benchmark(dataset_dir, router, experts, output_dir):
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    dataframe = discover_dataset(dataset_dir)
    results = []

    for row in dataframe.itertuples(index=False):
        try:
            result = predict_image(row.image_path, router, experts)
            result["true_category"] = row.true_category
            result["true_label"] = int(row.true_label)
            result["category_correct"] = (
                result["category"] == row.true_category
            )
            results.append(result)
        except Exception as error:
            print(f"Failed: {row.image_path}: {error}")

    result_df = pd.DataFrame(results)

    if result_df.empty:
        raise RuntimeError("No images were successfully evaluated.")

    y_true = result_df["true_label"].to_numpy()
    y_pred = (
        result_df["prediction"].eq("anomaly").astype(int).to_numpy()
    )
    y_score = result_df["anomaly_score"].to_numpy()

    try:
        auc = float(roc_auc_score(y_true, y_score))
    except ValueError:
        auc = None

    summary = {
        "repository": REPOSITORY,
        "device": str(DEVICE),
        "images_evaluated": len(result_df),
        "images_failed": len(dataframe) - len(result_df),
        "router_accuracy": float(
            result_df["category_correct"].mean()
        ),
        "anomaly_accuracy": float(accuracy_score(y_true, y_pred)),
        "anomaly_precision": float(
            precision_score(y_true, y_pred, zero_division=0)
        ),
        "anomaly_recall": float(
            recall_score(y_true, y_pred, zero_division=0)
        ),
        "anomaly_f1": float(
            f1_score(y_true, y_pred, zero_division=0)
        ),
        "anomaly_roc_auc": auc,
        "mean_latency_ms": float(result_df["latency_ms"].mean()),
        "median_latency_ms": float(result_df["latency_ms"].median()),
        "p95_latency_ms": float(result_df["latency_ms"].quantile(0.95))
    }

    result_df.to_csv(output / "predictions.csv", index=False)

    with open(output / "benchmark_summary.json", "w") as file:
        json.dump(summary, file, indent=2)

    print("\nBenchmark summary")
    print(json.dumps(summary, indent=2))

    print("\nClassification report")
    print(classification_report(
        y_true,
        y_pred,
        labels=[0, 1],
        target_names=["Normal", "Anomaly"],
        zero_division=0
    ))

    print("\nConfusion matrix")
    print(confusion_matrix(y_true, y_pred, labels=[0, 1]))

    return result_df, summary

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--image", type=str)
    parser.add_argument("--image-dir", type=str)
    parser.add_argument("--dataset-dir", type=str)
    parser.add_argument("--output-dir", default="benchmark_results")
    parser.add_argument("--model-dir", type=str)
    parser.add_argument("--hf-token", type=str)
    args = parser.parse_args()

    router, experts = load_models(args.model_dir, args.hf_token)

    print(f"Device: {DEVICE}")
    print(f"Loaded {len(experts)} experts.")

    if args.image:
        print(json.dumps(
            predict_image(args.image, router, experts),
            indent=2
        ))
        return

    if args.dataset_dir:
        benchmark(
            args.dataset_dir,
            router,
            experts,
            args.output_dir
        )
        return

    if args.image_dir:
        image_paths = [
            path for path in Path(args.image_dir).rglob("*")
            if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
        ]

        output = []

        for path in image_paths:
            try:
                output.append(predict_image(path, router, experts))
            except Exception as error:
                output.append({
                    "image_path": str(path),
                    "error": str(error)
                })

        Path(args.output_dir).mkdir(parents=True, exist_ok=True)

        pd.DataFrame(output).to_csv(
            Path(args.output_dir) / "predictions.csv",
            index=False
        )

        print(f"Processed {len(output)} images.")
        return

    parser.error(
        "Specify --image, --image-dir, or --dataset-dir."
    )

if __name__ == "__main__":
    main()
```

## 5. Run Predictions

### Single image

```bash
python app.py --image ./product.jpg
```

### Batch image directory

```bash
python app.py --image-dir ./product_images --output-dir ./results
```

The predictions will be saved to `results/predictions.csv`.

### Evaluate the VisA dataset

If the dataset is extracted to `/content/VisA` in Google Colab:

```bash
python app.py --dataset-dir /content/VisA --output-dir /content/benchmark_results
```

On Windows:

```powershell
python app.py --dataset-dir "E:\Datasets\VisA" --output-dir ".\benchmark_results"
```

The benchmark outputs include per-image predictions and aggregate metrics.

## 6. Use in Google Colab

Run these cells in order.

```python
!pip -q install torch torchvision huggingface_hub pillow numpy pandas scikit-learn tqdm
```

Upload `app.py` to Colab or create it using a notebook cell, then run:

```python
!python app.py --image "/content/product.jpg"
```

Benchmark all VisA images:

```python
!python app.py --dataset-dir "/content/VisA" --output-dir "/content/benchmark_results"
```

## 7. Private Hugging Face Repositories

If the repository is private, authenticate before running the application.

```python
from huggingface_hub import login
from kaggle_secrets import UserSecretsClient

secrets = UserSecretsClient()
token = secrets.get_secret("HF_091")

login(token=token)
```

Alternatively, set `HF_TOKEN` in the environment and pass it to `load_models()`. Do not commit access tokens to source control.

## 8. Interpreting Results

The application returns:

- `category`: predicted product category.
- `category_confidence`: router softmax score for the selected category.
- `prediction`: normal or anomaly.
- `anomaly_score`: expert softmax score for the anomaly class.
- `latency_ms`: inference time for the router and selected expert.

These scores are not necessarily calibrated probabilities. Set production thresholds using a separate validation dataset.

## 9. Important Evaluation Notes

The previously reported full-dataset results are preliminary because the evaluation included training images. For a reliable estimate of generalization, evaluate on held-out test images that were never used to train or select the models.

This system performs image-level anomaly classification. It does not currently produce pixel-level defect localization.

## 10. References

- Model repository: https://huggingface.co/bipulsardar/Anomaly_Classifier
- VisA dataset: https://github.com/amazon-science/spot-diff
- PyTorch: https://pytorch.org/
- Torchvision: https://pytorch.org/vision/
- Hugging Face Hub: https://huggingface.co/docs/huggingface_hub