用于视频回归任务的长期循环卷积网络

defpreprocess(movie_name, train=True, n_subclips=3, subclip_duration=30, frequency=45, verbose=False):
"""Preprocesses a movie trailer making subclips and then extracting sequences of frames:param movie_name: movie name:train: boolean flag to determine whether preprocessing is performed for training videos or not:param n_subclips: number of subclips to make from the trailer:param subclip_duration: duration of a subclip:param frequency: frequency of extracting frames from subclips:param verbose: increase verbosity if True"""name='.'.join(movie_name.split('.')[:-1])
format_=movie_name.split('.')[-1]
ifnotformat_ok(format_):
print('Skipping file, which is not a video...')
returniftrain==True:
ifnotos.path.isdir(TRAINING_PATH):
os.mkdir(TRAINING_PATH)
DEST=TRAINING_PATHelse:
ifnotos.path.isdir(PREDICTION_PATH):
os.mkdir(PREDICTION_PATH)
DEST=PREDICTION_PATHifformat_=='flv':   #Decorddoesnotworkwithflvformatformat_='mov'#Extractingsubclipfromtrailerbase=10os.makedirs(f"{FPATH}/{name}", exist_ok=True)
foriinrange(n_subclips):
ifverbose:
print(f"{i} iteration...")
print("....Making subclip....")
try:
ffmpeg_extract_subclip(f"{FPATH}/{movie_name}", base, base+subclip_duration, targetname=f"{FPATH}/{name}/{i}.{format_}")
base=base+subclip_durationexceptBaseException:
print(f"Some error occured during {i+1} extraction")
continue#Checkifallsubclipswerecorrectlycreatedtry:
video=moviepy.editor.VideoFileClip(f"{FPATH}/{name}/{i}.{format_}")
ifint(video.duration) <=subclip_duration//2:raiseDurationErrorexcept:
print(f"The {i} subclip was not correctly created, deleting...")
os.remove(f"{FPATH}/{name}/{i}.{format_}")
continue#Creatingframesifverbose:
print("....Extracting frames....")
os.makedirs(f"{DEST}/{name+'_'+str(i)}", exist_ok=True)   #CreatingdirectoryforTraindatasettry:
video_to_frames(f"{FPATH}/{name}/{i}.{format_}", f"{DEST}/{name+'_'+str(i)}", overwrite=False, every=frequency)
except:
print("Error occured while executing VIDEO_TO_FRAMES")
os.rmdir(f"{DEST}/{name+'_'+str(i)}/{i}")
os.rmdir(f"{DEST}/{name+'_'+str(i)}")
continue#Deletedirectorywithsubclipsifnameinos.listdir(f"{FPATH}"):  
shutil.rmtree(f"{FPATH}/{name}")

classCNN(torch.nn.Module):
def__init__(self, channels1=10, channels2=20, embedding_size=15, activation='relu'):
super(CNN, self).__init__()
self.layer1=torch.nn.Sequential(torch.nn.Conv2d(3, channels1, kernel_size=5, padding=2),
activation_func(activation), torch.nn.BatchNorm2d(num_features=channels1),
torch.nn.MaxPool2d(2, 2))
self.layer2=torch.nn.Sequential(torch.nn.Conv2d(channels1, channels2, kernel_size=3, padding=1),
torch.nn.Dropout2d(p=0.2), activation_func(activation),
torch.nn.BatchNorm2d(num_features=channels2), torch.nn.MaxPool2d(2, 2))
self.encoder=torch.nn.Sequential(self.layer1, self.layer2)
self.decoder=torch.nn.Sequential(torch.nn.Linear(56*56*channels2, 6000), activation_func(activation),
torch.nn.Linear(6000, 500), activation_func(activation), torch.nn.Linear(500, embedding_size))
defforward(self, x):
x=self.encoder(x)
#flattenheatmapbeforeutilizingdenselayersx=x.view(x.size(0), x.size(1) *x.size(2) *x.size(3))
out=self.decoder(x)
returnoutclassLRCN(torch.nn.Module):
def__init__(self, channels1=10, channels2=20, embedding_size=15, LSTM_size=64, LSTM_layers=1, activation='relu'):
super(LRCN, self).__init__()
self.cnn=CNN(channels1, channels2, embedding_size, activation)
#batchfirst: dataformattedin (batch, seq, feature)
self.rnn=torch.nn.LSTM(input_size=embedding_size, hidden_size=LSTM_size, num_layers=LSTM_layers, batch_first=True)
self.linear=torch.nn.Linear(LSTM_size, 1)
defforward(self, x):
heatmaps= []
forseqinx:
heatmaps.append(self.cnn.forward(seq))
heatmaps=torch.stack(heatmaps)
out, (_, _) =self.rnn(heatmaps)
out=self.linear(out)
returnout[:,-1,:]

deffit(self, dir_names, X_test, y_test, lr=3e-4, loss_name='mse', n_epoch=5, batch_size=10, device='cpu', saving_results=False, use_tensorb=False, verbose=False):
optimizer=torch.optim.Adam(self.parameters(), lr=lr)
loss=loss_choice(loss_name) #Specifyinglossfunctiondir_names=list(filter(lambdax: os.path.isdir(f"{TRAINING_PATH}/{x}"), dir_names)) #Filteringwastefilesrandom.shuffle(dir_names)
train_loss_history= []
test_loss_history= []
learning_dir_names=dir_names.copy()
#Trainingmodelprint('---------------TRAINING----------------')
forepochinrange(n_epoch):
dir_names=learning_dir_names.copy()
train_loss=0foriinrange(0, len(learning_dir_names), batch_size):
optimizer.zero_grad()
print(dir_names)
X_batch, y_batch, dir_names=load_data(dir_names, train=True, verbose=verbose, batch_size=batch_size)  
X_batch=X_batch.to(device).float()
y_batch=y_batch.to(device).float()
preds=self.forward(X_batch).view(y_batch.size()[0])
loss_value=loss(preds, y_batch)
loss_value.backward()
train_loss+=loss_value.data.cpu()
optimizer.step()
train_loss_history.append(train_loss)
withtorch.no_grad():
test_preds=self.forward(X_test).view(y_test.size()[0])
test_loss_history.append(loss(test_preds, y_test).data.cpu())
print(f"{epoch+1}: {loss_name} = {test_loss_history[-1]}")
ifsaving_results==True:
torch.save(self.state_dict(), MODEL_WEIGHTS)
print('---------------------------------------')
return [train_loss_history, test_loss_history]