Spaces:
Sleeping
Sleeping
| # 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() | |