Нейросеть на Pytorch не работает без Dataloader а
попытки подать на сетку данные без dataloader а не увенчались успехом ,хотя размер и тип данных совпадают
вот код
debug = True
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader
import torchvision.transforms as transforms
import numpy as np
import cv2
import pickle
import matplotlib
from matplotlib.pyplot import imshow
import time
print("Started!")
print('choosing devise for train')
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print("device = ", device)
file_path = "C:\AI/dataset_v3.pkl"
print('loading dataset file from path:{}',format(file_path))
with open(file_path, 'rb') as f:
X, Y = pickle.load(f)
print('load ok')
print('preparsing')
Y1 = Y[:, 0:1]
Y2 = Y[:, 1:2]
Y3 = Y[:, 2:3]
Y4 = Y[:, 3:4]
Y5 = Y[:, 2:3]
print('preparsing ok')
print("preparing all dataset for last test and display")
N1 = np.zeros(len(Y1))
N2 = np.zeros(len(Y2))
N3 = np.zeros(len(Y3))
N4 = np.zeros(len(Y4))
N5 = np.zeros(len(Y5))
for c in range (len(Y1)):
N1[c] = Y1[c, 0]
N2[c] = Y2[c, 0]
N3[c] = Y3[c, 0]
N4[c] = Y4[c, 0]
N5[c] = Y5[c, 0]
del(c)
print("preparing done")
def mp(x, in_min, in_max, out_min, out_max):
return (x - in_min) * (out_max - out_min) / (in_max - in_min) + out_min
print("want to see all dataset ? it will take about {} minutes ({} frames)".format(str(round(10*len(X)/6000,2)),len(X)))
inp=input("enter Y/N ")
if inp.upper().startswith("Y"):
print("press //space// to stop show")
for i in range(len(X)):
ext = cv2.resize(X[i], (640, 480))
x = N1[i]
y = N2[i]
m1 = N3[i]
m2 = N4[i]
dist = N5[i]
print(x, y)
x = int(mp(x, -256, 256, 10, 630))
y = int(mp(y, 256, -256, 10, 470))
m1 = int(mp(m1, 256, -256, 10, 470))
m2 = int(mp(m2, 256, -256, 10, 470))
dist = int(mp(dist, -256, 256, 10, 470))
ext = cv2.rectangle(ext, (0, 8), (10, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (305, 8), (315, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (325, 8), (335, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (x - 2, 450), (x + 2, 470), (0, 255, 0), 2)
ext = cv2.rectangle(ext, (0, y - 2), (10, y + 2), (0, 0, 255), 2)
ext = cv2.rectangle(ext, (305, m1 - 2), (315, m1 + 2), (0, 255, 255), 2)
ext = cv2.rectangle(ext, (325, m2 - 2), (335, m2 + 2), (0, 255, 255), 2)
ext = cv2.line(ext, (0, 240), (640, 240), (0, 0, 0), 1)
if cv2.waitKey(10) == 32:
break
cv2.imshow("test", ext)
cv2.destroyAllWindows()
print("what type of remapping all data to work with nerual network we will use ? 1)Y = X / 255 2)Y = (x+255) / 510. for default enter 1")
inp=input("enter 1 / 2 ")
if inp.upper().startswith("1"):
substr_type=1
Y1 = Y1 / 255.
Y2 = Y2 / 255.
Y3 = Y3 / 255.
Y4 = Y4 / 255.
else:
substr_type=2
Y1 = (Y1+255 )/ 510.
Y2 = (Y2+255 )/ 510.
Y3 = (Y3+255 )/ 510.
Y4 = (Y4+255) / 510.
X = X / 255.
print('X.shape: ', X.shape)
print('Y.shape: ', Y1.shape)
print("Test cutoff",Y1[0:10])
def unison_shuffled_copies(X, Y1, Y2):
assert len(X) == len(Y)
p = np.random.permutation(len(X))
# print(p)
return X[p], Y1[p], Y2[p]
class CNN(nn.Module):
def __init__(self, in_channels=3, num_classes=4):
super(CNN, self).__init__()
self.conv1 = nn.Conv2d(in_channels=3, out_channels=16, kernel_size=(
5, 5), stride=(2, 2), padding=(1, 1))
#self.pool = nn.MaxPool2d(kernel_size=(24, ), stride=(2, 2))
self.conv2 = nn.Conv2d(in_channels=16, out_channels=32, kernel_size=(
5, 5), stride=(2, 2), padding=(1, 1))
self.conv3 = nn.Conv2d(in_channels=32, out_channels=64, kernel_size=(
5, 5), stride=(2, 2), padding=(1, 1))
self.conv4 = nn.Conv2d(in_channels=64, out_channels=128, kernel_size=(
5, 5), stride=(2, 2), padding=(1, 1))
self.drop = nn.Dropout(p=0.5)
self.fc1 = nn.Linear(6912, 512)
self.fc2 = nn.Linear(512, num_classes)
# self.fc1 = nn.Linear(128*1*1, num_classes)
def forward(self, x):
# print("input", x.shape)
#print(x.shape)
x = F.relu(self.conv1(x))
#print("conv1 shape", x.shape)
#x = self.pool(x)
# print("pool shape", x.shape)
x = F.relu(self.conv2(x))
#print("conv2 shape", x.shape)
# x = self.pool(x)
# print("pool shape", x.shape)v
x = F.relu(self.conv3(x))
#print("conv3 shape", x.shape)
# x = self.pool(x)
x = F.relu(self.conv4(x))
#print("conv4 shape", x.shape)
#x = self.pool(x)
# print("conv4 pool shape", x.shape)
#x = self.drop(x)
#print("x shape", x.shape)
x = torch.flatten(x)
#print("xr shape", x.shape)
x = self.fc1(x)
#print("fc1", x.shape)
x = self.fc2(x)
#print("fc2 shape", x.shape)
return x
print("cuting")
train_cutoff = int(len(X) * .9)
val_cutoff = train_cutoff + int(len(X) * .1)
print("train cutoff len :{} validate cutoff len:{},summary:{}".format(train_cutoff, val_cutoff, len(X)))
train_X, train_Y1, train_Y2, train_Y3, train_Y4, train_Y5 = X[:train_cutoff], Y1[:train_cutoff], Y2[:train_cutoff], Y3[:train_cutoff], Y4[:train_cutoff], Y5[:train_cutoff]
val_X, val_Y1, val_Y2, val_Y3, val_Y4, val_Y5 = X[train_cutoff:val_cutoff], Y1[train_cutoff:val_cutoff], Y2[train_cutoff:val_cutoff], Y3[train_cutoff:val_cutoff], Y4[train_cutoff:val_cutoff], Y5[train_cutoff:val_cutoff]
print("cut ok")
my_x = []
my_y = []
print("get ready to wait....... it's will be long. Creating BIG Numpy tensor with dataset ")
for i in range(len(train_X)):
my_x.append(np.array(np.transpose(train_X[i], (2, 0, 1))))
append_list=train_Y1[i][0], train_Y2[i][0], train_Y3[i][0], train_Y4[i][0]
my_y .append(append_list)
print(round(mp(len(train_X)-1-i, 0, len(train_X)-1, 100, 0),2), "% ")
print("creating Numpy tensor Done")
print("creating Torch tensor")
tensor_x = torch.Tensor(my_x)
tensor_y = torch.Tensor(my_y)
print("Creating Torch tensor ok")
print("to clear memeory i will delete BIG old Numpy tensor")
del(my_x,my_y)
print("Del ok")
print("creating dataloader")
train_dataset = torch.utils.data.TensorDataset(tensor_x, tensor_y)
train_loader = torch.utils.data.DataLoader(
train_dataset, batch_size=1, shuffle=False)
print("creating dataloader done")
num_epochs = 2
in_channel = 1
num_classes = 10
learning_rate = 0.001
batch_size = 4
print("create net")
# Initialize network
model = CNN().to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
loss_list = []
loss = 0
for epoch in range(num_epochs):
i = 0
for data in train_loader:
i += 1
in_images, out_images = data
optimizer.zero_grad()
in_i = torch.FloatTensor(in_images.type(torch.FloatTensor)).to(device)
out_i = torch.FloatTensor(out_images.type(torch.FloatTensor)).to(device)
outputs = model(in_i) # torch Tensor of shape (32, num_classes)
loss = criterion(outputs, out_i)
loss.backward()
optimizer.step()
print("Loss = ", round(float(loss.cpu().detach().numpy()), 4), i)
loss_list.append(float(loss.cpu().detach().numpy()))
print(f'Epoch:{epoch} , loass:{loss}')
torch.save(model, "model_pytorch_0.plk")
time.sleep(5)
print("Here is loss schedule")
plt.plot(loss_list)
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train'], loc='upper left')
plt.show()
data1 = []
data2 = []
data3 = []
data4 = []
for i in range(len(X)):
img = np.transpose(X[i], (2, 0, 1))
#img = np.expand_dims(img, axis=0)
# print(img.shape)
tensor = torch.FloatTensor(img).to(device)
tensor = tensor.unsqueeze(0).to(device)
out = model(tensor)
out = out.cpu().detach().numpy()
x, y, m1, m2 = out
if substr_type==2:
x = (x[0][0]-0.5)*510
y = (y[0][0]-0.5)*510
m1 = (m1[0][0]-0.5)*510
m2 = (m2[0][0]-0.5)*510
if substr_type==1:
x *= 255
y *= 255
m1 *= 255
m2 *= 255
data1.append(x)
data2.append(y)
data3.append(m1)
data4.append(m2)
print(i)
P0 = np.array(data1)
P1 = np.array(data2)
P2 = np.array(data3)
P3 = np.array(data4)
print(P0.shape)
print(P0)
print(P0[0].shape)
if debug:
print("len", len(X))
plt.title('x')
plt.ylabel('ext')
plt.xlabel('sample number')
plt.grid(True)
plt.plot(data1)
plt.plot(N1)
plt.legend(['predicted', 'actual'], loc='upper left')
plt.show()
plt.title('y')
plt.ylabel('ext')
plt.xlabel('sample number')
plt.grid(True)
plt.plot(data2)
plt.plot(N2)
plt.legend(['predicted', 'actual'], loc='upper left')
plt.show()
plt.title('m1')
plt.ylabel('ext')
plt.xlabel('sample number')
plt.grid(True)
plt.plot(data3)
plt.plot(N3)
plt.legend(['predicted', 'actual'], loc='upper left')
plt.show()
plt.title('m2')
plt.ylabel('ext')
plt.xlabel('sample number')
plt.grid(True)
plt.plot(data4)
plt.plot(N4)
plt.legend(['predicted', 'actual'], loc='upper left')
plt.show()
print("want to see all dataset and nerual network results? it will take about {} minutes ({} frames)".format(str(round(10*len(X)/6000,2)),len(X)))
inp=input("enter Y/N ")
if inp.upper().startswith("Y"):
print("press //space// to stop show")
for i in range(len(X)):
ext = cv2.resize(X[i], (640, 480))
x = N1[i]
y = N2[i]
m1 = N4[i]
m2 = N3[i]
dist = Y5[i]
x_pred = data1[i]
y_pred = data2[i]
m1_pred = data3[i]
m2_pred = data4[i]
# print(data1[i],x_pred)
# print(x, y)
x = int(mp(x, -256, 256, 10, 630))
y = int(mp(y, 256, -256, 10, 470))
m1 = int(mp(m1, 256, -256, 10, 470))
m2 = int(mp(m2, 256, -256, 10, 470))
x_pred = int(mp(x_pred, -256, 256, 10, 630))
y_pred = int(mp(y_pred, 256, -256, 10, 470))
m1_pred = int(mp(m1_pred, 256, -256, 10, 470))
m2_pred = int(mp(m2_pred, 256, -256, 10, 470))
dist = int(mp(dist, -256, 256, 10, 470))
ext = cv2.rectangle(ext, (0, 8), (10, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (305, 8), (315, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (325, 8), (335, 472), (0, 0, 0), 1)
ext = cv2.rectangle(ext, (x - 2, 450), (x + 2, 470), (0, 255, 0), 2)
ext = cv2.rectangle(ext, (0, y - 2), (10, y + 2), (0, 0, 255), 2)
ext = cv2.rectangle(ext, (305, m1 - 2),
(315, m1 + 2), (0, 255, 255), 2)
ext = cv2.rectangle(ext, (325, m2 - 2),
(335, m2 + 2), (0, 255, 255), 2)
ext = cv2.rectangle(ext, (x_pred - 1, 450),
(x_pred + 1, 470), (255, 0, 0), 2)
ext = cv2.rectangle(ext, (0, y_pred - 1),
(10, y_pred + 1), (255, 0, 0), 2)
ext = cv2.rectangle(ext, (305, m1_pred - 1),
(315, m1_pred + 1), (255, 255, 0), 2)
ext = cv2.rectangle(ext, (325, m2_pred - 1),
(335, m2_pred + 1), (255, 255, 0), 2)
ext = cv2.line(ext, (0, 240), (640, 240), (0, 0, 0), 1)
cv2.imshow("test", ext)
if cv2.waitKey(10)==32:
break
датасет прилагается https://drive.google.com/file/d/1ogWA6SCBHw7uHEsyhPK0yDByy2xMgcGY/view?usp=sharing
со строки 220 обучение со строки 264 проверка