Spaces:
Paused
Paused
File size: 3,959 Bytes
9411c06 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
import copy
import os
import subprocess
from src.models.taskers.tasker import Tasker
from src.models.taskers.checker import Checker
from src.models.utils.dirs import *
from src.models.utils import get_logger
logger = get_logger(name='Normalizer', is_stream=True)
class Normalizer(Tasker):
FPS: int = 25
SR: int = 16_000
AV1_TO_H264_CMD = [
'ffmpeg', '-y',
'-loglevel', 'panic',
'-c:v', 'libaom-av1',
'-i', 'video_path',
'-c:v', 'libx264',
'-c:a', 'pcm_s16le',
'-ss', '0',
'-t', 'duration',
'-f', 'avi',
'output_path',
]
ANY_TO_H264_PCM_CMD = [
'ffmpeg', '-y',
'-loglevel', 'panic',
'-i', 'video_path',
'-map', '0:v:0',
'-map', '0:a:0',
'-c:v', 'libx264',
'-c:a', 'pcm_s16le',
'-ss', '0',
'-t', 'duration',
'-f', 'avi',
'output_path',
]
ADD_BLANK_SOUND_CMD = [
'ffmpeg', '-y',
'-loglevel', 'panic',
'-f', 'lavfi',
'-i', 'anullsrc=channel_layout=mono',
'-i', 'video_path',
'-r', '%d' % FPS,
'-ar', '%d' % SR,
'-c:v', 'copy',
'-c:a', 'pcm_s16le',
'-ss', '0',
'-t', 'duration',
'-f', 'avi',
'output_path'
]
ADD_BACKGROUND_VISUAL_CMD = [
'ffmpeg', '-y',
'-loglevel', 'panic',
'-f', 'lavfi',
'-i', 'color=c=black',
'-i', 'video_path',
'-c:v', 'libx264',
'-c:a', 'pcm_s16le',
'-ss', '0',
'-t', 'duration',
'-r', '%d' % FPS,
'-pix_fmt', 'yuv420p',
'-vf', 'scale=400:400',
'-f', 'avi',
'output_path',
]
def __init__(self):
super().__init__()
def do(self, metadata_dict: dict, checker: Checker=Checker()):
if metadata_dict['has_v']:
if metadata_dict['v_codec'] == 'av1':
metadata_dict['video_path'] = self._normalize(
cmd_=self.AV1_TO_H264_CMD,
meta_dict=metadata_dict,
output_name='nv',
fail_msg="Convert vcodec 'av1' -> 'libx264' fail."
)
elif metadata_dict['v_codec'] != 'h264':
metadata_dict['video_path'] = self._normalize(
cmd_=self.ANY_TO_H264_PCM_CMD,
meta_dict=metadata_dict,
output_name='nv',
fail_msg="Convert vcodec to 'libx264' fail."
)
if not metadata_dict['has_a']:
metadata_dict['video_path'] = self._normalize(
cmd_=self.ADD_BLANK_SOUND_CMD,
meta_dict=metadata_dict,
output_name='na',
fail_msg="Add blank sound fail."
)
else:
metadata_dict['video_path'] = self._normalize(
cmd_=self.ADD_BACKGROUND_VISUAL_CMD,
meta_dict=metadata_dict,
output_name='nv',
fail_msg="Add background visual fail."
)
tmp = checker.do(video_path=metadata_dict['video_path'])
tmp['result_video_path'] = metadata_dict.get('result_video_path', metadata_dict['video_path'])
return tmp
def _normalize(self, cmd_: list, meta_dict: dict, output_name: str, fail_msg: str = None) -> str:
cmd = copy.copy(cmd_)
output_path = os.path.join(
_EXTERNAL_DIR, f'{output_name}.mp4'
)
cmd[cmd.index('video_path')] = meta_dict['video_path']
cmd[-1] = output_path
if 'duration' in cmd:
cmd[cmd.index('duration')] = str(meta_dict['duration'])
subprocess.run(cmd, shell=False, capture_output=False, stdout=None)
if not os.path.isfile(output_path):
logger.critical(fail_msg)
raise RuntimeError()
return output_path |