app.py

# app.py  -- YOLOv8 live IP camera with threaded capture + motion throttling
from ultralytics import YOLO
from PIL import Image
import numpy as np
import cv2
import gradio as gr
import time
import threading

# -------------------------
# Model + IP camera config
# -------------------------
model = YOLO('best.pt')  # your trained weights
ip_url = None  # change to your IP webcam URL

# -------------------------
# Shared state (thread-safe-ish)
# -------------------------
cap = None
prev_gray = None
latest_frame = None       # PIL image (annotated) to show in UI
latest_result = "Waiting..."  # text summary
stop_flag = False
camera_thread_obj = None

# Tuning params (change these to adjust responsiveness / CPU)
MOTION_THRESHOLD = 50      # number of changed pixels to consider motion
COOLDOWN_SEC = 1.0          # min seconds between YOLO runs
RESIZE_TO = (640, 360)      # inference size used before passing to model (smaller -> faster)
POLL_INTERVAL = 0.4         # seconds between UI polls (gr.Timer interval)
SLEEP_BETWEEN_READS = 0.02  # small sleep inside camera thread to avoid tight loop



# -------------------------
# Inference helper (robust to None)
# -------------------------
def predict_yolov8(img: Image.Image):
    """
    Accepts a PIL.Image or None. Returns (PIL annotated image or placeholder, string result).
    """
    if img is None:
        # return placeholder
        placeholder = Image.new("RGB", RESIZE_TO, (0, 0, 0))
        return placeholder, "No image"

    try:
        img_np = np.array(img.convert('RGB'))
    except Exception as e:
        placeholder = Image.new("RGB", RESIZE_TO, (0, 0, 0))
        return placeholder, f"Bad image: {e}"

    # Run YOLO inference (batch size 1)
    # NOTE: if your model.predict(...) supports stream/inference kwargs to reduce overhead you can pass them.
    results = model.predict(img_np)

    img_draw = img_np.copy()
    preds_info = []

    # results[0].boxes may be empty
    for box in results[0].boxes:
        # x1, y1, x2, y2 (float) -> int
        xy = box.xyxy.squeeze().tolist()
        if isinstance(xy[0], list):  # handle edge-cases
            x1, y1, x2, y2 = [int(v) for v in xy[0]]
        else:
            x1, y1, x2, y2 = [int(v) for v in xy]

        class_id = int(box.cls.cpu().item()) if hasattr(box, "cls") else int(box.cls)
        conf = float(box.conf.cpu().item()) if hasattr(box, "conf") else float(box.conf)
        label_text = f"{model.model.names[class_id]} {conf:.2f}"

        # Draw rectangle + label
        cv2.rectangle(img_draw, (x1, y1), (x2, y2), (0, 255, 0), 2)
        cv2.putText(img_draw, label_text, (x1, max(15, y1 - 10)),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.45, (36, 255, 12), 2)

        preds_info.append({
            "bbox": [x1, y1, x2, y2],
            "class": model.model.names[class_id],
            "confidence": round(conf, 2)
        })

    out_img = Image.fromarray(img_draw)
    if preds_info:
        result_str = "\n".join([f"[{p['class']}] {p['bbox']}, conf={p['confidence']}" for p in preds_info])
    else:
        result_str = "No detections"
    return out_img, result_str

# -------------------------
# Camera thread: reads frames, detects motion, runs YOLO + updates shared state
# -------------------------
def camera_thread():
    global cap, prev_gray, latest_frame, latest_result, stop_flag

    try:
        cap = cv2.VideoCapture(ip_url)
    except Exception as e:
        latest_frame = Image.new("RGB", RESIZE_TO, (0, 0, 0))
        latest_result = f"Failed to open camera: {e}"
        return

    # warm-up read
    time.sleep(0.8)
    ret, frame = cap.read()
    if not ret or frame is None:
        latest_frame = Image.new("RGB", RESIZE_TO, (0, 0, 0))
        latest_result = "Camera opened but no frames received"
        cap.release()
        cap = None
        return

    prev_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    last_trigger = 0.0

    while not stop_flag:
        ret, frame = cap.read()
        if not ret or frame is None:
            # keep trying
            time.sleep(0.5)
            continue

        # motion detection (fast grayscale diff)
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        diff = cv2.absdiff(prev_gray, gray)
        thresh = cv2.threshold(diff, 25, 255, cv2.THRESH_BINARY)[1]
        motion_level = int(cv2.countNonZero(thresh))
        prev_gray = gray

        if motion_level < MOTION_THRESHOLD:
            # no meaningful motion; skip heavy processing
            time.sleep(SLEEP_BETWEEN_READS)
            continue

        # throttle YOLO inference
        now = time.time()
        if now - last_trigger < COOLDOWN_SEC:
            time.sleep(SLEEP_BETWEEN_READS)
            continue
        last_trigger = now

        # prepare frame for model (resize -> PIL)
        pil_frame = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)).resize(RESIZE_TO)

        # run inference (this is the heavy op)
        annotated, result_str = predict_yolov8(pil_frame)

        # update shared state for UI polling
        latest_frame = annotated
        latest_result = f"{result_str}  (motion={motion_level})"

        # tiny sleep to yield CPU
        time.sleep(0.005)

    # cleanup when stop_flag set
    if cap:
        cap.release()
        cap = None

# -------------------------
# Control functions for Gradio
# -------------------------
def start_live(ip):
    global stop_flag, camera_thread_obj, latest_result, latest_frame, ip_url
    
    ip_url = "http://192.168.1.4:8080/video"  # Construct the full URL with the provided IP
    
    # Try to open the connection and handle errors
    try:
        cap = cv2.VideoCapture(ip_url)
        if not cap.isOpened():
            raise Exception("Failed to connect to the camera.")

        # If connected successfully, start the camera thread
        if camera_thread_obj and camera_thread_obj.is_alive():
            return "Already running"
        
        stop_flag = False
        latest_result = "Starting camera..."
        camera_thread_obj = threading.Thread(target=camera_thread, daemon=True)
        camera_thread_obj.start()
        return "Live feed started"
    except Exception as e:
        # Handle connection failure
        latest_result = f"Failed to connect: {str(e)}"
        return latest_result


def stop_live():
    global stop_flag, camera_thread_obj
    stop_flag = True
    # camera thread will release the capture and exit
    return "Stopped"

def get_latest():
    """Called from UI timer to fetch latest annotated image + text."""
    if latest_frame is None:
        # placeholder when nothing yet
        placeholder = Image.new("RGB", RESIZE_TO, (20, 20, 20))
        return placeholder, latest_result
    return latest_frame, latest_result

# -------------------------
# Gradio UI
# -------------------------
css = "footer {display: none !important;}"

with gr.Blocks(theme=gr.themes.Soft(), css=css, title="YOLOv8 Detection Demo") as demo:
    gr.Markdown("# YOLOv8 Detection + Live IP Camera")

    with gr.Tabs():

        with gr.Tab("Image Upload"):
            with gr.Row():
                input_img = gr.Image(type="pil", label="Upload Image")
                out_img = gr.Image(type="pil", label="Detections")
            results_box = gr.Textbox(label="Detection Results")
            btn = gr.Button("Detect")
            btn.click(predict_yolov8, inputs=input_img, outputs=[out_img, results_box])

        with gr.Tab("Webcam"):
            webcam_input = gr.Image(type="pil", label="Webcam (browser)")
            webcam_out = gr.Image(type="pil", label="Detections")
            webcam_text = gr.Textbox(label="Detection Results")
            webcam_btn = gr.Button("Detect")
            webcam_btn.click(predict_yolov8, inputs=webcam_input, outputs=[webcam_out, webcam_text])

        with gr.Tab("Live IP Camera"):
            ip_input = gr.Textbox(label="IP Camera URL", placeholder="Enter ip address here")
            live_img = gr.Image(type="pil", label="Live Detection", height=480)
            live_txt = gr.Textbox(label="YOLO Results")
            start_btn = gr.Button("Start Live")
            stop_btn = gr.Button("Stop Live")

            start_btn.click(
                fn=start_live,
                inputs=ip_input,
                outputs=live_txt
            )

            stop_btn.click(stop_live, outputs=live_txt)

            # Poll for latest annotated frame every POLL_INTERVAL seconds
            timer = gr.Timer(POLL_INTERVAL)
    timer.tick(
        fn=get_latest,
        inputs=None,
        outputs=[live_img, live_txt]
    )


# Launch
if __name__ == "__main__":
    demo.launch()