CHAPTER 6

비디오 분석

시간 기반 비전 이해

비디오 분석 개요

비디오는 시간에 따라 변화하는 이미지의 시퀀스입니다. 비디오 분석은 공간적 정보뿐만 아니라 시간적 패턴도 활용하여 움직임, 활동, 이벤트를 이해합니다.

주요 작업

비디오 처리 기초

OpenCV로 비디오 읽기

import cv2
import numpy as np

# 비디오 캡처 객체 생성
cap = cv2.VideoCapture('video.mp4')

# 비디오 속성 확인
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

print(f"FPS: {fps}, 크기: {width}x{height}, 총 프레임: {total_frames}")

# 프레임별 처리
while True:
    ret, frame = cap.read()
    if not ret:
        break

    # 프레임 처리
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    # 프레임 표시
    cv2.imshow('Frame', frame)

    # 'q' 키로 종료
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

비디오 쓰기

# 비디오 라이터 생성
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter('output.mp4', fourcc, 30.0, (640, 480))

# 프레임 쓰기
cap = cv2.VideoCapture(0)  # 웹캠
while True:
    ret, frame = cap.read()
    if not ret:
        break

    # 프레임 처리
    processed = cv2.resize(frame, (640, 480))

    # 비디오에 쓰기
    out.write(processed)

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

cap.release()
out.release()

객체 추적

비디오에서 객체의 위치를 프레임 간 연속적으로 추적합니다.

OpenCV 추적기

# 추적기 초기화 (여러 알고리즘 사용 가능)
tracker = cv2.TrackerCSRT_create()  # CSRT, KCF, MOSSE 등

cap = cv2.VideoCapture('video.mp4')
ret, frame = cap.read()

# 추적할 영역 선택
bbox = cv2.selectROI("Frame", frame, False)
tracker.init(frame, bbox)

while True:
    ret, frame = cap.read()
    if not ret:
        break

    # 추적 업데이트
    success, bbox = tracker.update(frame)

    if success:
        # 바운딩 박스 그리기
        x, y, w, h = [int(v) for v in bbox]
        cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2)
    else:
        cv2.putText(frame, "추적 실패", (100, 80),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.75, (0, 0, 255), 2)

    cv2.imshow('추적', frame)

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

cap.release()
cv2.destroyAllWindows()

딥러닝 기반 다중 객체 추적

from ultralytics import YOLO

# YOLO 추적 모델
model = YOLO('yolov8n.pt')

# 비디오에서 추적
results = model.track(
    'video.mp4',
    persist=True,  # 프레임 간 ID 유지
    tracker='botsort.yaml',  # 추적 알고리즘
    conf=0.3,  # 신뢰도 임계값
    iou=0.5,   # IoU 임계값
)

for result in results:
    boxes = result.boxes
    for box in boxes:
        # 추적 ID
        track_id = box.id
        # 클래스
        cls = box.cls
        # 바운딩 박스
        x1, y1, x2, y2 = box.xyxy[0]

        print(f"ID: {track_id}, 클래스: {cls}, 위치: ({x1}, {y1}, {x2}, {y2})")

행동 인식

비디오에서 사람의 행동과 활동을 인식합니다.

3D CNN

시공간 특징을 학습하기 위해 3D 합성곱을 사용합니다.

import torch.nn as nn

class C3D(nn.Module):
    """3D CNN for action recognition"""
    def __init__(self, num_classes=101):
        super(C3D, self).__init__()

        # 3D 합성곱 레이어
        self.conv1 = nn.Conv3d(3, 64, kernel_size=(3, 3, 3), padding=(1, 1, 1))
        self.pool1 = nn.MaxPool3d(kernel_size=(1, 2, 2), stride=(1, 2, 2))

        self.conv2 = nn.Conv3d(64, 128, kernel_size=(3, 3, 3), padding=(1, 1, 1))
        self.pool2 = nn.MaxPool3d(kernel_size=(2, 2, 2), stride=(2, 2, 2))

        self.conv3a = nn.Conv3d(128, 256, kernel_size=(3, 3, 3), padding=(1, 1, 1))
        self.conv3b = nn.Conv3d(256, 256, kernel_size=(3, 3, 3), padding=(1, 1, 1))
        self.pool3 = nn.MaxPool3d(kernel_size=(2, 2, 2), stride=(2, 2, 2))

        # 완전 연결 레이어
        self.fc1 = nn.Linear(256 * 4 * 14 * 14, 4096)
        self.fc2 = nn.Linear(4096, 4096)
        self.fc3 = nn.Linear(4096, num_classes)

        self.dropout = nn.Dropout(0.5)
        self.relu = nn.ReLU()

    def forward(self, x):
        # x: [batch, channels, depth, height, width]
        x = self.relu(self.conv1(x))
        x = self.pool1(x)

        x = self.relu(self.conv2(x))
        x = self.pool2(x)

        x = self.relu(self.conv3a(x))
        x = self.relu(self.conv3b(x))
        x = self.pool3(x)

        # 평탄화
        x = x.view(x.size(0), -1)

        x = self.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.relu(self.fc2(x))
        x = self.dropout(x)
        x = self.fc3(x)

        return x

# 모델 생성
model = C3D(num_classes=101)

# 입력: [batch_size, 3, 16, 112, 112]
# 16프레임, 112x112 크기
dummy_input = torch.randn(1, 3, 16, 112, 112)
output = model(dummy_input)
print(f"출력 shape: {output.shape}")  # [1, 101]

Two-Stream Networks

공간 스트림(RGB)과 시간 스트림(Optical Flow)을 결합합니다.

class TwoStreamNetwork(nn.Module):
    def __init__(self, num_classes=101):
        super().__init__()

        # 공간 스트림 (RGB 프레임)
        self.spatial_stream = models.resnet50(pretrained=True)
        self.spatial_stream.fc = nn.Linear(2048, num_classes)

        # 시간 스트림 (Optical Flow)
        self.temporal_stream = models.resnet50(pretrained=True)
        # 첫 레이어를 광학 흐름 입력에 맞게 수정 (20채널)
        self.temporal_stream.conv1 = nn.Conv2d(20, 64, kernel_size=7, stride=2, padding=3)
        self.temporal_stream.fc = nn.Linear(2048, num_classes)

        # 융합 레이어
        self.fusion = nn.Linear(num_classes * 2, num_classes)

    def forward(self, rgb, flow):
        # RGB 스트림
        spatial_out = self.spatial_stream(rgb)

        # 광학 흐름 스트림
        temporal_out = self.temporal_stream(flow)

        # 융합
        combined = torch.cat([spatial_out, temporal_out], dim=1)
        output = self.fusion(combined)

        return output

광학 흐름 (Optical Flow)

연속 프레임 간의 픽셀 움직임을 나타냅니다.

import cv2

# 비디오 캡처
cap = cv2.VideoCapture('video.mp4')
ret, frame1 = cap.read()
prev_gray = cv2.cvtColor(frame1, cv2.COLOR_BGR2GRAY)

while True:
    ret, frame2 = cap.read()
    if not ret:
        break

    gray = cv2.cvtColor(frame2, cv2.COLOR_BGR2GRAY)

    # Farneback 광학 흐름 계산
    flow = cv2.calcOpticalFlowFarneback(
        prev_gray, gray,
        None,
        pyr_scale=0.5,
        levels=3,
        winsize=15,
        iterations=3,
        poly_n=5,
        poly_sigma=1.2,
        flags=0
    )

    # 흐름을 시각화 (HSV 색상 공간 사용)
    hsv = np.zeros_like(frame1)
    hsv[..., 1] = 255

    magnitude, angle = cv2.cartToPolar(flow[..., 0], flow[..., 1])
    hsv[..., 0] = angle * 180 / np.pi / 2
    hsv[..., 2] = cv2.normalize(magnitude, None, 0, 255, cv2.NORM_MINMAX)

    rgb = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)

    cv2.imshow('Optical Flow', rgb)

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

    prev_gray = gray

cap.release()
cv2.destroyAllWindows()

비디오 트랜스포머

TimeSformer

비디오 이해를 위한 순수 트랜스포머 아키텍처입니다.

class TimeSformer(nn.Module):
    def __init__(
        self,
        img_size=224,
        patch_size=16,
        num_frames=8,
        num_classes=400,
        embed_dim=768,
        depth=12,
        num_heads=12
    ):
        super().__init__()

        # 패치 임베딩
        self.patch_embed = nn.Conv3d(
            3, embed_dim,
            kernel_size=(1, patch_size, patch_size),
            stride=(1, patch_size, patch_size)
        )

        num_patches = (img_size // patch_size) ** 2

        # CLS 토큰
        self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))

        # 시공간 위치 임베딩
        self.pos_embed_spatial = nn.Parameter(
            torch.zeros(1, num_patches, embed_dim)
        )
        self.pos_embed_temporal = nn.Parameter(
            torch.zeros(1, num_frames, embed_dim)
        )

        # 트랜스포머 블록들
        self.blocks = nn.ModuleList([
            TransformerBlock(embed_dim, num_heads)
            for _ in range(depth)
        ])

        self.norm = nn.LayerNorm(embed_dim)
        self.head = nn.Linear(embed_dim, num_classes)

    def forward(self, x):
        # x: [batch, 3, num_frames, height, width]
        batch_size, _, num_frames, _, _ = x.shape

        # 패치 임베딩
        x = self.patch_embed(x)  # [batch, embed_dim, num_frames, H', W']
        x = x.flatten(2).transpose(1, 2)  # [batch, num_patches, embed_dim]

        # CLS 토큰 추가
        cls_token = self.cls_token.expand(batch_size, -1, -1)
        x = torch.cat([cls_token, x], dim=1)

        # 위치 임베딩
        x[:, 1:] += self.pos_embed_spatial
        x[:, 1:] += self.pos_embed_temporal.repeat_interleave(
            x.size(1) // num_frames, dim=1
        )

        # 트랜스포머 블록들
        for block in self.blocks:
            x = block(x)

        x = self.norm(x)
        cls_output = x[:, 0]

        return self.head(cls_output)

실시간 비디오 분석

프레임 스키핑

# 모든 프레임 대신 N번째 프레임만 처리
frame_skip = 3
frame_count = 0

cap = cv2.VideoCapture('video.mp4')

while True:
    ret, frame = cap.read()
    if not ret:
        break

    frame_count += 1

    # N번째 프레임만 처리
    if frame_count % frame_skip != 0:
        continue

    # 무거운 처리 수행
    results = model(frame)

cap.release()

비동기 처리

import threading
from queue import Queue

def capture_frames(cap, frame_queue):
    """프레임 캡처 스레드"""
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        frame_queue.put(frame)

def process_frames(frame_queue, model):
    """프레임 처리 스레드"""
    while True:
        if not frame_queue.empty():
            frame = frame_queue.get()
            results = model(frame)
            # 결과 처리

# 큐 생성
frame_queue = Queue(maxsize=30)

# 스레드 시작
cap = cv2.VideoCapture('video.mp4')
capture_thread = threading.Thread(target=capture_frames, args=(cap, frame_queue))
process_thread = threading.Thread(target=process_frames, args=(frame_queue, model))

capture_thread.start()
process_thread.start()

capture_thread.join()
process_thread.join()

비디오 응용

스포츠 분석

보안 및 감시

의료

자율주행

요약

  • 비디오 분석은 공간적 및 시간적 정보를 모두 활용합니다
  • 객체 추적은 프레임 간 객체의 위치를 연속적으로 추적합니다
  • 3D CNN과 Two-Stream 네트워크는 행동 인식에 효과적입니다
  • 광학 흐름은 픽셀 수준의 움직임 정보를 제공합니다
  • 비디오 트랜스포머(TimeSformer)는 시공간 관계를 학습합니다
  • 프레임 스키핑과 비동기 처리로 실시간 성능을 달성합니다
  • 스포츠, 보안, 의료, 자율주행 등 다양한 응용 분야가 있습니다
  • 弘益人間 - 비디오 분석으로 안전, 건강, 스포츠에서 인류에 기여합니다

복습 문제

  1. 비디오 분석이 단순 이미지 분류보다 어려운 이유는 무엇인가요?
  2. 객체 추적의 주요 도전 과제는 무엇이며, 어떻게 해결하나요?
  3. 3D CNN이 2D CNN보다 비디오 이해에 적합한 이유를 설명하세요.
  4. Two-Stream 네트워크의 공간 스트림과 시간 스트림의 역할은 무엇인가요?
  5. 광학 흐름이 무엇이며 행동 인식에 어떻게 사용되나요?
  6. 실시간 비디오 처리를 위한 최적화 기법은 무엇인가요?
  7. 비디오 분석이 자율주행에 어떻게 활용되는지 구체적으로 설명하세요.