HardNet 개발
하도 정리가 안되서 정리용 블로그 시작
이미지에서 segmentation 을 추출하는 모델
input 이미지의 width,height와 output 이미지의 width, height가 동일
input은 [channel, width, height] 라면
output 은 [classes, width, height] 로 각 픽셀별로 이미지화하여 나오게 됨
논문은 아래 링크
https://arxiv.org/abs/1909.00948
HarDNet: A Low Memory Traffic Network
State-of-the-art neural network architectures such as ResNet, MobileNet, and DenseNet have achieved outstanding accuracy over low MACs and small model size counterparts. However, these metrics might not be accurate for predicting the inference time. We sug
arxiv.org
기존에 mmdetection을 사용하여 segmentation 모델을 개발하였으나 너무 오래 걸려서 더 빠르고 가벼운 hardnet으로 변경
성능 테스트는 일단 학습시켜서 IOU 지표로 비교해보도록 하고
일단 목적은 image classification인데 정확도를 높이기 위해 segmentation을 먼저 앞에서 돌리고 classification을 진행
확실히 그냥 classification 모델 돌리는 것보다 정확도가 어마어마하게 차이 남 (90% 대에서 6~7% 차이)
이미지 예시나 정확한 프로젝트 용도는 회사 기밀이라 패스
pytorch을 사용하여 개발
모델 코드는 아래의 링크를 가져다 사용함
https://github.com/PingoLH/FCHarDNet/blob/master/ptsemseg/models/hardnet.py
PingoLH/FCHarDNet
Fully Convolutional HarDNet for Segmentation in Pytorch - PingoLH/FCHarDNet
github.com
평가 지표인 metrix로는 IOU를 사용함
코드는 아래
https://www.kaggle.com/iezepov/fast-iou-scoring-metric-in-pytorch-and-numpy
Fast IOU scoring metric in PyTorch and numpy
Explore and run machine learning code with Kaggle Notebooks | Using data from TGS Salt Identification Challenge
www.kaggle.com
참고로 0,1로만 이루어져있는 이미지에 대한 평가 지표라 class에 따른 평가를 수행하기 위해서 아래 코드와 같이 변경하였다
import torch
import numpy as np
SMOOTH = 1e-6
def iou_numpy(outputs: np.array, labels: np.array):
outputs = outputs.squeeze(1)
intersection = (outputs & labels).sum((1, 2))
union = (outputs | labels).sum((1, 2))
iou = (intersection + SMOOTH) / (union + SMOOTH)
thresholded = np.ceil(np.clip(20 * (iou - 0.5), 0, 10)) / 10
return thresholded # Or thresholded.mean()
def iou_pytorch(outputs: torch.Tensor, labels: torch.Tensor, device='cpu'):
# You can comment out this line if you are passing tensors of equal shape
# But if you are passing output from UNet or something it will most probably
# be with the BATCH x 1 x H x W shape
#outputs = outputs.squeeze(1) # BATCH x 1 x H x W => BATCH x H x W
outputs = torch.max(outputs, 1)[1]
intersection = torch.where(((outputs.int() != 0) | (labels.int() != 0)) & (outputs.int() == labels.int()), torch.Tensor([1]).to(device), torch.Tensor([0]).to(device))
intersection = intersection.float().sum((1, 2))
union = torch.where(((outputs.int() != 0) | (labels.int() != 0)), torch.Tensor([1]).to(device), torch.Tensor([0]).to(device))
union = union.float().sum((1, 2))
iou = (intersection + SMOOTH) / (union + SMOOTH) # We smooth our devision to avoid 0/0
thresholded = torch.clamp(20 * (iou - 0.5), 0, 10).ceil() / 10 # This is equal to comparing with thresolds
return thresholded.mean() # Or thresholded.mean() if you are interested in average across the batch
이미지 로딩
pytorch의 dataloader를 사용하여 custom
from torch.utils.data import Dataset
import torch
import cv2
import numpy as np
def HardNetLoader(img_list, label_img_list, label_list, batch_size):
hardnet_dataset = HardNetDataset(img_list, label_img_list, label_list)
hardnet_dataloader = torch.utils.data.DataLoader(hardnet_dataset, batch_size=batch_size)
return hardnet_dataloader
class HardNetDataset(Dataset):
def __init__(self, img_list, label_img_list, label_list):
self.img_list = img_list
self.label_img_list = label_img_list
self.label_list = label_list
self.samples = len(self.img_list)
def __len__(self):
return self.samples
def __getitem__(self, idx):
image = cv2.imread(self.img_list[idx], 1)/255
image = np.transpose(image, (2,0,1))
label_img = cv2.imread(self.label_img_list[idx], 0)
label_img[label_img != 0] = self.label_list[idx]
return image, label_img메모리가 부족하여 필요할 때마다 opencv 를 통하여 데이터 로딩
pytorch에 들어가는 데이터는 [channel, width, height]
하지만 opencv 는 [width, height, channel] 순으로 로딩되어서 transpose 로 순서를 바꿈
2020.02.05 : 바보같이 이미지를 255로 안 나누고 진행했었다..... 그래놓고 결과 테스트할 때는 255로 나눠서 넣었으니 안보였지
Loss function
코드는 아래
https://github.com/meetshah1995/pytorch-semseg/blob/master/ptsemseg/loss/loss.py
meetshah1995/pytorch-semseg
Semantic Segmentation Architectures Implemented in PyTorch - meetshah1995/pytorch-semseg
github.com
2 dimentional corss entropy 인데 해당 코드는 treshhold를 사용해서 일정 이하 값의 loss는 버리는 것으로 보인다 (굳이?)
나중에 추가적인 공부가 필요
optimizer
그냥 pytorch 의 SGD를 사용
optimizer = torch.optim.SGD(hard_net.parameters(), lr=0.02, weight_decay=0.0005, momentum=0.9)상무님이 SGD 대신 Adam 쓰라고 하셔서 코드 변경
optimizer = torch.optim.Adam(hard_net.parameters(), lr=0.0005, weight_decay=1e-4)SGD든 Adam 이든 튜닝의 영역이라 일단 기본인 Adam으로 쓰자고 하셨다
물론 learning rate는 0.02가 나도 높다고 생각했지만 일단 복붙한거라 안건드렸는데 0.0005로 수정
weight decay는 hardnet adam으로 검색해서 나온 코드에서의 파라미터를 넣었다
torch용 코드를 또 따로 짬
from tqdm import tqdm
from torch import nn
import numpy as np
import torch
import iou
import os
def load(model, path:str, slice_idx:int=0):
load_state_dict = torch.load(path)
model_state_dict = model.state_dict()
set_state_dict = {}
for k, v in load_state_dict.items():
key = k[slice_idx:]
if key in model_state_dict:
set_state_dict[key] = v
model_state_dict.update(set_state_dict)
model.load_state_dict(model_state_dict)
print("[LOAD] success at {}".format(path))
def predict(model, data, device='cpu') -> np.ndarray:
if not isinstance(model, nn.Module):
print('[E0RROR] predict : model isn\'t torch nn.Module')
raise TypeError
if not isinstance(data, torch.Tensor):
print('[ERROR] data isn\'t torch tensor')
model.eval()
with torch.no_grad():
model = model.to(device)
data_tensor = data.to(device, dtype=torch.float)
_pred_result = model(data_tensor)
return _pred_result.cpu().numpy()
def validate(model, data_loader, loss_fn, acc_fn=None, device='cpu') -> float:
if not isinstance(model, nn.Module):
print('[ERROR] validate : model isn\'t torch nn.Module')
raise TypeError
model.eval()
_valid_loss = 0.0
_acc_n = 0
if acc_fn is not None:
_acc_sum = 0.0
else:
_acc_sum = None
with torch.no_grad():
model = model.to(device)
print("[VALID] start")
for _data, _label in tqdm(data_loader):
_data_tensor = _data.to(device, dtype=torch.float)
_label_tensor = _label.to(device, dtype=torch.long)
_pred_result = model(_data_tensor)
_valid_loss += loss_fn(_pred_result, _label_tensor).item()
if acc_fn is not None:
_acc_sum += acc_fn(_pred_result, _label_tensor).item()
_acc_n += 1
return _valid_loss/_acc_n, _acc_sum/_acc_n
def train(model, train_data_loader, valid_data_loader=None, loss_fn=None, optimizer=None, epochs:int=1, acc_fn=None, weight_path:str='weight/', device='cpu'):
if not isinstance(model, nn.Module):
print('[ERROR] validate : model isn\'t torch nn.Module')
raise TypeError
model = model.to(device)
model.train()
for epoch in range(epochs):
_train_loss_sum = 0.0
_train_loss_n = 0
print("[TRAIN] start")
for _train_data, _train_label in tqdm(train_data_loader):
_train_data_tensor = _train_data.to(device, dtype=torch.float)
_train_label_tensor = _train_label.to(device, dtype=torch.long)
#_train_label_tensor = _train_label.long()
optimizer.zero_grad()
_train_result = model(_train_data_tensor)
_loss = loss_fn(_train_result, _train_label_tensor)
_loss.backward()
optimizer.step()
_train_loss_sum += _loss.item()
_train_loss_n += 1
#print('[batch] epoch {}: {}'.format(epoch, _loss.item()))
print('[train] epoch {}: {}'.format(epoch, _train_loss_sum/_train_loss_n))
_val_acc = None
if valid_data_loader:
_valid_loss, _val_acc = validate(model, valid_data_loader, loss_fn, acc_fn, device)
print('[valid] epoch {}: {}'.format(epoch, _valid_loss))
model.train()
if not os.path.isdir(weight_path):
os.makedirs(weight_path)
if _val_acc is not None:
weight_nm = os.path.join(weight_path, 'model_{}_acc_{:.4f}.pth'.format(epoch, _val_acc))
else:
weight_nm = os.path.join(weight_path, 'model_{}_loss_{:.4f}.pth'.format(epoch, _train_loss_sum/_train_loss_n))
torch.save(model.state_dict(), weight_nm)
print('[SAVE] weight save at {}'.format(weight_nm))
코드 중에서 맘에 안드는 부분도 있지만 일단 패스
나중에 수정해도 블로그 코드는 수정 안할꺼임
일단은 돌아는 가니 냅둔다
weight에서 'module.' 이 계속 키 값 앞에 붙어서 지우기 위해 새로 dict 만들고 update하는 번거로운 작업을 수행
메인코드
from torch.utils.data import TensorDataset, DataLoader
from torch import nn
import torch
import hardnet
import data_util
import torch_util
import hardnet_dataloader
from hardnet_loss import *
import random
import iou
import hardnet_loss
import cv2
import numpy as np
import os
import multiprocessing as mp
import time
device = "cuda" if torch.cuda.is_available() else "cpu"
hard_net = hardnet.HardNet(4)
hard_net = nn.DataParallel(hard_net)
hard_net = hard_net.to(device)
img_list, label_img_list, label_list = data_util.getDataPath('../data/images_ori/')
random.seed(100)
random.shuffle(img_list)
random.seed(100)
random.shuffle(label_img_list)
random.seed(100)
random.shuffle(label_list)
slice_idx = int(len(img_list)*0.8)
batch_size = 32
num_workers = 6
#optimizer = torch.optim.SGD(hard_net.parameters(), lr=0.0005, weight_decay=0.0005, momentum=0.9)
optimizer = torch.optim.Adam(hard_net.parameters(), lr=0.0005, weight_decay=1e-4)
train_loader = hardnet_dataloader.HardNetLoader(img_list[:slice_idx], label_img_list[:slice_idx], label_list[:slice_idx], batch_size, num_workers=num_workers, pin_memory=True)
valid_loader = hardnet_dataloader.HardNetLoader(img_list[slice_idx:], label_img_list[slice_idx:], label_list[slice_idx:], batch_size, num_workers=num_workers, pin_memory=True)
loss_fn = bootstrapped_cross_entropy2d().to(device)
torch_util.train(hard_net, train_loader, valid_loader, loss_fn, optimizer, epochs=20, acc_fn=iou.iou_pytorch, weight_path='weight/hardnet/',device=device)
원래 jupyter로 메인 코드 돌려서 하나씩 찍어보는데 일단은 나누는 건 위에서 알아서
위에 쓴 코드들은 다른 파일로 빼놓음 (optimizer 제외)
dataparallel 을 써도 느린건 여전하니 apex나 distribute parallel 을 찾아서 주피터 상에서 사용할 수 있도록 커스텀 예정
(망할 놈들이 무조건 main 파일을 terminal 상에서 돌리도록 예제를 만들어놔서 눈물을 흘리면서 코드 분석....)