Clean deploy to Hugging Face Space

Anomaly/resnet50_final.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1607839eb06288892d0201d3a2eefb6db8b1d27eba5d3a0d733ca7ed0e4b62f3
+size 330551545

anomaly_gradio.py

@@ -0,0 +1,56 @@
+import torch
+import cv2
+import os
+import tempfile
+from torchvision.models.detection import fasterrcnn_resnet50_fpn, FastRCNNPredictor
+from torchvision.transforms import functional as F
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+num_classes = 14
+
+def load_anomaly_model():
+    model = fasterrcnn_resnet50_fpn(pretrained=False)
+    in_features = model.roi_heads.box_predictor.cls_score.in_features
+    model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
+    checkpoint = torch.load("Anomaly/resnet50_final.pth", map_location=device)
+    model.load_state_dict(checkpoint["model_state_dict"])
+    model.to(device).eval()
+    return model
+
+model = load_anomaly_model()
+
+def draw_boxes(frame, outputs, threshold=0.5):
+    for box, label, score in zip(outputs['boxes'], outputs['labels'], outputs['scores']):
+        if score >= threshold:
+            x1, y1, x2, y2 = map(int, box.tolist())
+            cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
+            cv2.putText(frame, f'{label.item()}:{score:.2f}', (x1, y1 - 10),
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1)
+    return frame
+
+def detect_anomalies(video_file):
+    input_path = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+    input_path.write(video_file.read())
+    input_path.close()
+
+    cap = cv2.VideoCapture(input_path.name)
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    out_path = os.path.join(tempfile.gettempdir(), "anomaly_output.mp4")
+    out = cv2.VideoWriter(out_path, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))
+
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        img = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        tensor = F.to_tensor(img).to(device)
+        with torch.no_grad():
+            pred = model([tensor])[0]
+        result = draw_boxes(frame, pred)
+        out.write(result)
+
+    cap.release()
+    out.release()
+    return out_path

app.py

@@ -0,0 +1,28 @@
+# app.py
+import gradio as gr
+from anomaly_gradio import detect_anomalies
+from segmentation_gradio import segment_video
+
+anomaly_app = gr.Interface(
+    fn=detect_anomalies,
+    inputs=gr.Video(label="Upload video for anomaly detection"),
+    outputs=gr.Video(label="Processed anomaly video"),
+    title="Anomaly Detection with Faster R-CNN"
+)
+
+segmentation_app = gr.Interface(
+    fn=segment_video,
+    inputs=[
+        gr.Video(label="Upload video for segmentation"),
+        gr.Radio(["MobileNet", "ResNet34"], label="Choose segmentation model")
+    ],
+    outputs=gr.Video(label="Segmented output video"),
+    title="Semantic Segmentation with UNet"
+)
+
+demo = gr.TabbedInterface(
+    [anomaly_app, segmentation_app],
+    ["Anomaly Detection", "Segmentation"]
+)
+
+demo.launch()

pages/anomaly_detction.py

@@ -0,0 +1,104 @@
+import streamlit as st
+import cv2
+import torch
+import os
+import tempfile
+import numpy as np
+from torchvision.models.detection import fasterrcnn_resnet50_fpn, FastRCNNPredictor
+from torchvision.transforms import functional as F
+
+# ─────────────────────────────────────────────────────────────
+# Streamlit UI Setup
+st.set_page_config(page_title="Anomaly Detection", layout="centered")
+st.title("📹 Anomaly Detection with Faster R-CNN")
+
+# ─────────────────────────────────────────────────────────────
+# Model Setup
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+num_classes = 14  # 13 classes + background
+
+@st.cache_resource
+def load_model():
+    model = fasterrcnn_resnet50_fpn(pretrained=False)
+    in_features = model.roi_heads.box_predictor.cls_score.in_features
+    model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
+
+    checkpoint = torch.load("Anomaly/resnet50_final.pth", map_location=device)
+    model.load_state_dict(checkpoint["model_state_dict"])
+    model.to(device)
+    model.eval()
+    return model
+
+model = load_model()
+
+# ─────────────────────────────────────────────────────────────
+# Utility
+def draw_boxes(frame, outputs, threshold=0.5):
+    for box, label, score in zip(outputs['boxes'], outputs['labels'], outputs['scores']):
+        if score >= threshold:
+            x1, y1, x2, y2 = map(int, box.tolist())
+            cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
+            cv2.putText(frame, f'{label.item()}:{score:.2f}', (x1, y1 - 10),
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1)
+    return frame
+
+# ─────────────────────────────────────────────────────────────
+# Interface
+video_file = st.file_uploader("Upload a video for anomaly detection", type=["mp4", "mov", "avi"])
+
+if video_file:
+    st.video(video_file)
+    start_button = st.button("▶ Start Anomaly Detection")
+
+    if start_button:
+        stframe = st.empty()
+        progress = st.progress(0)
+
+        temp_input = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
+        temp_input.write(video_file.read())
+        temp_input.close()
+
+        cap = cv2.VideoCapture(temp_input.name)
+        fps = cap.get(cv2.CAP_PROP_FPS)
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+
+        output_path = os.path.join(tempfile.gettempdir(), "anomaly_output.mp4")
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+
+        frame_count = 0
+
+        while cap.isOpened():
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+            tensor = F.to_tensor(image).to(device)
+
+            with torch.no_grad():
+                prediction = model([tensor])[0]
+
+            annotated = draw_boxes(frame.copy(), prediction)
+            out.write(annotated)
+
+            resized = cv2.resize(annotated, (960, 540))
+            stframe.image(cv2.cvtColor(resized, cv2.COLOR_BGR2RGB), channels="RGB")
+
+            frame_count += 1
+            progress.progress(min(frame_count / total_frames, 1.0))
+
+        cap.release()
+        out.release()
+
+        st.success("✅ Anomaly detection complete!")
+
+        with open(output_path, "rb") as f:
+            st.download_button(
+                label="📥 Download Annotated Video",
+                data=f,
+                file_name="anomaly_output.mp4",
+                mime="video/mp4"
+            )

requirements.txt

@@ -0,0 +1,7 @@
+gradio
+torch==2.7.1
+torchvision==0.22.1
+segmentation-models-pytorch==0.3.4
+opencv-python-headless==4.8.0.76
+numpy==1.26.4
+Pillow

segmentation_gradio.py

@@ -0,0 +1,65 @@
+import torch
+import cv2
+import os
+import tempfile
+import numpy as np
+from PIL import Image
+from torchvision import transforms as T
+import segmentation_models_pytorch as smp
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+resize_w, resize_h = 640, 384
+mean = [0.485, 0.456, 0.406]
+std = [0.229, 0.224, 0.225]
+transform = T.Compose([
+    T.ToTensor(),
+    T.Normalize(mean, std)
+])
+color_map = np.random.RandomState(42).randint(0, 255, size=(23, 3), dtype=np.uint8)
+
+def apply_mask(image, mask):
+    mask_color = color_map[mask]
+    return cv2.addWeighted(image, 0.6, mask_color, 0.4, 0)
+
+def load_segmentation_model(name):
+    model = smp.Unet(
+        encoder_name="mobilenet_v2" if name == "MobileNet" else "resnet34",
+        encoder_weights=None,
+        in_channels=3,
+        classes=23
+    )
+    path = "unet_mobilenet_final_50.pt" if name == "MobileNet" else "unet_resnet34_final_50.pt"
+    model.load_state_dict(torch.load(path, map_location=device))
+    model.to(device).eval()
+    return model
+
+def segment_video(video_file, model_name):
+    model = load_segmentation_model(model_name)
+    input_path = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+    input_path.write(video_file.read())
+    input_path.close()
+
+    cap = cv2.VideoCapture(input_path.name)
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    out_path = os.path.join(tempfile.gettempdir(), "segmentation_output.mp4")
+    out = cv2.VideoWriter(out_path, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))
+
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        resized = cv2.resize(frame, (resize_w, resize_h))
+        rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB)
+        pil = Image.fromarray(rgb)
+        tensor = transform(pil).unsqueeze(0).to(device)
+        with torch.no_grad():
+            mask = torch.argmax(model(tensor), dim=1).squeeze().cpu().numpy()
+        resized_mask = cv2.resize(mask.astype(np.uint8), (width, height), interpolation=cv2.INTER_NEAREST)
+        overlay = apply_mask(frame, resized_mask)
+        out.write(overlay)
+
+    cap.release()
+    out.release()
+    return out_path

unet_mobilenet_final_50.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c977bc416a64415fd756eafecf9d07499350f64c35ef3106cc90f87899d30e25
+size 26813798

unet_resnet34_final_50.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0e3a9ee76eeb1bff57338dace892d0ab9f2f90f5829ddb6d8c7bff836248c2c7
+size 97936012