image_analysis.py

  1#!/usr/bin/env python3
  2"""
  3영상 분석 프로젝트 - 시뮬레이션 모드
  4Pi Camera를 사용한 영상 캡처, 객체 검출, 모션 감지
  5
  6시뮬레이션 모드로 실제 하드웨어 없이 실행 가능
  7"""
  8
  9import time
 10import json
 11import os
 12from datetime import datetime
 13from typing import List, Dict, Tuple, Optional
 14from dataclasses import dataclass
 15from enum import Enum
 16import threading
 17import io
 18import random
 19
 20# numpy는 시뮬레이션 모드에서 선택적
 21try:
 22    import numpy as np
 23    HAS_NUMPY = True
 24except ImportError:
 25    HAS_NUMPY = False
 26    print("[경고] numpy가 설치되지 않음. 시뮬레이션 모드로 실행합니다.")
 27    print("       실제 사용 시 'pip install numpy' 설치 필요\n")
 28
 29
 30# ==============================================================================
 31# 데이터 클래스
 32# ==============================================================================
 33
 34class DetectionClass(Enum):
 35    """검출 가능한 객체 클래스"""
 36    PERSON = "person"
 37    CAR = "car"
 38    DOG = "dog"
 39    CAT = "cat"
 40    BICYCLE = "bicycle"
 41    UNKNOWN = "unknown"
 42
 43
 44@dataclass
 45class BoundingBox:
 46    """바운딩 박스"""
 47    x1: int
 48    y1: int
 49    x2: int
 50    y2: int
 51
 52    @property
 53    def width(self) -> int:
 54        return self.x2 - self.x1
 55
 56    @property
 57    def height(self) -> int:
 58        return self.y2 - self.y1
 59
 60    @property
 61    def area(self) -> int:
 62        return self.width * self.height
 63
 64
 65@dataclass
 66class Detection:
 67    """객체 검출 결과"""
 68    class_name: str
 69    confidence: float
 70    bbox: BoundingBox
 71    timestamp: datetime
 72
 73
 74@dataclass
 75class MotionEvent:
 76    """모션 감지 이벤트"""
 77    regions: List[BoundingBox]
 78    area: int
 79    timestamp: datetime
 80    frame_id: int
 81
 82
 83# ==============================================================================
 84# 시뮬레이션 카메라
 85# ==============================================================================
 86
 87class SimulatedCamera:
 88    """시뮬레이션 카메라 (실제 Pi Camera 대체)"""
 89
 90    def __init__(self, resolution: Tuple[int, int] = (640, 480)):
 91        self.resolution = resolution
 92        self.is_running = False
 93        self.frame_count = 0
 94        print(f"[시뮬레이션] 카메라 초기화: {resolution[0]}x{resolution[1]}")
 95
 96    def start(self):
 97        """카메라 시작"""
 98        self.is_running = True
 99        print("[시뮬레이션] 카메라 시작")
100
101    def stop(self):
102        """카메라 중지"""
103        self.is_running = False
104        print("[시뮬레이션] 카메라 중지")
105
106    def capture_frame(self):
107        """프레임 캡처 (시뮬레이션)"""
108        if not self.is_running:
109            raise RuntimeError("카메라가 시작되지 않음")
110
111        # 랜덤 노이즈로 프레임 생성
112        width, height = self.resolution
113
114        if HAS_NUMPY:
115            frame = np.random.randint(0, 256, (height, width, 3), dtype=np.uint8)
116
117            # 간단한 패턴 추가 (움직이는 사각형)
118            x_offset = (self.frame_count * 5) % width
119            y_offset = 100
120            frame[y_offset:y_offset+50, x_offset:x_offset+50] = [255, 0, 0]
121        else:
122            # numpy 없을 때는 간단한 리스트로 시뮬레이션
123            frame = [[[random.randint(0, 255) for _ in range(3)]
124                      for _ in range(width)]
125                     for _ in range(height)]
126
127        self.frame_count += 1
128        return frame
129
130    def capture_image(self, filename: str):
131        """이미지 저장 (시뮬레이션)"""
132        frame = self.capture_frame()
133        print(f"[시뮬레이션] 이미지 저장: {filename} ({frame.shape})")
134        # 실제로는 PIL.Image.fromarray(frame).save(filename)
135
136
137# ==============================================================================
138# TFLite 객체 검출 (시뮬레이션)
139# ==============================================================================
140
141class TFLiteObjectDetector:
142    """TFLite 객체 검출기 (시뮬레이션)"""
143
144    COCO_LABELS = [
145        'person', 'bicycle', 'car', 'motorcycle', 'airplane', 'bus', 'train',
146        'truck', 'boat', 'traffic light', 'fire hydrant', 'stop sign',
147        'parking meter', 'bench', 'bird', 'cat', 'dog', 'horse', 'sheep',
148        'cow', 'elephant', 'bear', 'zebra', 'giraffe'
149    ]
150
151    def __init__(self, model_path: str = None, threshold: float = 0.5):
152        self.model_path = model_path or "simulated_model.tflite"
153        self.threshold = threshold
154        self.input_size = (300, 300)
155        print(f"[시뮬레이션] TFLite 모델 로드: {self.model_path}")
156        print(f"  - 입력 크기: {self.input_size}")
157        print(f"  - 임계값: {self.threshold}")
158
159    def preprocess(self, frame):
160        """전처리"""
161        # 리사이즈 시뮬레이션
162        target_h, target_w = self.input_size
163
164        if HAS_NUMPY:
165            # 실제로는 cv2.resize 사용
166            resized = np.random.randint(0, 256, (target_h, target_w, 3), dtype=np.uint8)
167            # 정규화
168            normalized = resized.astype(np.float32) / 255.0
169            return normalized
170        else:
171            # numpy 없을 때는 간단히 처리
172            return frame
173
174    def detect(self, frame) -> List[Detection]:
175        """객체 검출 (시뮬레이션)"""
176        # 전처리
177        input_data = self.preprocess(frame)
178
179        # 추론 시뮬레이션 (일부 확률로 객체 검출)
180        detections = []
181
182        # 랜덤하게 0-3개 객체 검출
183        num_objects = random.choice([0, 0, 0, 1, 1, 2])  # 대부분 0, 가끔 1-2개
184
185        if HAS_NUMPY:
186            h, w = frame.shape[:2]
187        else:
188            h, w = len(frame), len(frame[0]) if frame else 0
189
190        for _ in range(num_objects):
191            # 랜덤 클래스 선택
192            class_id = random.randint(0, min(len(self.COCO_LABELS), 10) - 1)
193            class_name = self.COCO_LABELS[class_id]
194
195            # 랜덤 신뢰도
196            confidence = random.uniform(self.threshold, 1.0)
197
198            # 랜덤 바운딩 박스
199            x1 = random.randint(0, w // 2)
200            y1 = random.randint(0, h // 2)
201            x2 = x1 + random.randint(50, w // 3)
202            y2 = y1 + random.randint(50, h // 3)
203
204            bbox = BoundingBox(
205                x1=min(x1, w-1),
206                y1=min(y1, h-1),
207                x2=min(x2, w-1),
208                y2=min(y2, h-1)
209            )
210
211            detection = Detection(
212                class_name=class_name,
213                confidence=confidence,
214                bbox=bbox,
215                timestamp=datetime.now()
216            )
217            detections.append(detection)
218
219        return detections
220
221    def draw_detections(self, frame, detections: List[Detection]):
222        """검출 결과 시각화 (시뮬레이션)"""
223        if HAS_NUMPY:
224            result = frame.copy()
225        else:
226            result = frame  # 시뮬레이션 모드에서는 복사하지 않음
227
228        for det in detections:
229            # 실제로는 cv2.rectangle, cv2.putText 사용
230            label = f"{det.class_name}: {det.confidence:.2f}"
231            print(f"  [검출] {label} at ({det.bbox.x1}, {det.bbox.y1})")
232
233        return result
234
235
236# ==============================================================================
237# 모션 감지
238# ==============================================================================
239
240class MotionDetector:
241    """모션 감지기"""
242
243    def __init__(self, threshold: int = 30, min_area: int = 500):
244        self.threshold = threshold
245        self.min_area = min_area
246        self.prev_frame = None
247        self.motion_count = 0
248        print(f"[시뮬레이션] 모션 감지기 초기화")
249        print(f"  - 임계값: {threshold}")
250        print(f"  - 최소 영역: {min_area} pixels")
251
252    def detect_motion(self, frame) -> Tuple[bool, List[BoundingBox]]:
253        """모션 감지"""
254        if HAS_NUMPY:
255            # 그레이스케일 변환 시뮬레이션
256            gray = np.mean(frame, axis=2).astype(np.uint8)
257
258            # 가우시안 블러 시뮬레이션
259            # 실제로는 cv2.GaussianBlur 사용
260
261            if self.prev_frame is None:
262                self.prev_frame = gray
263                return False, []
264
265            # 프레임 차이 계산
266            frame_delta = np.abs(gray.astype(np.int16) - self.prev_frame.astype(np.int16))
267
268            # 임계값 적용
269            thresh = (frame_delta > self.threshold).astype(np.uint8) * 255
270
271            # 변화 영역 비율 계산
272            motion_pixels = np.sum(thresh > 0)
273            total_pixels = thresh.size
274            motion_ratio = motion_pixels / total_pixels
275
276            self.prev_frame = gray
277
278            # 일정 비율 이상 변화 시 모션 감지
279            motion_detected = motion_ratio > 0.05  # 5% 이상 변화
280
281            h, w = frame.shape[:2]
282        else:
283            # numpy 없을 때 간단한 시뮬레이션
284            if self.prev_frame is None:
285                self.prev_frame = frame
286                return False, []
287
288            # 랜덤하게 모션 감지 (10% 확률)
289            motion_detected = random.random() < 0.1
290            h, w = len(frame), len(frame[0]) if frame else 0
291
292        regions = []
293        if motion_detected:
294            # 시뮬레이션: 랜덤 모션 영역 생성
295            num_regions = random.randint(1, 3)
296
297            for _ in range(num_regions):
298                x1 = random.randint(0, w // 2)
299                y1 = random.randint(0, h // 2)
300                x2 = x1 + random.randint(50, 150)
301                y2 = y1 + random.randint(50, 150)
302
303                bbox = BoundingBox(
304                    x1=min(x1, w-1),
305                    y1=min(y1, h-1),
306                    x2=min(x2, w-1),
307                    y2=min(y2, h-1)
308                )
309
310                if bbox.area >= self.min_area:
311                    regions.append(bbox)
312
313            self.motion_count += 1
314
315        return motion_detected, regions
316
317
318# ==============================================================================
319# 영상 스트리밍 (개념)
320# ==============================================================================
321
322class VideoStreamer:
323    """비디오 스트리머 (MJPEG over HTTP 개념)"""
324
325    def __init__(self, camera: SimulatedCamera, port: int = 8080):
326        self.camera = camera
327        self.port = port
328        self.is_streaming = False
329        self.frame_rate = 30
330        self.lock = threading.Lock()
331        self.current_frame = None
332        print(f"[시뮬레이션] 비디오 스트리머 초기화 (포트 {port})")
333
334    def _capture_loop(self):
335        """캡처 루프"""
336        frame_interval = 1.0 / self.frame_rate
337
338        while self.is_streaming:
339            start_time = time.time()
340
341            frame = self.camera.capture_frame()
342
343            # JPEG 인코딩 시뮬레이션
344            # 실제로는 PIL.Image.fromarray(frame).save(buffer, 'JPEG')
345
346            with self.lock:
347                self.current_frame = frame
348
349            # 프레임 레이트 유지
350            elapsed = time.time() - start_time
351            sleep_time = max(0, frame_interval - elapsed)
352            time.sleep(sleep_time)
353
354    def start_streaming(self):
355        """스트리밍 시작"""
356        if self.is_streaming:
357            return
358
359        self.is_streaming = True
360        self.thread = threading.Thread(target=self._capture_loop, daemon=True)
361        self.thread.start()
362        print(f"[시뮬레이션] 스트리밍 시작: http://0.0.0.0:{self.port}/video_feed")
363
364    def stop_streaming(self):
365        """스트리밍 중지"""
366        self.is_streaming = False
367        if hasattr(self, 'thread'):
368            self.thread.join(timeout=2)
369        print("[시뮬레이션] 스트리밍 중지")
370
371    def get_frame(self):
372        """현재 프레임 반환"""
373        with self.lock:
374            return self.current_frame
375
376
377# ==============================================================================
378# 결과 로깅
379# ==============================================================================
380
381class ResultLogger:
382    """검출 및 모션 결과 로깅"""
383
384    def __init__(self, log_dir: str = "logs"):
385        self.log_dir = log_dir
386        os.makedirs(log_dir, exist_ok=True)
387        self.log_file = os.path.join(log_dir, f"detection_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl")
388        print(f"[시뮬레이션] 로거 초기화: {self.log_file}")
389
390    def log_detection(self, detections: List[Detection], frame_id: int):
391        """객체 검출 로깅"""
392        log_entry = {
393            "type": "detection",
394            "timestamp": datetime.now().isoformat(),
395            "frame_id": frame_id,
396            "count": len(detections),
397            "objects": [
398                {
399                    "class": det.class_name,
400                    "confidence": det.confidence,
401                    "bbox": [det.bbox.x1, det.bbox.y1, det.bbox.x2, det.bbox.y2]
402                }
403                for det in detections
404            ]
405        }
406
407        with open(self.log_file, 'a') as f:
408            f.write(json.dumps(log_entry) + '\n')
409
410    def log_motion(self, event: MotionEvent):
411        """모션 이벤트 로깅"""
412        log_entry = {
413            "type": "motion",
414            "timestamp": event.timestamp.isoformat(),
415            "frame_id": event.frame_id,
416            "region_count": len(event.regions),
417            "total_area": event.area,
418            "regions": [
419                [r.x1, r.y1, r.x2, r.y2] for r in event.regions
420            ]
421        }
422
423        with open(self.log_file, 'a') as f:
424            f.write(json.dumps(log_entry) + '\n')
425
426    def get_statistics(self) -> Dict:
427        """로그 통계"""
428        if not os.path.exists(self.log_file):
429            return {}
430
431        detection_count = 0
432        motion_count = 0
433        object_classes = {}
434
435        with open(self.log_file, 'r') as f:
436            for line in f:
437                entry = json.loads(line)
438                if entry['type'] == 'detection':
439                    detection_count += 1
440                    for obj in entry['objects']:
441                        cls = obj['class']
442                        object_classes[cls] = object_classes.get(cls, 0) + 1
443                elif entry['type'] == 'motion':
444                    motion_count += 1
445
446        return {
447            "detection_events": detection_count,
448            "motion_events": motion_count,
449            "object_classes": object_classes
450        }
451
452
453# ==============================================================================
454# 성능 모니터링
455# ==============================================================================
456
457class PerformanceMonitor:
458    """성능 모니터"""
459
460    def __init__(self):
461        self.metrics = {
462            "fps": [],
463            "detection_time": [],
464            "frame_count": 0,
465            "start_time": None
466        }
467
468    def start(self):
469        """모니터링 시작"""
470        self.metrics["start_time"] = time.time()
471
472    def record_frame(self, processing_time: float):
473        """프레임 처리 시간 기록"""
474        self.metrics["frame_count"] += 1
475        self.metrics["detection_time"].append(processing_time)
476
477        if processing_time > 0:
478            fps = 1.0 / processing_time
479            self.metrics["fps"].append(fps)
480
481    def get_report(self) -> Dict:
482        """성능 보고서"""
483        if self.metrics["start_time"] is None:
484            return {}
485
486        elapsed = time.time() - self.metrics["start_time"]
487
488        if HAS_NUMPY:
489            avg_fps = np.mean(self.metrics["fps"]) if self.metrics["fps"] else 0
490            avg_detection_time = np.mean(self.metrics["detection_time"]) if self.metrics["detection_time"] else 0
491        else:
492            avg_fps = sum(self.metrics["fps"]) / len(self.metrics["fps"]) if self.metrics["fps"] else 0
493            avg_detection_time = sum(self.metrics["detection_time"]) / len(self.metrics["detection_time"]) if self.metrics["detection_time"] else 0
494
495        return {
496            "total_frames": self.metrics["frame_count"],
497            "elapsed_time": elapsed,
498            "average_fps": avg_fps,
499            "average_detection_time_ms": avg_detection_time * 1000,
500            "frames_per_second_actual": self.metrics["frame_count"] / elapsed if elapsed > 0 else 0
501        }
502
503
504# ==============================================================================
505# 통합 영상 분석 시스템
506# ==============================================================================
507
508class ImageAnalysisSystem:
509    """통합 영상 분석 시스템"""
510
511    def __init__(self, config: Optional[Dict] = None):
512        config = config or {}
513
514        # 카메라
515        resolution = config.get('resolution', (640, 480))
516        self.camera = SimulatedCamera(resolution)
517
518        # 객체 검출
519        self.detector = TFLiteObjectDetector(
520            threshold=config.get('detection_threshold', 0.5)
521        )
522
523        # 모션 감지
524        self.motion_detector = MotionDetector(
525            threshold=config.get('motion_threshold', 30),
526            min_area=config.get('min_motion_area', 500)
527        )
528
529        # 로거
530        self.logger = ResultLogger(log_dir=config.get('log_dir', 'logs'))
531
532        # 성능 모니터
533        self.perf_monitor = PerformanceMonitor()
534
535        # 스트리밍 (옵션)
536        self.enable_streaming = config.get('enable_streaming', False)
537        if self.enable_streaming:
538            self.streamer = VideoStreamer(self.camera, port=config.get('stream_port', 8080))
539
540        self.is_running = False
541
542        print("\n" + "="*60)
543        print("영상 분석 시스템 초기화 완료")
544        print("="*60)
545
546    def run(self, duration: float = 60, detect_objects: bool = True, detect_motion: bool = True):
547        """시스템 실행"""
548        print(f"\n시스템 시작 (실행 시간: {duration}초)")
549        print(f"  - 객체 검출: {'ON' if detect_objects else 'OFF'}")
550        print(f"  - 모션 감지: {'ON' if detect_motion else 'OFF'}")
551        print(f"  - 스트리밍: {'ON' if self.enable_streaming else 'OFF'}")
552        print()
553
554        self.camera.start()
555        self.perf_monitor.start()
556
557        if self.enable_streaming:
558            self.streamer.start_streaming()
559
560        self.is_running = True
561        start_time = time.time()
562        frame_id = 0
563
564        try:
565            while time.time() - start_time < duration and self.is_running:
566                frame_start = time.time()
567
568                # 프레임 캡처
569                frame = self.camera.capture_frame()
570                frame_id += 1
571
572                # 객체 검출
573                if detect_objects:
574                    detections = self.detector.detect(frame)
575                    if detections:
576                        print(f"[프레임 {frame_id}] 객체 검출: {len(detections)}개")
577                        for det in detections:
578                            print(f"  - {det.class_name} (신뢰도: {det.confidence:.2f})")
579                        self.logger.log_detection(detections, frame_id)
580
581                # 모션 감지
582                if detect_motion:
583                    motion_detected, regions = self.motion_detector.detect_motion(frame)
584                    if motion_detected:
585                        total_area = sum(r.area for r in regions)
586                        event = MotionEvent(
587                            regions=regions,
588                            area=total_area,
589                            timestamp=datetime.now(),
590                            frame_id=frame_id
591                        )
592                        print(f"[프레임 {frame_id}] 모션 감지: {len(regions)}개 영역 (면적: {total_area})")
593                        self.logger.log_motion(event)
594
595                # 성능 기록
596                processing_time = time.time() - frame_start
597                self.perf_monitor.record_frame(processing_time)
598
599                # 프레임 레이트 조절
600                sleep_time = max(0, 0.1 - processing_time)  # ~10 FPS
601                time.sleep(sleep_time)
602
603        except KeyboardInterrupt:
604            print("\n\n사용자 중단")
605
606        finally:
607            self._cleanup()
608
609    def _cleanup(self):
610        """정리"""
611        self.is_running = False
612        self.camera.stop()
613
614        if self.enable_streaming:
615            self.streamer.stop_streaming()
616
617        # 통계 출력
618        print("\n" + "="*60)
619        print("실행 완료 - 통계")
620        print("="*60)
621
622        # 성능 통계
623        perf_report = self.perf_monitor.get_report()
624        print("\n[성능]")
625        print(f"  총 프레임: {perf_report.get('total_frames', 0)}")
626        print(f"  실행 시간: {perf_report.get('elapsed_time', 0):.1f}초")
627        print(f"  평균 FPS: {perf_report.get('average_fps', 0):.1f}")
628        print(f"  평균 검출 시간: {perf_report.get('average_detection_time_ms', 0):.1f}ms")
629
630        # 로그 통계
631        log_stats = self.logger.get_statistics()
632        print("\n[검출 통계]")
633        print(f"  검출 이벤트: {log_stats.get('detection_events', 0)}회")
634        print(f"  모션 이벤트: {log_stats.get('motion_events', 0)}회")
635
636        object_classes = log_stats.get('object_classes', {})
637        if object_classes:
638            print("\n[검출된 객체]")
639            for cls, count in sorted(object_classes.items(), key=lambda x: x[1], reverse=True):
640                print(f"  {cls}: {count}회")
641
642        print("\n" + "="*60)
643
644
645# ==============================================================================
646# 메인 실행
647# ==============================================================================
648
649def main():
650    """메인 함수"""
651    print("영상 분석 프로젝트 - 시뮬레이션 모드")
652    print("="*60)
653    print("이 프로그램은 실제 Pi Camera 없이 시뮬레이션으로 동작합니다.")
654    print()
655
656    # 설정
657    config = {
658        'resolution': (640, 480),
659        'detection_threshold': 0.6,
660        'motion_threshold': 30,
661        'min_motion_area': 500,
662        'log_dir': 'logs',
663        'enable_streaming': False,  # 실제 Flask 서버는 실행하지 않음
664        'stream_port': 8080
665    }
666
667    # 시스템 생성
668    system = ImageAnalysisSystem(config)
669
670    # 실행 (테스트용으로 짧게 설정, 실제 사용시 duration 증가)
671    import sys
672    test_mode = '--test' in sys.argv
673    duration = 5 if test_mode else 30
674
675    system.run(
676        duration=duration,  # 테스트: 5초, 일반: 30초
677        detect_objects=True,
678        detect_motion=True
679    )
680
681
682if __name__ == "__main__":
683    main()