Initial push

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.pth
+data/MNIST
+models/
+.DS_Store

Dockerfile

@@ -0,0 +1,15 @@
+# Dockerfile
+
+FROM python:3.10-slim
+
+# Install dependencies
+RUN apt-get update && apt-get install -y git && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["python", "app.py"]

README.md

@@ -1,13 +1,52 @@
----
-title: Anomaly Detection As A Service
-emoji: 📈
-colorFrom: green
-colorTo: indigo
-sdk: gradio
-sdk_version: 5.33.2
-app_file: app.py
-pinned: false
-short_description: Anomaly Detection on MNIST
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🧠 Anomaly Detection as a Service
+
+A minimalistic anomaly detection microservice powered by an autoencoder trained on MNIST-style grayscale images.  
+This project provides both a backend API and a Gradio web interface to analyze images and identify abnormal patterns.
+
+## 🚀 Features
+
+- 🧪 **FastAPI** backend for anomaly score prediction
+- 🎛️ **Gradio** frontend with drag-and-drop image testing
+- 🔥 Visual output includes anomaly score, reconstruction, and heatmap
+- ⚙️ Pretrained autoencoder (trained on MNIST digit-style grayscale images)
+- 🧰 Easily extendable to new datasets and model architectures
+
+
+## ⚙️ Setup
+
+1. Install dependencies:
+```
+pip install -r requirements.txt
+```
+2. (Optional) Train the model from scratch:
+```
+python src/train.py
+```
+By default, the model will be saved to `models/autoencoder_mnist.pth`.
+
+## 🧪 Run the API Server (FastAPI)
+```
+uvicorn app.main:app --reload
+```
+
+This starts the API at http://127.0.0.1:8000
+
+You can test it using Swagger UI:
+http://127.0.0.1:8000/docs
+
+## 🎛️ Run the Gradio Frontend
+```
+python app.py
+```
+
+This launches an interactive browser interface for:
+
+- Uploading images
+- Viewing anomaly scores
+- Seeing reconstructions and heatmaps side-by-side
+
+## 📝 Notes
+- Input images must be grayscale, 28×28 pixels.
+- To adapt to a custom dataset, modify src/train.py accordingly and retrain.
+- The project is meant for educational/demo purposes and not intended for production deployment.
+

app.py

@@ -0,0 +1,76 @@
+import gradio as gr
+import torch
+from torchvision import transforms as T
+from PIL import Image
+import numpy as np
+import base64
+from io import BytesIO
+import matplotlib.pyplot as plt
+from src.train import load_model, loss_fn
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+model = load_model("./models/autoencoder_mnist.pth")
+model.eval()
+
+transform = T.Compose([
+    T.Grayscale(num_output_channels=1),
+    T.Resize((28, 28)),
+    T.ToTensor()
+])
+
+def predict(image):
+    original = image.convert("L")
+    tensor = transform(original).unsqueeze(0).to(DEVICE)
+    original_resized = original.resize((200, 200), Image.NEAREST)
+
+    with torch.no_grad():
+        decoded = model(tensor)
+        score = loss_fn(decoded, tensor).item()
+        diff = (decoded - tensor).squeeze().cpu().numpy() ** 2
+        reconstructed = decoded.squeeze().cpu().numpy()
+
+    # Anomaly interpretation
+    if score > 0.02:
+        verdict = "🚨 Anomaly detected!"
+    else:
+        verdict = "✅ Everything looks normal."
+
+    # Create heatmap
+    fig, ax = plt.subplots()
+    ax.imshow(diff, cmap="hot")
+    ax.axis("off")
+    buf1 = BytesIO()
+    plt.savefig(buf1, format="png", bbox_inches="tight", pad_inches=0)
+    plt.close(fig)
+    buf1.seek(0)
+    heatmap = Image.open(buf1)
+
+    # Reconstructed image
+    fig2, ax2 = plt.subplots()
+    ax2.imshow(reconstructed, cmap="gray")
+    ax2.axis("off")
+    buf2 = BytesIO()
+    plt.savefig(buf2, format="png", bbox_inches="tight", pad_inches=0)
+    plt.close(fig2)
+    buf2.seek(0)
+    reconstructed_img = Image.open(buf2)
+
+    return f"{score:.5f}", verdict, original_resized, reconstructed_img, heatmap
+
+
+iface = gr.Interface(
+    fn=predict,
+    inputs=gr.Image(type="pil", label="Upload MNIST-style Image"),
+    outputs=[
+        gr.Textbox(label="Anomaly Score"),
+        gr.Textbox(label="Verdict"),
+        gr.Image(label="Original", height=200),
+        gr.Image(label="Reconstruction", height=200),
+        gr.Image(label="Heatmap", height=200),
+    ],
+    allow_flagging="never",
+    title="🧠 Anomaly Detection",
+    description="Upload a grayscale image (28x28) to get anomaly score."
+)
+
+iface.launch(iface.launch(server_name="0.0.0.0", server_port=7860))

app/main.py

@@ -0,0 +1,95 @@
+# app/main.py
+
+import os
+from fastapi import FastAPI, UploadFile, HTTPException
+from fastapi.responses import JSONResponse
+from PIL import Image
+import torch
+from torchvision import transforms as T
+from src.train import load_model, loss_fn  # Load the model and loss function from src/train.py
+import io
+from io import BytesIO
+
+import base64
+import matplotlib.pyplot as plt
+
+import logging
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+from huggingface_hub import hf_hub_download
+
+app = FastAPI()
+
+# Device configuration
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+# Load the model once
+MODEL_PATH = hf_hub_download(repo_id="dmtschulz/anomaly-detection-model", filename="autoencoder_mnist.pth")
+
+log.info("Model loaded. App is ready.")
+
+model = load_model(MODEL_PATH)
+model.eval()
+
+# Transformations
+transform = T.Compose([
+    T.Grayscale(num_output_channels=1),
+    T.Resize((28, 28)),
+    T.ToTensor()
+])
+
+@app.get("/health")
+def health():
+    return {"status": "ok"}
+
+@app.post("/predict", summary="Predict anomaly score from image")
+async def predict(file: UploadFile):
+    try:
+        contents = await file.read()
+        image = Image.open(io.BytesIO(contents)).convert("L")
+        image = transform(image)
+        image = image.unsqueeze(0).to(DEVICE)
+
+        with torch.no_grad():
+            decoded = model(image)
+            score = loss_fn(decoded, image).item()
+            diff = (decoded - image).squeeze().cpu().numpy() ** 2
+
+        # Create heatmap image
+        fig, ax = plt.subplots()
+        ax.imshow(diff, cmap="hot")
+        ax.axis("off")
+        buf = BytesIO()
+        plt.savefig(buf, format="png", bbox_inches="tight", pad_inches=0)
+        plt.close(fig)
+        buf.seek(0)
+
+        # Encode to base64
+        heatmap_b64 = base64.b64encode(buf.read()).decode("utf-8")
+
+        # Create reconstructed image
+        decoded_img = decoded.squeeze().cpu().numpy()
+
+        # Visualize the decoded image as png
+        fig2, ax2 = plt.subplots()
+        ax2.imshow(decoded_img, cmap="gray")
+        ax2.axis("off")
+        buf2 = BytesIO()
+        plt.savefig(buf2, format="png", bbox_inches="tight", pad_inches=0)
+        plt.close(fig2)
+        buf2.seek(0)
+
+        # Decode to base64
+        decoded_b64 = base64.b64encode(buf2.read()).decode("utf-8")
+
+        # Return everything as JSON
+        return JSONResponse(content={
+            "anomaly_score": score,
+            "heatmap": heatmap_b64,
+            "decoded_image": decoded_b64
+        })
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}")
+

docker-compose.yml

@@ -0,0 +1,12 @@
+version: "3.8"
+
+services:
+  frontend:
+    build: .
+    ports:
+      - "7860:7860"
+    volumes:
+      - .:/app
+    environment:
+      - PYTHONUNBUFFERED=1
+    restart: unless-stopped

requirements.txt

@@ -0,0 +1,14 @@
+torch>=2.0.0
+fastapi
+gradio
+numpy
+pandas
+requests
+uvicorn
+pillow
+matplotlib
+plotly
+scikit-learn
+tqdm
+python-multipart
+torchvision

src/train.py

@@ -0,0 +1,125 @@
+# src/train.py
+
+import os
+import torch
+import numpy as np
+from tqdm import tqdm
+from torch import nn, optim
+from torch.utils.data import DataLoader, Subset
+from torchvision import transforms as T
+from torchvision.datasets import MNIST
+from torchvision.utils import save_image
+
+import logging
+
+logging.basicConfig(level=logging.INFO)
+log = logging.getLogger(__name__)
+
+# Device configuration
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+log.info("Using device: %s", DEVICE)
+
+# Constants
+DATA_ROOT = "./data"
+MODEL_DIR = "./models"
+
+os.makedirs(DATA_ROOT, exist_ok=True)
+os.makedirs(MODEL_DIR, exist_ok=True)
+
+# Autoencoder Model
+class Autoencoder(nn.Module):
+    def __init__(self, h: int = 32):
+        super().__init__()
+        self.encoder = nn.Sequential(
+            nn.Flatten(),
+            nn.Linear(784, 12000),
+            nn.Dropout(0.2),
+            nn.LeakyReLU(),
+            nn.Linear(12000, h),
+            nn.LeakyReLU(),
+        )
+        self.decoder = nn.Sequential(
+            nn.Linear(h, 12000),
+            nn.Dropout(0.2),
+            nn.LeakyReLU(),
+            nn.Linear(12000, 784),
+            nn.Sigmoid(),
+        )
+
+    def forward(self, x: torch.Tensor):
+        x = x.view(x.size(0), -1)  # (N, 784)
+        encoded = self.encoder(x)
+        decoded = self.decoder(encoded)
+        decoded = decoded.view(-1, 1, 28, 28)
+        return decoded
+
+# Loss function
+loss_fn = nn.MSELoss()
+
+# Train function
+def train(loader, h=32, epochs=5, save_path=None):
+    model = Autoencoder(h).to(DEVICE)
+    optimizer = optim.Adam(model.parameters(), lr=0.001)
+
+    for epoch in range(epochs):
+        log.info("Epoch %d", epoch + 1)
+        pbar = tqdm(loader)
+        for batch, _ in pbar:
+            batch = batch.to(DEVICE)
+            optimizer.zero_grad()
+            prediction = model(batch)
+            loss = loss_fn(prediction, batch)
+            loss.backward()
+            optimizer.step()
+            pbar.set_description(f"Loss: {loss.item():.4f}")
+
+    if save_path:
+        save_model(model, save_path)
+
+    return model
+
+# Save model
+def save_model(model, path):
+    torch.save(model.state_dict(), path)
+    log.info("Model saved to %s", path)
+
+
+# Load model
+def load_model(path, h=32):
+    model = Autoencoder(h).to(DEVICE)
+    model.load_state_dict(torch.load(path, map_location=DEVICE))
+    model.to(DEVICE)
+    model.eval()
+    log.info("Model loaded from %s", path)
+    return model
+
+# Get DataLoaders
+def get_data_loaders(batch_size=128, anomaly_class_threshold=None):
+    transform = T.ToTensor()
+    train_dataset = MNIST(DATA_ROOT, train=True, download=True, transform=transform)
+    test_dataset = MNIST(DATA_ROOT, train=False, download=True, transform=transform)
+
+    if anomaly_class_threshold is not None:
+        indices = np.where(train_dataset.targets < anomaly_class_threshold)[0]
+        train_dataset = Subset(train_dataset, indices)
+
+    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
+    test_loader = DataLoader(test_dataset, batch_size=batch_size)
+
+    img, _ = test_dataset[6]
+    save_image(img, "data/sample_mnist.png")
+
+    return train_loader, test_loader
+
+# Predict Anomaly Score
+def predict_anomaly_score(model, x):
+    model.eval()
+    with torch.no_grad():
+        reconstruction = model(x.to(DEVICE))
+        score = torch.mean((x - reconstruction.cpu()) ** 2, dim=(1, 2, 3))
+    return score.cpu().numpy()
+
+# Main for getting data and testing the model
+if __name__ == "__main__":
+    train_loader, test_loader = get_data_loaders()
+    model = train(train_loader, h=32, epochs=5, save_path=f"{MODEL_DIR}/autoencoder_mnist.pth")