webcam_yolo.py

from ultralytics import YOLO
import cv2
import time
from dataclasses import dataclass
from typing import List, Dict
import logging

# Configurar logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
logger = logging.getLogger(__name__)


@dataclass
class Detection:

    class_name: str
    confidence: float
    bbox: List[float]
    timestamp: float


class ObjectDetector:

    def __init__(self, model_path: str = 'yolov8n.pt', confidence_threshold: float = 0.5):
        self.model = YOLO(model_path)
        self.confidence_threshold = confidence_threshold
        self.detection_history: Dict[str, List[Detection]] = {}
        self.last_log_time = 0
        self.log_interval = 2

    def process_frame(self, frame):

        results = self.model(frame, verbose=False)
        detections = []

        for result in results:
            boxes = result.boxes
            if boxes is not None:
                for box in boxes:
                    confidence = box.conf.item()
                    if confidence >= self.confidence_threshold:
                        class_id = int(box.cls.item())
                        class_name = self.model.names[class_id]
                        bbox = box.xyxy[0].tolist()

                        detection = Detection(
                            class_name=class_name,
                            confidence=confidence,
                            bbox=bbox,
                            timestamp=time.time()
                        )
                        detections.append(detection)

        return detections

    def update_detection_history(self, detections: List[Detection]):

        current_time = time.time()

        for detection in detections:
            class_name = detection.class_name

            if class_name not in self.detection_history:
                self.detection_history[class_name] = []

            # Verificar si ya detectamos recientemente este objeto
            should_add = True
            if self.detection_history[class_name]:
                last_detection = self.detection_history[class_name][-1]

                if current_time - last_detection.timestamp < 1.0:
                    should_add = False

            if should_add:
                self.detection_history[class_name].append(detection)

    def log_detections(self, detections: List[Detection]):

        current_time = time.time()

        if current_time - self.last_log_time >= self.log_interval:
            if detections:
                detected_objects = set(det.class_name for det in detections)
                logger.info(f"Objetos detectados: {', '.join(detected_objects)}")
            else:
                logger.info("No se detectaron objetos")

            self.last_log_time = current_time


class VideoProcessor:

    def __init__(self, source: int = 0):
        self.cap = cv2.VideoCapture(source)
        self.is_running = False

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

    def read_frame(self):
        """Lee un frame de la cámara"""
        success, frame = self.cap.read()
        return success, frame

    def release(self):
        """Libera los recursos de la cámara"""
        if self.cap.isOpened():
            self.cap.release()
        cv2.destroyAllWindows()


class AnnotationEngine:

    @staticmethod
    def draw_detections(frame, detections: List[Detection]):
        annotated_frame = frame.copy()

        for detection in detections:
            x1, y1, x2, y2 = map(int, detection.bbox)

            # Dibujar rectángulo
            cv2.rectangle(annotated_frame, (x1, y1), (x2, y2), (0, 255, 0), 2)

            # Dibujar etiqueta
            label = f"{detection.class_name} {detection.confidence:.2f}"
            label_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 2)[0]

            cv2.rectangle(annotated_frame, (x1, y1 - label_size[1] - 10),
                          (x1 + label_size[0], y1), (0, 255, 0), -1)
            cv2.putText(annotated_frame, label, (x1, y1 - 5),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 2)

        return annotated_frame


def main():

    detector = ObjectDetector()
    annotation_engine = AnnotationEngine()

    with VideoProcessor(0) as video_processor:
        print("Iniciando detección de objetos. Presiona 'Q' para salir.")
        print("Las detecciones se mostrarán cada 2 segundos en el log.")

        while True:
            success, frame = video_processor.read_frame()
            if not success:
                logger.error("No se pudo leer el frame de la cámara")
                break

            detections = detector.process_frame(frame)

            # Actualizar historial y hacer log
            detector.update_detection_history(detections)
            detector.log_detections(detections)

            # Dibujar anotaciones
            annotated_frame = annotation_engine.draw_detections(frame, detections)

            # Mostrar frame
            cv2.imshow('YOLOv8 - Detección Optimizada', annotated_frame)

            # Salir con Q
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    # Resumen final
    print("\n--- Resumen de detecciones ---")
    for class_name, detections in detector.detection_history.items():
        print(f"{class_name}: {len(detections)} detecciones")


if __name__ == "__main__":
    main()