Deploy Anomaly Detection API

Dockerfile

@@ -0,0 +1,32 @@
+# Dockerfile for Hugging Face Spaces
+FROM python:3.10-slim
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies for OpenCV
+RUN apt-get update && apt-get install -y \
+    libgl1 \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install dependencies
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Download model checkpoint from Google Drive
+RUN pip install gdown && \
+    gdown --id 1ftzxTJUnlxpQFqPlaUozG_JUbl1Qi5tQ -O /app/model_checkpoint.ckpt
+
+# Copy all project files
+COPY . .
+
+# Expose port for Hugging Face Spaces
+EXPOSE 7860
+
+# Set environment variables
+ENV PYTHONUNBUFFERED=1
+ENV PORT=7860
+
+# Start Flask app (direct JSON responses)
+CMD ["python", "app.py"]

README.md

@@ -1,10 +1,98 @@
 ---
-title: Anomaly Detection Api
-emoji: 💻
-colorFrom: red
-colorTo: gray
+title: Anomaly Detection API
+emoji: 🔍
+colorFrom: blue
+colorTo: red
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🔍 Anomaly Detection API
+
+Real-time anomaly detection for electrical components using PatchCore + OpenCV classification.
+
+## 🚀 Quick Start
+
+### API Endpoint
+
+**POST** `/infer`
+
+**Request:**
+```json
+{
+  "image_url": "https://example.com/your-image.jpg"
+}
+```
+
+**Response:**
+```json
+{
+  "label": "Normal",
+  "boxed_url": "https://cloudinary.com/boxed_image.jpg",
+  "mask_url": "https://cloudinary.com/anomaly_mask.png",
+  "filtered_url": "https://cloudinary.com/filtered_anomalies.png",
+  "boxes": []
+}
+```
+
+### Example Usage
+
+```bash
+curl -X POST "https://YOUR_USERNAME-anomaly-detection-api.hf.space/infer" \
+  -H "Content-Type: application/json" \
+  -d '{"image_url": "https://example.com/test.jpg"}'
+```
+
+```python
+import requests
+
+response = requests.post(
+    "https://YOUR_USERNAME-anomaly-detection-api.hf.space/infer",
+    json={"image_url": "https://example.com/test.jpg"}
+)
+
+result = response.json()
+print(f"Classification: {result['label']}")
+print(f"Boxed Image: {result['boxed_url']}")
+```
+
+## 📋 Classification Labels
+
+- **Normal** - No anomalies detected
+- **Full Wire Overload** - Entire wire showing overload
+- **Point Overload (Faulty)** - Localized overload points
+
+## 🔧 Technical Details
+
+- **Model:** PatchCore (anomaly detection)
+- **Classification:** OpenCV-based heuristics
+- **Response Time:** ~5 seconds
+- **Max Image Size:** Unlimited (auto-resized)
+
+## 🌐 Endpoints
+
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/` | GET | API documentation |
+| `/health` | GET | Health check |
+| `/infer` | POST | Run inference |
+
+## 📦 Output Files
+
+All processed images are uploaded to Cloudinary:
+- **boxed_url:** Original image with bounding boxes
+- **mask_url:** Grayscale anomaly heatmap
+- **filtered_url:** Filtered image showing only anomalous regions
+
+## 🛠️ Built With
+
+- PyTorch 2.4.1
+- Anomalib (PatchCore)
+- OpenCV
+- Flask
+- Cloudinary
+
+## 📄 License
+
+MIT License

app.py

@@ -0,0 +1,82 @@
+"""
+Hugging Face Spaces API wrapper
+Provides direct JSON responses without job queues
+"""
+from flask import Flask, request, jsonify
+from inference_core import run_pipeline_for_image, download_image_from_url, upload_to_cloudinary
+import os
+
+app = Flask(__name__)
+
+
+@app.route("/", methods=["GET"])
+def home():
+    """Home page with API documentation"""
+    return jsonify({
+        "service": "Anomaly Detection API",
+        "version": "1.0",
+        "endpoints": {
+            "/health": "GET - Health check",
+            "/infer": "POST - Run inference on image URL"
+        },
+        "example_request": {
+            "method": "POST",
+            "url": "/infer",
+            "body": {
+                "image_url": "https://example.com/image.jpg"
+            }
+        }
+    })
+
+
+@app.route("/health", methods=["GET"])
+def health():
+    """Health check endpoint"""
+    return jsonify({"status": "healthy"}), 200
+
+
+@app.route("/infer", methods=["POST"])
+def infer():
+    """
+    Inference endpoint - returns direct JSON response
+    Request JSON: {"image_url": "https://..."}
+    """
+    try:
+        data = request.get_json()
+        if not data or "image_url" not in data:
+            return jsonify({"error": "Missing image_url"}), 400
+        
+        image_url = data["image_url"]
+        
+        # Download image
+        local_path = download_image_from_url(image_url)
+        
+        # Run pipeline
+        results = run_pipeline_for_image(local_path)
+        
+        # Upload outputs
+        boxed_url = upload_to_cloudinary(results["boxed_path"], folder="pipeline_outputs") if results["boxed_path"] else None
+        mask_url = upload_to_cloudinary(results["mask_path"], folder="pipeline_outputs") if results["mask_path"] else None
+        filtered_url = upload_to_cloudinary(results["filtered_path"], folder="pipeline_outputs") if results["filtered_path"] else None
+        
+        # Clean up
+        if os.path.exists(local_path):
+            os.remove(local_path)
+        
+        # Direct JSON response (no job queue wrapper)
+        return jsonify({
+            "label": results["label"],
+            "boxed_url": boxed_url,
+            "mask_url": mask_url,
+            "filtered_url": filtered_url,
+            "boxes": results.get("boxes", [])
+        })
+        
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+
+if __name__ == "__main__":
+    # For Hugging Face Spaces, use port 7860
+    port = int(os.environ.get("PORT", 7860))
+    app.run(host="0.0.0.0", port=port, debug=False)

configs/patchcore_transformers.yaml

@@ -0,0 +1,27 @@
+model:
+  class_path: anomalib.models.Patchcore
+  init_args:
+    backbone: "wide_resnet50_2"
+    layers: ["layer2", "layer3"]
+    coreset_sampling_ratio: 0.1
+    num_neighbors: 9
+
+data:
+  # You can also use: class_path: anomalib.data.Folder (re-export)
+  class_path: anomalib.data.datamodules.image.folder.Folder
+  init_args:
+    name: "transformers"
+    root: "./dataset"
+    normal_dir: "train/normal"       # ONLY normal images here
+    abnormal_dir: "test/faulty"      # faulty images for eval
+    normal_test_dir: "test/normal"   # a few normals for eval
+    train_batch_size: 4
+    eval_batch_size: 4
+    num_workers: 4
+    # (optional) add augmentations later via train_augmentations/val_augmentations/test_augmentations
+
+trainer:
+  max_epochs: 1
+  accelerator: "auto"
+  devices: 1
+  enable_checkpointing: true

inference_core.py

@@ -0,0 +1,232 @@
+"""
+Core inference module - contains model loading and inference functions
+Can be imported by both Flask app and RunPod handler
+"""
+import os
+import cv2
+import numpy as np
+from PIL import Image
+import torch
+import subprocess
+import sys
+import requests
+import tempfile
+import cloudinary
+import cloudinary.uploader
+
+# ---- Import your PatchCore API ----
+from scripts.patchcore_api_inference import Patchcore, config, device
+
+# ---- Output directories ----
+OUT_MASK_DIR = "api_inference_pred_masks_pipeline"
+OUT_FILTERED_DIR = "api_inference_filtered_pipeline"
+OUT_BOXED_DIR = "api_inference_labeled_boxes_pipeline"
+
+os.makedirs(OUT_MASK_DIR, exist_ok=True)
+os.makedirs(OUT_FILTERED_DIR, exist_ok=True)
+os.makedirs(OUT_BOXED_DIR, exist_ok=True)
+
+# ---- Cloudinary config ----
+cloudinary.config(
+    cloud_name="dtyjmwyrp",
+    api_key="619824242791553",
+    api_secret="l8hHU1GIg1FJ8rDgvHd4Sf7BWMk"
+)
+
+# ---- Load model once ----
+GDRIVE_URL = "1ftzxTJUnlxpQFqPlaUozG_JUbl1Qi5tQ"
+MODEL_CKPT_PATH = os.path.abspath("model_checkpoint.ckpt")
+try:
+    import gdown
+except ImportError:
+    subprocess.check_call([sys.executable, "-m", "pip", "install", "gdown"])
+    import gdown
+
+if not os.path.exists(MODEL_CKPT_PATH):
+    raise FileNotFoundError(f"Model checkpoint not found at {MODEL_CKPT_PATH}. Please rebuild the Docker image to include the model.")
+else:
+    print(f"[INFO] Model checkpoint already exists at {MODEL_CKPT_PATH}, skipping download.")
+
+model = Patchcore.load_from_checkpoint(MODEL_CKPT_PATH, **config.model.init_args)
+model.eval()
+model = model.to(device)
+print("[INFO] Model loaded and ready for inference")
+
+
+def infer_single_image_with_patchcore(image_path: str):
+    """PatchCore inference on a single image"""
+    fixed_path = os.path.abspath(os.path.normpath(image_path))
+    orig_img = Image.open(fixed_path).convert("RGB")
+    orig_w, orig_h = orig_img.size
+
+    img_resized = orig_img.resize((256, 256))
+    img_tensor = torch.from_numpy(np.array(img_resized)).permute(2, 0, 1).float() / 255.0
+    img_tensor = img_tensor.unsqueeze(0).to(device)
+
+    with torch.no_grad():
+        output = model(img_tensor)
+        if hasattr(output, "anomaly_map"):
+            anomaly_map = output.anomaly_map.squeeze().detach().cpu().numpy()
+        elif isinstance(output, (tuple, list)) and len(output) > 1:
+            anomaly_map = output[1].squeeze().detach().cpu().numpy()
+        else:
+            anomaly_map = None
+
+    base = os.path.splitext(os.path.basename(fixed_path))[0]
+    mask_path = None
+    filtered_path = None
+
+    if anomaly_map is not None:
+        norm_map = (255 * (anomaly_map - anomaly_map.min()) / (np.ptp(anomaly_map) + 1e-8)).astype(np.uint8)
+        if norm_map.ndim > 2:
+            norm_map = np.squeeze(norm_map)
+            if norm_map.ndim > 2:
+                norm_map = norm_map[0]
+
+        mask_img_256 = Image.fromarray(norm_map)
+        mask_img = mask_img_256.resize((orig_w, orig_h), resample=Image.BILINEAR)
+
+        mask_path = os.path.join(OUT_MASK_DIR, f"{base}_mask.png")
+        mask_img.save(mask_path)
+
+        bin_mask = np.array(mask_img) > 128
+        orig_np = np.array(orig_img)
+        filtered_np = np.zeros_like(orig_np)
+        filtered_np[bin_mask] = orig_np[bin_mask]
+        filtered_img = Image.fromarray(filtered_np)
+
+        filtered_path = os.path.join(OUT_FILTERED_DIR, f"{base}_filtered.png")
+        filtered_img.save(filtered_path)
+
+        print(f"[PatchCore] Saved mask -> {mask_path}")
+        print(f"[PatchCore] Saved filtered -> {filtered_path}")
+    else:
+        print("[PatchCore] No anomaly_map produced by model.")
+
+    return {
+        "orig_path": fixed_path,
+        "mask_path": mask_path,
+        "filtered_path": filtered_path,
+        "orig_size": (orig_w, orig_h),
+    }
+
+
+def classify_filtered_image(filtered_img_path: str):
+    """OpenCV heuristic classification on filtered image"""
+    img = cv2.imread(filtered_img_path)
+    if img is None:
+        raise FileNotFoundError(f"Could not read filtered image: {filtered_img_path}")
+
+    hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
+    
+    # Color masks
+    blue_mask   = cv2.inRange(hsv, (90, 50, 20), (130, 255, 255))
+    black_mask  = cv2.inRange(hsv, (0, 0, 0), (180, 255, 50))
+    yellow_mask = cv2.inRange(hsv, (20, 130, 130), (35, 255, 255))
+    orange_mask = cv2.inRange(hsv, (10, 100, 100), (25, 255, 255))
+    red_mask1   = cv2.inRange(hsv, (0, 100, 100), (10, 255, 255))
+    red_mask2   = cv2.inRange(hsv, (160, 100, 100), (180, 255, 255))
+    red_mask    = cv2.bitwise_or(red_mask1, red_mask2)
+
+    total = img.shape[0] * img.shape[1]
+    blue_count   = np.sum(blue_mask > 0)
+    black_count  = np.sum(black_mask > 0)
+    yellow_count = np.sum(yellow_mask > 0)
+    orange_count = np.sum(orange_mask > 0)
+    red_count    = np.sum(red_mask > 0)
+
+    label = "Unknown"
+    box_list, label_list = [], []
+
+    # Simplified classification logic (keeping only essential parts)
+    if (blue_count + black_count) / total > 0.8:
+        label = "Normal"
+    elif (red_count + orange_count + yellow_count) / total > 0.7:
+        label = "Full Wire Overload"
+        box_list.append((0, 0, img.shape[1], img.shape[0]))
+        label_list.append(label)
+    else:
+        # Point overloads detection (simplified)
+        min_area_faulty = 120
+        max_area = 0.05 * total
+        
+        contours, _ = cv2.findContours(red_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if min_area_faulty < area < max_area:
+                x, y, w, h = cv2.boundingRect(cnt)
+                box_list.append((x, y, w, h))
+                label_list.append("Point Overload (Faulty)")
+
+    return label, box_list, label_list, img
+
+
+def run_pipeline_for_image(image_path: str):
+    """Complete pipeline: PatchCore + classification + drawing"""
+    # 1) PatchCore inference
+    pc_out = infer_single_image_with_patchcore(image_path)
+    filtered_path = pc_out["filtered_path"]
+    orig_path = pc_out["orig_path"]
+
+    if filtered_path is None:
+        filtered_path = orig_path
+
+    # 2) Classify
+    label, boxes, labels, _filtered_bgr = classify_filtered_image(filtered_path)
+
+    # 3) Draw boxes on original image
+    draw_img = cv2.imread(orig_path)
+    if draw_img is None:
+        raise FileNotFoundError(f"Could not read original image: {orig_path}")
+
+    for (x, y, w, h), l in zip(boxes, labels):
+        cv2.rectangle(draw_img, (x, y), (x + w, y + h), (0, 0, 255), 2)
+        cv2.putText(draw_img, l, (x, max(0, y - 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)
+
+    if not boxes:
+        cv2.putText(draw_img, label, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)
+
+    base = os.path.splitext(os.path.basename(orig_path))[0]
+    ext = os.path.splitext(os.path.basename(orig_path))[1]
+    out_boxed_path = os.path.join(OUT_BOXED_DIR, f"{base}_boxed{ext if ext else '.png'}")
+    ok = cv2.imwrite(out_boxed_path, draw_img)
+    if not ok:
+        out_boxed_path = os.path.join(OUT_BOXED_DIR, f"{base}_boxed.png")
+        cv2.imwrite(out_boxed_path, draw_img)
+
+    print(f"[Pipeline] Classification label: {label}")
+    print(f"[Pipeline] Saved boxes-on-original -> {out_boxed_path}")
+    
+    return {
+        "label": label,
+        "boxed_path": out_boxed_path,
+        "mask_path": pc_out["mask_path"],
+        "filtered_path": pc_out["filtered_path"],
+        "boxes": [
+            {"box": [int(x), int(y), int(w), int(h)], "type": l}
+            for (x, y, w, h), l in zip(boxes, labels)
+        ]
+    }
+
+
+def download_image_from_url(url):
+    """Download image from URL to temp file"""
+    import requests
+    import tempfile
+    response = requests.get(url, stream=True)
+    if response.status_code != 200:
+        raise Exception(f"Failed to download image from {url}")
+    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".jpg")
+    for chunk in response.iter_content(1024):
+        tmp.write(chunk)
+    tmp.close()
+    return tmp.name
+
+
+def upload_to_cloudinary(file_path, folder=None):
+    """Upload file to Cloudinary"""
+    upload_opts = {"resource_type": "image"}
+    if folder:
+        upload_opts["folder"] = folder
+    result = cloudinary.uploader.upload(file_path, **upload_opts)
+    return result["secure_url"]

requirements.txt

@@ -0,0 +1,15 @@
+anomalib[full]
+cloudinary
+flask
+gdown
+numpy
+opencv-python
+omegaconf
+pillow
+python-dotenv
+pytorch-lightning==2.4.0
+requests
+torch==2.4.1
+torchvision
+gunicorn
+runpod

scripts/augment_patchcore_dataset.py

@@ -0,0 +1,71 @@
+import os
+from PIL import Image, ImageOps, ImageEnhance
+import random
+
+# Directory containing normal images for training
+NORMAL_DIR = 'dataset/train/normal'
+TARGET_COUNT = 100
+
+# List all images in the directory
+images = [f for f in os.listdir(NORMAL_DIR) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+current_count = len(images)
+print(f"[INFO] Found {current_count} images in {NORMAL_DIR}")
+
+if current_count == 0:
+    print("[ERROR] No images found to augment.")
+    exit(1)
+
+# Augmentation functions
+AUGS = [
+    lambda img: img.rotate(random.randint(-30, 30)),
+    lambda img: ImageOps.mirror(img),
+    lambda img: ImageOps.flip(img),
+    lambda img: ImageEnhance.Brightness(img).enhance(random.uniform(0.7, 1.3)),
+    lambda img: ImageEnhance.Contrast(img).enhance(random.uniform(0.7, 1.3)),
+    lambda img: ImageEnhance.Color(img).enhance(random.uniform(0.7, 1.3)),
+]
+
+aug_idx = 0
+while current_count < TARGET_COUNT:
+    for fname in images:
+        if current_count >= TARGET_COUNT:
+            break
+        img_path = os.path.join(NORMAL_DIR, fname)
+        img = Image.open(img_path).convert('RGB')
+        aug = random.choice(AUGS)
+        aug_img = aug(img)
+        aug_fname = f"aug_{aug_idx}_{fname}"
+        aug_img.save(os.path.join(NORMAL_DIR, aug_fname))
+        aug_idx += 1
+        current_count += 1
+        print(f"[AUG] Saved {aug_fname}")
+
+print(f"[DONE] {NORMAL_DIR} now contains {current_count} images.")
+
+# Repeat for faulty images in dataset/test/faulty
+FAULTY_DIR = 'dataset/test/faulty'
+TARGET_COUNT = 100
+images = [f for f in os.listdir(FAULTY_DIR) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+current_count = len(images)
+print(f"[INFO] Found {current_count} images in {FAULTY_DIR}")
+
+if current_count == 0:
+    print("[ERROR] No faulty images found to augment.")
+    exit(1)
+
+aug_idx = 0
+while current_count < TARGET_COUNT:
+    for fname in images:
+        if current_count >= TARGET_COUNT:
+            break
+        img_path = os.path.join(FAULTY_DIR, fname)
+        img = Image.open(img_path).convert('RGB')
+        aug = random.choice(AUGS)
+        aug_img = aug(img)
+        aug_fname = f"aug_{aug_idx}_{fname}"
+        aug_img.save(os.path.join(FAULTY_DIR, aug_fname))
+        aug_idx += 1
+        current_count += 1
+        print(f"[AUG] Saved {aug_fname}")
+
+print(f"[DONE] {FAULTY_DIR} now contains {current_count} images.")

scripts/balance_train_folders.py

@@ -0,0 +1,66 @@
+import os
+from PIL import Image, ImageOps, ImageEnhance
+import random
+import shutil
+
+TRAIN_NORMAL = 'dataset/train/normal'
+TRAIN_FAULTY = 'dataset/train/faulty'
+TEST_NORMAL = 'dataset/test/normal'
+TEST_FAULTY = 'dataset/test/faulty'
+TARGET_COUNT = 100
+
+os.makedirs(TRAIN_NORMAL, exist_ok=True)
+os.makedirs(TRAIN_FAULTY, exist_ok=True)
+
+# Helper: move images from test to train if needed
+def move_images(src, dst, needed):
+    imgs = [f for f in os.listdir(src) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+    moved = 0
+    for f in imgs:
+        if moved >= needed:
+            break
+        shutil.move(os.path.join(src, f), os.path.join(dst, f))
+        moved += 1
+    return moved
+
+# 1. Move images from test to train if train folders have < TARGET_COUNT
+normal_needed = TARGET_COUNT - len([f for f in os.listdir(TRAIN_NORMAL) if f.lower().endswith(('.jpg', '.jpeg', '.png'))])
+faulty_needed = TARGET_COUNT - len([f for f in os.listdir(TRAIN_FAULTY) if f.lower().endswith(('.jpg', '.jpeg', '.png'))])
+if normal_needed > 0:
+    move_images(TEST_NORMAL, TRAIN_NORMAL, normal_needed)
+if faulty_needed > 0:
+    move_images(TEST_FAULTY, TRAIN_FAULTY, faulty_needed)
+
+# 2. Augment if still not enough
+AUGS = [
+    lambda img: img.rotate(random.randint(-30, 30)),
+    lambda img: ImageOps.mirror(img),
+    lambda img: ImageOps.flip(img),
+    lambda img: ImageEnhance.Brightness(img).enhance(random.uniform(0.7, 1.3)),
+    lambda img: ImageEnhance.Contrast(img).enhance(random.uniform(0.7, 1.3)),
+    lambda img: ImageEnhance.Color(img).enhance(random.uniform(0.7, 1.3)),
+]
+
+def augment_to_count(folder, target):
+    images = [f for f in os.listdir(folder) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+    current_count = len(images)
+    aug_idx = 0
+    while current_count < target:
+        for fname in images:
+            if current_count >= target:
+                break
+            img_path = os.path.join(folder, fname)
+            img = Image.open(img_path).convert('RGB')
+            aug = random.choice(AUGS)
+            aug_img = aug(img)
+            aug_fname = f"aug_{aug_idx}_{fname}"
+            aug_img.save(os.path.join(folder, aug_fname))
+            aug_idx += 1
+            current_count += 1
+            print(f"[AUG] Saved {aug_fname} in {folder}")
+
+augment_to_count(TRAIN_NORMAL, TARGET_COUNT)
+augment_to_count(TRAIN_FAULTY, TARGET_COUNT)
+
+print(f"[DONE] {TRAIN_NORMAL} images: {len(os.listdir(TRAIN_NORMAL))}")
+print(f"[DONE] {TRAIN_FAULTY} images: {len(os.listdir(TRAIN_FAULTY))}")

scripts/batch_patchcore_infer.py

@@ -0,0 +1,54 @@
+import os
+import torch
+from torchvision import transforms
+from PIL import Image
+from anomalib.models.image.patchcore import Patchcore
+import pandas as pd
+
+# Path to your trained checkpoint
+CKPT_PATH = "results/Patchcore/transformers/v2/weights/lightning/model.ckpt"
+# Folder with images to score
+INPUT_DIR = "./dataset/test/faulty"  # Change as needed
+# Output CSV for scores
+OUTPUT_CSV = "patchcore_batch_scores.csv"
+
+# Load PatchCore model
+model = Patchcore.load_from_checkpoint(CKPT_PATH)
+model.eval()
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+model = model.to(device)
+model.eval()
+
+# Define image transforms (should match your training transforms)
+transform = transforms.Compose([
+    transforms.Resize((256, 256)),  # Change to your image_size if different
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+])
+
+def load_image(path):
+    img = Image.open(path).convert("RGB")
+    return transform(img)
+
+# Score all images in the input directory
+results = []
+for fname in sorted(os.listdir(INPUT_DIR)):
+    if not fname.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.tif', '.tiff')):
+        continue
+    fpath = os.path.join(INPUT_DIR, fname)
+    img_tensor = load_image(fpath).unsqueeze(0)  # Add batch dim
+    with torch.no_grad():
+           output = model(img_tensor.to(device))  # Move to device
+        # PatchCore returns a namedtuple with anomaly_score
+           if hasattr(output, 'anomaly_score'):
+               score = output.anomaly_score.item()
+           elif isinstance(output, (tuple, list)):
+               score = float(output[0])
+           else:
+               raise RuntimeError("Unknown PatchCore output type: {}".format(type(output)))
+    results.append({"image": fname, "anomaly_score": score})
+    print(f"{fname}: {score:.4f}")
+
+# Save results to CSV
+pd.DataFrame(results).to_csv(OUTPUT_CSV, index=False)
+print(f"Saved batch scores to {OUTPUT_CSV}")

scripts/classify_filtered_images_opencv.py

@@ -0,0 +1,276 @@
+import cv2
+import numpy as np
+import os
+
+# Directory containing filtered images
+dir_path = 'api_inference_filtered'
+output_dir = 'api_inference_labeled_boxes'
+os.makedirs(output_dir, exist_ok=True)
+
+# IOU function for non-max suppression
+def iou(boxA, boxB):
+    xA = max(boxA[0], boxB[0])
+    yA = max(boxA[1], boxB[1])
+    xB = min(boxA[0]+boxA[2], boxB[0]+boxB[2])
+    yB = min(boxA[1]+boxA[3], boxB[1]+boxB[3])
+    interW = max(0, xB - xA)
+    interH = max(0, yB - yA)
+    interArea = interW * interH
+    boxAArea = boxA[2] * boxA[3]
+    boxBArea = boxB[2] * boxB[3]
+    iou = interArea / float(boxAArea + boxBArea - interArea + 1e-6)
+    return iou
+
+# Merge close bounding boxes (same label, centers within dist_thresh)
+def merge_close_boxes(boxes, labels, dist_thresh=20):
+    merged = []
+    merged_labels = []
+    used = [False]*len(boxes)
+    for i in range(len(boxes)):
+        if used[i]:
+            continue
+        x1, y1, w1, h1 = boxes[i]
+        label1 = labels[i]
+        x2, y2, w2, h2 = x1, y1, w1, h1
+        for j in range(i+1, len(boxes)):
+            if used[j]:
+                continue
+            bx, by, bw, bh = boxes[j]
+            # If boxes are close (distance between centers < dist_thresh)
+            cx1, cy1 = x1 + w1//2, y1 + h1//2
+            cx2, cy2 = bx + bw//2, by + bh//2
+            if abs(cx1-cx2) < dist_thresh and abs(cy1-cy2) < dist_thresh and label1 == labels[j]:
+                # Merge boxes
+                x2 = min(x2, bx)
+                y2 = min(y2, by)
+                w2 = max(x1+w1, bx+bw) - x2
+                h2 = max(y1+h1, by+bh) - y2
+                used[j] = True
+        merged.append((x2, y2, w2, h2))
+        merged_labels.append(label1)
+        used[i] = True
+    return merged, merged_labels
+
+# Non-max suppression using IOU
+def non_max_suppression_iou(boxes, labels, iou_thresh=0.4):
+    if len(boxes) == 0:
+        return [], []
+    idxs = np.argsort([w*h for (x, y, w, h) in boxes])[::-1]
+    keep = []
+    keep_labels = []
+    while len(idxs) > 0:
+        i = idxs[0]
+        keep.append(boxes[i])
+        keep_labels.append(labels[i])
+        remove = [0]
+        for j in range(1, len(idxs)):
+            if iou(boxes[i], boxes[idxs[j]]) > iou_thresh:
+                remove.append(j)
+        idxs = np.delete(idxs, remove)
+    return keep, keep_labels
+
+# Filter out potential boxes that contain a faulty box inside
+def filter_faulty_inside_potential(boxes, labels):
+    filtered_boxes = []
+    filtered_labels = []
+    for i, (box, label) in enumerate(zip(boxes, labels)):
+        if label == 'Point Overload (Potential)':
+            # Check if any faulty box is inside this potential box
+            keep = True
+            for j, (fbox, flabel) in enumerate(zip(boxes, labels)):
+                if flabel == 'Point Overload (Faulty)':
+                    # Check if faulty box is inside potential box
+                    x, y, w, h = box
+                    fx, fy, fw, fh = fbox
+                    if fx >= x and fy >= y and fx+fw <= x+w and fy+fh <= y+h:
+                        keep = False
+                        break
+            if keep:
+                filtered_boxes.append(box)
+                filtered_labels.append(label)
+        else:
+            filtered_boxes.append(box)
+            filtered_labels.append(label)
+    return filtered_boxes, filtered_labels
+
+# Remove potential boxes that overlap with a faulty box (not just inside)
+def filter_faulty_overlapping_potential(boxes, labels):
+    # Remove potential boxes that overlap at all with a faulty box (any intersection)
+    filtered_boxes = []
+    filtered_labels = []
+    def is_overlapping(boxA, boxB):
+        xA = max(boxA[0], boxB[0])
+        yA = max(boxA[1], boxB[1])
+        xB = min(boxA[0]+boxA[2], boxB[0]+boxB[2])
+        yB = min(boxA[1]+boxA[3], boxB[1]+boxB[3])
+        return (xB > xA) and (yB > yA)
+    for i, (box, label) in enumerate(zip(boxes, labels)):
+        if label == 'Point Overload (Potential)':
+            keep = True
+            for j, (fbox, flabel) in enumerate(zip(boxes, labels)):
+                if flabel == 'Point Overload (Faulty)':
+                    if is_overlapping(box, fbox):
+                        keep = False
+                        break
+            if keep:
+                filtered_boxes.append(box)
+                filtered_labels.append(label)
+        else:
+            filtered_boxes.append(box)
+            filtered_labels.append(label)
+    return filtered_boxes, filtered_labels
+
+# Heuristic classification function
+def classify_image(img_path):
+    img = cv2.imread(img_path)
+    hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
+
+    # Color masks
+    blue_mask = cv2.inRange(hsv, (90, 50, 20), (130, 255, 255))
+    black_mask = cv2.inRange(hsv, (0, 0, 0), (180, 255, 50))
+    yellow_mask = cv2.inRange(hsv, (20, 130, 130), (35, 255, 255))  # increased threshold
+    orange_mask = cv2.inRange(hsv, (10, 100, 100), (25, 255, 255))
+    red_mask1 = cv2.inRange(hsv, (0, 100, 100), (10, 255, 255))
+    red_mask2 = cv2.inRange(hsv, (160, 100, 100), (180, 255, 255))
+    red_mask = cv2.bitwise_or(red_mask1, red_mask2)
+
+    total = img.shape[0] * img.shape[1]
+    blue_count = np.sum(blue_mask > 0)
+    black_count = np.sum(black_mask > 0)
+    yellow_count = np.sum(yellow_mask > 0)
+    orange_count = np.sum(orange_mask > 0)
+    red_count = np.sum(red_mask > 0)
+
+    label = 'Unknown'
+    box_list = []
+    label_list = []
+
+    # Full image checks
+    if (blue_count + black_count) / total > 0.8:
+        label = 'Normal'
+    elif (red_count + orange_count) / total > 0.5:
+        label = 'Full Wire Overload'
+    elif (yellow_count) / total > 0.5:
+        label = 'Full Wire Overload'
+    # Check for full wire overload (entire image reddish or yellowish)
+    full_wire_thresh = 0.7  # 70% of image is reddish or yellowish
+    if (red_count + orange_count + yellow_count) / total > full_wire_thresh:
+        label = 'Full Wire Overload'
+        # Add a box covering the whole image
+        box_list.append((0, 0, img.shape[1], img.shape[0]))
+        label_list.append(label)
+    else:
+        # Small spot checks (improved: filter tiny spots, merge overlapping boxes)
+        min_area_faulty = 120  # increased min area for red/orange (faulty)
+        min_area_potential = 1000  # much higher min area for yellow (potential)
+        max_area = 0.05 * total
+        # Faulty (red/orange) spots
+        for mask, spot_label, min_a in [
+            (red_mask, 'Point Overload (Faulty)', min_area_faulty),
+            (yellow_mask, 'Point Overload (Potential)', min_area_potential)
+        ]:
+            contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+            for cnt in contours:
+                area = cv2.contourArea(cnt)
+                if min_a < area < max_area:
+                    x, y, w, h = cv2.boundingRect(cnt)
+                    box_list.append((x, y, w, h))
+                    label_list.append(spot_label)
+        # Middle area checks
+        h, w = img.shape[:2]
+        center = img[h//4:3*h//4, w//4:3*w//4]
+        center_hsv = cv2.cvtColor(center, cv2.COLOR_BGR2HSV)
+        center_yellow = cv2.inRange(center_hsv, (20, 130, 130), (35, 255, 255))
+        center_orange = cv2.inRange(center_hsv, (10, 100, 100), (25, 255, 255))
+        center_red1 = cv2.inRange(center_hsv, (0, 100, 100), (10, 255, 255))
+        center_red2 = cv2.inRange(center_hsv, (160, 100, 100), (180, 255, 255))
+        center_red = cv2.bitwise_or(center_red1, center_red2)
+        if np.sum(center_red > 0) + np.sum(center_orange > 0) > 0.1 * center.size:
+            label = 'Loose Joint (Faulty)'
+            box_list.append((w//4, h//4, w//2, h//2))
+            label_list.append(label)
+        elif np.sum(center_yellow > 0) > 0.1 * center.size:
+            label = 'Loose Joint (Potential)'
+            box_list.append((w//4, h//4, w//2, h//2))
+            label_list.append(label)
+    # Always check for tiny spots, even if image is labeled as Normal
+    min_area_tiny = 10
+    max_area_tiny = 30
+    for mask, spot_label in [
+        (red_mask, 'Tiny Faulty Spot'),
+        (yellow_mask, 'Tiny Potential Spot')
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if min_area_tiny < area < max_area_tiny:
+                x, y, w, h = cv2.boundingRect(cnt)
+                box_list.append((x, y, w, h))
+                label_list.append(spot_label)
+    # Detect wire-shaped (long, thin) regions for wire overloads only
+    aspect_ratio_thresh = 5
+    min_strip_area = 0.01 * total
+    wire_boxes = []
+    wire_labels = []
+    for mask, strip_label in [
+        (red_mask, 'Wire Overload (Red Strip)'),
+        (yellow_mask, 'Wire Overload (Yellow Strip)'),
+        (orange_mask, 'Wire Overload (Orange Strip)')
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if area > min_strip_area:
+                x, y, w, h = cv2.boundingRect(cnt)
+                aspect_ratio = max(w, h) / (min(w, h) + 1e-6)
+                if aspect_ratio > aspect_ratio_thresh:
+                    wire_boxes.append((x, y, w, h))
+                    wire_labels.append(strip_label)
+    # Add wire overloads to box_list/label_list
+    box_list = wire_boxes[:]
+    label_list = wire_labels[:]
+    # For point overloads, do not require wire shape
+    min_area_faulty = 120
+    min_area_potential = 1000
+    max_area = 0.05 * total
+    for mask, spot_label, min_a in [
+        (red_mask, 'Point Overload (Faulty)', min_area_faulty),
+        (yellow_mask, 'Point Overload (Potential)', min_area_potential)
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if min_a < area < max_area:
+                x, y, w, h = cv2.boundingRect(cnt)
+                box_list.append((x, y, w, h))
+                label_list.append(spot_label)
+    # Remove overlapping boxes using IOU
+    box_list, label_list = non_max_suppression_iou(box_list, label_list, iou_thresh=0.4)
+    box_list, label_list = filter_faulty_inside_potential(box_list, label_list)
+    box_list, label_list = filter_faulty_overlapping_potential(box_list, label_list)
+    box_list, label_list = merge_close_boxes(box_list, label_list, dist_thresh=100)
+    return label, box_list, label_list, img
+
+# Batch process all images in the directory
+for fname in os.listdir(dir_path):
+    if not fname.lower().endswith(('.jpg', '.jpeg', '.png')):
+        continue
+    label, box_list, label_list, img = classify_image(os.path.join(dir_path, fname))
+    # Load the original (unfiltered) image for drawing boxes
+    orig_dir = 'api_inference_pred_masks'  # or the directory with original images
+    orig_img_path = os.path.join(orig_dir, fname)
+    if os.path.exists(orig_img_path):
+        draw_img = cv2.imread(orig_img_path)
+        if draw_img is None:
+            draw_img = img.copy()
+    else:
+        draw_img = img.copy()
+    # Draw bounding boxes and labels on the original image
+    for (x, y, w, h), l in zip(box_list, label_list):
+        cv2.rectangle(draw_img, (x, y), (x+w, y+h), (0, 0, 255), 2)
+        cv2.putText(draw_img, l, (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)
+    if not box_list:
+        cv2.putText(draw_img, label, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)
+    out_path = os.path.join(output_dir, fname)
+    cv2.imwrite(out_path, draw_img)
+    print(f"{fname}: {label} (saved with boxes on original image)")

scripts/classify_single_image_opencv.py

@@ -0,0 +1,276 @@
+import cv2
+import numpy as np
+import os
+
+# Directory containing filtered images
+dir_path = 'api_inference_filtered_pipeline'
+output_dir = 'api_inference_labeled_boxes_pipeline'
+os.makedirs(output_dir, exist_ok=True)
+
+# IOU function for non-max suppression
+def iou(boxA, boxB):
+    xA = max(boxA[0], boxB[0])
+    yA = max(boxA[1], boxB[1])
+    xB = min(boxA[0]+boxA[2], boxB[0]+boxB[2])
+    yB = min(boxA[1]+boxA[3], boxB[1]+boxB[3])
+    interW = max(0, xB - xA)
+    interH = max(0, yB - yA)
+    interArea = interW * interH
+    boxAArea = boxA[2] * boxA[3]
+    boxBArea = boxB[2] * boxB[3]
+    iou = interArea / float(boxAArea + boxBArea - interArea + 1e-6)
+    return iou
+
+# Merge close bounding boxes (same label, centers within dist_thresh)
+def merge_close_boxes(boxes, labels, dist_thresh=20):
+    merged = []
+    merged_labels = []
+    used = [False]*len(boxes)
+    for i in range(len(boxes)):
+        if used[i]:
+            continue
+        x1, y1, w1, h1 = boxes[i]
+        label1 = labels[i]
+        x2, y2, w2, h2 = x1, y1, w1, h1
+        for j in range(i+1, len(boxes)):
+            if used[j]:
+                continue
+            bx, by, bw, bh = boxes[j]
+            # If boxes are close (distance between centers < dist_thresh)
+            cx1, cy1 = x1 + w1//2, y1 + h1//2
+            cx2, cy2 = bx + bw//2, by + bh//2
+            if abs(cx1-cx2) < dist_thresh and abs(cy1-cy2) < dist_thresh and label1 == labels[j]:
+                # Merge boxes
+                x2 = min(x2, bx)
+                y2 = min(y2, by)
+                w2 = max(x1+w1, bx+bw) - x2
+                h2 = max(y1+h1, by+bh) - y2
+                used[j] = True
+        merged.append((x2, y2, w2, h2))
+        merged_labels.append(label1)
+        used[i] = True
+    return merged, merged_labels
+
+# Non-max suppression using IOU
+def non_max_suppression_iou(boxes, labels, iou_thresh=0.4):
+    if len(boxes) == 0:
+        return [], []
+    idxs = np.argsort([w*h for (x, y, w, h) in boxes])[::-1]
+    keep = []
+    keep_labels = []
+    while len(idxs) > 0:
+        i = idxs[0]
+        keep.append(boxes[i])
+        keep_labels.append(labels[i])
+        remove = [0]
+        for j in range(1, len(idxs)):
+            if iou(boxes[i], boxes[idxs[j]]) > iou_thresh:
+                remove.append(j)
+        idxs = np.delete(idxs, remove)
+    return keep, keep_labels
+
+# Filter out potential boxes that contain a faulty box inside
+def filter_faulty_inside_potential(boxes, labels):
+    filtered_boxes = []
+    filtered_labels = []
+    for i, (box, label) in enumerate(zip(boxes, labels)):
+        if label == 'Point Overload (Potential)':
+            # Check if any faulty box is inside this potential box
+            keep = True
+            for j, (fbox, flabel) in enumerate(zip(boxes, labels)):
+                if flabel == 'Point Overload (Faulty)':
+                    # Check if faulty box is inside potential box
+                    x, y, w, h = box
+                    fx, fy, fw, fh = fbox
+                    if fx >= x and fy >= y and fx+fw <= x+w and fy+fh <= y+h:
+                        keep = False
+                        break
+            if keep:
+                filtered_boxes.append(box)
+                filtered_labels.append(label)
+        else:
+            filtered_boxes.append(box)
+            filtered_labels.append(label)
+    return filtered_boxes, filtered_labels
+
+# Remove potential boxes that overlap with a faulty box (not just inside)
+def filter_faulty_overlapping_potential(boxes, labels):
+    # Remove potential boxes that overlap at all with a faulty box (any intersection)
+    filtered_boxes = []
+    filtered_labels = []
+    def is_overlapping(boxA, boxB):
+        xA = max(boxA[0], boxB[0])
+        yA = max(boxA[1], boxB[1])
+        xB = min(boxA[0]+boxA[2], boxB[0]+boxB[2])
+        yB = min(boxA[1]+boxA[3], boxB[1]+boxB[3])
+        return (xB > xA) and (yB > yA)
+    for i, (box, label) in enumerate(zip(boxes, labels)):
+        if label == 'Point Overload (Potential)':
+            keep = True
+            for j, (fbox, flabel) in enumerate(zip(boxes, labels)):
+                if flabel == 'Point Overload (Faulty)':
+                    if is_overlapping(box, fbox):
+                        keep = False
+                        break
+            if keep:
+                filtered_boxes.append(box)
+                filtered_labels.append(label)
+        else:
+            filtered_boxes.append(box)
+            filtered_labels.append(label)
+    return filtered_boxes, filtered_labels
+
+# Heuristic classification function
+def classify_image(img_path):
+    img = cv2.imread(img_path)
+    hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
+
+    # Color masks
+    blue_mask = cv2.inRange(hsv, (90, 50, 20), (130, 255, 255))
+    black_mask = cv2.inRange(hsv, (0, 0, 0), (180, 255, 50))
+    yellow_mask = cv2.inRange(hsv, (20, 130, 130), (35, 255, 255))  # increased threshold
+    orange_mask = cv2.inRange(hsv, (10, 100, 100), (25, 255, 255))
+    red_mask1 = cv2.inRange(hsv, (0, 100, 100), (10, 255, 255))
+    red_mask2 = cv2.inRange(hsv, (160, 100, 100), (180, 255, 255))
+    red_mask = cv2.bitwise_or(red_mask1, red_mask2)
+
+    total = img.shape[0] * img.shape[1]
+    blue_count = np.sum(blue_mask > 0)
+    black_count = np.sum(black_mask > 0)
+    yellow_count = np.sum(yellow_mask > 0)
+    orange_count = np.sum(orange_mask > 0)
+    red_count = np.sum(red_mask > 0)
+
+    label = 'Unknown'
+    box_list = []
+    label_list = []
+
+    # Full image checks
+    if (blue_count + black_count) / total > 0.8:
+        label = 'Normal'
+    elif (red_count + orange_count) / total > 0.5:
+        label = 'Full Wire Overload'
+    elif (yellow_count) / total > 0.5:
+        label = 'Full Wire Overload'
+    # Check for full wire overload (entire image reddish or yellowish)
+    full_wire_thresh = 0.7  # 70% of image is reddish or yellowish
+    if (red_count + orange_count + yellow_count) / total > full_wire_thresh:
+        label = 'Full Wire Overload'
+        # Add a box covering the whole image
+        box_list.append((0, 0, img.shape[1], img.shape[0]))
+        label_list.append(label)
+    else:
+        # Small spot checks (improved: filter tiny spots, merge overlapping boxes)
+        min_area_faulty = 120  # increased min area for red/orange (faulty)
+        min_area_potential = 1000  # much higher min area for yellow (potential)
+        max_area = 0.05 * total
+        # Faulty (red/orange) spots
+        for mask, spot_label, min_a in [
+            (red_mask, 'Point Overload (Faulty)', min_area_faulty),
+            (yellow_mask, 'Point Overload (Potential)', min_area_potential)
+        ]:
+            contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+            for cnt in contours:
+                area = cv2.contourArea(cnt)
+                if min_a < area < max_area:
+                    x, y, w, h = cv2.boundingRect(cnt)
+                    box_list.append((x, y, w, h))
+                    label_list.append(spot_label)
+        # Middle area checks
+        h, w = img.shape[:2]
+        center = img[h//4:3*h//4, w//4:3*w//4]
+        center_hsv = cv2.cvtColor(center, cv2.COLOR_BGR2HSV)
+        center_yellow = cv2.inRange(center_hsv, (20, 130, 130), (35, 255, 255))
+        center_orange = cv2.inRange(center_hsv, (10, 100, 100), (25, 255, 255))
+        center_red1 = cv2.inRange(center_hsv, (0, 100, 100), (10, 255, 255))
+        center_red2 = cv2.inRange(center_hsv, (160, 100, 100), (180, 255, 255))
+        center_red = cv2.bitwise_or(center_red1, center_red2)
+        if np.sum(center_red > 0) + np.sum(center_orange > 0) > 0.1 * center.size:
+            label = 'Loose Joint (Faulty)'
+            box_list.append((w//4, h//4, w//2, h//2))
+            label_list.append(label)
+        elif np.sum(center_yellow > 0) > 0.1 * center.size:
+            label = 'Loose Joint (Potential)'
+            box_list.append((w//4, h//4, w//2, h//2))
+            label_list.append(label)
+    # Always check for tiny spots, even if image is labeled as Normal
+    min_area_tiny = 10
+    max_area_tiny = 30
+    for mask, spot_label in [
+        (red_mask, 'Tiny Faulty Spot'),
+        (yellow_mask, 'Tiny Potential Spot')
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if min_area_tiny < area < max_area_tiny:
+                x, y, w, h = cv2.boundingRect(cnt)
+                box_list.append((x, y, w, h))
+                label_list.append(spot_label)
+    # Detect wire-shaped (long, thin) regions for wire overloads only
+    aspect_ratio_thresh = 5
+    min_strip_area = 0.01 * total
+    wire_boxes = []
+    wire_labels = []
+    for mask, strip_label in [
+        (red_mask, 'Wire Overload (Red Strip)'),
+        (yellow_mask, 'Wire Overload (Yellow Strip)'),
+        (orange_mask, 'Wire Overload (Orange Strip)')
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if area > min_strip_area:
+                x, y, w, h = cv2.boundingRect(cnt)
+                aspect_ratio = max(w, h) / (min(w, h) + 1e-6)
+                if aspect_ratio > aspect_ratio_thresh:
+                    wire_boxes.append((x, y, w, h))
+                    wire_labels.append(strip_label)
+    # Add wire overloads to box_list/label_list
+    box_list = wire_boxes[:]
+    label_list = wire_labels[:]
+    # For point overloads, do not require wire shape
+    min_area_faulty = 120
+    min_area_potential = 1000
+    max_area = 0.05 * total
+    for mask, spot_label, min_a in [
+        (red_mask, 'Point Overload (Faulty)', min_area_faulty),
+        (yellow_mask, 'Point Overload (Potential)', min_area_potential)
+    ]:
+        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        for cnt in contours:
+            area = cv2.contourArea(cnt)
+            if min_a < area < max_area:
+                x, y, w, h = cv2.boundingRect(cnt)
+                box_list.append((x, y, w, h))
+                label_list.append(spot_label)
+    # Remove overlapping boxes using IOU
+    box_list, label_list = non_max_suppression_iou(box_list, label_list, iou_thresh=0.4)
+    box_list, label_list = filter_faulty_inside_potential(box_list, label_list)
+    box_list, label_list = filter_faulty_overlapping_potential(box_list, label_list)
+    box_list, label_list = merge_close_boxes(box_list, label_list, dist_thresh=100)
+    return label, box_list, label_list, img
+
+# Batch process all images in the directory
+for fname in os.listdir(dir_path):
+    if not fname.lower().endswith(('.jpg', '.jpeg', '.png')):
+        continue
+    label, box_list, label_list, img = classify_image(os.path.join(dir_path, fname))
+    # Load the original (unfiltered) image for drawing boxes
+    orig_dir = 'api_inference_pred_masks'  # or the directory with original images
+    orig_img_path = os.path.join(orig_dir, fname)
+    if os.path.exists(orig_img_path):
+        draw_img = cv2.imread(orig_img_path)
+        if draw_img is None:
+            draw_img = img.copy()
+    else:
+        draw_img = img.copy()
+    # Draw bounding boxes and labels on the original image
+    for (x, y, w, h), l in zip(box_list, label_list):
+        cv2.rectangle(draw_img, (x, y), (x+w, y+h), (0, 0, 255), 2)
+        cv2.putText(draw_img, l, (x, y-10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)
+    if not box_list:
+        cv2.putText(draw_img, label, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)
+    out_path = os.path.join(output_dir, fname)
+    cv2.imwrite(out_path, draw_img)
+    print(f"{fname}: {label} (saved with boxes on original image)")

scripts/crop_train_images.py

@@ -0,0 +1,21 @@
+import os
+from PIL import Image
+
+TRAIN_NORMAL = 'dataset/train/normal'
+TRAIN_FAULTY = 'dataset/train/faulty'
+CROP_PERCENT = 0.125  # 12.5%
+
+for folder in [TRAIN_NORMAL, TRAIN_FAULTY]:
+    for fname in os.listdir(folder):
+        if not fname.lower().endswith(('.jpg', '.jpeg', '.png')):
+            continue
+        fpath = os.path.join(folder, fname)
+        img = Image.open(fpath)
+        w, h = img.size
+        crop_w = int(w * CROP_PERCENT)
+        # Crop 12.5% from left and right
+        cropped = img.crop((crop_w, 0, w - crop_w, h))
+        cropped.save(fpath)
+        print(f"[CROP] {fpath} -> size {cropped.size}")
+
+print("[DONE] All training images cropped 12.5% from both sides.")

scripts/fill_test_from_TX.py

@@ -0,0 +1,35 @@
+import os
+import shutil
+import random
+
+TEST_NORMAL = 'dataset/test/normal'
+TEST_FAULTY = 'dataset/test/faulty'
+TX_DIR = 'TX'
+TARGET_COUNT = 100
+
+os.makedirs(TEST_NORMAL, exist_ok=True)
+os.makedirs(TEST_FAULTY, exist_ok=True)
+
+# Helper to collect images from TX folders
+def collect_images(src_pattern, dst_folder, needed):
+    collected = 0
+    for root, dirs, files in os.walk(TX_DIR):
+        if src_pattern in root:
+            imgs = [f for f in files if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+            random.shuffle(imgs)
+            for f in imgs:
+                if collected >= needed:
+                    return
+                src_path = os.path.join(root, f)
+                dst_path = os.path.join(dst_folder, f)
+                if not os.path.exists(dst_path):
+                    shutil.copyfile(src_path, dst_path)
+                    collected += 1
+                    print(f"[COPY] {src_path} -> {dst_path}")
+
+# Fill test/normal and test/faulty to TARGET_COUNT from TX (do not touch train)
+collect_images('normal', TEST_NORMAL, TARGET_COUNT)
+collect_images('faulty', TEST_FAULTY, TARGET_COUNT)
+
+print(f"[DONE] {TEST_NORMAL} images: {len(os.listdir(TEST_NORMAL))}")
+print(f"[DONE] {TEST_FAULTY} images: {len(os.listdir(TEST_FAULTY))}")

scripts/patchcore_api_inference.py

@@ -0,0 +1,93 @@
+import os
+import torch
+import numpy as np
+from PIL import Image
+from omegaconf import OmegaConf
+from anomalib.models import Patchcore
+from anomalib.data import Folder
+from pytorch_lightning import Trainer
+# --- Load config ---
+CONFIG_PATH = "configs/patchcore_transformers.yaml"
+CKPT_PATH = "results/Patchcore/transformers/v7/weights/lightning/model.ckpt"
+
+OUT_MASK_DIR = "api_inference_pred_pipeline"
+OUT_FILTERED_DIR = "api_inference_filtered_pipeline"
+
+os.makedirs(OUT_MASK_DIR, exist_ok=True)
+os.makedirs(OUT_FILTERED_DIR, exist_ok=True)
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+
+# Load config
+config = OmegaConf.load(CONFIG_PATH)
+
+if __name__ == "__main__":
+    os.makedirs(OUT_MASK_DIR, exist_ok=True)
+    os.makedirs(OUT_FILTERED_DIR, exist_ok=True)
+    # Setup datamodule for prediction (use test set)
+    # Use arguments matching the YAML config and Folder datamodule signature
+    data_module = Folder(
+        name=config.data.init_args.name,
+        root=config.data.init_args.root,
+        normal_dir=config.data.init_args.normal_dir,
+        abnormal_dir=config.data.init_args.abnormal_dir,
+        normal_test_dir=config.data.init_args.normal_test_dir,
+        train_batch_size=config.data.init_args.train_batch_size,
+        eval_batch_size=config.data.init_args.eval_batch_size,
+        num_workers=config.data.init_args.num_workers,
+    )
+    data_module.setup()
+
+    # Load model
+    model = Patchcore.load_from_checkpoint(CKPT_PATH, **config.model.init_args)
+    model.eval()
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    model = model.to(device)
+
+    # Inference loop
+    for batch in data_module.test_dataloader():
+        img = batch.image.to(device)
+        fname = batch.image_path[0]
+        with torch.no_grad():
+            output = model(img)
+            # PatchCore returns (anomaly_score, anomaly_map, ...)
+            if hasattr(output, 'anomaly_map'):
+                anomaly_map = output.anomaly_map.squeeze().cpu().numpy()
+            elif isinstance(output, (tuple, list)) and len(output) > 1:
+                anomaly_map = output[1].squeeze().cpu().numpy()
+            else:
+                anomaly_map = None
+        if anomaly_map is not None:
+            # Normalize to 0-255 for visualization
+            norm_map = (255 * (anomaly_map - anomaly_map.min()) / (np.ptp(anomaly_map) + 1e-8)).astype(np.uint8)
+            # Ensure norm_map is 2D for PIL
+            if norm_map.ndim > 2:
+                norm_map = np.squeeze(norm_map)
+                if norm_map.ndim > 2:
+                    norm_map = norm_map[0]
+            mask_img = Image.fromarray(norm_map)
+            out_name = os.path.splitext(os.path.basename(fname))[0] + "_mask.png"
+            mask_img.save(os.path.join(OUT_MASK_DIR, out_name))
+            print(f"Saved mask for {fname}")
+
+            # Save filtered (masked) part of the original transformer image
+            orig_img = Image.open(fname).convert("RGB")
+            # Resize mask to match original image size if needed
+            if mask_img.size != orig_img.size:
+                mask_img_resized = mask_img.resize(orig_img.size, resample=Image.BILINEAR)
+            else:
+                mask_img_resized = mask_img
+            # Binarize mask (threshold at 128)
+            bin_mask = np.array(mask_img_resized) > 128
+            # Apply mask to original image
+            orig_np = np.array(orig_img)
+            filtered_np = np.zeros_like(orig_np)
+            filtered_np[bin_mask] = orig_np[bin_mask]
+            filtered_img = Image.fromarray(filtered_np)
+            filtered_name = os.path.splitext(os.path.basename(fname))[0] + "_filtered.png"
+            filtered_img.save(os.path.join(OUT_FILTERED_DIR, filtered_name))
+            print(f"Saved filtered image for {fname}")
+        else:
+            print(f"No mask generated for {fname}")
+
+    print(f"All masks saved to {OUT_MASK_DIR}")

scripts/patchcore_inference.py

@@ -0,0 +1,61 @@
+import os
+import torch
+from torchvision import transforms
+from PIL import Image
+import numpy as np
+import matplotlib.pyplot as plt
+
+# PatchCore import for your config (update if needed for your Anomalib version)
+from anomalib.models.image.patchcore import Patchcore
+
+# Config from your YAML
+CKPT_PATH = "results/Patchcore/transformers/v2/weights/lightning/model.ckpt"
+IMAGE_SIZE = (256, 256)  # Update if you used a different size in training
+
+# Inference directory (change as needed)
+INFER_DIR = "./dataset/test/faulty"
+
+# Output directory for masks
+OUT_MASK_DIR = "inference_masks"
+os.makedirs(OUT_MASK_DIR, exist_ok=True)
+
+# Load model
+model = Patchcore.load_from_checkpoint(CKPT_PATH)
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+model = model.to(device)
+model.eval()
+
+# Transforms (should match your training config)
+transform = transforms.Compose([
+    transforms.Resize(IMAGE_SIZE),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+])
+
+# Inference loop
+for fname in sorted(os.listdir(INFER_DIR)):
+    if not fname.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.tif', '.tiff')):
+        continue
+    fpath = os.path.join(INFER_DIR, fname)
+    img = Image.open(fpath).convert("RGB")
+    img_tensor = transform(img).unsqueeze(0).to(device)
+    with torch.no_grad():
+        output = model(img_tensor)
+        # PatchCore returns (anomaly_score, anomaly_map, ...)
+        if hasattr(output, 'anomaly_map'):
+            anomaly_map = output.anomaly_map.squeeze().cpu().numpy()
+        elif isinstance(output, (tuple, list)) and len(output) > 1:
+            anomaly_map = output[1].squeeze().cpu().numpy()
+        else:
+            anomaly_map = None
+    # Save mask as PNG
+    if anomaly_map is not None:
+        # Normalize to 0-255 for visualization
+        norm_map = (255 * (anomaly_map - anomaly_map.min()) / (np.ptp(anomaly_map) + 1e-8)).astype(np.uint8)
+        mask_img = Image.fromarray(norm_map)
+        mask_img.save(os.path.join(OUT_MASK_DIR, f"{os.path.splitext(fname)[0]}_mask.png"))
+        print(f"Saved mask for {fname}")
+    else:
+        print(f"No mask generated for {fname}")
+
+print(f"All masks saved to {OUT_MASK_DIR}")

scripts/patchcore_single_image.py

@@ -0,0 +1,74 @@
+import sys
+from patchcore_api_inference import Patchcore, config, device
+from PIL import Image
+import torch
+import numpy as np
+import os
+
+# Output directories (should match those in patchcore_api_inference.py)
+OUT_MASK_DIR = "api_inference_pred_masks_pipeline"
+OUT_FILTERED_DIR = "api_inference_filtered_pipeline"
+os.makedirs(OUT_MASK_DIR, exist_ok=True)
+os.makedirs(OUT_FILTERED_DIR, exist_ok=True)
+
+# Load model
+model = Patchcore.load_from_checkpoint(
+    "results/Patchcore/transformers/v7/weights/lightning/model.ckpt",
+    **config.model.init_args
+)
+model.eval()
+model = model.to(device)
+
+def infer_single_image(image_path):
+    # Load and preprocess image
+    # Normalize path for cross-platform compatibility
+    import os
+    fixed_path = os.path.normpath(image_path)
+    fixed_path = os.path.abspath(fixed_path)
+    orig_img = Image.open(fixed_path).convert("RGB")
+    # Use the same transforms as in training (resize, normalize)
+    # If you have a transform pipeline, import and use it here
+    img_resized = orig_img.resize((256, 256))  # Change if your model uses a different size
+    img_tensor = torch.from_numpy(np.array(img_resized)).permute(2, 0, 1).float() / 255.0
+    img_tensor = img_tensor.unsqueeze(0).to(device)
+
+    with torch.no_grad():
+        output = model(img_tensor)
+        if hasattr(output, 'anomaly_map'):
+            anomaly_map = output.anomaly_map.squeeze().cpu().numpy()
+        elif isinstance(output, (tuple, list)) and len(output) > 1:
+            anomaly_map = output[1].squeeze().cpu().numpy()
+        else:
+            anomaly_map = None
+    if anomaly_map is not None:
+        norm_map = (255 * (anomaly_map - anomaly_map.min()) / (np.ptp(anomaly_map) + 1e-8)).astype(np.uint8)
+        if norm_map.ndim > 2:
+            norm_map = np.squeeze(norm_map)
+            if norm_map.ndim > 2:
+                norm_map = norm_map[0]
+        mask_img = Image.fromarray(norm_map)
+        out_name = os.path.splitext(os.path.basename(image_path))[0] + "_mask.png"
+        mask_img.save(os.path.join(OUT_MASK_DIR, out_name))
+        print(f"Saved mask for {image_path}")
+
+        # Resize mask to match original image size if needed
+        if mask_img.size != orig_img.size:
+            mask_img_resized = mask_img.resize(orig_img.size, resample=Image.BILINEAR)
+        else:
+            mask_img_resized = mask_img
+        bin_mask = np.array(mask_img_resized) > 128
+        orig_np = np.array(orig_img)
+        filtered_np = np.zeros_like(orig_np)
+        filtered_np[bin_mask] = orig_np[bin_mask]
+        filtered_img = Image.fromarray(filtered_np)
+        filtered_name = os.path.splitext(os.path.basename(image_path))[0] + "_filtered.png"
+        filtered_img.save(os.path.join(OUT_FILTERED_DIR, filtered_name))
+        print(f"Saved filtered image for {image_path}")
+    else:
+        print(f"No mask generated for {image_path}")
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python patchcore_single_image.py <image_path>")
+        sys.exit(1)
+    infer_single_image(sys.argv[1])

scripts/prepare_patchcore_dataset.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# 1. Gather all normal images from TX folders into train/normal and test/normal
+mkdir -p dataset/train/normal
+mkdir -p dataset/test/normal
+
+# Move/copy all normal images from TX folders to train/normal (for training)
+find TX/ -type f -name '*.jpg' -path '*/normal/*' -exec cp {} dataset/train/normal/ \;
+
+# Optionally, split some normal images into test/normal (for evaluation)
+# Here, move 20% of images to test/normal (adjust as needed)
+cd dataset/train/normal
+mkdir -p ../test/normal
+count=$(ls -1 | wc -l)
+test_count=$((count / 5))
+ls | shuf | head -n $test_count | xargs -I{} mv {} ../test/normal/
+cd ../../..
+
+# 2. Gather all faulty images from TX folders into test/faulty (for evaluation)
+mkdir -p dataset/test/faulty
+find TX/ -type f -name '*.jpg' -path '*/faulty/*' -exec cp {} dataset/test/faulty/ \;
+
+# 3. (Optional) Remove duplicates between train/normal and test/normal
+# This step assumes filenames are unique. If not, use a more robust deduplication method.
+cd dataset/test/normal
+for f in *; do
+  [ -e "../../train/normal/$f" ] && rm -f "../../train/normal/$f"
+done
+cd ../../..
+
+# 4. Print summary
+train_n=$(ls dataset/train/normal | wc -l)
+test_n=$(ls dataset/test/normal | wc -l)
+test_f=$(ls dataset/test/faulty | wc -l)
+echo "[✓] Normal images in train: $train_n"
+echo "[✓] Normal images in test:  $test_n"
+echo "[✓] Faulty images in test:  $test_f"

scripts/recreate_train_from_TX.py

@@ -0,0 +1,38 @@
+import os
+import shutil
+import random
+
+TX_DIR = 'TX'
+TRAIN_NORMAL = 'dataset/train/normal'
+TRAIN_FAULTY = 'dataset/train/faulty'
+TARGET_COUNT = 100
+
+# Clean train folders
+shutil.rmtree(TRAIN_NORMAL, ignore_errors=True)
+os.makedirs(TRAIN_NORMAL, exist_ok=True)
+shutil.rmtree(TRAIN_FAULTY, ignore_errors=True)
+os.makedirs(TRAIN_FAULTY, exist_ok=True)
+
+# Helper to collect images from TX folders
+def collect_images(src_pattern, dst_folder, needed):
+    collected = 0
+    for root, dirs, files in os.walk(TX_DIR):
+        if src_pattern in root:
+            imgs = [f for f in files if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+            random.shuffle(imgs)
+            for f in imgs:
+                if collected >= needed:
+                    return
+                src_path = os.path.join(root, f)
+                dst_path = os.path.join(dst_folder, f)
+                if not os.path.exists(dst_path):
+                    shutil.copyfile(src_path, dst_path)
+                    collected += 1
+                    print(f"[COPY] {src_path} -> {dst_path}")
+
+# Fill train/normal and train/faulty to TARGET_COUNT from TX
+collect_images('normal', TRAIN_NORMAL, TARGET_COUNT)
+collect_images('faulty', TRAIN_FAULTY, TARGET_COUNT)
+
+print(f"[DONE] {TRAIN_NORMAL} images: {len(os.listdir(TRAIN_NORMAL))}")
+print(f"[DONE] {TRAIN_FAULTY} images: {len(os.listdir(TRAIN_FAULTY))}")

scripts/run_anomalib.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# 0) Environment
+python3 -V || true
+echo "[*] Installing python3-full and python3-venv if needed…"
+sudo apt update && sudo apt install -y python3-full python3.12-venv
+echo "[*] Creating virtual environment if not exists…"
+if [ ! -d ~/anomalib_env ]; then
+  python3 -m venv ~/anomalib_env
+  source ~/anomalib_env/bin/activate
+  pip install -U pip
+  pip install "anomalib[full]" flask requests cloudinary pillow numpy opencv-python omegaconf torch
+else
+  source ~/anomalib_env/bin/activate
+  pip install -U pip
+  pip install "anomalib[full]" flask requests cloudinary pillow numpy opencv-python omegaconf torch
+fi
+
+
+# # 1) Train (PatchCore builds the memory bank from normals)
+# anomalib train \
+#   --config configs/patchcore_transformers.yaml
+#
+# CKPT=$(ls -1t results/transformers/patchcore/*/weights/*.ckpt | head -n 1)
+# echo "[*] Using checkpoint: $CKPT"
+#
+# # 2) Test/Eval on test/{normal,faulty}
+# anomalib test \
+#   --config configs/patchcore_transformers.yaml \
+#   --ckpt_path "$CKPT"
+
+echo
+echo "[✓] Done. Check:"
+echo "  • results/transformers/patchcore/**/images/    (heatmaps & overlays)"
+echo "  • results/transformers/patchcore/**/metrics.csv (AUROC/F1 etc.)"