提交项目文件

src/image_object_detection/camera_detector.py

@@ -0,0 +1,98 @@
+# camera_detector.py
+import cv2
+import time
+
+class CameraDetector:
+    def __init__(self, detection_engine):
+        self.engine = detection_engine
+
+    def detect_camera(self):
+        """检测摄像头实时画面"""
+        try:
+            # 打开摄像头
+            print("正在打开摄像头...")
+            cap = self._open_camera()
+
+            if cap is None:
+                return
+
+            print("摄像头已打开，开始实时检测。")
+            print("在显示窗口中按 'q' 键退出，或在终端中按 Ctrl+C 中断程序。\n")
+
+            self._run_camera_loop(cap)
+
+        except KeyboardInterrupt:
+            print("\n\n用户按下了 Ctrl+C，强制退出摄像头检测...")
+            print("程序已安全退出。")
+        except Exception as e:
+            print(f"摄像头检测过程中发生错误: {e}")
+        finally:
+            self._cleanup_resources(cap)
+
+    def _open_camera(self):
+        """打开摄像头"""
+        cap = cv2.VideoCapture(0)
+
+        if not cap.isOpened():
+            print("错误: 无法打开摄像头")
+            return None
+
+        return cap
+
+    def _run_camera_loop(self, cap):
+        """摄像头检测主循环"""
+        # 记录上次输出时间，控制输出频率
+        last_output_time = time.time()
+        output_interval = 1.0  # 每秒最多输出一次检测信息
+
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                print("无法读取摄像头画面，退出...")
+                break
+
+            results = self._perform_detection(frame)
+            annotated_frame = results[0].plot()
+
+            # 控制输出频率，避免终端被刷屏
+            self._handle_output_frequency(results, last_output_time, output_interval)
+            last_output_time = time.time()
+
+            cv2.imshow('YOLO 检测结果 - 摄像头', annotated_frame)
+
+            # 按 'q' 键退出
+            key = cv2.waitKey(1) & 0xFF
+            if key == ord('q'):
+                print("\n用户按下 'q' 键，退出摄像头检测...")
+                break
+
+    def _perform_detection(self, frame):
+        """执行检测"""
+        annotated_frame, results = self.engine.detect(frame)
+        return results
+
+    def _handle_output_frequency(self, results, last_output_time, output_interval):
+        """处理输出频率"""
+        current_time = time.time()
+        if current_time - last_output_time >= output_interval:
+            detected_count = len(results[0].boxes)
+            if detected_count > 0:
+                # 构建检测对象字符串
+                detected_objects = []
+                for box in results[0].boxes:
+                    cls_index = int(box.cls)
+                    cls_name = self.engine.model.names[cls_index]
+                    confidence = box.conf.item()
+                    detected_objects.append(f"{cls_name}({confidence:.2f})")
+
+                print(f"检测到 {detected_count} 个对象: {', '.join(detected_objects)}")
+
+    def _cleanup_resources(self, cap):
+        """清理资源"""
+        print("释放摄像头资源...")
+        try:
+            cap.release()
+            cv2.destroyAllWindows()
+        except:
+            pass
+        print("摄像头检测已停止。")

src/image_object_detection/config.py

@@ -0,0 +1,15 @@
+# config.py
+class Config:
+    def __init__(self):
+        # 测试图像路径
+        self.test_image_path = r"C:\Users\apple\OneDrive\桌面\test.jpg"
+
+        # YOLO模型配置
+        self.model_path = "yolov8n.pt"
+        self.confidence_threshold = 0.25
+
+        # 摄像头配置
+        self.camera_index = 0
+
+        # 输出频率控制（秒）
+        self.output_interval = 1.0

src/image_object_detection/detection_engine.py

@@ -0,0 +1,46 @@
+# detection_engine.py
+from ultralytics import YOLO
+import io
+import sys
+
+class DetectionEngine:
+    def __init__(self, model_path="yolov8n.pt", conf_threshold=0.25):
+        self.model_path = model_path
+        self.conf_threshold = conf_threshold
+        self.model = self._load_model()
+
+    def _load_model(self):
+        """加载YOLO模型并抑制输出"""
+        old_stdout = sys.stdout
+        old_stderr = sys.stderr
+        sys.stdout = io.StringIO()
+        sys.stderr = io.StringIO()
+
+        try:
+            model = YOLO(self.model_path)
+        finally:
+            sys.stdout = old_stdout
+            sys.stderr = old_stderr
+
+        return model
+
+    def detect(self, frame):
+        """
+        对输入帧进行检测。
+        返回: (annotated_frame: np.ndarray, results: List[Results])
+        即使无检测框，annotated_frame 也是原始图像（HWC格式）。
+        """
+        old_stdout = sys.stdout
+        old_stderr = sys.stderr
+        sys.stdout = io.StringIO()
+        sys.stderr = io.StringIO()
+
+        try:
+            results = self.model(frame, conf=self.conf_threshold, verbose=False)
+        finally:
+            sys.stdout = old_stdout
+            sys.stderr = old_stderr
+
+        # YOLOv8 always returns at least one result
+        annotated_frame = results[0].plot()  # numpy array (H, W, C)
+        return annotated_frame, results

src/image_object_detection/image_detector.py

@@ -0,0 +1,112 @@
+# image_detector.py
+import cv2
+import numpy as np
+import traceback
+
+class ImageDetector:
+    def __init__(self, detection_engine):
+        self.engine = detection_engine
+
+    def detect_static_image(self, image_path):
+        """检测静态图像"""
+        print(f"正在加载图像: {image_path}")
+
+        if not self._check_image_exists(image_path):
+            return
+
+        frame = self._load_image(image_path)
+        if frame is None or frame.size == 0:
+            print("错误: 无法读取图像文件或图像为空")
+            return
+
+        print("正在检测图像...")
+        results, annotated_frame = self._perform_detection(frame)
+
+        if annotated_frame is None:
+            print("错误: 检测未返回有效图像")
+            return
+
+        # 显示检测结果文本
+        self._display_results(results)
+
+        # 显示图像窗口（带固定初始大小）
+        self._show_image(annotated_frame, 'YOLO 检测结果 - 静态图像')
+
+    def _check_image_exists(self, image_path):
+        import os
+        if not os.path.exists(image_path):
+            print(f"错误: 图像文件不存在 - {image_path}")
+            return False
+        return True
+
+    def _load_image(self, image_path):
+        try:
+            with open(image_path, "rb") as f:
+                bytes_data = bytearray(f.read())
+            nparr = np.frombuffer(bytes_data, np.uint8)
+            frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+            return frame
+        except Exception as e:
+            print(f"加载图像时出错: {e}")
+            return None
+
+    def _perform_detection(self, frame):
+        try:
+            annotated_frame, results = self.engine.detect(frame)
+            return results, annotated_frame
+        except Exception as e:
+            print(f"检测过程中发生错误: {e}")
+            traceback.print_exc()
+            return [], None
+
+    def _display_results(self, results):
+        if not results:
+            print("未检测到任何对象（results 为空）")
+            return
+
+        result = results[0]
+        boxes = result.boxes
+        if len(boxes) == 0:
+            print("未检测到任何对象")
+            return
+
+        print("检测完成，正在显示结果...")
+        print(f"检测到 {len(boxes)} 个对象:")
+
+        names_list = self.engine.model.names
+
+        for i, box in enumerate(boxes):
+            try:
+                cls_index = int(box.cls.item())
+                confidence = box.conf.item()
+
+                if 0 <= cls_index < len(names_list):
+                    cls_name = names_list[cls_index]
+                else:
+                    cls_name = f"unknown_class_{cls_index}"
+                    print(f"  ⚠️ 警告：类别索引 {cls_index} 超出范围（共 {len(names_list)} 类）")
+
+                print(f"  {i+1}. {cls_name} (置信度: {confidence:.2f})")
+            except Exception as e:
+                print(f"  ⚠️ 解析第 {i+1} 个检测框时出错: {e}")
+
+    def _show_image(self, annotated_frame, window_name):
+        """显示图像，并设置窗口为可调整的小尺寸"""
+        if annotated_frame is None or annotated_frame.size == 0:
+            print("错误: 标注帧无效，无法显示")
+            return
+
+        print("\n图像已显示。按任意键关闭窗口...")
+        try:
+            # 创建可调整大小的窗口
+            cv2.namedWindow(window_name, cv2.WINDOW_NORMAL)
+            # 设置初始窗口大小为 800x600（你可以按需修改）
+            cv2.resizeWindow(window_name, 800, 600)
+            # 显示图像（OpenCV 会自动适应窗口）
+            cv2.imshow(window_name, annotated_frame)
+            cv2.waitKey(0)
+            cv2.destroyAllWindows()
+            print("窗口已关闭。")
+        except Exception as e:
+            print(f"显示图像时发生错误: {e}")
+            traceback.print_exc()

src/image_object_detection/main.py

@@ -0,0 +1,11 @@
+# main.py
+from ui_handler import UIHandler
+from config import Config
+
+def main():
+    config = Config()
+    ui_handler = UIHandler(config)
+    ui_handler.run()
+
+if __name__ == "__main__":
+    main()

src/image_object_detection/requirements.txt

@@ -0,0 +1,15 @@
+# YOLOv8核心库（必须）
+ultralytics>=8.0.0  # 推荐8.2.0（稳定版）
+# 图像处理（必须）
+opencv-python>=4.5.0  # 推荐4.8.1.78
+# 可视化（必须）
+matplotlib>=3.0.0  # 推荐3.7.1
+# PyTorch核心（ultralytics依赖，必须）
+torch>=2.0.0  # 推荐2.0.1或2.1.0
+torchvision>=0.15.0  # 与torch版本匹配
+# 基础数值计算（间接依赖，建议指定）
+numpy>=1.21.0  # 推荐1.24.3
+# 可选（补充依赖，避免隐性报错）
+pillow>=8.0.0  # 推荐9.5.0
+psutil>=5.8.0  # ultralytics监控系统资源用
+pyyaml>=6.0  # 自定义训练时解析yaml配置文件

src/image_object_detection/ui_handler.py

@@ -0,0 +1,75 @@
+# ui_handler.py
+import cv2
+from detection_engine import DetectionEngine
+from image_detector import ImageDetector
+from camera_detector import CameraDetector
+import sys
+import os
+
+class UIHandler:
+    def __init__(self, config):
+        self.config = config
+        self.detection_engine = DetectionEngine()
+        self.image_detector = ImageDetector(self.detection_engine)
+        self.camera_detector = CameraDetector(self.detection_engine)
+
+    def print_debug(self, message):
+        """调试输出函数"""
+        print(f"[DEBUG] {message}")
+        sys.stdout.flush()  # 强制刷新输出缓冲区
+
+    def display_menu(self):
+        """显示主菜单"""
+        print("=" * 60)
+        print("           欢迎使用 YOLO 对象检测系统")
+        print("=" * 60)
+
+        print(f"\n测试图像路径: {self.config.test_image_path}")
+        print(f"图像文件存在: {os.path.exists(self.config.test_image_path)}")
+
+        print("\n请选择检测模式:")
+        print("1. 检测静态图像")
+        print("2. 检测摄像头实时画面")
+        print("3. 退出程序")
+
+    def handle_choice(self, choice):
+        """处理用户选择"""
+        if choice == '1':
+            print("\n您选择了: 检测静态图像")
+            self.image_detector.detect_static_image(self.config.test_image_path)
+            return True
+        elif choice == '2':
+            print("\n您选择了: 检测摄像头实时画面")
+            print("注意：要退出摄像头模式，请按 Ctrl+C ！")
+            self.camera_detector.detect_camera()
+            return True
+        elif choice == '3':
+            print("\n程序已退出。")
+            return False
+        else:
+            print("\n无效选择，请输入 1、2 或 3")
+            return True
+
+    def run(self):
+        """运行主程序"""
+        self.print_debug("程序开始执行")
+        self.display_menu()
+
+        running = True
+        while running:
+            try:
+                choice = input("\n请输入您的选择 (1/2/3): ").strip()
+                self.print_debug(f"用户输入: {choice}")
+                running = self.handle_choice(choice)
+
+            except KeyboardInterrupt:
+                print("\n\n程序被用户中断。")
+                break
+            except EOFError:
+                print("\n输入结束，程序退出。")
+                break
+            except Exception as e:
+                print(f"\n发生错误: {e}")
+                break
+
+        self.print_debug("程序结束")