# 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()