import torch
import time
import cv2
import traceback
import subprocess
from moviepy.editor import VideoFileClip, AudioFileClip
class Encoder:
def __init__(self, log_dir, logger):
self.logger = logger
def write_video(self, frame_list, video_path, video_output_path):
try :
last_frame = frame_list[-1]
cap = cv2.VideoCapture(video_path)
width = cap.get(cv2.CAP_PROP_FRAME_WIDTH) # float
height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT) # float
fps = cap.get(cv2.CAP_PROP_FPS)
vid_writer = cv2.VideoWriter(
video_output_path, cv2.VideoWriter_fourcc(*'DIVX'), fps, (int(width), int(height))
)
cnt = 0
if cap.isOpened() :
while True:
cnt+=1
ret, frame = cap.read()
if ret == False :
break
if cnt < 0:
continue
num_frame = round(cap.get(cv2.CAP_PROP_POS_FRAMES))
if int(num_frame) not in frame_list :
continue
print('num_frame',num_frame)
vid_writer.write(frame)
ch = cv2.waitKey(1)
if ch == 27 or ch == ord("q") or ch == ord("Q"):
break
if int(num_frame) == int(last_frame) :
break
vid_writer.release()
return True, fps
except Exception as e :
print(f"[write_viddeo] {e}\n{traceback.format_exc()}")
return False, None
def write_audio(self, segments, fps, video_path, audio_output_path) :
try :
temp_files = []
temp_concat_file = 'temp_concat.txt'
# 각 구간별로 오디오 추출
for segment in segments:
start_frame = segment['start_frame']
end_frame = segment['end_frame']
output_file = segment['output_file']
# ffmpeg 명령어 실행하여 오디오 추출
command = f'ffmpeg -i {video_path} -ss {start_frame/fps} -to {end_frame/fps} -vn -acodec libmp3lame -ab 192k {output_file}'
subprocess.call(command, shell=True)
# 추출된 오디오 파일을 임시 파일 리스트에 추가
temp_files.append(output_file)
# 오디오 파일 합치기 위한 임시 파일 목록 생성
with open(temp_concat_file, 'w') as f:
for temp_file in temp_files:
f.write(f"file '{temp_file}'\n")
# ffmpeg 명령어 실행하여 오디오 파일 합치기
command = f'ffmpeg -f concat -safe 0 -i {temp_concat_file} -c copy {audio_output_path}'
subprocess.call(command, shell=True)
# 임시 파일 및 임시 파일 목록 삭제
for temp_file in temp_files:
subprocess.call(f'rm {temp_file}', shell=True)
subprocess.call(f'rm {temp_concat_file}', shell=True)
return True
except Exception as e :
print(f"[write_audio] {e}\n{traceback.format_exc()}")
return False
def merge_file(self, video_output_path, audio_output_path, merged_output_file):
try :
# Load the video file
video = VideoFileClip(video_output_path)
# Load the audio file
audio = AudioFileClip(audio_output_path)
# # Set the audio duration to match the video duration
# audio = audio.set_duration(video.duration)
# Set the audio to the video
video = video.set_audio(audio)
# Write the final video file
video.write_videofile(merged_output_file, codec="libx264", audio_codec="aac")
return True
except Exception as e :
print(f"[merge_file] {e}\n{traceback.format_exc()}")
return False