利用UNet进行回归任务

利用UNet进行回归任务

分析

from matplotlib import pyplot as plt
import cv2
import numpy as np
test_img = cv2.imread(r"D:\workplace\python\UNet2LP\data\imgi\1.png", 0)
predicted_img= np.loadtxt(r"D:\workplace\python\UNet2LP\data\vi\1.txt")


#plt.imshow(test_img[:, :, 0], cmap='gray')
predicted_img = predicted_img
plt.imshow(predicted_img, cmap='jet')
plt.colorbar()
plt.show()

预测

from simple_multi_unet_model import multi_unet_model # Uses softmax
import numpy as np
from tensorflow.keras.utils import normalize
import os
import glob
import cv2
import numpy as np
from matplotlib import pyplot as plt
import pandas as pd

#n_classes = 3 # Number of classes for segmentation

# Capture training image info as a list
X_train = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\imgi"):
for img_path in glob.glob(os.path.join(directory_path, "*.png")):
img = cv2.imread(img_path, 0)
X_train.append(img)

# Convert list to array for machine learning processing
X_train = np.array(X_train)
X_train= np.array(X_train,dtype=np.float32)/255.0
X_train= abs(X_train-1)

# Capture mask/label info as a list
X_test = []
for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\imgt"):
for mask_path in glob.glob(os.path.join(directory_path, "*.png")):
mask = cv2.imread(mask_path, 0)
X_test.append(mask)

# Convert list to array for machine learning processing
X_test = np.array(X_test)
X_test = np.array(X_test,dtype=np.float32)/255.0
X_test= abs(X_test-1)

#label
y_train = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\vi"):

for vt_path in glob.glob(os.path.join(directory_path, "*.txt")):
X = np.loadtxt(vt_path)

y_train.append(X)
y_train = np.array(y_train,dtype=np.float32)/255.0

y_test = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\vt"):

for ve_path in glob.glob(os.path.join(directory_path, "*.txt")):
Y = np.loadtxt(ve_path)

y_test.append(Y)

y_test = np.array(y_test,dtype=np.float32)/255.0
###############################################################
#对image处理
X_train = np.expand_dims(X_train, axis=3) #用于扩展数组的形状 增加维度
#X_train = normalize(X_train, axis=1)
X_test = np.expand_dims(X_test, axis=3) #用于扩展数组的形状 增加维度
#X_test = normalize(X_test, axis=1)
y_train = np.expand_dims(y_train, axis=3) #用于扩展数组的形状 增加维度
y_train = normalize(y_train, axis=1)
y_test= np.expand_dims(y_test, axis=3) #用于扩展数组的形状 增加维度
y_test = normalize(y_test, axis=1)




def get_model():
return multi_unet_model(IMG_HEIGHT=128, IMG_WIDTH=128, IMG_CHANNELS=1)


model = get_model()

# model = get_model()
model.load_weights('sandstone_50_epochs_catXentropy_acc.hdf5')
# model.load_weights('sandstone_50_epochs_catXentropy_acc_with_weights.hdf5')

# IOU
y_pred = model.predict(X_test)
# y_pred_argmax = np.array(y_pred, axis=3)

plt.figure(figsize=(12, 8))


for i in range(len(y_pred)):
#plt.imshow(test_img[:, :, 0], cmap='gray')
predicted_img = np.squeeze(y_pred[i], 2)
plt.subplot(231)
plt.title('predicted_img')
plt.imshow(predicted_img, cmap='jet')
plt.colorbar()



label = np.squeeze(y_test[i], 2)
plt.subplot(233)
plt.title('label')
plt.imshow(label, cmap='jet')
plt.colorbar()
plt.show()

模型

from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Conv2D, MaxPooling2D, UpSampling2D, concatenate, Conv2DTranspose, BatchNormalization, \
Dropout, Lambda
from tensorflow.keras.layers import LeakyReLU

################################################################
def multi_unet_model(n_classes=1, IMG_HEIGHT=128, IMG_WIDTH=128, IMG_CHANNELS=1):
# Build the model
inputs = Input((IMG_HEIGHT, IMG_WIDTH, IMG_CHANNELS))
# s = Lambda(lambda x: x / 255)(inputs) #No need for this if we normalize our inputs beforehand
s = inputs
#keras.layers.Conv2D(filters, kernel_size, strides=(1, 1), padding='valid', data_format=None, dilation_rate=(1, 1), activation=None, use_bias=True, kernel_initializer='glorot_uniform', bias_initializer='zeros', kernel_regularizer=None, bias_regularizer=None, activity_regularizer=None, kernel_constraint=None, bias_constraint=None)
#filters: 过滤器数量,图片等高维数据每经过一个卷积层，深度都会增加，并且等于过滤器的数量。
#kernel_size: 一个整数，或者 2 个整数表示的元组或列表， 指明 2D 卷积窗口的宽度和高度。 可以是一个整数，为所有空间维度指定相同的值。
#strides: 一个整数，或者 2 个整数表示的元组或列表， 指明卷积沿宽度和高度方向的步长。 可以是一个整数，为所有空间维度指定相同的值。 指定任何 stride 值 != 1 与指定 dilation_rate 值 != 1 两者不兼容。
#padding: "valid" 或 "same" (大小写敏感)。"valid"不填充 "same"填充，当滑动步长大于1时：填充数=K-I%S（K:卷积核边长，I：输入图像边长，S：滑动步长），滑动步长为1时，填充数是卷积核边长减1，eg:5*5的图用3*3的核，步长为1时same填充之后是7*7代表保留边界处的卷积结果，通常会导致输出shape与输入shape相同，因为卷积核移动时在边缘会出现大小不够的情况。
#activation: 要使用的激活函数(详见 activations)。 如果你不指定，则不使用激活函数(即线性激活： a(x) = x)。
#kernel_initializer: kernel 权值矩阵的初始化器。he_normal:He 均匀方差缩放初始化器。它从 [-limit，limit] 中的均匀分布中抽取样本， 其中 limit 是 sqrt(6 / fan_in)， 其中 fan_in 是权值张量中的输入单位的数量。
# Contraction path
c1 = Conv2D(16, (3, 3), kernel_initializer='he_normal', padding='same')(s)
c1 = LeakyReLU(alpha=0.05)(c1)
c1 = Dropout(0.1)(c1) #防止过拟合
c1 = Conv2D(16, (3, 3), kernel_initializer='he_normal', padding='same')(c1)
c1 = LeakyReLU(alpha=0.05)(c1)
p1 = MaxPooling2D((2, 2))(c1)

c2 = Conv2D(32, (3, 3), kernel_initializer='he_normal', padding='same')(p1)
c2 = LeakyReLU(alpha=0.05)(c2)
c2 = Dropout(0.1)(c2)
c2 = Conv2D(32, (3, 3), kernel_initializer='he_normal', padding='same')(c2)
c2 = LeakyReLU(alpha=0.05)(c2)
p2 = MaxPooling2D((2, 2))(c2)

c3 = Conv2D(64, (3, 3), kernel_initializer='he_normal', padding='same')(p2)
c3 = LeakyReLU(alpha=0.05)(c3)
c3 = Dropout(0.2)(c3)
c3 = Conv2D(64, (3, 3), kernel_initializer='he_normal', padding='same')(c3)
c3 = LeakyReLU(alpha=0.05)(c3)
p3 = MaxPooling2D((2, 2))(c3)

c4 = Conv2D(128, (3, 3), kernel_initializer='he_normal', padding='same')(p3)
c4 = LeakyReLU(alpha=0.05)(c4)
c4 = Dropout(0.2)(c4)
c4 = Conv2D(128, (3, 3), kernel_initializer='he_normal', padding='same')(c4)
c4 = LeakyReLU(alpha=0.05)(c4)
p4 = MaxPooling2D(pool_size=(2, 2))(c4)

c5 = Conv2D(256, (3, 3), kernel_initializer='he_normal', padding='same')(p4)
c5 = LeakyReLU(alpha=0.05)(c5)
c5 = Dropout(0.3)(c5)
c5 = Conv2D(256, (3, 3), kernel_initializer='he_normal', padding='same')(c5)
c5 = LeakyReLU(alpha=0.05)(c5)
# Expansive path
#Conv2DTranspose反卷积
u6 = Conv2DTranspose(128, (2, 2), strides=(2, 2), padding='same')(c5)
u6 = concatenate([u6, c4])
c6 = Conv2D(128, (3, 3), kernel_initializer='he_normal', padding='same')(u6)
c6 = LeakyReLU(alpha=0.05)(c6)
c6 = Dropout(0.2)(c6)
c6 = Conv2D(128, (3, 3), kernel_initializer='he_normal', padding='same')(c6)
c6 = LeakyReLU(alpha=0.05)(c6)

u7 = Conv2DTranspose(64, (2, 2), strides=(2, 2), padding='same')(c6)
u7 = concatenate([u7, c3])
c7 = Conv2D(64, (3, 3), kernel_initializer='he_normal', padding='same')(u7)
c7 = LeakyReLU(alpha=0.05)(c7)
c7 = Dropout(0.2)(c7)
c7 = Conv2D(64, (3, 3), kernel_initializer='he_normal', padding='same')(c7)
c7 = LeakyReLU(alpha=0.05)(c7)

u8 = Conv2DTranspose(32, (2, 2), strides=(2, 2), padding='same')(c7)
u8 = concatenate([u8, c2])
c8 = Conv2D(32, (3, 3), kernel_initializer='he_normal', padding='same')(u8)
c8 = LeakyReLU(alpha=0.05)(c8)
c8 = Dropout(0.1)(c8)
c8 = Conv2D(32, (3, 3), kernel_initializer='he_normal', padding='same')(c8)
c8 = LeakyReLU(alpha=0.05)(c8)

u9 = Conv2DTranspose(16, (2, 2), strides=(2, 2), padding='same')(c8)
u9 = concatenate([u9, c1], axis=3)
c9 = Conv2D(16, (3, 3), kernel_initializer='he_normal', padding='same')(u9)
c9 = LeakyReLU(alpha=0.05)(c9)
c9 = Dropout(0.1)(c9)
c9 = Conv2D(16, (3, 3), kernel_initializer='he_normal', padding='same')(c9)
c9 = LeakyReLU(alpha=0.05)(c9)

#outputs = Conv2D(n_classes, (1, 1), activation='softmax')(c9)
outputs = Conv2D(1,(1,1), kernel_initializer='he_normal')(c9)
model = Model(inputs=[inputs], outputs=[outputs])

# NOTE: Compile the model in the main program to make it easy to test with various loss functions
# model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])

# model.summary()

return model#import pandas as pd
#df = pd.read_csv('D:/velocity/28.txt', sep=' ', header=None, dtype=str, na_filter=False)
from simple_multi_unet_model import multi_unet_model # Uses softmax


from tensorflow.keras.utils import normalize
import os
import glob
import cv2
import numpy as np
from matplotlib import pyplot as plt
import pandas as pd

#n_classes = 3 # Number of classes for segmentation

# Capture training image info as a list
X_train = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\imgi"):
for img_path in glob.glob(os.path.join(directory_path, "*.png")):
img = cv2.imread(img_path, 0)
X_train.append(img)

# Convert list to array for machine learning processing
X_train = np.array(X_train)
X_train= np.array(X_train,dtype=np.float32)/255.0
X_train= abs(X_train-1)

# Capture mask/label info as a list
X_test = []
for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\imgt"):
for mask_path in glob.glob(os.path.join(directory_path, "*.png")):
mask = cv2.imread(mask_path, 0)
X_test.append(mask)

# Convert list to array for machine learning processing
X_test = np.array(X_test)
X_test = np.array(X_test,dtype=np.float32)/255.0
X_test= abs(X_test-1)

#label
y_train = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\vi"):

for vt_path in glob.glob(os.path.join(directory_path, "*.txt")):
X = np.loadtxt(vt_path)

y_train.append(X)
y_train = np.array(y_train,dtype=np.float32)/255.0

y_test = []

for directory_path in glob.glob(r"D:\workplace\python\UNet2LP\data\vt"):

for ve_path in glob.glob(os.path.join(directory_path, "*.txt")):
Y = np.loadtxt(ve_path)

y_test.append(Y)

y_test = np.array(y_test,dtype=np.float32)/255.0
###############################################################
#对image处理
X_train = np.expand_dims(X_train, axis=3) #用于扩展数组的形状 增加维度
#X_train = normalize(X_train, axis=1)
X_test = np.expand_dims(X_test, axis=3) #用于扩展数组的形状 增加维度
#X_test = normalize(X_test, axis=1)
y_train = np.expand_dims(y_train, axis=3) #用于扩展数组的形状 增加维度
y_train = normalize(y_train, axis=1)
y_test= np.expand_dims(y_test, axis=3) #用于扩展数组的形状 增加维度
y_test = normalize(y_test, axis=1)
###############################################################
IMG_HEIGHT = 128
IMG_WIDTH = 128
IMG_CHANNELS = 1

def get_model():
return multi_unet_model(IMG_HEIGHT=IMG_HEIGHT, IMG_WIDTH=IMG_WIDTH, IMG_CHANNELS=IMG_CHANNELS)


model = get_model()
model.compile(optimizer='adam', loss='mean_absolute_error', metrics=['accuracy'])
model.summary()

# If starting with pre-trained weights.
# model.load_weights('???.hdf5')

history = model.fit(X_train, y_train,
batch_size=2,
verbose=1,
epochs=50,
validation_data=(X_test, y_test),
# class_weight=class_weights,
shuffle=False)

#model.save('test.hdf5')
model.save('sandstone_50_epochs_catXentropy_acc.hdf5')
############################################################
# Evaluate the model
# evaluate model
_, acc = model.evaluate(X_test, y_test)
print("Accuracy is = ", (acc * 100.0), "%")

###
# plot the training and validation accuracy and loss at each epoch
loss = history.history['loss']
val_loss = history.history['val_loss']
epochs = range(1, len(loss) + 1)
plt.plot(epochs, loss, 'y', label='Training loss')
plt.plot(epochs, val_loss, 'r', label='Validation loss')
plt.title('Training and validation loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.legend()
plt.show()

acc = history.history['accuracy']
val_acc = history.history['val_accuracy']

plt.plot(epochs, acc, 'y', label='Training Accuracy')
plt.plot(epochs, val_acc, 'r', label='Validation Accuracy')
plt.title('Training and validation Accuracy')
plt.xlabel('Epochs')
plt.ylabel('Accuracy')
plt.legend()
plt.show()

##################################
# model = get_model()
model.load_weights('sandstone_50_epochs_catXentropy_acc.hdf5')
# model.load_weights('sandstone_50_epochs_catXentropy_acc_with_weights.hdf5')

# IOU
y_pred = model.predict(X_test)
y_pred_argmax = np.argmax(y_pred, axis=3)

##################################################

# Using built in keras function
from tensorflow.keras.metrics import MeanIoU

import random

test_img_number = random.randint(1, len(X_test))
test_img = X_test[test_img_number]
ground_truth = y_test[test_img_number]
test_img_norm = test_img[:, :, 0][:, :, None]
test_img_input = np.expand_dims(test_img_norm, 0)
prediction = (model.predict(test_img_input))
predicted_img = np.argmax(prediction, axis=3)[0, :, :]
print(test_img_number)
plt.figure(figsize=(12, 8))
plt.subplot(231)
plt.title('Testing Image')
plt.imshow(test_img[:, :, 0], cmap='gray')

plt.subplot(232)
plt.title('Testing Label')
plt.imshow(ground_truth[:, :, 0], cmap='jet')
plt.colorbar()


plt.subplot(233)
plt.title('Prediction on test image')
plt.imshow(predicted_img, cmap='jet')
plt.colorbar()
plt.show()

print(test_img[:, :, 0])
print(ground_truth[:, :, 0])
print(predicted_img)
#######################################################################
import random

train_img_number = random.randint(0, len(X_train))
train_img = X_train[train_img_number]
#print(test_img)
ground_truth = y_train[train_img_number]
train_img_norm = train_img[:, :, 0][:, :, None]
train_img_input = np.expand_dims(train_img_norm, 0)
prediction = (model.predict(train_img_input))
predicted_img = np.argmax(prediction, axis=3)[0, :, :]
print(train_img_number)

plt.figure(figsize=(12, 8))
plt.subplot(231)
plt.title('Training Image')
plt.imshow(train_img[:, :, 0], cmap='gray')

plt.subplot(232)
plt.title('Training Label')
plt.imshow(ground_truth[:, :, 0], cmap='jet')
plt.colorbar()


plt.subplot(233)
plt.title('Prediction on train image')
plt.imshow(predicted_img, cmap='jet')
plt.colorbar()
plt.show()

print(train_img[:, :, 0])
print(ground_truth[:, :, 0])
print(predicted_img)