stf/stf-api-tools/src/stf_tools/.ipynb_checkpoints/silent-checkpoint.py

from concurrent.futures import ThreadPoolExecutor

from pydub import AudioSegment


def create_silent_video(template, writer):
    reader = iter(template._get_reader(num_skip_frames=0))
    audio_segment = AudioSegment.silent(10000)
    pivot = 0

    with ThreadPoolExecutor(4) as p:
        try:
            while True:
                gen_infer = template.gen_infer_concurrent(
                    p,
                    audio_segment,
                    pivot,
                )
                for idx, (it, chunk) in enumerate(gen_infer, pivot):
                    frame = next(reader)
                    composed = template.compose(idx, frame, it)
                    writer.video_writer.write(composed)
                    writer.audio_writer.write(chunk)
                pivot = idx + 1
        except StopIteration as e:
            pass

    writer.finish(forced=False)