실시간 흉기 난동 상황 탐지 서비스 개발기

devstone·2024년 5월 21일

Capstone Project

목록 보기

2/2

Intro

좌충우돌 모델 선정기

흉기 난동 상황 탐지 서비스를 위한 객체 탐지 모델을 선정하는 과정에서 다양한 모델들을 검토하게 되었다. 보통은 이런 서비스를 개발할 때에 흔히 YOLO모델을 많이 사용하던데, 이 지점에서 의문이 생겼다. 이런 지도 학습 모델을 사용하려면 대량의 데이터가 필요할텐데 우리가 과연 그걸 해낼 수 있을까 → 졸업 프로젝트 단에서 이런 방식으로 프로덕트를 개발하기엔 공수가 너무 클 것이라고 느꼈고, 우리는 비지도학습 기반 객체 탐지 모델들 중 어떤 것을 채택할 지를 고민하였다. 후보군은 아래와 같았다.

Few-shot Learning 모델

적은 양의 학습 데이터로 모델을 학습시키는 방법이다. 기존의 학습된 모델을 기반으로 적은 양의 새로운 데이터를 통해 빠르게 학습할 수 있다
- Meta R-CNN
  - 특징: Meta-learning을 사용하여 적은 수의 예제로부터 객체 탐지를 학습.
  - 장점: 새로운 클래스에 대한 빠른 적응.
  - 단점: 여전히 약간의 학습 데이터가 필요.
Zero-shot Learning 모델

학습된 객체 외의 새로운 객체를 인식할 수 있는 방법이다. 모델은 학습 데이터 외에도 객체의 속성이나 설명을 통해 새로운 객체를 인식한다. 이러한 방식은 추가적인 데이터 없이도 새로운 객체를 인식할 수 있어 매우 유연하다.
- OWLv2
  - 특징: Open vocabulary 학습을 기반으로 하는 제로샷 객체 탐지 모델로, 다양한 객체를 인식할 수 있음. 학습된 범주 외의 새로운 객체를 탐지하는 데 강점을 가짐.
  - 장점: 다양한 비전 작업에 적용 가능, 추가 학습 데이터 없이 새로운 객체 인식 가능, 유연한 객체 인식 능력.
  - 단점: 실시간 성능이 떨어질 수 있음, 복잡한 설정과 최적화가 필요할 수 있음.
- DETIC (Detection Transformer for Image Classification)
  - 특징: 최대 21,000개의 클래스를 탐지할 수 있는 모델로, 매우 광범위한 객체 인식이 가능함. 제로샷 객체 탐지 모델로 사전 학습된 범주 외의 새로운 객체를 인식할 수 있음.
  - 장점: 광범위한 클래스 탐지, 높은 정확도, 다양한 응용 분야에 쉽게 적용 가능, 추가 학습 데이터 없이도 새로운 객체 인식 가능.
  - 단점: 실시간 성능이 떨어질 수 있음, 대규모 클래스 탐지로 인해 계산 자원이 많이 필요함.
- Grounding DINO
  - 특징: 제로샷 객체 탐지 모델로, 입력된 카테고리 이름이나 지시 표현을 기반으로 객체를 탐지함. 텍스트와 비전 모달리티의 융합을 통해 새로운 객체를 인식함.
  - 장점: 다양한 객체 인식, 추가 학습 데이터 없이도 새로운 객체 인식 가능, 실시간 대응 가능.
  - 단점: 일부 특수한 경우에 대한 탐지 성능이 떨어질 수 있음.

Grounding DINO 모델 선정 이유

기본적으로 제로샷 러닝 모델들이 흥미롭다고 느껴졌고, 최소의 데이터도 필요하지 않다는 것이 우리 상황에서 매력적이라고 느껴 제로샷 러닝 모델을 사용해보고자 하였다.

Grounding DINO 모델을 선정한 이유는 다음과 같다

language + vision 융합
- Grounding DINO는 텍스트 설명을 기반으로 객체를 탐지할 수 있는 능력을 가지고 있다. 이는 다양한 흉기(예: 칼, 망치 등)를 텍스트로 명시하여 탐지할 수 있다는 의미이다. 예를 들어, "칼을 든 사람"이라는 텍스트 입력을 통해 해당 상황을 인식할 수 있다.
제로샷 학습 능력
- Grounding DINO는 사전 학습된 범주 외의 새로운 객체를 인식할 수 있다. 이는 흉기 난동과 같은 예기치 않은 상황에서도 유연하게 대응할 수 있게 한다. 새로운 유형의 위협이 발생하더라도 추가적인 학습 없이 즉시 인식할 수 있다.
실시간 대응
- Grounding DINO는 높은 성능과 빠른 응답 속도를 자랑하여 실시간 객체 탐지가 가능하다. 흉기 난동과 같은 긴급 상황에서는 실시간 대응이 필수적이다. Grounding DINO의 빠른 탐지 능력은 이러한 요구사항을 충족한다.
고성능 백본
- Swin Transformer-B 백본을 사용하여 높은 성능과 정확성을 보장한다. 이는 저화질 CCTV 영상에서도 효과적으로 객체를 탐지할 수 있도록 한다.
데이터 효율성
- 추가적인 학습 데이터 없이도 새로운 객체를 인식할 수 있기 때문에, 데이터 수집과 라벨링에 드는 시간을 절약할 수 있다. 이는 특히 데이터가 부족한 상황에서도 효과적으로 사용할 수 있다.

다른 제로샷 모델들과 비교했던 지점들이 분명 있었다.

OWLv2 → 다양한 객체를 인식할 수 있는 능력을 가지고 있으나, Grounding DINO만큼 실시간 성능과 언어-비전 융합 측면에서의 강점을 가지지 않는다.

DETIC → 최대 21,000개의 클래스를 탐지할 수 있지만, Grounding DINO와 비교했을 때 실시간 성능이 떨어질 수 있으며, 텍스트 기반의 객체 탐지 기능이 상대적으로 부족하다.

Grounding Dino 모델의 종류와 특징

그렇게 우리가 선정한 Grounding Dino 모델에도 몇 가지 종류가 있다.

Grounding DINO-T (Swin-T 백본):
- Swin Transformer-Tiny를 기반으로 한 모델이다.
- 모델 크기가 작고 계산 비용이 적은 버전이다.
- Zero-shot 및 Fine-tune 성능이 뛰어나다.
Grounding DINO-B (Swin-B 백본):
- Swin Transformer-Base를 기반으로 한 모델이다.
- 더 큰 모델 크기와 더 많은 계산 비용을 가지며, 일반적으로 더 높은 성능을 제공한다.
Grounding DINO-R50 (ResNet-50 백본):
- ResNet-50을 기반으로 한 모델이다.
- 전통적인 CNN 기반 모델의 강력한 성능을 제공한다.
- 주로 Scratch에서 학습하여 높은 유연성을 제공한다.

우리는 연산 비용 대비 성능이 뛰어나다고 평가되는 Grounding DINO-T를 사용하기로 결정하였다.

Grounding DINO-T 모델을 활용한 실시간 흉기난동 서비스 개발 튜토리얼

1. 환경 설정

먼저 필요한 라이브러리와 패키지를 설치한다.

pip install torch torchvision fastapi uvicorn opencv-python transformers

2. Grounding DINO-T 모델 준비

모델을 로드하고 필요한 설정을 한다.

import torch
from transformers import AutoTokenizer, AutoModelForObjectDetection

# 모델과 토크나이저 로드
tokenizer = AutoTokenizer.from_pretrained("GroundingDINO-T")
model = AutoModelForObjectDetection.from_pretrained("GroundingDINO-T")

# 디바이스 설정
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)

3. 객체 탐지 함수 구현

총기, 칼, 도끼, 낫, 유리병, 돌, 몽둥이, 공구를 든 사람과 그 주변의 쓰러진 사람을 감지하는 객체 탐지 함수를 구현한다.

경찰청의 범행도구 통계자료에 기반하여 우리는 총기, 칼, 도끼, 낫, 유리병, 돌, 몽둥이, 공구를 명시적인 흉기로 정의하였다.

import cv2
import numpy as np

def detect_objects(frame, model, tokenizer, device):
    # 이미지 전처리
    inputs = tokenizer(images=frame, return_tensors="pt").to(device)
    outputs = model(**inputs)
    
    # 결과 처리
    boxes = outputs.logits.argmax(-1)
    scores = outputs.logits.max(-1).values
    labels = outputs.labels

    relevant_labels = ["gun", "knife", "axe", "sickle", "glass_bottle", "stone", "club", "tool", "fallen_person"]
    results = []
    for box, score, label in zip(boxes, scores, labels):
        if score > 0.5 and label in relevant_labels:
            results.append({
                "box": box.cpu().numpy(),
                "score": score.cpu().numpy(),
                "label": label.cpu().numpy()
            })
    return results

4. FastAPI 서버 구현

결국 모델의 결과 값을 api 인터페이스를 통해 통신해야 하므로, FastAPI를 사용하여 객체 탐지 결과를 보내는 api 서버를 구축하였다.

from fastapi import FastAPI, UploadFile, File
from fastapi.responses import StreamingResponse
import io

app = FastAPI()

@app.post("/detect/")
async def detect(file: UploadFile = File(...)):
    contents = await file.read()
    np_img = np.frombuffer(contents, np.uint8)
    img = cv2.imdecode(np_img, cv2.IMREAD_COLOR)

    results = detect_objects(img, model, tokenizer, device)
    
    for result in results:
        box = result["box"]
        label = result["label"]
        cv2.rectangle(img, (box[0], box[1]), (box[2], box[3]), (0, 255, 0), 2)
        cv2.putText(img, label, (box[0], box[1]-10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (36,255,12), 2)
    
    _, img_encoded = cv2.imencode('.jpg', img)
    return StreamingResponse(io.BytesIO(img_encoded.tobytes()), media_type="image/jpeg")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

5. 실시간 영상 송출

실시간으로 들어오는 영상에 대한 처리가 필요하기 때문에 socket을 활용하여 실시간 영상 처리를 구현하였다.

import socket
import threading

def video_streaming():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('0.0.0.0', 8080))
    server_socket.listen(1)
    print('Listening for connection...')

    connection, address = server_socket.accept()
    print('Connected to:', address)

    try:
        while True:
            data = connection.recv(1024)
            if not data:
                break

            frame = np.frombuffer(data, dtype=np.uint8).reshape(480, 640, 3)
            results = detect_objects(frame, model, tokenizer, device)
            
            for result in results:
                box = result["box"]
                label = result["label"]
                cv2.rectangle(frame, (box[0], box[1]), (box[2], box[3]), (0, 255, 0), 2)
                cv2.putText(frame, label, (box[0], box[1]-10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (36,255,12), 2)
            
            connection.sendall(frame.tobytes())
    finally:
        connection.close()
        server_socket.close()

thread = threading.Thread(target=video_streaming)
thread.start()