DETR
DETR(End-to-End Object Detection with Transformers) 은 2020년 나온 이전 논문이지만 CV에서 transformer 구조를 접목한 최초의 논문 중 하나이다.
이전 object detection 기법에서 필수적인 Anchor 및 NMS와 같은 post process가 없어 간단하게 End-to-End object detection 을 수행하였다.
해당 논문의 contribution은 높으며 이후 DETR-Like의 논문들이 지속해서 나오고 있다, 그 중 최신 기법인 CO-DETR은 COCO 데이터 셋에서 가장 높은 성능을 보여주고 있다.
Transformer의 구조적 특성으로 인하여 multi-modal에 사용하기 적합하며 Grounding DINO 같은 기법의 open-set object detection의 기초이기도 하다.
또한 해당 논문 코드의 완성도도 높기에 코드 레벨까지 공부할 필요성이 있어 해당 포스트에서 논문 설명과 실제 코드가 어떻게 작동되는지 알아보고자 한다.
단 해당 포스트는 detection 부분 위주로 설명할 것이며, segmentation 부분은 생략하겠다.
만약 논문을 잘 알고 있다면 논문 부분을 skip하고 바로 Code 부분으로 넘어가도 무방하다.
논문
해당 논문은 주요 contributions은 다음과 같다.
- Transformer encoder-decoder 구조를 object detection에 적용하고 또한 기존 autoregressive가 아닌 parallel하게 예측.
- Bipartite matching을 사용한 새로운 set기반 목적 함수를 설계하여 직접 예측하여 기존의 surrogate task을 회피.
Introduction
객체 탐지의 목표는 관심 있는 각 객체에 대한 바운딩박스와 카테고리 레이블를 예측하는 것.
이를 위해 모던 detector들은 (대규모 proposals set, anchors 및 window centers) 간접적인 regression과 classification으로 예측 문제를 해결.
하지만 이러한 기법의 성능은 post-processing의 중복 제거에 큰 영향을 받음.
해당 논문은 이러한 부분 없이 direct하게 set을 예측함으로써 prior knowledge없이 예측한다.(물론 prior knowledge 주입이 없기에 성능이 최신 기법대비 낮으며, 학습 속도 또한 느리다.)
상위 그림에서 보이듯이,
1. DETR에서는 먼저 CNN통해 이미지 features을 추출,
2. 해당 features을 transformer encoder에 입력으로 사용하여 global 정보 추출,
3. (그림에는 없지만) Query와 global 정보를 tansformer decoder의 입력으로 사용, hidden feature 추출,
4. 추출된 query들 중 이분 매칭을 사용하여 GT와 예측들을 유니크하게 매칭, 남은 예측은 no object로 할당하여 일대일 매칭 진행.
DETR은 한번에 모든 예측을 수행하며 학습 과정중 예측과 GT의 이분 매칭 set based loss function으로 학습.
DETR은 어떤 특이한 커스텀 layer가 없어 standard framework(PyTorch)로 다음과 같은 심플한 코드로 구현 할 수 있다.
(PyTorch framework을 사용하여 약 50줄로 DETR 구조를 구성할 수 있다.)
class DETRdemo(nn.Module):
def __init__(self, num_classes, hidden_dim=256, nheads=8,
num_encoder_layers=6, num_decoder_layers=6):
super().__init__()
# create ResNet-50 backbone
self.backbone = resnet50()
del self.backbone.fc
# create conversion layer
self.conv = nn.Conv2d(2048, hidden_dim, 1)
# create a default PyTorch transformer
self.transformer = nn.Transformer(
hidden_dim, nheads, num_encoder_layers, num_decoder_layers)
# prediction heads, one extra class for predicting non-empty slots
# note that in baseline DETR linear_bbox layer is 3-layer MLP
self.linear_class = nn.Linear(hidden_dim, num_classes + 1)
self.linear_bbox = nn.Linear(hidden_dim, 4)
# output positional encodings (object queries)
self.query_pos = nn.Parameter(torch.rand(100, hidden_dim))
# spatial positional encodings
# note that in baseline DETR we use sine positional encodings
self.row_embed = nn.Parameter(torch.rand(50, hidden_dim // 2))
self.col_embed = nn.Parameter(torch.rand(50, hidden_dim // 2))
def forward(self, inputs):
# propagate inputs through ResNet-50 up to avg-pool layer
x = self.backbone.conv1(inputs)
x = self.backbone.bn1(x)
x = self.backbone.relu(x)
x = self.backbone.maxpool(x)
x = self.backbone.layer1(x)
x = self.backbone.layer2(x)
x = self.backbone.layer3(x)
x = self.backbone.layer4(x)
# convert from 2048 to 256 feature planes for the transformer
h = self.conv(x)
# construct positional encodings
H, W = h.shape[-2:]
pos = torch.cat([
self.col_embed[:W].unsqueeze(0).repeat(H, 1, 1),
self.row_embed[:H].unsqueeze(1).repeat(1, W, 1),
], dim=-1).flatten(0, 1).unsqueeze(1)
# propagate through the transformer
h = self.transformer(pos + 0.1 * h.flatten(2).permute(2, 0, 1),
self.query_pos.unsqueeze(1)).transpose(0, 1)
# finally project transformer outputs to class labels and bounding boxes
return {'pred_logits': self.linear_class(h),
'pred_boxes': self.linear_bbox(h).sigmoid()}가장 널리 사용되는 COCO 데이터셋에서 Faster R-CNN와 비슷한 성능을 보여주고 있다.
물론 해당 성능은 당시 SOTA 성능보다는 약 10 포인트 정도로 낮앗지만, DETR은 기존 기법의 여러 문제점(spatial anchors, NMS, ...)을 해결하였으며, 이후 DETR-like 모델이 지속적으로 발표되었고 있다.
앞서 말했듯이 CO-DETR은 아래 그림과 같이 Object detection에서 가장 좋은 성능을 보여주고 있다.
The DETR model
앞선 설명에서 말했듯이, DETR의 중요 부분은 1) 집합기반 예측 손실 함수, 2) transformer 구조이다.
다음 섹션에서 이 두 부분을 설명하겠다.
Object detection set prediction loss
DETR은 고정된 N개의 예측을 실행한다. 여기서 N은 일반 영상속에 있는 객체보다 큰 숫자이다(실제 구현 N=100).
훈련의 주요 어려움 중 하나는 예측된 객체(클래스, 위치, 크기)를 실제 객체와 비교하여 점수를 매기는 것.
본 논문의 손실 함수는 예측된 객체와 실제 객체 간의 최적 이분 매칭 진행, 객체 별(분류 및 바운딩 박스) 손실을 최적화.
여기서 를 ground truth의 객체 집합이고 을 예측이라고 정의하자.
그리고 N은 실제 이미지속의 객체 숫자보다도 크다, 역시도 부족한 부분을 (no object) 로 패딩하여 N개의 사이즈로 만든다.
이 두 집합 사이의 이진 매칭을 구하기 위해 N개의 요소에 대한 순열 중에서 비용이 가장 낮은 것을 찾는다:
여기서 는 GT 및 예측 인덱스 간의 pair-wise 매칭 cost 이다.
최적의 할당은 헝가리안 알고리듬으로 효율적으로 계산할 수 있다.
헝가리안 알고리듬
해당 방법은 두 그룬의 요수를 최대(최소) 값으로 일대일로 매칭하는 방법이다.
예로 다음과 어떠한 사람에게 어떠한 작업을 매칭하는게 비용을 최소화 하는지를 알아는 것이다.
이 부분에 대한 실제 작동 이해가 충분하기 않기에, 추후 시간이 된다면 해당 알고리듬 포스트를 작성하여 설명하겠다.
코드 레벨에서는scipy라이브러리를 사용하여scipy.optimize.linear_sum_assignment사용하여 대응 인덱스 튜플을 반환하여 사용한다.
객체 탐지에서는 상위 그림의 y 축을 모델의 예측이라고 볼수 있으며, x축은 GT라고 볼 수 있다.
우리가 원하는 것은 해당 매트릭스에서 각 GT(테스크)마다 가장 cost가 낮은 매칭을 찾는 것이다.
여기에서의 매칭 cost는 class prediction 및 bbox 회귀 두 파트의 합으로 계산된다.
객체 집합의 각 요소 는 로 표현될 수 있으며, 여기서 는 타켓 클래스 레이블( 일 수도 있음)이고 는 이미지 크기에 대한 상대적인 실제 바운딩 박스 중심 좌표와 높이, 너비로 정의하는 벡터이다.
인덱스 를 갖는 예측에 대해 클래스 의 확률을 로, 예측된 바운딩 박스를 로 정의한다.
이러한 표기법의 매칭 손실 를 다음과 같이 정의된다.
여기서 최적 할당은 1번 수식의 계산으로 구할 수 있다.
헝가리안 알고리듬을 통해 우리는 각 예측마다 가장 알 맞는 매칭을 구성 할 수 있다.
즉 각 집합마다 1대1 매칭을 하여 set-based loss를 정의한다.
이제 상위 스텝에서 찾은 매칭의 hungarian loss를 사용하여 해당 매칭의 손실 함수를 최소화하고자 한다.
해당 로스는 negative log-likelihood와 box 로스의 linear combination이다.
재밌는 점은 2번 수식과 3번 수식의 클래스 부분 손실 부분이 과 으로 다르다는 점이다.
이는 매칭이 최대한 분류와 회귀 파트가 적절하게 동시에 고려되길 기대하기 때문이다.
다음 그림과 같이 NLL은 큰 오차를 발생하게 생성한다.
만약 매칭 부분에서 NLL을 사용한다면, 2번 수식은 너무 분류에 치우치는 매칭이 될것이며 이는 밸런스싱이 잡힌 매칭이 아니다.
따라서 같은 스케일인 확률로 바꾸어 매칭이 최대한 두 부분을 동시 고려한다.
그리고 확고한 매칭되었다면 일반적으로 NLL으로 최대한 목적에 가깝게 하는 것이 합리적이다.
추가적으로 실제 bouding box 손실함수는 두가지 부분으로 구성된다. 실제 거리와 giou 손실함수이다.
giou 손실 함수는 scale-invariant 함수하여, 작은 객체의 탐지의 성능을 높이기위해 파트이며, 실험의 작은 디테일이라고 볼 수 있겠다.
현재는 D(Distance)iou 혹은 종합적인 C(Complete)iou가 더 좋은 선택이 것이다.
이렇게 구성한 loss는 기존의 post processing을 제거하여 end-to-end object detection을 학습할 수 있게 된다.
DETR architecture
전체 구조는 상위 그림으로 확인 할 수 있다. 해당 구조는 3가지 부분으로 구성되어 있다. 1) feature representation을 획득하는 CNN backbone, 2) encode-decoder transformer, 3) 최종 객체를 예측하기위한 simple FFN.
1. Backbone
기존 이미지 를 전통적인 CNN backbone을 통과하여 low-resolution의 feature 맵 회득.
2.1. Transformer encoder
먼저 1x1 convolution을 사용하여 높은 차원을 작은 차원 로 매핑하여 새로운 feature map 회득.
Encoder 인터페이스는 일련의 sequance 입력을 기대하기 때문에 공간 차원을 1차원 데이터 flatten하여 획득.
이후 multi-head self-attention과 FFN으로 구성된 표준 encoder layers을 통과.
Transformer는 permutation-invariant 하기때문에 (self-attention의 값은 입력의 위치가 중요하기 않고 와 간의 내적(유사도)으로 결정) 고정된 position encoding 정보는 각 attention layer 입력 전에 add하여 전달.
2.2. Transformer decoder
Decoder는 표준 transformer decoder 따른다.
Multi-head self- 및 encoder-decoder(cross) attention 메커니즘을 사용하여 크기 개 크기의 hidden feature 획득.
원래 transformer와 차이점은 의 객체의 parallel하게 처리한다는 점이다.
Decoder에서 별도의 object query을 입력으로 사용하여 객체를 검출한다.
앞선 self- 및 cross-attention으로 DETR은 queries 각각 자체/영상 정보를 상호 교류하면서 곂치치않고 영상 전체영역에서 객체를 탐지 할 수 있게 된다.
2.3. Prediction feed-forward networks (FFNs)
마지막 예측 부분은 3층 짜리 MLP와 linear projection layer로 각각 개의 객체의 바운딩 박스와 클래스를 예측한다.
앞서 설명한 로스로 인하여 실제 영상 객체와 대응이 안되는 queries는 special class label no object를 예측한다.
Experiments
같은 데이터 증강 및 학습 방법으로 재 학습된 faster RCNN와 DETR의 간의 성능이 비슷한 걸을 상위 표로 확인 할 수 있다.
단 보이듯이 작은 물체 탐지성능은 떨어지며 large 객체 탐지 성능은 기존 RCNN대비 높은 것을 확인 할 수 있다.
Encoder는 영상 전체에서 정보를 획득하기때문에 큰 객체를 탐지 능력은 우수하다고 볼 수 있다.
작은 객체 같은 경우, backbone에서 입력 영상을 저 해상도로 낮추어 작은 객체의 정보가 소실되어 성능이 저하된걸로 판단된다.
추후 deformable-DETR에서 multi-scale features을 사용하여 개선하였다.
상위 그림은 각 포인트마다 대응되는 마지막 인코더 레이어의 attention maps을 시각화하였다. 보이듯이 encoder는 이미 instance들을 잘 분리하고 있다. 이는 decoder의 객체 추출 및 localization을 간소화 할 수 있다.
이는 encoder가 global scene의 추론하여 객 물체의 얽힘을 분리 할 수 있다는 것을 확인 할 수 있다.
상위 표에서 인코더 layer의 증가 할 수록 성능이 지속적으로 증가하는 것을 볼 수 있다.
당연하게 layer가 증가 할 수록 계산량도 증가한다.
어떻게보면 너무나 당연한 실험을 한것이지만 해당 논문은 처음으로 transformer을 영상에서 사용하였기에 해당 실험을 보다 디테일하게 작성한것이다.
Encoder에서 이미 객체를 충분히 구분하는 것을 시각화로 확인 할 수 있다. 그렇다면 docoder는 어떤 역활을 하고 실제로는 어떤한 것을 학습하고 있을까?
상위 그림의 decoder의 query의 attention 맵을 보면, decoder은 각 물체의 바운딩 박스를 결정하는 객체의 외각 부분(뒷 코끼리의 코, 다리, 꼬리)에 집중 하는 것을 확인 할 수 있다.
객체의 외각 부분의 정보를 중적적으로 획득하여 바운딩 박스의 위치를 결정하는 것을 확인 할 수 있다.
그리고 query는 어떤 것들을 학습하였을까?
학습된 object query의 그림에서 볼 수 있듯이 각 query들은 영상의 특정 지역의 물체 존재 여부를 확인하고 있는 것이다.
예로 들어 상위 그림의 빨간색의 박스들의 query들은 각각 영상의 좌하단의 작은 물체와 중앙부분/오른쪽의 작은 객체와 중앙 부분을 집중적으로 확인 하고 있는 점이다.
이는 기존 predefined anchor와 비슷하며 단 query들은 이를 직접 학습 한다는 점이다.
Conclusion
해당 논문은 transformer와 bipartite matching loss을 기반으로 한 새로운 객체 검출 기법 DETR을 제안하였다.
이 접근법은 Faster R-CNN 기준선과 비슷한 결과를 달성.
DETR은 구현이 간단하고 유연한 아키텍처를 가지며, 다른 task로 쉽게 확장가능하며, 실제로 이후 많은 후속 연구가 진행되고 있다.
다음 부분은 해당 기법의 코드를 원리를 파악하면 상위 방법의 실 구현을 확인 하도록하겠다.
Code
DETR의 original 구현은 해당 링크에서 확인 할 수 있다.
다음 색션에서는 간단하게 실제 구현을 확인하겠다.
전체 구조는 다음 그림과 같다.
가장 중요한 main.py 부터 라인마다 필요한 부분을 설명하면서 진행하겠다.
main.py
def get_args_parser():
parser = argparse.ArgumentParser('Set transformer detector', add_help=False)
...
# * Segmentation
parser.add_argument('--masks', action='store_true',
help="Train segmentation head if the flag is provided")
...
# dataset parameters
parser.add_argument('--dataset_file', default='coco')
# 데이터 입력 부분
parser.add_argument('--coco_path', type=str, default=r'path/to/COCO')
parser.add_argument('--coco_panoptic_path', type=str)
parser.add_argument('--remove_difficult', action='store_true')
parser.add_argument('--output_dir', default='',
help='path where to save, empty for no saving')
parser.add_argument('--device', default='cuda',
help='device to use for training / testing')
parser.add_argument('--seed', default=42, type=int)
parser.add_argument('--resume', default='', help='resume from checkpoint')
parser.add_argument('--start_epoch', default=0, type=int, metavar='N',
help='start epoch')
parser.add_argument('--eval', action='store_true')
parser.add_argument('--num_workers', default=2, type=int)
# distributed training parameters
parser.add_argument('--world_size', default=1, type=int,
help='number of distributed processes')
parser.add_argument('--dist_url', default='env://', help='url used to set up distributed training')
return parser모델의 학습/추론의 필요한 arguments을 받는 부분이다.
arguments 변수 이름에서 쉽게 해당 변수의 용도를 알 수 있을 것이다.
우리는 object detection이니 masks는 비활성이 되어 있다.
알단 학습에 필요한 부분 dataset parameters을 보자.
그 중 중요한 부분은 COCO 데이터셋 path이며 이부분은 다운로드 받은 데이터셋으로으로 설정 해주면 된다.
데이터는 https://cocodataset.org/#download에서 받을 수 있으며 다음 그림의 빨간박스 부분의 링크를 클릭하여 다운로드.
그리고 데이터 셋 구축 부분은 다음과 같다.
dataset_train = build_dataset(image_set='train', args=args)
dataset_val = build_dataset(image_set='val', args=args)이 부분의 내부 코드는 build_coco(image_set, args)이며 실 코드는 다음과 같다.
def build(image_set, args):
root = Path(args.coco_path)
assert root.exists(), f'provided COCO path {root} does not exist'
mode = 'instances'
PATHS = {
"train": (root / "train2017", root / "annotations" / f'{mode}_train2017.json'),
"val": (root / "val2017", root / "annotations" / f'{mode}_val2017.json'),
}
img_folder, ann_file = PATHS[image_set]
dataset = CocoDetection(img_folder, ann_file, transforms=make_coco_transforms(image_set), return_masks=args.masks)
return dataset상위 코드는 어려운 부분이 없으며 중요한 부분은 CocoDetection 클래스이다.
해당 부분을 확인해보자.
class CocoDetection(torchvision.datasets.CocoDetection):
def __init__(self, img_folder, ann_file, transforms, return_masks):
super().__init__(img_folder, ann_file)
self._transforms = transforms
self.prepare = ConvertCocoPolysToMask(return_masks)
def __getitem__(self, idx):
# 상속 class 의 __getitem__ 사용하여 image 및 target 획득
img, target = super().__getitem__(idx)
image_id = self.ids[idx]
target = {'image_id': image_id, 'annotations': target}
img, target = self.prepare(img, target)
if self._transforms is not None:
img, target = self._transforms(img, target)
return img, target보이듯이 torchvision.datasets.CocoDetection을 상속 받아 COCO dataset을 쉽게 처리 할 수 있다.
super에 대해서 자세하게 알고 싶다면 해당 링크 참조.
이후 get_item에서 target 중에서 필요한 부분만 filtering 하는 self.prepare 와 img, target에 같은 transform을 진행 할 수 있는 self._transforms이 추가되어 있다.
torchvision의 최신 라이브러리는 이미지와 타켓에 일관된 transforms 적용할 수 있다.
단 DETR이 발표된 2020년도에서는 이러한 일관된 transforms가 지원되지 않았다.
이에 따라 저자들은 custom tranforms을 설계하였다.(에초에 저자중 한명은 PyTorch 저자이기에 여기서 작성된 것을 다시 이식 한것을 판단됨.)
최신 transforms v2 을 사용한다면 이미지와 bbox에 같은 transform을 적용 할 수 있으니 참고.
단 본 포스트는 일단 원본 코드를 따라 갈 것이며 좀 더 low level에서 어떻게 tensor 조작하는 확인해보자.
img은 전형적인 PIL.Image, target의 type은 list[dict]이다.
target은 len은 해당 이미지의 객체 수이며 각 element는 객체에 대한 정보이다.
해당 img와 target을 prepare로 처리하여 transform의 기대하는 형태으로 출력한다.
class ConvertCocoPolysToMask(object):
def __init__(self, return_masks=False):
self.return_masks = return_masks
def __call__(self, image, target):
# 이미지 w, h획득
w, h = image.size
# 해당 이미지 idx을 Tensor로 전환
image_id = target["image_id"]
image_id = torch.tensor([image_id])
# target["annotations"]은 list[dict]
anno = target["annotations"]
# 해당 리스트중 dict중 key값중에 'iscrowd`가 없거나, 'iscrowd'이 0이면 리스트 추가
# 즉 iscrowd != 0 이면 제거, 특수 케이스 제거
anno = [obj for obj in anno if 'iscrowd' not in obj or obj['iscrowd'] == 0]
# 마찬가지로 bbox가 있는 객체만 유지하고 해당 value만 리스트로
boxes = [obj["bbox"] for obj in anno]
# guard against no boxes via resizing
# boxes.shape = [객체수, 4]
boxes = torch.as_tensor(boxes, dtype=torch.float32).reshape(-1, 4)
# xywh format -> xyxy format으로 전환
# 처음 두개는 topleft 포인트이고 뒤 2개는 wh이기에 더하여
# topleft, bottomright로 변환
boxes[:, 2:] += boxes[:, :2]
# bbox 포인트가 이미지 사이즈를 넘지않아야 하기에 clamp으로 보장
boxes[:, 0::2].clamp_(min=0, max=w)
boxes[:, 1::2].clamp_(min=0, max=h)
# class label
classes = [obj["category_id"] for obj in anno]
classes = torch.tensor(classes, dtype=torch.int64)
# object detection에서 필요 없는 segmenation mask
if self.return_masks:
segmentations = [obj["segmentation"] for obj in anno]
masks = convert_coco_poly_to_mask(segmentations, h, w)
# 마찬가지인 keypoint(인가 포즈 측정 데이터)
keypoints = None
if anno and "keypoints" in anno[0]:
keypoints = [obj["keypoints"] for obj in anno]
keypoints = torch.as_tensor(keypoints, dtype=torch.float32)
num_keypoints = keypoints.shape[0]
if num_keypoints:
keypoints = keypoints.view(num_keypoints, -1, 3)
# filtering 부분
# top left 및 botton right 포인트간의 실제 로직이 맞는 점들만 keep
keep = (boxes[:, 3] > boxes[:, 1]) & (boxes[:, 2] > boxes[:, 0])
boxes = boxes[keep]
classes = classes[keep]
if self.return_masks:
masks = masks[keep]
if keypoints is not None:
keypoints = keypoints[keep]
# 처리된 Tensor들을 dict 형태의 수집 및 출력
target = {}
target["boxes"] = boxes
target["labels"] = classes
if self.return_masks:
target["masks"] = masks
target["image_id"] = image_id
if keypoints is not None:
target["keypoints"] = keypoints
# for conversion to coco api
area = torch.tensor([obj["area"] for obj in anno])
iscrowd = torch.tensor([obj["iscrowd"] if "iscrowd" in obj else 0 for obj in anno])
target["area"] = area[keep]
target["iscrowd"] = iscrowd[keep]
target["orig_size"] = torch.as_tensor([int(h), int(w)])
target["size"] = torch.as_tensor([int(h), int(w)])
return image, target출력된 결과는 다음과 같다.
그리고 출력된 결과는 다음 transforms 처리를 진행한다.
형태는 기존 torchvision.transforms과 같기에 어떠한 작업인지 직관적이다.
차이점은 기존은 영상만 가능했으나, 아래 코드는 영상 및 target을 동시에 처러한다는 점이다.
def make_coco_transforms(image_set):
normalize = T.Compose([
T.ToTensor(),
T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])
scales = [480, 512, 544, 576, 608, 640, 672, 704, 736, 768, 800]
if image_set == 'train':
return T.Compose([
T.RandomHorizontalFlip(),
T.RandomSelect(
T.RandomResize(scales, max_size=1333),
T.Compose([
T.RandomResize([400, 500, 600]),
T.RandomSizeCrop(384, 600),
T.RandomResize(scales, max_size=1333),
])
),
normalize,
])
if image_set == 'val':
return T.Compose([
T.RandomResize([800], max_size=1333),
normalize,
])
raise ValueError(f'unknown {image_set}')해당 함수들은 datasets/transforms.py에서 확인이 가능.
성명된 class들은 아래 function들의 간단한 wraper이기에 function 위주로 확인.
def crop(image, target, region):
# 이미지는 torchvision 내장 crop 함수 그대로 사용
cropped_image = F.crop(image, *region)
target = target.copy()
# i: The starting vertical coordinate (top) of the crop.
# j: The starting horizontal coordinate (left) of the crop.
# h: The target height of the crop.
# w: The target width of the crop.
i, j, h, w = region
# should we do something wrt the original size?
target["size"] = torch.tensor([h, w])
fields = ["labels", "area", "iscrowd"]
if "boxes" in target:
boxes = target["boxes"]
max_size = torch.as_tensor([w, h], dtype=torch.float32)
cropped_boxes = boxes - torch.as_tensor([j, i, j, i])
# croped_boxes의 포인트는 (0, 0) <= (tx, ty) <= (w, h)
# broadcasting 사용
cropped_boxes = torch.min(cropped_boxes.reshape(-1, 2, 2), max_size)
cropped_boxes = cropped_boxes.clamp(min=0)
area = (cropped_boxes[:, 1, :] - cropped_boxes[:, 0, :]).prod(dim=1)
# cropped boxes 다시 할당
target["boxes"] = cropped_boxes.reshape(-1, 4)
target["area"] = area
fields.append("boxes")
if "masks" in target:
# FIXME should we update the area here if there are no boxes?
target['masks'] = target['masks'][:, i:i + h, j:j + w]
fields.append("masks")
# remove elements for which the boxes or masks that have zero area
if "boxes" in target or "masks" in target:
# favor boxes selection when defining which elements to keep
# this is compatible with previous implementation
if "boxes" in target:
cropped_boxes = target['boxes'].reshape(-1, 2, 2)
keep = torch.all(cropped_boxes[:, 1, :] > cropped_boxes[:, 0, :], dim=1)
else:
keep = target['masks'].flatten(1).any(1)
for field in fields:
target[field] = target[field][keep]
return cropped_image, targetdef hflip(image, target):
flipped_image = F.hflip(image)
w, h = image.size
target = target.copy()
'''
1. 바운딩 박스 좌표를 [x_min, y_min, x_max, y_max]에서 [x_max, y_min, x_min, y_max]로
2. * torch.as_tensor([-1, 1, -1, 1]): x 좌표값에 -1을 곱하여 플립
3. + torch.as_tensor([w, 0, w, 0]): 이미지 너비 w를 더하여 반전된 x 좌표를 조정
'''
if "boxes" in target:
boxes = target["boxes"]
boxes = boxes[:, [2, 1, 0, 3]] * torch.as_tensor([-1, 1, -1, 1]) + torch.as_tensor([w, 0, w, 0])
target["boxes"] = boxes
if "masks" in target:
target['masks'] = target['masks'].flip(-1)
return flipped_image, target
def resize(image, target, size, max_size=None):
# size can be min_size (scalar) or (w, h) tuple
def get_size_with_aspect_ratio(image_size, size, max_size=None):
w, h = image_size
# max_size가 제공된 경우 크기가 조정된 이미지가 최대 크기를 초과하지 않도록 크기를 조정
if max_size is not None:
min_original_size = float(min((w, h)))
max_original_size = float(max((w, h)))
# 만약 큰 치수의 크기를 'size'로 조정하여 max_size을 초과한다면
# size'를 조정하여 최대 차원이 max_size가 되도록 함
if max_original_size / min_original_size * size > max_size:
size = int(round(max_size * min_original_size / max_original_size))
# 작은 차원이 size와 같다면 비율을 조절할 필요가 없기에 바로 반환
if (w <= h and w == size) or (h <= w and h == size):
return (h, w)
# w가 h보다 작은 경우 너비를 'size'로 조정하고 높이를 조정하여 가로 세로 비율을 유지
if w < h:
ow = size
oh = int(size * h / w)
# 반대 케이스
else:
oh = size
ow = int(size * w / h)
return (oh, ow)
def get_size(image_size, size, max_size=None):
# transforms 코드에서는 size가 scalar이기에 전 부분은 크게 의미 없음,
# 만약 입력이 Sequence이면 (w,h )를 기대히여 단순 reverse하여 (h, w) 반환
if isinstance(size, (list, tuple)):
return size[::-1]
else:
return get_size_with_aspect_ratio(image_size, size, max_size)
size = get_size(image.size, size, max_size)
rescaled_image = F.resize(image, size)
if target is None:
return rescaled_image, None
# 적용된 w, h의 변환 ratio 각각 계산
ratios = tuple(float(s) / float(s_orig) for s, s_orig in zip(rescaled_image.size, image.size))
ratio_width, ratio_height = ratios
target = target.copy()
# 원본 bbox에 비율 적용, resize된 bbox 할당
if "boxes" in target:
boxes = target["boxes"]
scaled_boxes = boxes * torch.as_tensor([ratio_width, ratio_height, ratio_width, ratio_height])
target["boxes"] = scaled_boxes
if "area" in target:
area = target["area"]
scaled_area = area * (ratio_width * ratio_height)
target["area"] = scaled_area
# resize 된 h,w 다시 할당
h, w = size
target["size"] = torch.tensor([h, w])
if "masks" in target:
target['masks'] = interpolate(
target['masks'][:, None].float(), size, mode="nearest")[:, 0] > 0.5
return rescaled_image, targetclass Normalize(object):
def __init__(self, mean, std):
self.mean = mean
self.std = std
def __call__(self, image, target=None):
image = F.normalize(image, mean=self.mean, std=self.std)
if target is None:
return image, None
target = target.copy()
h, w = image.shape[-2:]
if "boxes" in target:
boxes = target["boxes"]
boxes = box_xyxy_to_cxcywh(boxes)
boxes = boxes / torch.tensor([w, h, w, h], dtype=torch.float32)
target["boxes"] = boxes
return image, target 마지막으로 Normalize경우, bbox 부분을 xyxy format 에서 xywh fomat변경후, 0~1로 normalize 진행.
transforms.py의 다른 부분은 쉽게 이해가 될 것이라 생각되기에 skip하겠다.
원본 영상과 출력 결과는 다음과 같다.
보이듯이 bbox와 이미지가 같이 형태로 transforms된 것을 확인 할 수 있다.
이제 build_dataset(image_set, args=args)이 어떻게 작동되는 확인하였다.
이제 다시 main.py로 돌아가자.
if args.distributed:
sampler_train = DistributedSampler(dataset_train)
sampler_val = DistributedSampler(dataset_val, shuffle=False)
else:
sampler_train = torch.utils.data.RandomSampler(dataset_train)
sampler_val = torch.utils.data.SequentialSampler(dataset_val)
batch_sampler_train = torch.utils.data.BatchSampler(
sampler_train, args.batch_size, drop_last=True)
data_loader_train = DataLoader(dataset_train, batch_sampler=batch_sampler_train,
collate_fn=utils.collate_fn, num_workers=args.num_workers)
data_loader_val = DataLoader(dataset_val, args.batch_size, sampler=sampler_val,
drop_last=False, collate_fn=utils.collate_fn, num_workers=args.num_workers)해당 부분은 Dataloader부분으며 특이한점이 굳이 찾자면 sampler와 batch_sampler 객체를 따로 성명하여 DataLoader의 인자로 입력한다.
이 부분은 Dataloader의 인자, shuffle 와 batch_size을 입력한다면 같은 효과이다.
다음 DataLoader 코드의 일부분을 확인하자.
# (DataLoader 코드중)
...
if batch_size is not None and batch_sampler is None:
# auto_collation without custom batch_sampler
batch_sampler = BatchSampler(sampler, batch_size, drop_last)
...util.misc.py
그 다음으로 중요한 점은 datasets의 출력들을 하나의 NestedTensor로 묶는 utils.collate_fn 부분이다.
collate_fn 함수는 서로 다른 크기의 이미지와 그와 관련된 정보를 모델에 입력하기 위해 한 묶음으로 결합하는데 사용된다.
def collate_fn(batch):
batch = list(zip(*batch))
batch[0] = nested_tensor_from_tensor_list(batch[0])
return tuple(batch)DataLoader은 일반적으로 list의 각 요소들을 자동적으로 하나의 Tensor로 만들어준다.
하지만 우리의 dataset의 영상 부분인 img 는 각각 다른 크기의 영상 사이즈가 출력된다.
이러한 다른 사이즈 영상들을 자동적으로 하나의 Tensor로 만들 수 가 없다.
일반적으로 같은 사이즈로 resize하여 처리하는게 일반적이다.
단 그렇게 된다면 원본 영상 정보(ratio, 객체 사이즈)가 왜곡이 되는 문제가 발생된다.
nested_tensor_from_tensor_list은 객체 탐지와 같이 서로 다른 크기 이미지들을 하나의 Tensor로 만들기 위해 사용된다.
DataLoader의 collate_fn으로 해당 함수로 받는다.
이제 실제 코드에 대해서 알아보자.
batch = list(zip(*batch)): 이 코드 라인은 이미지와 target를 따로 가져오기 위해 배치의 샘플을 재정렬하여 두 개의 다른 리스트에 담는다.
이제 batch[0]에는 모든 이미지가 포함되고 batch[1]에는 모든 target이 포함된다.
nested_tensor_from_tensor_list 함수를 사용하여 이미지 list을 NestedTensor로 변환하여 크기가 다른 이미지를 하나로 묶는지 다음 코드에서 확인 할 수 있다.
def _max_by_axis(the_list):
# type: (List[List[int]]) -> List[int]
maxes = the_list[0]
for sublist in the_list[1:]:
for index, item in enumerate(sublist):
maxes[index] = max(maxes[index], item)
return maxes
def nested_tensor_from_tensor_list(tensor_list: List[Tensor]):
# TODO make this more general
if tensor_list[0].ndim == 3:
if torchvision._is_tracing():
# nested_tensor_from_tensor_list() does not export well to ONNX
# call _onnx_nested_tensor_from_tensor_list() instead
return _onnx_nested_tensor_from_tensor_list(tensor_list)
# TODO make it support different-sized images
max_size = _max_by_axis([list(img.shape) for img in tensor_list])
# min_size = tuple(min(s) for s in zip(*[img.shape for img in tensor_list]))
batch_shape = [len(tensor_list)] + max_size
b, c, h, w = batch_shape
dtype = tensor_list[0].dtype
device = tensor_list[0].device
tensor = torch.zeros(batch_shape, dtype=dtype, device=device)
mask = torch.ones((b, h, w), dtype=torch.bool, device=device)
for img, pad_img, m in zip(tensor_list, tensor, mask):
pad_img[: img.shape[0], : img.shape[1], : img.shape[2]].copy_(img)
m[: img.shape[1], :img.shape[2]] = False
else:
raise ValueError('not supported')
return NestedTensor(tensor, mask)
the_list은 일련의 Tensor의 입력 정확히는 Tensor.shape(일반적으로 (3, H, W))을 기대한다.
먼저 maxes을 리스트의 첫번째 원소로 초기화하고, 리스트의 각 원소를 방문하며 각 차원마다 가장 큰 값으로 할당하여 해당 batch의 각 차원별 가장 큰 차원을 획득한다.
상위 영상의 사이즈의 예시로 본다면 출력 결과는 (3, 894, 749)이다
Batch 사이즈을 앞에 추가하여 (2, 3, 894, 749)의 Tensor을 0으로 초기화한다.
또한 mask또한 0으로 초기화한다.
그리고 각 b 마다 영상의 복사하며 실제 영상이 어디인지 같은 위치의 mask 값을 1로 변환하여 마킹한다.
마지막으로 해당 tensor와 mask을 NestedTensor로 하나의 클래스로 묶는다.
NestedTensor은 기본적으로 데이터을 한데 묶는 tuple와 비슷하며 간단한 method가 추가되어 있다.
해당 method들은 간단하기에 쉽게 이해 할 수 있다고 판단된다.
class NestedTensor(object):
def __init__(self, tensors, mask: Optional[Tensor]):
self.tensors = tensors
self.mask = mask
def to(self, device):
# type: (Device) -> NestedTensor # noqa
cast_tensor = self.tensors.to(device)
mask = self.mask
if mask is not None:
assert mask is not None
cast_mask = mask.to(device)
else:
cast_mask = None
return NestedTensor(cast_tensor, cast_mask)
def decompose(self):
return self.tensors, self.mask
def __repr__(self):
return str(self.tensors)다음 코드로 loader에서 생성된 결과를 확인해보자.
from torch.utils.data import DataLoader
from torch.utils.data import RandomSampler, BatchSampler
sampler = RandomSampler(dataset_train)
batch_sampler = BatchSampler(sampler, args.batch_size, drop_last=True)
data_loader = DataLoader(dataset_train, batch_sampler=batch_sampler, collate_fn=utils.collate_fn, num_workers=args.num_workers)
for X, y in data_loader:
breakX의 타입은 NestedTensor이며, 크기는 다음과 같다.
X의 각 batch의 영상과 mask을 출력하여 확인해보자.
보이듯이 비록 두 영상의 크기는 다르지만 같은 Tensor에 할당되어 있으며, 영상의 무효한 부분은 mask에서 확인 할 수 있다.
이제 main.py 의 optimizer, scheduler 생성 부분을 확인해보자.
model, criterion, postprocessors 구성은 이후 모델의 forward 부분에서 같이 설명하겠다.
model, criterion, postprocessors = build_model(args)
model.to(device)
model_without_ddp = model
if args.distributed:
model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[args.gpu])
model_without_ddp = model.module
n_parameters = sum(p.numel() for p in model.parameters() if p.requires_grad)
print('number of params:', n_parameters)
param_dicts = [
{"params": [p for n, p in model_without_ddp.named_parameters() if "backbone" not in n and p.requires_grad]},
{
"params": [p for n, p in model_without_ddp.named_parameters() if "backbone" in n and p.requires_grad],
"lr": args.lr_backbone,
},
]
optimizer = torch.optim.AdamW(param_dicts, lr=args.lr,
weight_decay=args.weight_decay)
lr_scheduler = torch.optim.lr_scheduler.StepLR(optimizer, args.lr_drop)DETR은 backbone과 transformer 두 파트로 나누어진다. 그리고 이 둘의 학습 파라미터를 다르게 가져간다. Optimizer는 매개변수별 옵션 지정도 지원한다.
Variables의 이터러블을 전달하는 대신 dicts의 이터러블을 전달하면 된다.
좀 더 정확한 설명은 공식 docs을 참고 바란다.
scheduler은 StepLR로 설정하였다. 단 요즘에는 CosineAnnealingLR 더 나아가, transformers.get_cosine_schedule_with_warmup가 일반적으로 더 좋은 방법이다.
engine.py
panoptic segmentation, resume 및 evaluate 부분은 우리의 관심이 아니기에 skip.
이제 epoch안에 train_one_epoch을 확인하겠다.
print("Start training")
start_time = time.time()
for epoch in range(args.start_epoch, args.epochs):
if args.distributed:
sampler_train.set_epoch(epoch)
train_stats = train_one_epoch(
model, criterion, data_loader_train, optimizer, device, epoch,
args.clip_max_norm)
lr_scheduler.step()def train_one_epoch(model: torch.nn.Module, criterion: torch.nn.Module,
data_loader: Iterable, optimizer: torch.optim.Optimizer,
device: torch.device, epoch: int, max_norm: float = 0):
model.train()
criterion.train()
metric_logger = utils.MetricLogger(delimiter=" ")
metric_logger.add_meter('lr', utils.SmoothedValue(window_size=1, fmt='{value:.6f}'))
metric_logger.add_meter('class_error', utils.SmoothedValue(window_size=1, fmt='{value:.2f}'))
header = 'Epoch: [{}]'.format(epoch)
print_freq = 10
for samples, targets in metric_logger.log_every(data_loader, print_freq, header):
samples = samples.to(device)
targets = [{k: v.to(device) for k, v in t.items()} for t in targets]
outputs = model(samples)
loss_dict = criterion(outputs, targets)
weight_dict = criterion.weight_dict
losses = sum(loss_dict[k] * weight_dict[k] for k in loss_dict.keys() if k in weight_dict)
# reduce losses over all GPUs for logging purposes
loss_dict_reduced = utils.reduce_dict(loss_dict)
loss_dict_reduced_unscaled = {f'{k}_unscaled': v
for k, v in loss_dict_reduced.items()}
loss_dict_reduced_scaled = {k: v * weight_dict[k]
for k, v in loss_dict_reduced.items() if k in weight_dict}
losses_reduced_scaled = sum(loss_dict_reduced_scaled.values())
loss_value = losses_reduced_scaled.item()
if not math.isfinite(loss_value):
print("Loss is {}, stopping training".format(loss_value))
print(loss_dict_reduced)
sys.exit(1)
optimizer.zero_grad()
losses.backward()
if max_norm > 0:
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm)
optimizer.step()
metric_logger.update(loss=loss_value, **loss_dict_reduced_scaled, **loss_dict_reduced_unscaled)
metric_logger.update(class_error=loss_dict_reduced['class_error'])
metric_logger.update(lr=optimizer.param_groups[0]["lr"])
# gather the stats from all processes
metric_logger.synchronize_between_processes()
print("Averaged stats:", metric_logger)
return {k: meter.global_avg for k, meter in metric_logger.meters.items()}상위 코드에서 metric_logger 외 거의 대부분의 코드는 일반적은 PyTorch의 일반적인 학습 코드와 같다.
물론 metric_logger은 잘만든 로깅이나 DETR의 핵심과 크게 연관이 없다.
다른 TorchMetric 같은 라이브러리를 통해 로깅을 할 수 있으니, 일단 스킵하겠다(내가 아직 이 부분을 제대로 보지 않았다...).
for samples, targets in metric_logger.log_every(data_loader, print_freq, header):
samples = samples.to(device)
targets = [{k: v.to(device) for k, v in t.items()} for t in targets]
outputs = model(samples)
loss_dict = criterion(outputs, targets)
...
optimizer.zero_grad()
losses.backward()결국 상위 부분이 중요하고 핵심은 model(samples) 및 criterion(outputs, targets)이다.
model.py
이제 model이 어떻게 구성 되는 확인하자.
def build(args):
# the `num_classes` naming here is somewhat misleading.
# it indeed corresponds to `max_obj_id + 1`, where max_obj_id
# is the maximum id for a class in your dataset. For example,
# COCO has a max_obj_id of 90, so we pass `num_classes` to be 91.
# As another example, for a dataset that has a single class with id 1,
# you should pass `num_classes` to be 2 (max_obj_id + 1).
# For more details on this, check the following discussion
# https://github.com/facebookresearch/detr/issues/108#issuecomment-650269223
# detection은 num_classes = 91 이다
num_classes = 20 if args.dataset_file != 'coco' else 91
if args.dataset_file == "coco_panoptic":
# for panoptic, we just add a num_classes that is large enough to hold
# max_obj_id + 1, but the exact value doesn't really matter
num_classes = 250
device = torch.device(args.device)
# backbone 은 resnet51
backbone = build_backbone(args)
# transfomer 빌드
transformer = build_transformer(args)
# 생성된 backbone 과 tansformer로 DETR 구성
model = DETR(
backbone,
transformer,
num_classes=num_classes,
num_queries=args.num_queries,
aux_loss=args.aux_loss,
)
...
matcher = build_matcher(args)
weight_dict = {'loss_ce': 1, 'loss_bbox': args.bbox_loss_coef}
weight_dict['loss_giou'] = args.giou_loss_coef
...
# TODO this is a hack
if args.aux_loss:
aux_weight_dict = {}
for i in range(args.dec_layers - 1):
aux_weight_dict.update({k + f'_{i}': v for k, v in weight_dict.items()})
weight_dict.update(aux_weight_dict)
losses = ['labels', 'boxes', 'cardinality']
...
criterion = SetCriterion(num_classes, matcher=matcher, weight_dict=weight_dict,
eos_coef=args.eos_coef, losses=losses)
criterion.to(device)
postprocessors = {'bbox': PostProcess()}
...
return model, criterion, postprocessors보이듯이 모델의 중요 부분은 backbone과 transformer을 빌드하고
DETR의 인자로 입력한다.
먼저 가장 큰 부분인 DETR을 화보자.
detr.py
class DETR(nn.Module):
""" This is the DETR module that performs object detection """
def __init__(self, backbone, transformer, num_classes, num_queries, aux_loss=False):
""" Initializes the model.
Parameters:
backbone: torch module of the backbone to be used. See backbone.py
transformer: torch module of the transformer architecture. See transformer.py
num_classes: number of object classes
num_queries: number of object queries, ie detection slot. This is the maximal number of objects
DETR can detect in a single image. For COCO, we recommend 100 queries.
aux_loss: True if auxiliary decoding losses (loss at each decoder layer) are to be used.
"""
super().__init__()
self.num_queries = num_queries
self.transformer = transformer
hidden_dim = transformer.d_model
self.class_embed = nn.Linear(hidden_dim, num_classes + 1)
# 3층 hiddim layer을 가진 단순한 MLP이다.
self.bbox_embed = MLP(hidden_dim, hidden_dim, 4, 3)
self.query_embed = nn.Embedding(num_queries, hidden_dim)
self.input_proj = nn.Conv2d(backbone.num_channels, hidden_dim, kernel_size=1)
self.backbone = backbone
self.aux_loss = aux_loss
def forward(self, samples: NestedTensor):
""" The forward expects a NestedTensor, which consists of:
- samples.tensor: batched images, of shape [batch_size x 3 x H x W]
- samples.mask: a binary mask of shape [batch_size x H x W], containing 1 on padded pixels
It returns a dict with the following elements:
- "pred_logits": the classification logits (including no-object) for all queries.
Shape= [batch_size x num_queries x (num_classes + 1)]
- "pred_boxes": The normalized boxes coordinates for all queries, represented as
(center_x, center_y, height, width). These values are normalized in [0, 1],
relative to the size of each individual image (disregarding possible padding).
See PostProcess for information on how to retrieve the unnormalized bounding box.
- "aux_outputs": Optional, only returned when auxilary losses are activated. It is a list of
dictionnaries containing the two above keys for each decoder layer.
"""
# 입력이 NestedTensor가 아니라면 해당 타입으로 전환
if isinstance(samples, (list, torch.Tensor)):
samples = nested_tensor_from_tensor_list(samples)
# backbone으로 CNN feature 및 position embedding 출력
features, pos = self.backbone(samples)
# features의 출력은 NestedTensor의 list이다. 그리고 detection에서는 마지막 요소가 backbone 마지막 layer의 출력이다.
src, mask = features[-1].decompose()
assert mask is not None
# 2048 체널을 512로 변환하고 transformer 모델에 입력으로 사용
hs = self.transformer(self.input_proj(src), mask, self.query_embed.weight, pos[-1])[0]
# [6, 100, b, hidden_features]의 출력을 simple FFN으로 원하는 출력 즉[6, 100, b, 92], [100, b, 4]로 만듬.
# 6개인 이유는 decoder이 6개 decoder layer의 출력을 concat
outputs_class = self.class_embed(hs)
outputs_coord = self.bbox_embed(hs).sigmoid()
# 마지막이 마지막 레이어, 즉 DETR의 마지막 결과이기에 저장
out = {'pred_logits': outputs_class[-1], 'pred_boxes': outputs_coord[-1]}
if self.aux_loss:
out['aux_outputs'] = self._set_aux_loss(outputs_class, outputs_coord)
return out
@torch.jit.unused
def _set_aux_loss(self, outputs_class, outputs_coord):
# this is a workaround to make torchscript happy, as torchscript
# doesn't support dictionary with non-homogeneous values, such
# as a dict having both a Tensor and a list.
# zip으로 로짓과 bbox 좌표 짝을 짓는다
# 그렇게 구성된 dicts을 리스트로 구성
return [{'pred_logits': a, 'pred_boxes': b}
for a, b in zip(outputs_class[:-1], outputs_coord[:-1])]DETR 클래스는 backbone 네트워크, transformer 모듈, classification 및 regression를 위한 header 포함하는 object detection 모델을 정의한다.
상위 코드의 흐름은 다음 그림과 같다.
이제 중요한 구성 요소 backbone 및 transformer에 대애서 알아보자.
backbone.py
build_backbone은 다음과 같다.
def build_backbone(args):
position_embedding = build_position_encoding(args)
train_backbone = args.lr_backbone > 0 # 일반적으로 True
return_interm_layers = args.masks # False
backbone = Backbone(args.backbone, train_backbone, return_interm_layers, args.dilation)
model = Joiner(backbone, position_embedding)
model.num_channels = backbone.num_channels
return model중요한 구성 요소는 Backbone 및 Joiner에 대해서 알아보자.
class BackboneBase(nn.Module):
def __init__(self, backbone: nn.Module, train_backbone: bool, num_channels: int, return_interm_layers: bool):
super().__init__()
for name, parameter in backbone.named_parameters():
# train_backbone이 False이거나 backbone에 'layer2', 'layer3', 'layer4'가 없는 경우, gradient를 False하여 동결
if not train_backbone or 'layer2' not in name and 'layer3' not in name and 'layer4' not in name:
parameter.requires_grad_(False)
# segmentation 일 경우, 다양한 크기의 feature 추출
if return_interm_layers:
return_layers = {"layer1": "0", "layer2": "1", "layer3": "2", "layer4": "3"}
else:
return_layers = {'layer4': "0"}
# IntermediateLayerGetter는 원래 모델에서 특정 이름의 feature을 추출
# detection은 마지막 layer인 layer 4의 feature 추출
# resnet은 일반적으로 layer 1, 2, 3, 4로 큰 블럭 구성
self.body = IntermediateLayerGetter(backbone, return_layers=return_layers)
self.num_channels = num_channels
def forward(self, tensor_list: NestedTensor):
# nn.Module의 입력은 Tensor을 기대하기때문에 NestedTensor의 tensors사용
xs = self.body(tensor_list.tensors)
# 출력 xs는 dict형태로 구성되어 있으며 결과 형태는 {`0`: values0, '1': values1, ... } 이다.
out: Dict[str, NestedTensor] = {}
for name, x in xs.items():
m = tensor_list.mask
assert m is not None
# seems mask.shape is [b, h, w], so add new axis for batch processing.
# 앞서 영상중에 무효한 부분이 존재하기에 mask를 출력 사이즈로 resize하여 대응되는 feature의 무효 부분 마킹,
# 그리고 다시 NestedTensor로 성명한다, 출력된 mask의 크기는 다음 그림 참조
mask = F.interpolate(m[None].float(), size=x.shape[-2:]).to(torch.bool)[0]
out[name] = NestedTensor(x, mask)
return out
class Backbone(BackboneBase):
"""ResNet backbone with frozen BatchNorm."""
def __init__(self, name: str,
train_backbone: bool,
return_interm_layers: bool,
dilation: bool):
backbone = getattr(torchvision.models, name)(
replace_stride_with_dilation=[False, False, dilation],
pretrained=is_main_process(), norm_layer=FrozenBatchNorm2d)
num_channels = 512 if name in ('resnet18', 'resnet34') else 2048 # seems we don't use it...
# Backbone은 BaclboneBase에서 상속 받으며, 차이점은 torchvision.models의 일반적인 resnet을 사용한다
super().__init__(backbone, train_backbone, num_channels, return_interm_layers)출력된 resize된 마스크의 크기는 다음과 같다.
Backbone에서 backbone = getattr(torchvision.models, name)( replace_stride_with_dilation=[False, False, dilation], pretrained=is_main_process(), norm_layer=FrozenBatchNorm2d)코드중 FrozenBatchNorm2d이라는 특이한 custom norm_layer로 기존 batch norm을 대치한다.
먼저 FrozenBatchNorm2d을 확인하고 그 이유를 알아보자.
class FrozenBatchNorm2d(torch.nn.Module):
"""
BatchNorm2d where the batch statistics and the affine parameters are fixed.
Copy-paste from torchvision.misc.ops with added eps before rqsrt,
without which any other models than torchvision.models.resnet[18,34,50,101]
produce nans.
"""
def __init__(self, n):
super(FrozenBatchNorm2d, self).__init__()
self.register_buffer("weight", torch.ones(n))
self.register_buffer("bias", torch.zeros(n))
self.register_buffer("running_mean", torch.zeros(n))
self.register_buffer("running_var", torch.ones(n))
def _load_from_state_dict(self, state_dict, prefix, local_metadata, strict,
missing_keys, unexpected_keys, error_msgs):
num_batches_tracked_key = prefix + 'num_batches_tracked'
if num_batches_tracked_key in state_dict:
del state_dict[num_batches_tracked_key]
super(FrozenBatchNorm2d, self)._load_from_state_dict(
state_dict, prefix, local_metadata, strict,
missing_keys, unexpected_keys, error_msgs)
def forward(self, x):
# move reshapes to the beginning
# to make it fuser-friendly
w = self.weight.reshape(1, -1, 1, 1)
b = self.bias.reshape(1, -1, 1, 1)
rv = self.running_var.reshape(1, -1, 1, 1)
rm = self.running_mean.reshape(1, -1, 1, 1)
eps = 1e-5
scale = w * (rv + eps).rsqrt()
bias = b - rm * scale
return x * scale + bias해당 FrozenBatchNorm2d은 기존 batchnorm과의 차이점은 무엇일까?
일반 PyTorch의 Norm의 간단하게 확인해보자.
보이듯이 일반 Norm에서 weight와 bias는 Parameter이기에 학습중 이 값들은 실제로 계속해서 변환한다.
하지만 FrozenBatchNorm2d에서 해당 값을을 로딩하고 register_buffer로 학습하지 값으로 살수로 고정한다.
custom layer 사용 여부의 모델의 batch norm의 학습 파라미터가 없다는 걸 다음 코드에서 알 수 있다.
이렇게 한다면 클래스 이름 그대로 Frozen이 되어 기존 이미지넷의 정규화 상수 값 그대로 사용하되 갱신되지 않는다.
그렇다면 왜 이 값을 고정할까? 그 이유는 object detection의 실제 학습 batch 크기가 작기때문이다.
args.batch_size=2이기에 각 gpu당 배치 크기는 상당히 작다.
이로 인해 batch의 통계 값이 큰 batch 대비 불안정하기에 정규화의 의미가 퇘색된다.
따라서 backbone 부분의 batch 정교화는 갱신하지 않는게 오히려 좋은 선택이기에 해당 부분을 Parameter에서 register_buffer로 바꾸어 상수로 사용한다.
이제 남은 Joiner을 확인해보자. 해당 class는 NestedTensor용 Sequential이며 간단하다.
용도는 backbone에서 출력된 결과의 공간 정보의 position_encoding을 진행하여 같이 출력하는 역활이다.
class Joiner(nn.Sequential):
def __init__(self, backbone, position_embedding):
super().__init__(backbone, position_embedding)
def forward(self, tensor_list: NestedTensor):
xs = self[0](tensor_list)
out: List[NestedTensor] = []
pos = []
for name, x in xs.items():
out.append(x)
# position encoding
pos.append(self[1](x).to(x.tensors.dtype))
return out, posposition_embedding.py
이제 position_embedding에 대해서 알아보자.
position_embedding은 sine과 learned 두 방식있으나 우리는 기본 세팅인 sin 형태를 알아본다.
attention is all you need에 나온 position encoding은 다음 그림과 같다.
DETR에서는 데이터는 공간 영상이기에 2차원으로 확장하여 변화를 준다.
기존 세팅의 hidden_dim은 256이다. 그중 앞 128은 x축의 position 정보, 뒷 부분은 y축position 정보를 추가한다.
즉 코드에서 상위 그림을 각각 x, y축으로 hidden_dim //2 만큼만 진행한다.
그리고 아래 코드중 pos = torch.cat((pos_y, pos_x), dim=3).permute(0, 3, 1, 2)에서 두 축의 정보를 나누어 hidden_dim 형태로 구성한다.
def build_position_encoding(args):
N_steps = args.hidden_dim // 2 # hidden_dim = 256이기에 N_step = 128
if args.position_embedding in ('v2', 'sine'):
# TODO find a better way of exposing other arguments
position_embedding = PositionEmbeddingSine(N_steps, normalize=True)
elif args.position_embedding in ('v3', 'learned'):
position_embedding = PositionEmbeddingLearned(N_steps)
else:
raise ValueError(f"not supported {args.position_embedding}")
return position_embedding
class PositionEmbeddingSine(nn.Module):
"""
This is a more standard version of the position embedding, very similar to the one
used by the Attention is all you need paper, generalized to work on images.
"""
def __init__(self, num_pos_feats=64, temperature=10000, normalize=False, scale=None):
super().__init__()
self.num_pos_feats = num_pos_feats
self.temperature = temperature
# 기본 세팅은 True이다 즉 영상의 w, h을 각각 2 pi로 정규화하여 영상 크기 대비 상대적인 차이로 정의한다.
self.normalize = normalize
if scale is not None and normalize is False:
raise ValueError("normalize should be True if scale is passed")
if scale is None:
scale = 2 * math.pi
self.scale = scale
def forward(self, tensor_list: NestedTensor):
x = tensor_list.tensors
mask = tensor_list.mask
assert mask is not None
not_mask = ~mask # 영상 영역을 1로
# 각 축으로 누적하여 유효 영역 인덱스 구함, 마스크인 부분은 0이기에 최대 누적값 그대로 계승
y_embed = not_mask.cumsum(1, dtype=torch.float32)
x_embed = not_mask.cumsum(2, dtype=torch.float32)
# 정규화시, 가장 마지막 인덱스(마지막은 최대 누적 값)(즉 범위를 0~1로 정구화)로 나누고 2 * pi 을 곱함
if self.normalize:
eps = 1e-6
y_embed = y_embed / (y_embed[:, -1:, :] + eps) * self.scale
x_embed = x_embed / (x_embed[:, :, -1:] + eps) * self.scale
# 원래 공식의 i을 차원별로 계산
dim_t = torch.arange(self.num_pos_feats, dtype=torch.float32, device=x.device)
# 공식 적용 sin과 cos 교차하기에 2 * dim_t // 2로 2개씩 같은 값으로
dim_t = self.temperature ** (2 * (dim_t // 2) / self.num_pos_feats)
# 브로드 캐스팅을 위해 마지막에 새로운 dimension 추가
# ex: [2, 21, 27] -> [2, 21, 27, 1]
# [2, 21, 27, 1] / [128] -> [2, 21, 27, 128]
pos_x = x_embed[:, :, :, None] / dim_t
pos_y = y_embed[:, :, :, None] / dim_t
# sin, cos을 각각 계산하여 새로운 dimension stack하고 다시 flatten
# [2, 21, 27, 64, 2] -> [2, 21, 27, 128]
pos_x = torch.stack((pos_x[:, :, :, 0::2].sin(), pos_x[:, :, :, 1::2].cos()), dim=4).flatten(3)
pos_y = torch.stack((pos_y[:, :, :, 0::2].sin(), pos_y[:, :, :, 1::2].cos()), dim=4).flatten(3)
# x, y축 결합하고 transformer 입력 행태로 전환
# [2, 21, 27, 128] * 2 -> [2, 256, 21, 27]
pos = torch.cat((pos_y, pos_x), dim=3).permute(0, 3, 1, 2)
return postransformer.py
이렇게 얻게된 src(features)와 pos를 이제 transformer의 입력한다.
앞서 설명했듯이 src와 pos의 리스트이고 마지막 요소만 사용한다.
features, pos = self.backbone(samples)
src, mask = features[-1].decompose()
assert mask is not None
hs = self.transformer(self.input_proj(src), mask, self.query_embed.weight, pos[-1])[0]이제 transformer의 코드를 아래 그림과 main.py부분에 transformer 부분의 변수를 함께 확인해보자.
class Transformer(nn.Module):
def __init__(self, d_model=512, nhead=8, num_encoder_layers=6,
num_decoder_layers=6, dim_feedforward=2048, dropout=0.1,
activation="relu", normalize_before=False,
return_intermediate_dec=False):
super().__init__()
encoder_layer = TransformerEncoderLayer(d_model, nhead, dim_feedforward,
dropout, activation, normalize_before)
encoder_norm = nn.LayerNorm(d_model) if normalize_before else None
self.encoder = TransformerEncoder(encoder_layer, num_encoder_layers, encoder_norm)
decoder_layer = TransformerDecoderLayer(d_model, nhead, dim_feedforward,
dropout, activation, normalize_before)
decoder_norm = nn.LayerNorm(d_model)
self.decoder = TransformerDecoder(decoder_layer, num_decoder_layers, decoder_norm,
return_intermediate=return_intermediate_dec)
self._reset_parameters()
self.d_model = d_model
self.nhead = nhead
def _reset_parameters(self):
for p in self.parameters():
if p.dim() > 1:
nn.init.xavier_uniform_(p)
def forward(self, src, mask, query_embed, pos_embed):
# flatten NxCxHxW to HWxNxC
bs, c, h, w = src.shape
src = src.flatten(2).permute(2, 0, 1) # [bs, c, h, w] -> [h * w, bs, c]
pos_embed = pos_embed.flatten(2).permute(2, 0, 1) # same
query_embed = query_embed.unsqueeze(1).repeat(1, bs, 1) # [100, c] -> [100, bs, c] repeat query embed for batch
mask = mask.flatten(1) # [bs, h, w] -> [bs, h * w]
tgt = torch.zeros_like(query_embed) # 첫번째 입력은 제로이지만 실제로는 query_embed 정보가 같이 추가되어 들어감, 논문 참조
memory = self.encoder(src, src_key_padding_mask=mask, pos=pos_embed)
hs = self.decoder(tgt, memory, memory_key_padding_mask=mask,
pos=pos_embed, query_pos=query_embed)
return hs.transpose(1, 2), memory.permute(1, 2, 0).view(bs, c, h, w)보이듯이 코드는 상위 그림을 충실히 따르고 있다. 추가적인 디테일이 있다면 mask부분일 것이다.
앞서 설명했듯이, 영상의 사이즈를 그대로 유지하되 같은 Tensor로 처리하기 위해 유효하지 않은 영역이 포함되고 있다.
Backbone을 처리된 feature역시 resize된 mask로 유요하지 않는 영역을 표기한다.
Transformer의 multi-head attention은 각 token(feature 포인트)마다 내적 유사도 계산을 진행한다.(이부분 추후 Layer안에서 더 정밀하게 알아보겠다.)
하지만 이러한 불필요한 영역을 attention 값을 -inf로 바꾸어 softmax의 값을 0이 되게 한다.
이러한 방법으로 불필요한 feature들이 반영이 안되게 처리 할 수 있다.
이 부분을 각 TransformerEncoderLayer 및 TransformerDecoderLayer에서 확인 할 것이다.
먼저 TransformerEncoderLayer을 확인해보자.
class TransformerEncoderLayer(nn.Module):
def __init__(self, d_model, nhead, dim_feedforward=2048, dropout=0.1,
activation="relu", normalize_before=False):
super().__init__()
self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout)
# Implementation of Feedforward model
self.linear1 = nn.Linear(d_model, dim_feedforward)
self.dropout = nn.Dropout(dropout)
self.linear2 = nn.Linear(dim_feedforward, d_model)
self.norm1 = nn.LayerNorm(d_model)
self.norm2 = nn.LayerNorm(d_model)
self.dropout1 = nn.Dropout(dropout)
self.dropout2 = nn.Dropout(dropout)
self.activation = _get_activation_fn(activation)
self.normalize_before = normalize_before
def with_pos_embed(self, tensor, pos: Optional[Tensor]):
return tensor if pos is None else tensor + pos
def forward_post(self,
src,
src_mask: Optional[Tensor] = None,
src_key_padding_mask: Optional[Tensor] = None,
pos: Optional[Tensor] = None):
q = k = self.with_pos_embed(src, pos)
src2 = self.self_attn(q, k, value=src, attn_mask=src_mask,
key_padding_mask=src_key_padding_mask)[0]
src = src + self.dropout1(src2)
src = self.norm1(src)
src2 = self.linear2(self.dropout(self.activation(self.linear1(src))))
src = src + self.dropout2(src2)
src = self.norm2(src)
return src
def forward_pre(self, src,
src_mask: Optional[Tensor] = None,
src_key_padding_mask: Optional[Tensor] = None,
pos: Optional[Tensor] = None):
src2 = self.norm1(src)
q = k = self.with_pos_embed(src2, pos)
src2 = self.self_attn(q, k, value=src2, attn_mask=src_mask,
key_padding_mask=src_key_padding_mask)[0]
src = src + self.dropout1(src2)
src2 = self.norm2(src)
src2 = self.linear2(self.dropout(self.activation(self.linear1(src2))))
src = src + self.dropout2(src2)
return src
def forward(self, src,
src_mask: Optional[Tensor] = None,
src_key_padding_mask: Optional[Tensor] = None,
pos: Optional[Tensor] = None):
if self.normalize_before:
return self.forward_pre(src, src_mask, src_key_padding_mask, pos)
return self.forward_post(src, src_mask, src_key_padding_mask, pos)해당 구조는 standard한 구조이다. Linear, Dropout, LayerNorm들은 익숙한 클래스일 것이다.
단 약간의 부연 설명이 필요한 부분은 아마 MultiheadAttention일 것이다.
이 부분이 바로 attention을 계산하지 부분이다. 해당 작동 코드는 고도의 최적화가 되어있어 read 자체가 생각보다 하드코어하다...
물론 해당 논문을 보고 있는 reader라면 attention 연산의 어떻게 작동되는지 다들 알고 있을 것이라 믿고 있지만 복기 차원에서 간단한 구한을 코드를 만들어 놓았다.
class MyMultiHeadAttention(nn.Module):
def __init__(self, embed_dim, num_heads):
super().__init__()
self.embed_dim = embed_dim
self.num_heads = num_heads
self.head_dim = embed_dim // num_heads
self.qkv_proj = nn.Linear(embed_dim, 3 * embed_dim)
self.out_proj = nn.Linear(embed_dim, embed_dim)
self.scale = 1.0 / (self.head_dim ** 0.5)
def forward(self, input, attention_mask=None):
B, N, C = input.shape
qkv = self.qkv_proj(input)
qkv = qkv.reshape(B, N, 3, self.num_heads, self.head_dim)
qkv = qkv.permute(2, 0, 3, 1, 4)
q, k, v = qkv.unbind(dim=0)
attn = (q @ k.transpose(-2, -1)) * self.scale
if attention_mask is not None:
attn = attn.masked_fill(attention_mask, float('-inf'))
attn = attn.softmax(dim=-1)
output = attn @ v
output = output.transpose(1, 2).reshape(B, N, C)
output = self.out_proj(output)
return output일단은 공홈 또는 클래스 docstring에 기반하에 해당 class의 입력과 출력에 대해서 알아보자.
실제 구현중 코드중 중요한 부분은 아래 두 부분이다.
self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout=dropout),
src2 = self.self_attn(q, k, value=src2, attn_mask=src_mask, key_padding_mask=src_key_padding_mask)[0]
성명 부분은 크게 어렵지 않을 것이지만, 문제는 후자 forward 입력 부분이다.
이에 대응되는 interface 설명은 다음 그림과 같다.
여기서 개인적으로 모호한 부분은 두 mask 인자이다.
설명에서 보이듯이 key_padding_mask의 용도는 대부분 padding에 대한것이다.
이는 자연어에서 온 개념이지안 DETR의 NestedTensor에서도 확장이 가능하다.
먼저 자연어를 예시로 생각하자.
hello my name is wood spoon 과 I am a software developer 이라는 두 문장을 batch 로 묶어 처리해야 한다면
같은 크기의 Tensor로 만들어야한다.
보이듯이 두번째 문장은 마지막 토큰 부분은 0으로 padding하고 해당 부분이 무효한 것을 inputs['attention_mask']으로 알 수 있다.
입력에 해당 부분을 key_padding_mask의 입력으로 넣어 무효한 부분을 -inf으로 채워 attention score가 0이 되로록하는 것이다.
마찬가지로 NestedTensor에세도 영상 무효한 부분을 mask을 보관하고 있다. 이 부분을 MultiAttentionHead의 key_padding_mask의 입력으로 사용한다.
실제 batch 영상 Tensor의 mask는 다음 그림과 같다.
그맇다면 att_mask 은 무엇을까?
이 부분은 autoregressive model에서 사용되는 개념이다.
트랜스포머 디코더에서 삼각형 마스크는 추론 시간을 시뮬레이션하고 "future" 위치에 대한 attention을 방지하기 위해 사용된다. 이것이 일반적으로 att_mask가 사용되는 용도입니다.
이는 다음 그림으로 쉽게 이해 할 수 있다.
(출처: https://peterbloem.nl/blog/transformers)
이둘을 구별함으로써 query와 key의 다른 masking을 적용하여 앞선 두가지 용도를 분리하여 관리 할 수 있다.
하지만 DETR은 autoregressive model가 아니다. 즉 모든 queries가 동시에 병렬적으로 처리되어 결과는 얻어내는 기법이기에 attn_mask 인자 사용이 구현이 되어 있으나 실제로는 사용되지 않는다.
이는 아래 Transformer 클래스 forward 부분 코드에서 확인 할 수 있다.
memory = self.encoder(src, src_key_padding_mask=mask, pos=pos_embed),
hs = self.decoder(tgt, memory, memory_key_padding_mask=mask, pos=pos_embed, query_pos=query_embed)
보이듯이, key_padding_mask만 사용하고 있다.
이제 TransformerDecoder부분을 확인해보자.
앞선 설명과 같이 마스킹은 오로지 memory_key_padding_mask만 사용하고 있다.
Encoder 대비 차이점은
1. target이 queries,
2. self- 및 cross-attention 포함,
3. 매 decoder layer hidden feature을 저장 및 stack하여 하나의 Tensor([6, 100, 2, 256])로 만든다는 것이다.
class TransformerDecoder(nn.Module):
def __init__(self, decoder_layer, num_layers, norm=None, return_intermediate=False):
super().__init__()
self.layers = _get_clones(decoder_layer, num_layers)
self.num_layers = num_layers
self.norm = norm
self.return_intermediate = return_intermediate
def forward(self, tgt, memory,
tgt_mask: Optional[Tensor] = None,
memory_mask: Optional[Tensor] = None,
tgt_key_padding_mask: Optional[Tensor] = None,
memory_key_padding_mask: Optional[Tensor] = None,
pos: Optional[Tensor] = None,
query_pos: Optional[Tensor] = None):
output = tgt
intermediate = []
for layer in self.layers:
output = layer(output, memory, tgt_mask=tgt_mask,
memory_mask=memory_mask,
tgt_key_padding_mask=tgt_key_padding_mask,
memory_key_padding_mask=memory_key_padding_mask,
pos=pos, query_pos=query_pos)
if self.return_intermediate:
intermediate.append(self.norm(output))
if self.norm is not None:
output = self.norm(output)
if self.return_intermediate:
intermediate.pop()
intermediate.append(output)
if self.return_intermediate:
return torch.stack(intermediate)
return output.unsqueeze(0)그리고 이렇게 나온 hs을 분류 및 bbox 회귀 head에 각각 입력으로 사용하여 우리가 원하는 Tensor 형태로 만들어 준다.
마지막 layer의 출력을 따로 'pred_logits' 및 'pred_boxes'로 출력하고 'aux_outputs'는 zip연산과 list로 각 layer의 결과 매칭하고 리스트의 원소로 출력한다.
outputs_class = self.class_embed(hs)
outputs_coord = self.bbox_embed(hs).sigmoid()
out = {'pred_logits': outputs_class[-1], 'pred_boxes': outputs_coord[-1]}
if self.aux_loss:
out['aux_outputs'] = self._set_aux_loss(outputs_class, outputs_coord)
return out
@torch.jit.unused
def _set_aux_loss(self, outputs_class, outputs_coord):
# this is a workaround to make torchscript happy, as torchscript
# doesn't support dictionary with non-homogeneous values, such
# as a dict having both a Tensor and a list.
return [{'pred_logits': a, 'pred_boxes': b}
for a, b in zip(outputs_class[:-1], outputs_coord[:-1])]자 이로써, DETR의 모델은 코드는 설명이 끝났으며, 이제 헝가리안 Loss을 확인해보자.
Criterion
손실함수는 SetCriterion에 정의 되어있다.
해당 클래스는 nn.Module을 상속 받기에 가장 중요한 init과 forward을 먼저 이해하고 다른 method들은 필요할때 마다 확인하자.
class SetCriterion(nn.Module):
""" This class computes the loss for DETR.
The process happens in two steps:
1) we compute hungarian assignment between ground truth boxes and the outputs of the model
2) we supervise each pair of matched ground-truth / prediction (supervise class and box)
"""
def __init__(self, num_classes, matcher, weight_dict, eos_coef, losses):
""" Create the criterion.
Parameters:
num_classes: number of object categories, omitting the special no-object category
matcher: module able to compute a matching between targets and proposals
weight_dict: dict containing as key the names of the losses and as values their relative weight.
eos_coef: relative classification weight applied to the no-object category
losses: list of all the losses to be applied. See get_loss for list of available losses.
"""
super().__init__()
self.num_classes = num_classes
self.matcher = matcher
self.weight_dict = weight_dict
self.eos_coef = eos_coef
self.losses = losses
empty_weight = torch.ones(self.num_classes + 1)
empty_weight[-1] = self.eos_coef
self.register_buffer('empty_weight', empty_weight)
...여기서 가장 중요한 부분은 matcher이다. 이 부분이 바로 100개 quries와 label을 매칭 시켜주는 부분이다.
mathcer.py
따라서 HungarianMatcher 코드를 확인해보자.
class HungarianMatcher(nn.Module):
"""This class computes an assignment between the targets and the predictions of the network
For efficiency reasons, the targets don't include the no_object. Because of this, in general,
there are more predictions than targets. In this case, we do a 1-to-1 matching of the best predictions,
while the others are un-matched (and thus treated as non-objects).
"""
def __init__(self, cost_class: float = 1, cost_bbox: float = 1, cost_giou: float = 1):
"""Creates the matcher
Params:
cost_class: This is the relative weight of the classification error in the matching cost
cost_bbox: This is the relative weight of the L1 error of the bounding box coordinates in the matching cost
cost_giou: This is the relative weight of the giou loss of the bounding box in the matching cost
"""
super().__init__()
self.cost_class = cost_class
self.cost_bbox = cost_bbox
self.cost_giou = cost_giou
assert cost_class != 0 or cost_bbox != 0 or cost_giou != 0, "all costs cant be 0"초기화 부분은 심플하다. 매칭에 사용된 손실 함수의 계수를 정의하기만한다. 이제 실제 계산 부분 forward을 봐보자.
@torch.no_grad()
def forward(self, outputs, targets):
""" Performs the matching
Params:
outputs: This is a dict that contains at least these entries:
"pred_logits": Tensor of dim [batch_size, num_queries, num_classes] with the classification logits
"pred_boxes": Tensor of dim [batch_size, num_queries, 4] with the predicted box coordinates
targets: This is a list of targets (len(targets) = batch_size), where each target is a dict containing:
"labels": Tensor of dim [num_target_boxes] (where num_target_boxes is the number of ground-truth
objects in the target) containing the class labels
"boxes": Tensor of dim [num_target_boxes, 4] containing the target box coordinates
Returns:
A list of size batch_size, containing tuples of (index_i, index_j) where:
- index_i is the indices of the selected predictions (in order)
- index_j is the indices of the corresponding selected targets (in order)
For each batch element, it holds:
len(index_i) = len(index_j) = min(num_queries, num_target_boxes)
"""
# outputs["pred_logits"]의 shape는 [2, 100, 92]
bs, num_queries = outputs["pred_logits"].shape[:2]
# We flatten to compute the cost matrices in a batch
# logits에 softmax을 적용하여 확률함수로 바꿈
out_prob = outputs["pred_logits"].flatten(0, 1).softmax(-1) # [batch_size * num_queries, num_classes]
out_bbox = outputs["pred_boxes"].flatten(0, 1) # [batch_size * num_queries, 4]
# batch의 객체들을 하나로 결합
# Also concat the target labels and boxes
tgt_ids = torch.cat([v["labels"] for v in targets]) # [2, 4] -> [6]
tgt_bbox = torch.cat([v["boxes"] for v in targets]) # [[2, 4], [4, 4]] -> [6, 4]
# Compute the classification cost. Contrary to the loss, we don't use the NLL,
# but approximate it in 1 - proba[target class].
# The 1 is a constant that doesn't change the matching, it can be ommitted.
# 우리는 92개의 클래스중 실제 영상에 존재하는 타켓에 확률만 알면 된다.
# 단 batch 처리에 용이하기 위해 단일 영상 대신 batch 영상의 타켓의 분류 전체를 추출
# 해당 예시에서는 92개중 [2, 4] = [6] 만 뽑아 비교한다.
# 해당 인덱스만 뽑아 분류 cost 구성
# NLL을 사용하지 이유는 앞선 논문 설명 부분 참고 바람
cost_class = -out_prob[:, tgt_ids] # [200, 6]
# 두 행 벡터 컬렉션의 각 쌍 사이의 p-norm 거리를 일괄 계산
# 여기도 마찬가지로 전체 batch 타겟을 하나로 뭉처 계산
# Compute the L1 cost between boxes
cost_bbox = torch.cdist(out_bbox, tgt_bbox, p=1) # [200, 4], [6, 4] -> [200, 6]
# Compute the giou cost betwen boxes
cost_giou = -generalized_box_iou(box_cxcywh_to_xyxy(out_bbox), box_cxcywh_to_xyxy(tgt_bbox))
# 수식 2번을 weight을 조절하여 구성
# Final cost matrix
C = self.cost_bbox * cost_bbox + self.cost_class * cost_class + self.cost_giou * cost_giou
# scipy.optimize.linear_sum_assignment에 적용하기 위해 cpu로 전환
C = C.view(bs, num_queries, -1).cpu()
# 앞선 matching cost는 [2, 100, 6]이다. 하지만 실제 matching cost는 각각
# [100, 2], [100, 4]이다. 아래에서 Tensor 조작으로 해당 목표를 이룬다.
# C.split(sizes, -1)로 ([2, 100, 2], [2, 100, 4]) 로 나눈다.
# 하지만 첫번째 영상은 첫번째 요소의 첫번째 batch 만 필요로 하고, 두번째는 두번째 요소의 두번째 batch만 필요로 한다.
# 즉 [0, 100, 2], [1, 100, 4]이 필요.
# 아래 코드를 통해 각 요소마다 각 배치의 매칭 관계 list[tuple[list, list]]] 로 획득할 수 있다.
sizes = [len(v["boxes"]) for v in targets]
indices = [linear_sum_assignment(c[i]) for i, c in enumerate(C.split(sizes, -1))]
return [(torch.as_tensor(i, dtype=torch.int64), torch.as_tensor(j, dtype=torch.int64)) for i, j in indices]앞선 코드로 우리는 매칭완료, 이렇게 얻어진 매칭으로 손실 함수를 계산 할 수 있다.
다시 SetCriterion의 forward 부분을 확인해보자.
def forward(self, outputs, targets):
""" This performs the loss computation.
Parameters:
outputs: dict of tensors, see the output specification of the model for the format
targets: list of dicts, such that len(targets) == batch_size.
The expected keys in each dict depends on the losses applied, see each loss' doc
"""
# 마지막 layer의 아웃풋으로 매칭 계산
outputs_without_aux = {k: v for k, v in outputs.items() if k != 'aux_outputs'}
# Retrieve the matching between the outputs of the last layer and the targets
indices = self.matcher(outputs_without_aux, targets)
# Compute the average number of target boxes accross all nodes, for normalization purposes
num_boxes = sum(len(t["labels"]) for t in targets)
num_boxes = torch.as_tensor([num_boxes], dtype=torch.float, device=next(iter(outputs.values())).device)
if is_dist_avail_and_initialized():
torch.distributed.all_reduce(num_boxes)
num_boxes = torch.clamp(num_boxes / get_world_size(), min=1).item()
# Compute all the requested losses
losses = {}
# 계산된 인덱스로 ['labels', 'boxes', 'cardinality'(실제 손실 함수로 안쓰임)]
# 에 대응되는 손실 함수를 계산
for loss in self.losses:
losses.update(self.get_loss(loss, outputs, targets, indices, num_boxes))
...
def get_loss(self, loss, outputs, targets, indices, num_boxes, **kwargs):
loss_map = {
'labels': self.loss_labels,
'cardinality': self.loss_cardinality,
'boxes': self.loss_boxes,
'masks': self.loss_masks
}
assert loss in loss_map, f'do you really want to compute {loss} loss?'
return loss_map[loss](outputs, targets, indices, num_boxes, **kwargs)그렇다면 각 매핑된 손실 함수중 loss_label을 봐보자.
해당 부분이 self._get_src_permutation_idx이 사용되기에 해당 코드도 같이 첨부한다.
self._get_src_permutation_idx에서는 실제 batch에서 매칭되는 2차원 인덱스를 얻는 method이다.
def loss_labels(self, outputs, targets, indices, num_boxes, log=True):
"""Classification loss (NLL)
targets dicts must contain the key "labels" containing a tensor of dim [nb_target_boxes]
"""
assert 'pred_logits' in outputs
src_logits = outputs['pred_logits']
idx = self._get_src_permutation_idx(indices)
...
def _get_src_permutation_idx(self, indices):
# permute predictions following indices
# 만약 첫번째/두번째 영상에 각각 2/4개 객체가 있다면
# batch_idx [0, 0, 1, 1, 1, 1]
batch_idx = torch.cat([torch.full_like(src, i) for i, (src, _) in enumerate(indices)])
# indices각 [(tensor([60, 72]), tensor([0, 1])),
# (tensor([11, 17, 51, 76]), tensor([0, 1, 2, 3]))] 이라면
# src_idx는 [60, 72, 11, 17, 51, 76]
src_idx = torch.cat([src for (src, _) in indices])
# 각 배치의 실제 매칭되는 부분의 인덱스를 tuple형식으로 출력
return batch_idx, src_idx계산된 batch_idx, src_idx를 제외한 나머지 인덱스는 다 no_object와 매칭될 것이다.
...
idx = self._get_src_permutation_idx(indices)
# batch 타켓을 하나의 tensor로 만든다
target_classes_o = torch.cat([t["labels"][J] for t, (_, J) in zip(targets, indices)])
# 일단 전체 타켓 매칭을 no_object로 초기화된 tensor로 만든다.
# [2, 100] 모든 데이터는 91로 초기화
target_classes = torch.full(src_logits.shape[:2], self.num_classes,
dtype=torch.int64, device=src_logits.device)
# (tensor([0, 0, 1, 1, 1, 1]), tensor([60, 72, 11, 17, 51, 76]))
# 각 배치의 매칭되는 queries 값을 타켓 값으로 변경하여 매칭관계 정립
target_classes[idx] = target_classes_o
# 이제 weighted cross entropy 계산
# no_object가 많기에 해당 부분 가중치는 다른 타켓보다 낮게 설정
loss_ce = F.cross_entropy(src_logits.transpose(1, 2), target_classes, self.empty_weight)
losses = {'loss_ce': loss_ce}
if log:
# TODO this should probably be a separate loss, not hacked in this one here
losses['class_error'] = 100 - accuracy(src_logits[idx], target_classes_o)[0]
return losses비슷환 알고리즘이 바운딩 박스에도 적용된다. 이제 loss_bbox을 봐보자
def loss_boxes(self, outputs, targets, indices, num_boxes):
"""Compute the losses related to the bounding boxes, the L1 regression loss and the GIoU loss
targets dicts must contain the key "boxes" containing a tensor of dim [nb_target_boxes, 4]
The target boxes are expected in format (center_x, center_y, w, h), normalized by the image size.
"""
assert 'pred_boxes' in outputs
idx = self._get_src_permutation_idx(indices)
# 매칭된 인덱스의 bbox 값만 추출
src_boxes = outputs['pred_boxes'][idx]
# 매칭된 타켓들만 추출
target_boxes = torch.cat([t['boxes'][i] for t, (_, i) in zip(targets, indices)], dim=0)
# 손실 함수 계산
loss_bbox = F.l1_loss(src_boxes, target_boxes, reduction='none')
losses = {}
losses['loss_bbox'] = loss_bbox.sum() / num_boxes
loss_giou = 1 - torch.diag(box_ops.generalized_box_iou(
box_ops.box_cxcywh_to_xyxy(src_boxes),
box_ops.box_cxcywh_to_xyxy(target_boxes)))
losses['loss_giou'] = loss_giou.sum() / num_boxes
return losses
이제 이러한 방법을 aux_ouputs에서도 똑같이 적용하여 전체 loss 계산.
다시 train_one_epoch에서
optimizer.zero_grad()
losses.backward()
if max_norm > 0:
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm)
optimizer.step()계산된 손실함수의 역전파하여 모델 학습한다.
이로써 train_one_epoch부분 분석완료하였다. 다른 부분은 모델 평가, 로깅 및 저장 부분과 같은 마이너한 요소.
이제 마지막인 후처리 과정 PostProcess을 봐보자.
class PostProcess(nn.Module):
""" This module converts the model's output into the format expected by the coco api"""
@torch.no_grad()
def forward(self, outputs, target_sizes):
""" Perform the computation
Parameters:
outputs: raw outputs of the model
target_sizes: tensor of dimension [batch_size x 2] containing the size of each images of the batch
For evaluation, this must be the original image size (before any data augmentation)
For visualization, this should be the image size after data augment, but before padding
"""
out_logits, out_bbox = outputs['pred_logits'], outputs['pred_boxes']
assert len(out_logits) == len(target_sizes)
assert target_sizes.shape[1] == 2
# 확률 함수로 전환
prob = F.softmax(out_logits, -1)
# no_object를 제외하고 가장 높은 predicted label 과 score 계산
scores, labels = prob[..., :-1].max(-1)
# convert to [x0, y0, x1, y1] format
boxes = box_ops.box_cxcywh_to_xyxy(out_bbox)
# and from relative [0, 1] to absolute [0, height] coordinates
img_h, img_w = target_sizes.unbind(1)
scale_fct = torch.stack([img_w, img_h, img_w, img_h], dim=1)
boxes = boxes * scale_fct[:, None, :]
results = [{'scores': s, 'labels': l, 'boxes': b} for s, l, b in zip(scores, labels, boxes)]
return results이로써 전체 DETR 코드를 확인해보았다.
결론
기존의 휴리스틱 NMS을 대체하는 헝가리안 손실함수는 개인적으로 아주 효과적인 접근방법이라고 생각한다.
2020년에 발표된 기법이지만 여전히 굉장히 우아한 코드로 작성되어 있다.
스마트한 Tensor 조작으로 많은 for ... in ... 을 생략할 수 잇다.
텐서 조작이 처음에는 어렵게 느껴질수 있지만, 습관이 된다면 이는 매우 효과/직관적 방법이다. 또한 무엇가 멋진
이후 DETR의 학습이 느리다는 단점을 보완하는 deformable DETR, DAB-DETR 및 Denoise-DETR이 리뷰할 것이다.
이상 DETR 논문 리뷰 끝!
