CV语义分割实践指南！
遥感技术已成为获取地表覆盖信息最为行之有效的手段，已经成功应用于地表覆盖检测、植被面积检测和建筑物检测任务。本文以天池学习赛地表建筑物识别为例，对语义分割类项目的实践全流程进行了解析。具体流程如下：
赛题理解
赛题名称：零基础入门语义分割-地表建筑物识别 赛题地址：https://tianchi.aliyun.com/competition/entrance/531872/information
1.1 赛题数据
本赛题使用航拍数据，需要参赛选手完成地表建筑物识别，将地表航拍图像素划分为有建筑物和无建筑物两类。
如下图，左边为原始航拍图，右边为对应的建筑物标注。
1.2 数据标签
赛题为语义分割任务，因此具体的标签为图像像素类别。在赛题数据中像素属于2类（无建筑物和有建筑物），因此标签为有建筑物的像素。赛题原始图片为jpg格式，标签为RLE编码的字符串。
RLE全称（run-length encoding），翻译为游程编码或行程长度编码，对连续的黑、白像素数以不同的码字进行编码。RLE是一种简单的非破坏性资料压缩法，经常用在在语义分割比赛中对标签进行编码。
RLE与图片之间的转换代码详见本文第二节Baseline代码解析。
1.3 评价指标
赛题使用Dice coefficient来衡量选手结果与真实标签的差异性，Dice coefficient可以按像素差异性来比较结果的差异性。Dice coefficient的具体计算方式如下：
其中是预测结果， 为真实标签的结果。当与完全相同时Dice coefficient为1，排行榜使用所有测试集图片的平均Dice coefficient来衡量，分数值越大越好。
1.4 解题思路
由于本次赛题是一个典型的语义分割任务，因此可以直接使用语义分割的模型来完成：
步骤1：使用FCN模型模型跑通具体模型训练过程，并对结果进行预测提交； 步骤2：在现有基础上加入数据扩增方法，并划分验证集以监督模型精度； 步骤3：使用更加强大模型结构（如Unet和PSPNet）或尺寸更大的输入完成训练； 步骤4：训练多个模型完成模型集成操作；
Baseline代码分析
Ⅰ.将图片编码为rle格式
import numpy as np
import pandas as pd
import cv2
# 将图片编码为rle格式
def rle_encode(im):
'''
im: numpy array, 1 - mask, 0 - background
Returns run length as string formated
'''
pixels = im.flatten(order = 'F')
pixels = np.concatenate([[0], pixels, [0]])
runs = np.where(pixels[1:] != pixels[:-1])[0] + 1
runs[1::2] -= runs[::2]
return ' '.join(str(x) for x in runs)Ⅱ.将rle格式进行解码为图片
# 将rle格式进行解码为图片
def rle_decode(mask_rle, shape=(512, 512)):
'''
mask_rle: run-length as string formated (start length)
shape: (height,width) of array to return
Returns numpy array, 1 - mask, 0 - background
'''
s = mask_rle.split()
starts, lengths = [np.asarray(x, dtype=int) for x in (s[0:][::2], s[1:][::2])]
starts -= 1
ends = starts + lengths
img = np.zeros(shape[0]*shape[1], dtype=np.uint8)
for lo, hi in zip(starts, ends):
img[lo:hi] = 1
return img.reshape(shape, order='F')
RLE编码的时候返回的时候每两个数字有空格为间隔，利用s = mask_rle.split()将空格去掉。
s[0:][::2]表示（从1开始的）索引，s[1:][::2]表示个数。于是starts存的是索引，lengths存的是个数，两者为一一对应关系。
starts \-= 1转化为（从0开始的）索引。
后续就是创建一副全0的一维序列，填充1，再按列排序，转为二维的二值图，就解码成图片了。
如果输入的mask_rle是空的，那么返回的就是全为0的mask，可以观察数据发现，部分图片的地表建筑不存在，他们的rle标签也就是空的。
Ⅲ.定义数据集
class TianChiDataset(D.Dataset):
def __init__(self, paths, rles, transform, test_mode=False):
self.paths = paths
self.rles = rles
self.transform = transform
self.test_mode = test_mode
self.len = len(paths)
self.as_tensor = T.Compose([
T.ToPILImage(),
T.Resize(IMAGE_SIZE),
T.ToTensor(),
T.Normalize([0.625, 0.448, 0.688],
[0.131, 0.177, 0.101]),
])
# get data operation
def __getitem__(self, index):
#img = cv2.imread(self.paths[index])
img = np.array(Image.open(self.paths[index]))
if not self.test_mode:
mask = rle_decode(self.rles[index])
augments = self.transform(image=img, mask=mask)
return self.as_tensor(augments['image']), augments['mask'][None]#（3，256，256），（1，256，256）
else:
return self.as_tensor(img), ''
def __len__(self):
"""
Total number of samples in the dataset
"""
return self.len
定义数据集，主要作了数据的预处理。其中，我将opencv的读取图片换成了PIL读取,因为路径中包含中文augments['mask'][None]中的[None],将（256，256）的mask形状转为（1，256，256），起到升维作用。
Ⅳ.可视化一下效果
这一步主要是为了验证上述的代码。用了rle_encode(rle_decode(RLE标签))==RLE标签来验证之前写的RLE编码和解码正确性。
train_mask = pd.read_csv('数据集/train_mask.csv', sep='\t', names=['name', 'mask'])
train_mask['name'] = train_mask['name'].apply(lambda x: '数据集/train/' + x)
img = cv2.imread(train_mask['name'].iloc[0])
mask = rle_decode(train_mask['mask'].iloc[0])
print(rle_encode(mask) == train_mask['mask'].iloc[0])
train_mask['name'].apply(lambda x: '数据集/train/' + x)这一步就是在图片前补全下路径
0 KWP8J3TRSV.jpg
1 DKI3X4VFD3.jpg
2 AYPOE51XNI.jpg
3 1D9V7N0DGF.jpg
4 AWXXR4VYRI.jpg
0 数据集/train/KWP8J3TRSV.jpg
1 数据集/train/DKI3X4VFD3.jpg
2 数据集/train/AYPOE51XNI.jpg
3 数据集/train/1D9V7N0DGF.jpg
4 数据集/train/AWXXR4VYRI.jpg
实例化数据集
dataset = TianChiDataset(
train_mask['name'].values,
train_mask['mask'].fillna('').values,
trfm, False
)
fillna('')起到补全缺失值为''的作用
可视化
image, mask = dataset[0]
plt.figure(figsize=(16,8))
plt.subplot(121)
plt.imshow(mask[0], cmap='gray')
plt.subplot(122)
plt.imshow(image[0])
plt.show()# 补上
看一下第二张图片
image, mask = dataset[1]
没有建筑物，mask全黑。
Ⅴ.加载数据集
#定义数据集
train_mask = pd.read_csv('数据集/train_mask.csv', sep='\t', names=['name', 'mask'])
train_mask['name'] = train_mask['name'].apply(lambda x: '数据集/train/' + x)
dataset = TianChiDataset(
train_mask['name'].values,
train_mask['mask'].fillna('').values,
trfm, False
)
#划分数据集(按index手动去划分)
valid_idx, train_idx = [], []
for i in range(len(dataset)):
if i % 7 == 0:
valid_idx.append(i)
else:
# elif i % 7 == 1:
train_idx.append(i)
train_ds = D.Subset(dataset, train_idx)
valid_ds = D.Subset(dataset, valid_idx)
# print(len(dataset))#30000
# print(len(train_ds))#4286
# print(len(valid_ds))#4286
# define training and validation data loaders
loader = D.DataLoader(
train_ds, batch_size=BATCH_SIZE, shuffle=True, num_workers=0)
vloader = D.DataLoader(
valid_ds, batch_size=BATCH_SIZE, shuffle=False, num_workers=0)
D.subset是按照索引序列来划分数据集的， 于是按照每7个数据里面，1个当作验证集，6个当作训练集。最后放入数据加载器中。
Ⅵ.定义模型、优化器、损失函数
# 定义模型
model = get_model()
model.to(DEVICE)
#model.load_state_dict(torch.load("model_best.pth"))
#定义优化器
optimizer = torch.optim.AdamW(model.parameters(),
lr=1e-4, weight_decay=1e-3)
#定义损失函数
bce_fn = nn.BCEWithLogitsLoss()
dice_fn = SoftDiceLoss()
def loss_fn(y_pred, y_true):
bce = bce_fn(y_pred, y_true)
dice = dice_fn(y_pred.sigmoid(), y_true)
return 0.8 * bce + 0.2 * dice
Ⅶ.进行训练
header = r'''
Train | Valid
Epoch | Loss | Loss | Time, m
'''
# Epoch metrics time
raw_line = '{:6d}' + '\u2502{:7.3f}' * 2 + '\u2502{:6.2f}'
print(header)
EPOCHES = 10
best_loss = 10
for epoch in range(1, EPOCHES + 1):
losses = []
start_time = time.time()
model.train()
for image, target in tqdm(loader):#取消了tqdm
image, target = image.to(DEVICE), target.float().to(DEVICE)
optimizer.zero_grad()
output = model(image)['out']
loss = loss_fn(output, target)
loss.backward()
optimizer.step()
losses.append(loss.item())
# print(loss.item())
vloss = validation(model, vloader, loss_fn)
print(raw_line.format(epoch, np.array(losses).mean(), vloss,
(time.time() - start_time) / 60 ** 1))
losses = []
if vloss < best_loss:
best_loss = vloss
torch.save(model.state_dict(), 'model_best.pth')
print("save successful!")
Ⅷ.使用模型对测试集进行预测
trfm = T.Compose([
T.ToPILImage(),
T.Resize(IMAGE_SIZE),
T.ToTensor(),
T.Normalize([0.625, 0.448, 0.688],
[0.131, 0.177, 0.101]),
])
subm = []
model.load_state_dict(torch.load("./model_best.pth"))
model.eval()
test_mask = pd.read_csv('数据集/test_a_samplesubmit.csv', sep='\t', names=['name', 'mask'])
test_mask['name'] = test_mask['name'].apply(lambda x: '数据集/test_a/' + x)
for idx, name in enumerate(tqdm(test_mask['name'].iloc[:])):
image = np.array(Image.open(name))#改成PIL
image = trfm(image)
with torch.no_grad():
image = image.to(DEVICE)[None]
score = model(image)['out'][0][0]
score_sigmoid = score.sigmoid().cpu().numpy()
score_sigmoid = (score_sigmoid > 0.5).astype(np.uint8)
score_sigmoid = cv2.resize(score_sigmoid, (512, 512))
# break
subm.append([name.split('/')[-1], rle_encode(score_sigmoid)])
subm = pd.DataFrame(subm)
subm.to_csv('./tmp.csv', index=None, header=None, sep='\t')
Ⅸ.可视化模型预测结果
from file1 import rle_decode
from PIL import Image
import pandas as pd
import numpy as np
subm = pd.read_csv("./tmp.csv",sep="\t",names=["name","mask"])
def show_predict_pic(num=0):
plt.figure(figsize=(16,8))
plt.subplot(121)
plt.imshow(rle_decode(subm.fillna('').iloc[num,1]), cmap='gray')
plt.subplot(122)
plt.imshow(np.array(Image.open('数据集/test_a/' + subm.iloc[num,0])))
plt.show()
if __name__ == '__main__':
show_predict_pic(num=10)
查看第10张图片的预测结果
评论