// tech blog · ecology × technology

아직 궁금한게
너무 많다

생태학을 중심으로, 수학 · 물리학 · 머신러닝 · 공간정보가 교차하는 지점을 탐구합니다. 자연을 데이터와 코드로 이해하는 기술 블로그.

🌿 Ecology 🗺️ Geospatial 🧠 Data Science ∑ Fundamentals
~/about.json
{
  "blog": "아직 궁금한게 너무 많다",
  "bio": "자연을 알고 싶은 사람",
  "core": "Ecology",
  "fields": [
    "GIS", "RS", "ML/DL",
    "Math", "Physics"
  ],
  "status": "exploring"
}

SpeciesNet #2: 야생동물 객체 탐지 및 분류 모델 실습

안녕하세요:) 이번 포스팅에서는 SpeciesNet(AI 기반 야생동물 객체 탐지 및 종 분류 모델)에 대해 알아보고, 실제로 동물을 잘 찾아내고 분류해 주는지 실습을 통해 알아보겠습니다.

 

SpeciesNet이란?

SpeciesNet은 구글(Google)과 전 세계 주요 생태 보전 기관들이 협력하여 개발한 야생동물 AI 이미지 분류 모델입니다. 국립공원 등 보호지역에 설치된 무인 센서 카메라(Camera Trap)에 사진이 찍히면, AI가 이를 분석해 사진 속에 동물이 있는지, 있다면 전 세계 2,000여 종의 야생동물 중 어떤 종에 해당하는지를 자동으로 판별해 줍니다.

 

최근 센서 카메라 기술의 발전으로 연구자들은 숲속 생태계의 수천만 장의 사진을 쉽게 얻게 되었습니다. 하지만 역설적으로 이를 분류할 인력이 턱없이 부족해 데이터 마비(Data Bottleneck) 상태에 빠졌습니다. 매일 모니터 앞에 앉아 수만 장의 사진을 넘기며 어떤 동물인지 눈으로 확인하고 분류하는 것은 엄청난 고통과 시간을 요구하는 반복 노동이었습니다(저도 그런 노동자 중에 한명이었다는).

 

이러한 고충을 해결하기 위해 전 세계 주요 생태 보전 기관들이 힘을 합쳤고, 여기에 Google이 기술 파트너로 참여하여 연구자들이 수년간 직접 라벨링 한 방대한 양의 초고품질 데이터를 AI에게 학습시켰고, 그 결과물이 SpeciesNet입니다.

 

현재 이 거대한 프로젝트는 Wildlife Insights 플랫폼을 통해 전 세계의 데이터를 하나로 모으고 있으며, 이 모델을 구동하는 핵심 아키텍처와 기술들은 전 세계 보전 기술자(Conservation Technologist) 커뮤니티인 Wildlabs를 통해 투명하게 공유되고 발전하고 있습니다.

Home | Wildlife Insights

 

Home | Wildlife Insights

Bringing Cutting-Edge Technology to Wildlife Conservation Wildlife Insights streamlines decision-making by providing machine learning models and other tools to manage, analyze and share camera trap data. With access to reliable data, everyone can make bett

www.wildlifeinsights.org

WILDLABS.NET

 

WILDLABS.NET

 

wildlabs.net


딥러닝 아키텍처(Architecture)와 모델(Model)의 구분

기술적인 내용을 다루기에 앞서, 우리가 사용할 도구들의 뼈대인 아키텍처와 실체인 모델을 구분해 보겠습니다.

  • 아키텍처: 데이터가 거쳐 갈 층(Layer), 노드 수, 활성화 함수 등 연산 흐름만을 설계한 빈 구조입니다. 아직 학습되지 않았기 때문에, 내부의 가중치 행렬 \(W\)와 편향 벡터 \(b\)가 최적화되지 않은 미지수 상태로 남아있습니다(예: YOLOv5x6, EfficientNetV2-M).
  • 모델: 빈 아키텍처에 데이터를 주입하고 오차 역전파(Backpropagation) 과정을 거쳐 학습을 마친 최종 결과물입니다. 미지수였던 \(W\)와 \(b\)가 오차를 줄이는 방향으로 깍여나가 구체적인 상수(Constants)로 채워진 것을 볼 수 있습니다(예: MegaDetector v5, SpeciesNet Classifier).


2-Stage Pipeline

SpeciesNet 시스템은 하나의 AI 모델이 분석의 모든 과정을 혼자 처리하지 않습니다. 대신 객체 탐지(Detection)와 종 분류(Classification)가 순차적으로 이루어지는 2단계 파이프라인 구조를 채택하고 있습니다. 왜 굳이 사진을 한 번에 분석하지 않고 번거롭게 두 번에 걸쳐 처리할까요?

 

해답은 바로 야생 카메라 트랩 사진이 가진 태생적인 노이즈 때문입니다. 구글 연구진의 논문은 카메라 트랩 이미지 전체를 한 번에 분류하는 것보다, 동물을 먼저 잘라낸(Crop) 후 분류하는 것이 더 정확할 것이라는 가설을 세우고 이를 검증했습니다. 숲속 생태계는 배경이 매우 복잡하고, 동물이 사진 귀퉁이에 아주 작게 찍히거나 멀리서 찍히는 경우가 빈번합니다. 만약 사진 전체를 한 번에 분류기(Whole-image classifier)에 넣게 되면, AI가 동물의 고유한 특징을 학습하는 대신, 특정 숲의 배경이나 카메라에 찍힌 타임스탬프 같은 불필요한 정보를 정답의 기준으로 잘못 학습해버리는 거짓 상관성(spurious correlations)의 함정에 빠지는 치명적인 부작용이 생깁니다.

 

하지만 1단계에서 탐지기를 이용해 동물만 정확히 찾아내어 네모나게 잘라낸 후 2단계 분류기에게 넘겨주면 AI가 배경 노이즈의 방해를 받지 않고, 동물의 고해상도 특징에만 온전히 집중할 수 있게 됩니다. 실제 연구 결과, 2단계 방식이 대규모 데이터셋에서 분류 정확도(Macro-average F1)를 무려 약 25%나 상승시켰습니다. [1]

 

그럼 이 파이프라인이 어떻게 동작하는지 1단계부터 살펴보겠습니다.

 

1단계: 탐지기(MegaDetector v5)

이 탐지기는 컴퓨터 비전의 객체 탐지(Object Detection) 분야에서 속도와 정확도의 혁신을 이룬 YOLOv5(You Only Look Once) 아키텍처, 그중에서도 가장 깊은 신경망 구조인 YOLOv5x6를 기반으로 합니다. 덕분에 Faster R-CNN 방식을 사용하던 이전 버전(v4) 대비 처리 속도가 비약적으로 향상되었습니다. 이 모델은 이것이 고라니인가?와 같은 종 분류를 수행하지 않습니다. 대신 이미지 내에서 배경이 아닌 객체(동물, 사람, 차량)의 위치를 정확히 찾아내어 바운딩 박스(Bounding Box)를 설정하고 해당 이미지를 크롭(Crop)하는 전처리 역할에 집중합니다.

 

기반이 된 YOLOv5는 본래 80가지 사물(COCO dataset)을 구분할 수 있지만, MegaDetector 개발팀은 전 세계 어떤 야생 환경에서도 동물을 찾아내는 탐지 능력을 극대화하기 위해, 출력 클래스를 단 3가지(Animal, Person, Vehicle)로 재설계하고 집중적으로 학습시켰습니다.

 

MegaDetector 공식 깃허브 https://github.com/agentmorris/MegaDetector

 

GitHub - agentmorris/MegaDetector: MegaDetector is an AI model that helps conservation folks spend less time doing boring things

MegaDetector is an AI model that helps conservation folks spend less time doing boring things with camera trap images. - agentmorris/MegaDetector

github.com

 

2단계: 분류기(SpeciesNet Classifier)

1단계 탐지기가 객체의 위치를 찾아 이미지를 잘라주면, 분류기(SpeciesNet)는 그것이 어떤 생물인지 구체적인 종(Species)을 판별합니다. 이 모델의 뼈대는 이미지 분류에 탁월한 구글의 EfficientNetV2-M 아키텍처입니다.

 

구글 공식 GitHub 자료와 Wildlife insights 플랫폼에 따르면, SpeciesNet은 단순한 인터넷 크롤링 이미지를 사용한 것이 아니라 전 세계 주요 생태 기관의 연구자들이 수년간 직접 검증한 라벨링된 6,500만 장 이상의 초고품질 카메라 트랩 데이터가 주입되었습니다. 그 결과, 현재 SpeciesNet은 전 세계 2,000종 이상의 야생동물을 약 94.5%의 높은 정확도로 분류해 내는 강력한 분류기로 거듭났습니다. 이 정도면 휴먼보다 난 것 같단 ^^;;

 

Google CameraTrapAI (SpeciesNet) 공식 깃허브 https://github.com/google/cameratrapai

 

GitHub - google/cameratrapai: AI models trained by Google to classify species in images from motion-triggered wildlife cameras.

AI models trained by Google to classify species in images from motion-triggered wildlife cameras. - google/cameratrapai

github.com


이제 지난 포스팅에 이어 실습을 진행해보겠습니다.

 

vscode를 실행하여 [File] > [Open Folder] 를 클릭하고, speciesnet 폴더를 선택합니다. 다음은 프로젝트 전용 가상환경(speciesnet_venv)을 활성화 해주고, 터미널에서 확인해줍니다.

 

가상환경 준비를 마쳤으니 코드 작성에 앞서 프로젝트의 디텍러리 구조부터 잡고 가겠습니다. 단순한 연습용 코드가 아니라, 현업에서 수만 장의 야생동물 사진을 안정적으로 처리하려면 확장성과 유지보수성이 핵심입니다. 향후 MegaDetector 모델을 최신 버전으로 업그레이드하거나, 분류기를 완전히 다른 알고리즘으로 교체하거나, 분석 결과를 사내 데이터베이스(DB) 양식에 맞춰 연동해야 한다면 어떨까요? 모든 처리 로직이 단일 스크립트에 섞여 있다면, 기능 하나를 추가할 때마다 코드 전체를 뜯어고쳐야 하는 대참사가 발생합니다.(한 스크립트에서 모델을 만들고 시각화하면서 몇천줄을 뜯어 고치는 건 꽤나 괴롭습니다 ^^;;;)

 

이러한 문제를 방지하기 위해, 데이터, 모델, 로직을 분리하는 모듈화(Modularization) 설계를 적용합니다. 프로젝트 폴더(speciesnet) 안에 아래처럼 빈 폴더와 파일만 먼저 만들어보겠습니다.

speciesnet/
 ├── configs/                # 설정파일 폴더
 ├── data/                   # 데이터
 ├── models/                 # 오프라인/로컬 환경용 학습된 가중치 파일
 ├── src/                    # 소스코드
 ├── main.py                 # 메인 실행 컨트롤러
 ├── notebooks/              # 실험실
 ├── .gitignore              # 깃(Git) 업로드 시 제외할 파일 목록
 ├── README.md               # 프로젝트 설명서
 └── requirements.txt        # 프로젝트 의존성 패키지 목록

 

탐색기 패널에서 우클릭 > [New Folder...] > 폴더 이름 입력 혹은 우클릭 > [New File...] > 파일 이름을 입력합니다.

 

디렉터리 구조가 잡혔으니, 파이프라인을 구동할 딥러닝 엔진을 설치할 차례입니다. 우리가 사용할 MegaDetector와 SpeciesNet은 현재 모두 Pytorch를 기반으로 구동됩니다. Pytorch와 같은 거대한 딥러닝 프레임워크는 사용자의 컴퓨터 환경(그래픽카드 유무 및 CUDA 버전)에 맞춰 개별적으로 세팅해 주어야 합니다.

 

내 컴퓨터의 그래픽카드(CUDA) 버전은 아래 명령어를 입력하여 확인합니다.

nvidia-smi

 

표가 출력되면 우측 상단에 CUDA Version: 13.0과 같은 숫자가 적혀 있습니다. 이 숫자는 무조건 13.0 전용을 설치해야 한다는 의미는 아니고, 내 그래픽카드가 소화할 수 있는 최대 상한선을 의미합니다. 즉, 13.0 이하의 버전을 선택하면 호환됩니다. 이제 Pytorch 공식 홈페이지 https://pytorch.org/get-started/locally/에 접속하여 Pytorch 안정화 버전 설치 명령어를 확인해보겠습니다. 저는 Python 3.9 버전과 호환성이 좋은 CUDA 12.6 버전용 Pytorch를 설치하겠습니다.

 

Get Started

Set up PyTorch easily with local installation or supported cloud platforms.

pytorch.org

만약 내 pc의 CUDA 상한선이 낮다면, Previous Versions 메뉴에서 확인하시면 됩니다. https://pytorch.org/get-started/previous-versions/

 

각자 버전을 확인하시고 터미널에 명령어를 입력합니다.

 

GPU 프로세스(CUDA)

pip3 install torch torchvision --index-url https://download.pytorch.org/whl/cu126

 

설치가 완료됐다면, 딥러닝 엔진이 그래픽카드를 인식하고 있는지 확인해봅니다. True가 나오면 인식이 된 상태입니다 :)

python -c "import torch; print('GPU Available:', torch.cuda.is_available())"

 

혹은 pc에 NVIDIA 그래픽카드가 없거나, 가볍게 테스트만 해보고 싶다면 CPU 전용 명령어를 입력하세요.

 

CPU 프로세스

pip3 install torch torchvision

딥러닝 엔진이 준비되었으니, 이제 모델을 구동할 유틸리티와 핵심 패키지를 갖출 차례입니다. requrements.txt 파일에 아래 내용을 입력하고 저장해주세요. 추후, 프로젝트가 진행되면서 필요한 패키지들을 이 파일에 넣어두면, 복잡한 의존성 문제를 한 번에 해결할 수 있습니다.

# ------------------------------------------
# SpeciesNet 2-Stage Pipeline Requirements
# ------------------------------------------

# 1. Google Official Camera Trap AI
speciesnet>=5.0.0  # MegaDetector & SpeciesNet 통합 패키지

# 2. Utilities & Configuration
PyYAML>=6.0    # config.yaml 설정 파일을 읽기 위한 도구
pandas>=2.0.0  # 최종 분석된 종(Species) 결과 데이터를 CSV로 예쁘게 저장
tqdm>=4.65.0   # 수만 장의 사진이 분석되는 진행률을 텍스트 bar로 보여줌

 

파일을 저장했다면, 터미널에 아래 명령어를 입력하여 한 번에 설치를 마무리합니다.

pip install -r requirements.txt

 

다음은 코드 내부에 경로나 설정값을 직접 적어두는 하드코딩을 방지하기 위해, config.yaml 파일을 세팅할 차례입니다. configs 폴더 안에 config.yaml 파일을 새로 만들고, 아래의 설정값을 복사해서 붙여넣습니다.

# ---------------------------------------------------------
# SpeciesNet Pipeline Configuration
# ---------------------------------------------------------

# Directory Paths (I/O)
paths:
  input_image_dir: "./data/raw_images"  # 추론 대상 원본 이미지 보관소
  input_video_dir: "./data/raw_videos"  # 프레임 추출 대상 원본 동영상 보관소
  output_csv_dir: "./data/results_csv"  # 파싱된 최종 데이터셋(CSV) 저장 경로
  output_viz_dir: "./data/results_viz"  # 바운딩 박스가 렌더링된 시각화 결과물 저장 경로

# Model Settings
model:
  name: "kaggle:google/speciesnet/pyTorch/v4.0.2a/1"  # 메모리에 적재할 모델 가중치 버전
  geofence: true                                      # 타대륙 특산종 오탐지(False Positive) 방어벽 활성화

# Location Information
location:
  country: "KOR"  # 지오펜스 필터링의 기준이 되는 국가 코드 (반드시 ISO 3166-1 alpha-3 포맷 사용)

# Video Preprocessing (MOG2 Background Subtraction)
video_params:
  min_contour_area: 1000  # 유효 움직임으로 간주할 최소 픽셀 면적. 야간(IR) 노이즈나 곤충을 무시하고 타겟 동물만 잡기 위한 수치
  frames_per_sec: 1       # 모션 감지 시 초당 최대 추출 프레임 수. 동일한 동물이 너무 많이 캡처되는 것을 제한
  var_threshold: 16       # 배경 대비 픽셀 변화 감지 민감도. 낮을수록 작은 움직임에도 민감하게 반응함
  history: 500            # 배경 모델을 학습하고 업데이트하는 데 참고할 과거 프레임의 개수

# Visualization 
visualizer_params:
  conf_threshold: 0.3     # 바운딩 박스 렌더링 최소 임계값. 모델의 확신도가 30% 미만인 노이즈성 예측(돌멩이, 나뭇가지 등)은 그리지 않음

 

다음은 config.yaml 파일에 지정해 둔 경로에 맞춰, 테스트해 볼 야생동물 사진과 동영상을 준비할 차례입니다. data 폴더 안에 하위 폴더를(raw_images, raw_videos) 생성하여 테스트할 사진들과 영상들을 넣어주겠습니다.(참고로 결과물이 저장될 results_csv와 results_viz 폴더는 이후에 작성할 파이썬 코드가 알아서 생성해 주므로, 지금은 원본 데이터를 넣을 폴더 두 개만 만들면 충분합니다.)

 

테스트 파일 링크: https://drive.google.com/file/d/1Gc2Otcut2q4isC1NMarILbBKYIyyh7Pz/view?usp=sharing

 

raw.zip

 

drive.google.com


데이터 준비까지 마쳤다면, src 폴더 안에 모듈들을 만들어 보겠습니다. 먼저 __init__.py 라는 빈 파일을 하나 만들어 줍니다. 이 빈 파일은 파이썬 인터프리터에게 "이 src 폴더는 단순한 디렉터리가 아니라, import 해서 가져다 쓸 수 있는 패키지(package)야" 라고 알려주는 '마커' 역할을 합니다.

 

이제 이 src 폴더 안에 첫 번째 부품인 동영상 전처리 모듈을 만들어 보겠습니다.

 

동영상 전처리 모듈 (video_extractor.py)

카메라 트랩에 찍힌 야생동물 영상은 보통 10초에서 30초 분량입니다. 이 영상을 1프레임 단위로 모두 쪼개어 AI 모델에 밀어 넣는 것은 막대한 서버 디스크 I/O와 GPU 추론 비용을 발생시키는 안티 패턴입니다. 따라서 AI에게 영상을 넘기기 전에, 의미 있는 움직임(Motion)이 있는 프레임만 가볍게 선별해 내는 전처리 과정이 필수적입니다. 이를 위해 OpenCV의 배경 차분 기법인 'MOG2(Mixture of Gaussians 2)' 알고리즘을 활용합니다.

 

특히 야간 적외선(IR) 카메라 트랩 영상에서는 동물의 움직임이 뚜렷하지 않고 흐릿한 회색 그림자처럼 잡히는 경우가 많습니다. MOG2 알고리즘은 가짜 움직임(바람에 흔들리는 풀)을 지워내는 동시에, 아주 미세한 회색빛 움직임도 뚜렷한 흰색 윤곽선으로 강제 이진화하여 잡아내는 생태학 도메인 특화 로직이 숨어있습니다. 이 과정에서 MOG2 알고리즘은 매 프레임마다 각 픽셀의 배경 분포를 학습하여 조명 변화가 심한 야외에서도 안정적으로 동물을 분리해 내는 검증된 방식입니다. [2]

 

src 폴더 안에 video_extractor.py 파일을 생성하고 아래의 코드를 붙여 넣습니다.

"""
Video Frame Extraction Module
- Processes raw video files to extract static frames based on motion detection.
- Utilizes OpenCV MOG2 (Mixture of Gaussians) for dynamic background subtraction.
- Implements frame-rate throttling to optimize I/O and reduce downstream GPU inference costs.
"""

import cv2
from pathlib import Path
from typing import List

def extract_frames(
    video_path: Path, 
    output_dir: Path, 
    min_contour_area: int = 1000, 
    frames_per_sec: int = 1,
    var_threshold: int = 16,
    history: int = 500
) -> List[Path]:
    """
    동영상 스트림을 분석하여 유효한 객체(동물)의 움직임이 포함된 프레임만 선별적으로 추출합니다.
    
    Args:
        video_path: 처리할 원본 동영상 파일의 절대 경로
        output_dir: 추출된 프레임이 저장될 임시 워크스페이스 경로
        min_contour_area: 유효 모션으로 판별할 최소 픽셀 면적 (바람에 흔들리는 풀 등 환경 노이즈 필터링)
        frames_per_sec: 초당 최대 추출 프레임 수 (중복 데이터 생성 방지 및 추론 처리량 조절)
        var_threshold: MOG2 배경 모델의 분산 임계값 (낮을수록 미세한 움직임 감지)
        history: 배경 모델 업데이트에 사용될 과거 프레임 수
        
    Returns:
        디스크에 저장이 완료된 프레임 이미지들의 절대 경로 리스트
    """
    cap = cv2.VideoCapture(str(video_path))
    if not cap.isOpened():
        print(f"[-] Failed to open video: {video_path.name}")
        return []

    # 원본 영상의 디폴트 FPS를 기준으로 타겟 추출 간격(Interval)을 계산하여 불필요한 I/O 방지
    original_fps = round(cap.get(cv2.CAP_PROP_FPS)) or 30
    frame_interval = max(1, original_fps // frames_per_sec)

    video_name = video_path.stem
    save_dir = output_dir / video_name
    save_dir.mkdir(parents=True, exist_ok=True)

    extracted_paths = []
    frame_count = 0
    saved_count = 0
    last_saved_frame = -frame_interval

    # MOG2 배경 차분기 인스턴스 초기화 (Config 기반 동적 파라미터 주입)
    back_sub = cv2.createBackgroundSubtractorMOG2(
        history=history, 
        varThreshold=var_threshold, 
        detectShadows=True
    )

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

        fg_mask = back_sub.apply(frame)
        
        # 전처리: MOG2가 인식한 부드러운 그림자(회색) 영역을 제거하고 뚜렷한 전경(흰색)만 이진화
        # 야간 카메라 트랩의 IR(적외선) 저조도 영상에서 객체의 윤곽을 명확히 확보하기 위한 조치
        _, fg_mask = cv2.threshold(fg_mask, 127, 255, cv2.THRESH_BINARY)
        contours, _ = cv2.findContours(fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        
        # 형태학적 필터링: 설정된 최소 면적(min_contour_area) 이상의 객체 존재 여부 검증
        motion_detected = any(cv2.contourArea(c) > min_contour_area for c in contours)

        # 유효 객체가 감지되었고, 이전 추출 시점으로부터의 제한 간격(Throttle)을 충족했을 때만 저장
        if motion_detected and (frame_count - last_saved_frame) >= frame_interval:
            frame_filename = save_dir / f"{video_name}_frame_{saved_count:04d}.jpg"
            cv2.imwrite(str(frame_filename), frame)
            extracted_paths.append(frame_filename)
            
            saved_count += 1
            last_saved_frame = frame_count

        frame_count += 1

    cap.release()
    return extracted_paths

영상 전처리를 마쳤으니, AI 모델이 뱉어낸 복잡한 결괏값을 우리가 다루기 편한 csv 형태로 가공할 차례입니다.

 

데이터 파싱 모듈 (parser.py)

AI 모델(SpeciesNet)의 추론 결괏값은 중첩된 딕셔너리(JSON 형태) 구조를 띄고 있습니다. 따라서 이 데이터 구조를 Pandas를 이용해 2차원 표(Data Frame) 형태로 평탄화(Flatten)하여 csv 파일로 저장합니다.

 

이 모듈은 확신도가 낮은 오탐지(Ghost object)를 걸러내고, 여러 장의 프레임 중 확신도가 가장 높은 단 1장의 베스트 컷만 남기는 정제 작업을 수행하여 마릿수(Count)의 데이터 정합성을 지켜냅니다.

 

이전처럼 parser.py 파일을 생성하고 아래의 코드를 붙여 넣습니다.

"""
Prediction Parsing Module
- Parses raw JSON-like prediction dictionaries from SpeciesNet.
- Applies confidence thresholding to maintain data integrity (Ghost object removal).
- Handles video frame deduplication, preserving only the highest confidence frame per media.
"""

import pandas as pd
from pathlib import Path
from datetime import datetime
from typing import Dict

def parse_and_save_results(preds: Dict, output_dir: Path, conf_threshold: float = 0.3, prefix: str = "results") -> Path:
    """
    원시 예측 데이터를 평탄화(Flatten)하여 CSV 포맷으로 직렬화합니다.
    
    Args:
        preds: 모델 추론 결과 딕셔너리
        output_dir: CSV 파일이 저장될 디렉터리 경로
        conf_threshold: 유효 객체로 인정할 최소 확신도 (시각화 모듈과 동일한 기준 적용)
        prefix: 출력 파일명에 추가할 접두사 (예: images, videos)
    """
    predictions_list = next(iter(preds.values())) if preds else []
    parsed_data = []
    
    for p in predictions_list:
        # 모델 예측 문자열 구조: "계;문;강;목;과;속;종(일반명)"
        pred_str = p.get('prediction', '')
        parts = pred_str.split(';')
        
        common_name = parts[-1] if len(parts) >= 7 else pred_str
        scientific_name = f"{parts[4]} {parts[5]}" if len(parts) >= 6 else "unknown"
        
        # 노이즈 필터링: 임계값 미만의 예측(돌멩이, 나뭇가지 등)을 제외하여 실제 동물 마릿수 산정
        raw_detections = p.get('detections', [])
        valid_detections = [
            d for d in raw_detections 
            if d.get('conf', d.get('prediction_score', 0)) >= conf_threshold
        ]
        animal_count = len(valid_detections)
        
        file_path = p.get('filepath', '')
        file_name = Path(file_path).name
        
        # 동영상 프레임 그룹화 키 생성 (예: 'videoA_frame_001.jpg' -> 'videoA')
        is_video_frame = '_frame_' in file_name
        group_key = file_name.split('_frame_')[0] if is_video_frame else file_name

        parsed_data.append({
            'source_media': group_key,
            'file_name': file_name,
            'common_name': common_name,
            'scientific_name': scientific_name,
            'confidence': round(p.get('prediction_score', 0), 4),
            'animal_count': animal_count,
            'file_path': file_path,
            'is_video_frame': is_video_frame,
            # 동물이 검출되지 않았거나, 모델 예측이 blank인 경우 빈 화면으로 간주
            'is_blank': (animal_count == 0) or (common_name.lower() == 'blank')
        })
        
    df = pd.DataFrame(parsed_data)
    
    if not df.empty:
        # 동영상 프레임 중복 제거 (Deduplication)
        # 동일한 동영상 내에서 동물이 찍힌 프레임을 우선하고, 그 중 확신도가 가장 높은 1장만 데이터셋에 남김
        df = df.sort_values(by=['is_blank', 'confidence'], ascending=[True, False])
        df = df.drop_duplicates(subset=['source_media'], keep='first')
        
        df = df.sort_values(['source_media'])
        df.drop(columns=['is_video_frame', 'is_blank'], inplace=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    csv_path = output_dir / f"speciesnet_{prefix}_{timestamp}.csv"
    df.to_csv(csv_path, index=False, encoding='utf-8')
    
    return csv_path

파싱 모듈을 통해 csv 형태의 정량적 데이터를 얻었다면, 이번에는 모델이 실제로 대상을 잘 찾았는지 두 눈으로 확인하는 정성적 평가 단계입니다.

 

시각화 모듈 (visualizer.py)

사진 위에 그림을 그리는 것을 넘어, 데이터 분석가가 모델의 성능을 직관적으로 검증하고 파라미터를 튜닝할 수 있게 돕는 핵심 모듈입니다. 다음 두 가지 핵심 기능이 포함됩니다.

  • 오탐지 검증(False Positive Check): 모델이 숫자로 뱉어낸 좌표와 확신도(Confidence Score)를 원본 미디어 위에 시각화합니다. 이를 통해 모델이 진짜 야생동물을 찾은 것인지, 아니면 주변의 바위나 그림자를 동물로 착각(Hallucination)한 것인지 연구자가 직접 눈으로 교차 검증할 수 있습니다.
  • 최적의 임계값(Threshold) 탐색: 카메라 트랩 데이터는 촬영 환경(야간, 비, 눈)에 따라 노이즈가 심합니다. 시각화된 결과물들을 비교해 보면서, 앞서 세팅한 config.yaml의 conf_threshold 값을 올릴지 직관적으로 결정할 수 있는 판단 기준이 됩니다.

이제 visualizer.py 파일을 생성하고 아래의 코드를 작성합니다.

"""
Bounding Box Visualization Module
- Renders bounding boxes and confidence scores on original media.
- Integrates confidence thresholding to filter out low-probability detections.
- Implements dynamic UI scaling with clamping to ensure legibility across varying resolutions (e.g., VGA to 4K).
"""

import cv2
from pathlib import Path
from typing import Dict

# 렌더링 디자인 시스템 상수 (OpenCV 포맷에 맞춰 BGR 순서 사용)
BOX_COLOR = (0, 255, 0)
TEXT_BG_COLOR = (0, 255, 0)
TEXT_COLOR = (0, 0, 0)
FONT = cv2.FONT_HERSHEY_SIMPLEX

# 동적 스케일링 상한/하한 제어 상수 (초고해상도 이미지에서의 UI 비대화 방지)
BASE_RESOLUTION = 1000.0
MIN_FONT_SCALE = 0.5
MAX_FONT_SCALE = 1.5
MIN_THICKNESS = 2
MAX_THICKNESS = 4
MIN_PADDING = 8
MAX_PADDING = 20

def draw_predictions(image_path: str, pred_data: Dict, output_dir: Path, conf_threshold: float = 0.3) -> None:
    """
    추론 결과를 바탕으로 원본 이미지 위에 Bounding Box와 텍스트 라벨을 렌더링합니다.
    
    Args:
        image_path: 렌더링 대상 원본 이미지의 절대 경로
        pred_data: 단일 이미지에 대한 모델 예측 결과 딕셔너리
        output_dir: 렌더링이 완료된 이미지를 저장할 디렉터리 경로
        conf_threshold: 유효 객체로 판별할 최소 확신도 (이 값 미만의 노이즈는 렌더링 제외)
    """
    img = cv2.imread(image_path)
    if img is None:
        print(f"[-] Failed to load image: {image_path}")
        return

    height, width, _ = img.shape
    
    # 이미지 해상도에 비례하는 동적 스케일링(Dynamic Scaling) 적용
    # min/max를 활용한 클램핑(Clamping) 기법으로 UI 요소가 원본 객체를 과도하게 가리는 현상 방어
    scale_factor = width / BASE_RESOLUTION
    
    font_scale = max(MIN_FONT_SCALE, min(scale_factor, MAX_FONT_SCALE))
    thickness = max(MIN_THICKNESS, min(int(2 * scale_factor), MAX_THICKNESS))
    padding = max(MIN_PADDING, min(int(10 * scale_factor), MAX_PADDING))

    detections = pred_data.get('detections', [])
    image_level_pred = pred_data.get('prediction', 'unknown')
    
    for det in detections:
        score = det.get('conf', det.get('prediction_score', 0))
        
        # 설정된 임계값(Threshold) 미만의 노이즈성 예측(오탐지) 렌더링 제외
        if score < conf_threshold:
            continue
            
        bbox = det.get('bbox')
        if not bbox or len(bbox) != 4:
            continue
            
        # SpeciesNet 표준 포맷인 정규화된 상대 좌표 [xmin, ymin, width, height]를 절대 픽셀 좌표로 변환
        xmin, ymin, box_w, box_h = bbox
        
        xmax = xmin + box_w
        ymax = ymin + box_h
        
        start_point = (int(xmin * width), int(ymin * height))
        end_point = (int(xmax * width), int(ymax * height))
        
        cv2.rectangle(img, start_point, end_point, BOX_COLOR, thickness)
        
        # 개별 객체 단위의 예측값이 누락된 경우, Fallback으로 이미지 전체 수준의 예측 라벨 사용
        pred_str = det.get('prediction', image_level_pred)
        label_name = pred_str.split(';')[-1] if ';' in pred_str else pred_str
        label_text = f"{label_name}: {score:.2f}"
        
        (text_width, text_height), _ = cv2.getTextSize(label_text, FONT, font_scale, thickness)
        
        text_bg_start = (start_point[0], start_point[1] - text_height - padding)
        text_bg_end = (start_point[0] + text_width, start_point[1])
        
        # 방어 로직: 객체가 이미지 최상단 테두리에 위치하여 라벨 텍스트가 화면 밖으로 잘리는 현상(Clipping) 방지
        if text_bg_start[1] < 0:
            text_bg_start = (start_point[0], start_point[1] + padding)
            text_bg_end = (start_point[0] + text_width, start_point[1] + text_height + (padding * 2))
            text_pos = (start_point[0], start_point[1] + text_height + int(padding * 1.5))
        else:
            text_pos = (start_point[0], start_point[1] - int(padding / 2))

        cv2.rectangle(img, text_bg_start, text_bg_end, TEXT_BG_COLOR, cv2.FILLED)
        cv2.putText(img, label_text, text_pos, FONT, font_scale, TEXT_COLOR, thickness)

    filename = Path(image_path).name
    save_path = output_dir / filename
    cv2.imwrite(str(save_path), img)

드디어 모든 모듈을 하나로 엮어, 실제로 시스템을 가동할 마지막 단계입니다.

 

메인 컨트롤러 (main.py)

이 파일은 복잡한 내부 로직을 일일이 수정할 필요 없이, 설정 파일(config.yaml)만 살짝 바꾸는 것만으로도 수백 개의 야생동물 사진과 영상을 한꺼번에 분석해 주는 역할을 합니다. 복잡한 설명 없이, main.py 파일을 생성하고 아래 코드를 붙여 넣겠습니다.

"""
SpeciesNet Inference Pipeline Controller
- Orchestrates the end-to-end batch processing for both images and video frames.
- Ensures I/O isolation between raw data and processing workspaces.
- Injects dynamic configuration parameters (MOG2, Visualization) into sub-modules.
"""

import argparse
import yaml
from pathlib import Path

from speciesnet import SpeciesNet
from src.visualizer import draw_predictions
from src.parser import parse_and_save_results
from src.video_extractor import extract_frames

PROJECT_ROOT = Path(__file__).resolve().parent

def load_config(config_path: Path) -> dict:
    """YAML 설정 파일을 파이썬 딕셔너리로 안전하게 로드합니다."""
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)

def process_batch(model: SpeciesNet, file_paths: list, csv_out_dir: Path, viz_out_dir: Path, loc_config: dict, conf_threshold: float, batch_name: str = "images") -> None:
    """
    이미지 및 비디오 프레임에 대한 공통 추론, 파싱, 시각화 파이프라인을 실행합니다.
    (DRY: Don't Repeat Yourself 원칙 적용)
    """
    if not file_paths:
        return
        
    print(f"[*] Running inference on {len(file_paths)} {batch_name}...")
    
    preds = model.predict(
        filepaths=file_paths, 
        country=loc_config.get('country'), 
        admin1_region=loc_config.get('admin1_region')
    )
    
    for key in preds.keys():
        preds[key] = list(preds[key])
    
    csv_path = parse_and_save_results(preds, csv_out_dir, conf_threshold=conf_threshold, prefix=batch_name)
    print(f"    - Results saved to: {csv_path}")

    predictions_list = next(iter(preds.values())) if preds else []
    for pred_data in predictions_list:
        draw_predictions(
            image_path=pred_data['filepath'], 
            pred_data=pred_data, 
            output_dir=viz_out_dir,
            conf_threshold=conf_threshold
        )
        
    print(f"    - Visualizations saved to: {viz_out_dir}")

def main(args: argparse.Namespace) -> None:
    config_file = PROJECT_ROOT / args.config
    if not config_file.exists():
        raise FileNotFoundError(f"Config file not found: {config_file}")
        
    config = load_config(config_file)
    
    # =====================================================================
    # I/O Directory Setup & Isolation
    # =====================================================================
    video_dir = PROJECT_ROOT / config['paths']['input_video_dir']
    image_dir = PROJECT_ROOT / config['paths']['input_image_dir']
    
    csv_base = PROJECT_ROOT / config['paths']['output_csv_dir']
    viz_base = PROJECT_ROOT / config['paths']['output_viz_dir']
    
    csv_img_dir, csv_vid_dir = csv_base / "images", csv_base / "videos"
    viz_img_dir, viz_vid_dir = viz_base / "images", viz_base / "videos"
    
    frames_workspace = PROJECT_ROOT / "data" / "workspace" / "video_frames"
    
    for d in [video_dir, image_dir, csv_img_dir, csv_vid_dir, viz_img_dir, viz_vid_dir, frames_workspace]:
        d.mkdir(parents=True, exist_ok=True)

    # =====================================================================
    # Configuration Parameter Loading
    # =====================================================================
    loc_config = config.get('location', {})
    model_config = config.get('model', {})
    video_config = config.get('video_params', {})
    viz_config = config.get('visualizer_params', {})

    model_name = model_config.get('name', 'kaggle:google/speciesnet/pyTorch/v4.0.2a/1')
    use_geofence = model_config.get('geofence', True)
    conf_threshold = viz_config.get('conf_threshold', 0.3)

    print(f"[*] Initializing SpeciesNet Model (Geofence: {use_geofence})...")
    model = SpeciesNet(model_name=model_name, geofence=use_geofence) 

    valid_exts = {'.jpg', '.jpeg', '.png'}

    # =====================================================================
    # Original Images Processing
    # =====================================================================
    image_paths = [
        str(p) for p in image_dir.rglob("*") 
        if p.is_file() and p.suffix.lower() in valid_exts and "_frame_" not in p.name
    ]
    
    if image_paths:
        print("\n" + "-"*40 + "\n[Original Images Processing]\n" + "-"*40)
        process_batch(model, image_paths, csv_img_dir, viz_img_dir, loc_config, conf_threshold, batch_name="images")
    else:
        print("\n[*] No original images found. Skipping.")

    # =====================================================================
    # Video Frames Processing (MOG2 Pipeline)
    # =====================================================================
    video_exts = {'.mp4', '.avi', '.mov'}
    video_paths = [p for p in video_dir.rglob("*") if p.is_file() and p.suffix.lower() in video_exts]
    
    if video_paths:
        print("\n" + "-"*40 + "\n[Video Frames Processing]\n" + "-"*40)
        print(f"[*] Extracting motion frames from {len(video_paths)} video(s)...")
        
        min_area = video_config.get('min_contour_area', 1000)
        fps = video_config.get('frames_per_sec', 1)
        var_thresh = video_config.get('var_threshold', 16)
        history = video_config.get('history', 500)

        for vp in video_paths:
            extract_frames(
                video_path=vp, 
                output_dir=frames_workspace, 
                min_contour_area=min_area, 
                frames_per_sec=fps,
                var_threshold=var_thresh,
                history=history
            )
        
        frame_paths = [str(p) for p in frames_workspace.rglob("*") if p.is_file() and p.suffix.lower() in valid_exts]
        
        if frame_paths:
            process_batch(model, frame_paths, csv_vid_dir, viz_vid_dir, loc_config, conf_threshold, batch_name="videos")
        else:
            print("[*] No motion detected in videos. Skipping inference.")
    else:
        print("\n[*] No videos found. Skipping.")

    print("\n[*] Pipeline completed.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="SpeciesNet Inference Pipeline")
    parser.add_argument("--config", type=str, default="configs/config.yaml")
    main(parser.parse_args())

 

이제 터미널을 열고 python main.py 를 입력하면, 데이터 파이프라인이 유기적으로 작동하여 정제된 데이터와 시각화 리포트를 쏟아낼 것입니다.

 

Python main.py 실행이 끝나면 results_csv 폴더에 생성된 표를 확인하실 수 있습니다. 여기서 아주 흥미롭고도 당황스러운 결과를 발견하게 됩니다. 분명 우리나라에서 찍힌 사슴인데, 모델은 단순히 "사슴"이라 답하지 않고 "사슴과(Cervidae)" 혹은 "유럽 노루(Capreolus capreolus)" 처럼 계층적인 분류 결과를 보여줍니다.

 

이는 SpeciesNet이 생태학적 계층 분류(Taxonomic Classification)를 기반으로 설계되었기 때문입니다.

  • 데이터 한계 극복: 카메라 트랩 사진이 흐릿하거나 각도가 애매할 때, 모델은 무리하게 '종(Species)'을 확정 짓기보다 상위 단계인 '과(Family)' 단위에서 높은 확신도를 가짐으로써 분석의 신뢰도를 높입니다.
  • 분석 유연성: 연구자들은 때로 특정 종이 아니더라도 이 지역에 사슴과 동물이 얼마나 서식하는가? 라는 광범위한 질문에 답해야 합니다. 계층적 결과물은 이러한 통계적 접근을 훨씬 수월하게 해줍니다.

 

하지만 한계도 명확합니다. 우리가 넣은 사슴이나 너구리가 한국의 자생종이 아닌 유럽이나 북미의 유사종으로 예측되는 현상을 보셨을 겁니다. 이는 현재 우리가 사용한 모델이 전 세계의 광범위한 데이터로 학습된 범용 모델이기 때문입니다.

 

이번 시간에는 이미 학습된(Pre-trained) 강력한 SpeciesNet 모델을 가져와, 실무 환경에 맞게 전처리부터 시각화까지 이어지는 전체 파이프라인을 구축해 보았습니다.

 

비록 지금은 우리 환경에 완벽히 들어맞지 않는 것처럼 보일 수 있지만, 더 정교한 분석으로 나아가기 위한 첫걸음입니다. 다음 포스팅에서는 왜 이런 현상이 발생하는지 근본적인 원인을 파헤치기 위해 SpeciesNet의 딥러닝 아키텍처를 해부해 보겠습니다.

 

 

 

References

[1]  T. Gadot et al., "To crop or not to crop: Comparing whole-image and cropped classification on a large dataset of camera trap images," IET Computer Vision, vol. 18, no. 8, pp. 1193–1208, 2024.

[2]  Z. Zivkovic, "Improved adaptive Gaussian mixture model for background subtraction," Proceedings of the 17th International Conference on Pattern Recognition (ICPR), vol. 2, pp. 28–31, 2004.