이번 시간에는 object detection 모델을 통해 주변에 다른 차나 사람이 가까이 있는지 확인한 후 멈출 수 있는 자율주행 시스템을 만들어 본다. 하지만 자율주행 시스템은 아직 완전하지 않기 때문에, 위험한 상황에서는 운전자가 직접 운전할 수 있도록 하거나 판단이 어려운 상황에서는 멈추도록 설계된다. 우리도 같은 구조를 가진 미니 자율주행 시스템을 만들어 볼 것이다.
먼저 전체적인 시스템을 구성하기 위해서 보조장치의 역할과 이를 학습하기 위한 데이터셋 전처리를 수행한다. Detection 모델을 학습시키기 위한 전체 파이프라인을 직접 제작하기에는 많은 시간이 들기 때문에 RetinaNet이라는 1-stage detector를 미리 학습시킨 라이브러리를 활용하도록 한다.
- 바운딩 박스(bounding box) 데이터셋을 전처리할 수 있다.
- Object detection 모델을 학습할 수 있다.
- Detection 모델을 활용한 시스템을 만들 수 있다.
RetinaNet Github에 보면 tensorflow 2.3.0, tf.keras 2.4.0을 사용할 것을 권장하며, 그보다 높은 버전에서는 오류가 발생한다고 안내하고 있다. 만약 가상환경에 설치된 Tensorflow 버전이 2.3.0보다 높을 경우 2.3.0에 맞추어 재설치하거나, 별도의 가상환경을 생성 후 해당 환경에서 jupyter notebook으로 실습을 진행할 것을 권장한다.
$ git clone https://github.com/fizyr/keras-retinanet.git
$ cd keras-retinanet && python setup.py build_ext --inplace
$ pip install tensorflow_datasets tqdm
$ pip install -r requirements.txt
$ pip install .이번 시간에 만들어 볼 자율주행 보조장치는 카메라에 사람이 가까워졌을 때, 그리고 차가 가까워져서 탐지된 크기가 일정 이상일 때 멈춰야 한다.
- 자율주행 보조장치 object detection 요구사항
- 사람이 카메라에 감지되면 정지
- 차량이 일정 크기 이상으로 감지되면 정지
이번 시간에는 tensorflow_datasets에서 제공하는 KITTI 데이터셋을 사용해볼 것이다. KITTI 데이터셋은 자율주행을 위한 데이터셋으로 2D objecte detection 뿐만 아니라 깊이까지 포함한 3D object detection 라벨 등을 제공하고 있다.
# TensorFlow and tf.keras
import tensorflow as tf
from tensorflow import keras
# Helper libraries
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import tensorflow_datasets as tfds
import copy
import cv2
from PIL import Image, ImageDraw(주의) 이 데이터셋은 11GB가 넘어가는 대용량 데이터셋입니다. 다운로드 소요시간이 수시간 이상 지속될 수 있으니, 다운로드는 가급적 별도의 jupyter notebook에서 수행하는 것을 권장한다.
# 다운로드에 매우 긴 시간이 소요됩니다.
import urllib3
urllib3.disable_warnings()
(ds_train, ds_test), ds_info = tfds.load(
'kitti',
split=['train', 'test'],
shuffle_files=True,
with_info=True,
)다운로드한 KITTI 데이터셋을 tfds.show_examples를 통해 보도록 한다. 우리가 일반적으로 보는 사진보다 광각으로 촬영되어 다양한 각도의 물체를 확인할 수 있다.
fig = tfds.show_examples(ds_train, ds_info)데이터 다운로드 시 담아둔 ds_info에서는 불러온 데이터셋의 정보를 확인할 수 있다. 오늘 사용할 데이터셋은 7,481개의 학습 데이터(training data), 711개의 평가용 데이터(test data), 423개의 검증용 데이터(validation data)로 구성되어 있다. 라벨에는 alpha, bbox, dimensions, location, occluded, rotation_y, truncated 등의 정보가 있다.
ds_info
tfds.core.DatasetInfo(
name='kitti',
full_name='kitti/3.2.0',
description="""
Kitti contains a suite of vision tasks built using an autonomous driving
platform. The full benchmark contains many tasks such as stereo, optical flow,
visual odometry, etc. This dataset contains the object detection dataset,
including the monocular images and bounding boxes. The dataset contains 7481
training images annotated with 3D bounding boxes. A full description of the
annotations can be found in the readme of the object development kit readme on
the Kitti homepage.
""",
homepage='http://www.cvlibs.net/datasets/kitti/',
data_path='/home/ssac10/tensorflow_datasets/kitti/3.2.0',
download_size=11.71 GiB,
dataset_size=5.27 GiB,
features=FeaturesDict({
'image': Image(shape=(None, None, 3), dtype=tf.uint8),
'image/file_name': Text(shape=(), dtype=tf.string),
'objects': Sequence({
'alpha': tf.float32,
'bbox': BBoxFeature(shape=(4,), dtype=tf.float32),
'dimensions': Tensor(shape=(3,), dtype=tf.float32),
'location': Tensor(shape=(3,), dtype=tf.float32),
'occluded': ClassLabel(shape=(), dtype=tf.int64, num_classes=4),
'rotation_y': tf.float32,
'truncated': tf.float32,
'type': ClassLabel(shape=(), dtype=tf.int64, num_classes=8),
}),
}),
supervised_keys=None,
splits={
'test': <SplitInfo num_examples=711, num_shards=4>,
'train': <SplitInfo num_examples=6347, num_shards=64>,
'validation': <SplitInfo num_examples=423, num_shards=4>,
},
citation="""@inproceedings{Geiger2012CVPR,
author = {Andreas Geiger and Philip Lenz and Raquel Urtasun},
title = {Are we ready for Autonomous Driving? The KITTI Vision Benchmark Suite},
booktitle = {Conference on Computer Vision and Pattern Recognition (CVPR)},
year = {2012}
}""",
)이번에는 데이터셋을 직접 확인해보자. ds_train.take(1)을 통해서 데이터셋을 하나씩 뽑아볼 수 있는 TakeDataset을 얻을 수 있다. 이렇게 뽑은 데이터에는 image 등의 정보가 포함되어 있다.
TakeDataset = ds_train.take(1)
for example in TakeDataset:
print('------Example------')
print(list(example.keys())) # example is `{'image': tf.Tensor, 'label': tf.Tensor}`
image = example["image"]
filename = example["image/file_name"].numpy().decode('utf-8')
objects = example["objects"]
print('------objects------')
print(objects)
img = Image.fromarray(image.numpy())
img
이미지와 라벨을 얻는 방법을 알게 되었다. 그렇다면 이렇게 얻은 이미지의 바운딩 박스(bounding box, bbox)를 확인하기 위해서는 어떻게 해야 할 것인가? 아래는 KITTI에서 제공하는 데이터셋에 대한 설명이다.
데이터셋 이해를 위한 예시
Values Name Description
----------------------------------------------------------------------------
1 type Describes the type of object: 'Car', 'Van', 'Truck',
'Pedestrian', 'Person_sitting', 'Cyclist', 'Tram',
'Misc' or 'DontCare'
1 truncated Float from 0 (non-truncated) to 1 (truncated), where
truncated refers to the object leaving image boundaries
1 occluded Integer (0,1,2,3) indicating occlusion state:
0 = fully visible, 1 = partly occluded
2 = largely occluded, 3 = unknown
1 alpha Observation angle of object, ranging [-pi..pi]
4 bbox 2D bounding box of object in the image (0-based index):
contains left, top, right, bottom pixel coordinates
3 dimensions 3D object dimensions: height, width, length (in meters)
3 location 3D object location x,y,z in camera coordinates (in meters)
1 rotation_y Rotation ry around Y-axis in camera coordinates [-pi..pi]
1 score Only for results: Float, indicating confidence in
detection, needed for p/r curves, higher is better.
[KITTI 원본이미지 예시]
[KITTI 원본이미지 예시]
def visualize_bbox(input_image, object_bbox):
input_image = copy.deepcopy(input_image)
draw = ImageDraw.Draw(input_image)
# 바운딩 박스 좌표(x_min, x_max, y_min, y_max) 구하기
width, height = input_image.size
print('width:', width, ' height:', height)
print(object_bbox.shape)
x_min = object_bbox[:,1] * width
x_max = object_bbox[:,3] * width
y_min = height - object_bbox[:,0] * height
y_max = height - object_bbox[:,2] * height
# 바운딩 박스 그리기
rects = np.stack([x_min, y_min, x_max, y_max], axis=1)
for _rect in rects:
print(_rect)
draw.rectangle(_rect, outline=(255,0,0), width=2)
print(input_image)
return input_image
visualize_bbox(img, objects['bbox'].numpy())
- Focal Loss for Dense Object Detection
- kimcando94님의 Object Detection에 대하여_01: Overall Object detection flow
- 김홍배님의 Focal loss의 응용(Detection & Classification)
RetinaNet은 Focal Loss for Dense Object Detection 논문을 통해 공개된 detection 모델이다. Detection 모델을 직접 만들기에는 많은 시간이 소요되기 때문에, 미리 모델을 구현한 라이브러리를 가져와 커스텀 데이터셋에 학습시키고 빠르게 사용해 보자.
1-stage detector 모델인 YOLO와 SSD는 2-stage detector인 Faster-RCNN 등보다 속도는 빠르지만 성능이 낮은 문제를 가지고 있었다. 이를 해결하기 위해서 RetinaNet에서는 focal loss와 FPN(Feature Pyramid Network) 를 적용한 네트워크를 사용한다.
Focal loss는 기존의 1-stage detection 모델들(YOLO, SSD)이 물체 전경과 배경을 담고 있는 모든 그리드(grid)에 대해 한 번에 학습됨으로 인해서 생기는 클래스 간의 불균형을 해결하고자 도입되었다. 여기서 그리드(grid)와 픽셀(pixel)이 혼란스러울 수 있겠는데, 위 그림 왼쪽 7x7 feature level에서는 한 픽셀이고, 오른쪽의 image level(자동차 사진)에서 보이는 그리드는 각 픽셀의 receptive field이다.
그림에서 보이는 것처럼 우리가 사용하는 이미지는 물체보다는 많은 배경을 학습하게 된다. 논문에서는 이를 해결하기 위해서 Loss를 개선하여 정확도를 높였다.
Focal loss는 우리가 많이 사용해왔던 교차 엔트로피를 기반으로 만들어졌다. 위 그림을 보면 Focal loss는 그저 교차 엔트로피 CE(
교차 엔트로피의 개형을 보면 ground truth class에 대한 확률이 높으면 잘 분류된 것으로 판단되므로 손실이 줄어드는 것을 볼 수 있다. 하지만 확률이 1에 매우 가깝지 않은 이상 상당히 큰 손실로 이어진다.
이 상황은 물체 검출 모델을 학습시키는 과정에서 문제가 될 수 있다. 대부분의 이미지에서는 물체보다 배경이 많다. 따라서 이미지는 극단적으로 배경의 class가 많은 class imbalanced data라고 할 수 있다. 이렇게 너무 많은 배경 class에 압도되지 않도록 modulating factor로 손실을 조절해준다. 람다를 0으로 설정하면 modulating factor가 0이 되어 일반적인 교차 엔트로피가 되고 람다가 커질수록 modulating이 강하게 적용되는 것을 확인할 수 있다.
FPN은 특성을 피라미드처럼 쌓아서 사용하는 방식이다. CNN 백본 네트워크에서는 다양한 레이어의 결과값을 특성 맵(feature map)으로 사용할 수 있다. 이때 컨볼루션 연산은 커널을 통해 일정한 영역을 보고 몇 개의 숫자로 요약해 내기 때문에, 입력 이미지를 기준으로 생각하면 입력 이미지와 먼 모델의 뒷쪽의 특성 맵일수록 하나의 "셀(cell)"이 넓은 이미지 영역의 정보를 담고 있고, 입력 이미지와 가까운 앞쪽 레이어의 특성 맵일수록 좁은 범위의 정보를 담고 있다. 이를 receptive field라고 한다. 레이어가 깊어질 수록 pooling을 거쳐 넓은 범위의 정보(receptive field)를 갖게 되는 것이다.
FPN은 백본의 여러 레이어를 한꺼번에 쓰는데에 의의가 있다. SSD가 각 레이어의 특성 맵에서 다양한 크기에 대한 결과를 얻는 방식을 취했다면 RetinaNet에서는 receptive field가 넓은 뒷쪽의 특성 맵을 upsampling(확대)하여 앞단의 특성 맵과 더해서 사용했다. 레이어가 깊어질수록 feature map의 w, h방향의 receptive field가 넓어지는 것인데, 넓게 보는 것과 좁게 보는 것을 같이 쓰겠다는 목적인 것이다.
- Upsampling 참고: CS231n - Lecture 11 | Detection and Segmentation
위 그림은 RetinaNet 논문에서 FPN 구조가 어떻게 적용되었는지를 설명하는 그림이다. FPN은 각 level이 256채널로 이루어지는데, RetinaNet에서는 FPN의 P3부터 P7까지의 Pyramid level을 사용한다. 이를 통해 Classification Subnet과 Box Regression Subnet 2개의 Subnet을 구성하게 되는데, Anchor의 갯수를 A라고 하면 최종적으로 Classification Subnet은 K개 class에 대해 KA개의 채널을, Box Regression Subnet은 4A개 채널을 사용하게 된다.
Keras RetinaNet은 케라스(Keras) 라이브러리로 구현된 RetinaNet이다. 현재는 텐서플로우 2 버전을 지원하는 Repository도 만들어졌으나 아직 커스텀 데이터셋을 학습하는 방법을 공식 문서로 제시하지 않고 있다. 지금은 우선 Keras RetineNet을 이용해보도록 한다.
우리가 가진 tensorflow_dataset의 KITTI 데이터셋을 그대로 사용해서 Keras RetinaNet을 학습시키기 위해서는 라이브러리를 수정해야 한다. 하지만 이보다 더 쉬운 방법은 해당 모델을 훈련할 수 있는 공통된 데이터셋 포맷인 CSV 형태로 모델을 변경해주는 방법이다.
우리는 tensorflow_dataset의 API를 사용해 이미지와 각 이미지에 해당하는 바운딩 박스 라벨의 정보를 얻을 수 있었다. 그렇다면 API를 활용하여 데이터를 추출, 이를 포맷팅 하여 CSV 형태로 한 줄씩 저장해 보자.
한 라인에 이미지 파일의 위치, 바운딩 박스 위치, 그리고 클래스 정보를 가지는 CSV 파일을 작성하도록 코드를 작성하고, 이를 사용해 CSV 파일을 생성한다. 우리가 생각하는 브레이크 시스템은 차와 사람을 구분해야 하는 점을 유의하자. 데이터셋 포맷은 아래를 참고한다.
# 데이터셋 형식
path/to/image.jpg,x1,y1,x2,y2,class_name
# Example
/data/imgs/img_001.jpg,837,346,981,456,cow
/data/imgs/img_002.jpg,215,312,279,391,cat
/data/imgs/img_002.jpg,22,5,89,84,bird
/data/imgs/img_003.jpg,,,,,CSV로 저장할 때는 아래 코드를 참고한다.
# 데이터셋 저장 시 참고
df_test = parse_dataset(ds_test, total=ds_info.splits['test'].num_examples)
df_test.to_csv('./kitti_test.csv', sep=',',index = False, header=False)import os
data_dir = os.getenv('HOME')+'/aiffel/object_detection/data'
img_dir = os.getenv('HOME')+'/kitti_images'
train_csv_path = data_dir + '/kitti_train.csv'
# parse_dataset 함수를 구현해 주세요.
def parse_dataset(dataset, img_dir="kitti_images", total=0):
if not os.path.exists(img_dir):
os.mkdir(img_dir)
# Dataset의 claas를 확인하여 class에 따른 index를 확인해둡니다.
# 저는 기존의 class를 차와 사람으로 나누었습니다.
type_class_map = {
0: "car",
1: "car",
2: "car",
3: "person",
4: "person",
5: "person",
}
# Keras retinanet을 학습하기 위한 dataset을 csv로 parsing하기 위해서 필요한 column을 가진 pandas.DataFrame을 생성합니다.
df = pd.DataFrame(columns=["img_path", "x1", "y1", "x2", "y2", "class_name"])
for item in tqdm(dataset, total=total):
filename = item['image/file_name'].numpy().decode('utf-8')
img_path = os.path.join(img_dir, filename)
img = Image.fromarray(item['image'].numpy())
img.save(img_path)
object_bbox = item['objects']['bbox']
object_type = item['objects']['type'].numpy()
width, height = img.size
# tf.dataset의 bbox좌표가 0과 1사이로 normalize된 좌표이므로 이를 pixel좌표로 변환합니다.
x_min = object_bbox[:,1] * width
x_max = object_bbox[:,3] * width
y_min = height - object_bbox[:,2] * height
y_max = height - object_bbox[:,0] * height
# 한 이미지에 있는 여러 Object들을 한 줄씩 pandas.DataFrame에 append합니다.
rects = np.stack([x_min, y_min, x_max, y_max], axis=1).astype(np.int)
for i, _rect in enumerate(rects):
_type = object_type[i]
if _type not in type_class_map.keys():
continue
df = df.append({
"img_path": img_path,
"x1": _rect[0],
"y1": _rect[1],
"x2": _rect[2],
"y2": _rect[3],
"class_name": type_class_map[_type]
}, ignore_index=True)
break
return df
df_train = parse_dataset(ds_train, img_dir, total=ds_info.splits['train'].num_examples)
df_train.to_csv(train_csv_path, sep=',',index = False, header=False)테스트 데이터셋에 대해서도 동일하게 parse_dataset()을 적용해서 DataFrame을 생성한다.
test_csv_path = data_dir + '/kitti_test.csv'
df_test = parse_dataset(ds_test, img_dir, total=ds_info.splits['test'].num_examples)
df_test.to_csv(test_csv_path, sep=',',index = False, header=False)데이터셋에서 클래스는 문자열(string)으로 표시되지만, 모델에게 데이터를 알려줄 때에는 숫자를 사용해 클래스를 표시해야 한다. 이때 모두 어떤 클래스가 있고 각 클래스가 어떤 인덱스(index)에 맵핑(mapping)될지 미리 정하고 저장해 두어야 학습을 한 후 추론(inference)을 할 때에도 숫자 인덱스로 나온 정보를 클래스 이름으로 바꾸어 해석할 수 있다.
class_txt_path = data_dir + '/classes.txt'
def save_class_format(path="./classes.txt"):
class_type_map = {
"car" : 0,
"person": 1
}
with open(path, mode='w', encoding='utf-8') as f:
for k, v in class_type_map.items():
f.write(f"{k},{v}\n")
save_class_format(class_txt_path)준비가 완료되었다면 아래 스크립트를 참고하셔서 위에서 변환한 데이터셋으로 학습을 시작한다. 학습이 잘 되기 위해서는 환경에 따라 batch_size나 worker, epoch를 조절해야 한다.
훈련 이미지 크기 또는 batch_size가 너무 크면 GPU에서 out-of-memory 에러가 날 수 있으니 적절히 조정해야 한다. 원 개발자는 8G 메모리도 RetinaNet을 훈련시키기에는 부족할 수 있다고 설명한다. (참고)
# RetinaNet 훈련이 시작됩니다!! 50epoch 훈련에 1시간 이상 소요될 수 있습니다.
!cd ~/aiffel/object_detection && python keras-retinanet/keras_retinanet/bin/train.py --gpu 0 --multiprocessing --workers 4 --batch-size 2 --epochs 50 --steps 195 csv data/kitti_train.csv data/classes.txt아래 코드를 사용해 학습된 모델을 추론을 위해 실행할 수 있는 케라스 모델로 변환한다.
!cd ~/aiffel/object_detection && python keras-retinanet/keras_retinanet/bin/convert_model.py snapshots/resnet50_csv_50.h5 snapshots/resnet50_csv_50_infer.h5이제 위에서 변환한 모델을 load하고 추론 및 시각화를 해보자. 아래에 load된 모델을 통해 추론을 하고 시각화를 하는 함수를 작성한다. 일정 점수 이하는 경우를 제거해야 함을 유의한다.
%matplotlib inline
# automatically reload modules when they have changed
%load_ext autoreload
%autoreload 2
# import keras
import keras
# import keras_retinanet
from keras_retinanet import models
from keras_retinanet.models import load_model
from keras_retinanet.utils.image import read_image_bgr, preprocess_image, resize_image
from keras_retinanet.utils.visualization import draw_box, draw_caption
from keras_retinanet.utils.colors import label_color
from keras_retinanet.utils.gpu import setup_gpu
# import miscellaneous modules
import matplotlib.pyplot as plt
import cv2
import os
import numpy as np
import time
gpu = '0'
setup_gpu(gpu)
dir_path = os.getenv('HOME') + '/aiffel/object_detection/'
model_path = os.path.join(dir_path, 'snapshots', 'resnet50_csv_50_infer.h5')
model = load_model(model_path, backbone_name='resnet50')import os
img_path = os.getenv('HOME')+'/aiffel/object_detection/test_set/go_1.png'
# inference_on_image 함수를 구현해 주세요.
def inference_on_image(model, img_path="./test_set/go_1.png", visualize=True):
image = read_image_bgr(img_path)
# copy to draw on
draw = image.copy()
draw = cv2.cvtColor(draw, cv2.COLOR_BGR2RGB)
color_map = {
0: (0, 0, 255), # blue
1: (255, 0, 0) # red
}
# preprocess image for network
image = preprocess_image(image)
image, scale = resize_image(image)
# process image
boxes, scores, labels = model.predict_on_batch(np.expand_dims(image, axis=0))
# correct for image scale
boxes /= scale
# display images
if visualize:
for box, score, label in zip(boxes[0], scores[0], labels[0]):
print(box)
if score < 0.5:
break
b = box.astype(int)
draw_box(draw, b, color=color_map[label])
caption = "{:.3f}".format(score)
draw_caption(draw, b, caption)
plt.figure(figsize=(15, 15))
plt.axis('off')
plt.imshow(draw)
plt.show()
inference_on_image(model, img_path=img_path)
img_path = os.getenv('HOME')+'/aiffel/object_detection/test_set/stop_1.png'
inference_on_image(model, img_path=img_path)
위에서 만든 모델을 통해 아래의 조건을 만족하는 함수를 만들어 보자.
- 입력으로 이미지 경로를 받는다.
- 정지조건에 맞는 경우 "Stop" 아닌 경우 "Go"를 반환한다.
- 조건은 다음과 같다.
- 사람이 한 명 이상 있는 경우
- 차량의 크기(width or height)가 300px이상인 경우
img_path = os.getenv('HOME')+'/aiffel/object_detection/test_set/stop_1.png'
def self_drive_assist(img_path, size_limit=300):
image = read_image_bgr(img_path)
color_map = {
0: (0, 0, 255), # blue
1: (255, 0, 0) # red
}
# preprocess image for network
image = preprocess_image(image)
image, scale = resize_image(image)
# process image
boxes, scores, labels = model.predict_on_batch(np.expand_dims(image, axis=0))
# correct for image scale
boxes /= scale
# 코드 구현
# 정지조건에 맞으면 return "Stop"
# 아닌 경우 return "Go"
for box, score, label in zip(boxes[0], scores[0], labels[0]):
if score < 0.5:
break
if label == 1:
return "Stop"
if box[2] - box[0] >= size_limit or box[3] - box[1] >= size_limit:
return "Stop"
return "Go"
print(self_drive_assist(img_path))아래 test_system() 를 통해서 위에서 만든 함수를 평가해보자. 10장에 대해 Go와 Stop을 맞게 반환하는지 확인하고 100점 만점으로 평가해준다.
import os
def test_system(func):
work_dir = os.getenv('HOME')+'/aiffel/object_detection'
score = 0
test_set=[
("test_set/stop_1.png", "Stop"),
("test_set/stop_2.png", "Stop"),
("test_set/stop_3.png", "Stop"),
("test_set/stop_4.png", "Stop"),
("test_set/stop_5.png", "Stop"),
("test_set/go_1.png", "Go"),
("test_set/go_2.png", "Go"),
("test_set/go_3.png", "Go"),
("test_set/go_4.png", "Go"),
("test_set/go_5.png", "Go"),
]
for image_file, answer in test_set:
image_path = work_dir + '/' + image_file
pred = func(image_path)
if pred == answer:
score += 10
print(f"{score}점입니다.")
test_system(self_drive_assist)- Object Detection의 개념 자체는 이해했지만, 모델의 작동 원리를 이해하는 데 조금 어려웠다.
- 다운로드에 너무 오랜 시간이 걸렸다. 이런 부분은 미리 서버에서 받을 수 있도록 해주면 좋을 것 같다.
- 자율주행 보조 시스템을 Object Detection을 이용하여 간단하게 만들어볼 수 있었다.