batch_renderer.py

# batch_renderer.py

import os
import subprocess

from pathlib import Path

from renderer import render

from usage_tracker import (
    build_usage_map,
    consume_scene,
)

from config import (
    OUTPUT_DIR,
)


# =====================================================
# CHUNK LIST
# =====================================================

def chunk_list(
    items,
    size
):

    for i in range(
        0,
        len(items),
        size
    ):

        yield items[
            i:i+size
        ]


# =====================================================
# BUILD PART PATH
# =====================================================

def build_part_path(
    index
):

    return (

        OUTPUT_DIR

        /

        f"part_{index:03d}.mp4"

    )


# =====================================================
# RENDER PARTS
# =====================================================

def render_parts(
    timeline,
    batch_size=50
):

    if not timeline:
        raise ValueError(
            "Timeline is empty"
        )

    usage_map = build_usage_map(
        timeline
    )

    part_files = []

    batches = list(

        chunk_list(
            timeline,
            batch_size
        )

    )

    total_batches = len(
        batches
    )

    print()

    print(
        f"[Batch] "
        f"Total Batches: "
        f"{total_batches}"
    )

    print()

    for idx, batch in enumerate(
        batches,
        start=1
    ):

        part_path = build_part_path(
            idx
        )

        print(
            f"[Batch] "
            f"Rendering "
            f"{idx}/{total_batches}"
        )

        render(
            batch,
            part_path
        )

        part_files.append(
            str(part_path)
        )

        # -------------------------
        # CLEANUP USED VARIANTS
        # -------------------------

        deleted = 0

        for scene in batch:

            delete_file = consume_scene(
                usage_map,
                scene
            )

            if not delete_file:
                continue

            try:

                if os.path.exists(
                    scene["path"]
                ):

                    os.remove(
                        scene["path"]
                    )

                    deleted += 1

            except Exception as e:

                print(
                    f"[Cleanup] "
                    f"{scene['path']}"
                )

                print(e)

        print(
            f"[Batch] "
            f"Deleted "
            f"{deleted} clips"
        )

        print()

    return part_files


# =====================================================
# CONCAT PARTS
# =====================================================

def concat_parts(
    part_files
):

    if not part_files:

        raise ValueError(
            "No part files"
        )

    concat_txt = (
        OUTPUT_DIR
        / "parts.txt"
    )

    with open(
        concat_txt,
        "w",
        encoding="utf-8"
    ) as f:

        for part in part_files:

            path = (
                Path(part)
                .resolve()
                .as_posix()
            )

            f.write(
                f"file '{path}'\n"
            )

    final_video = (
        OUTPUT_DIR
        / "video_only.mp4"
    )

    cmd = [

        "ffmpeg",
        "-y",

        "-f",
        "concat",

        "-safe",
        "0",

        "-i",
        str(concat_txt),

        "-c",
        "copy",

        str(final_video)

    ]

    subprocess.run(
        cmd,
        check=True,
        capture_output=True
    )

    # -------------------------
    # DELETE PART FILES
    # -------------------------

    for part in part_files:

        try:

            os.remove(
                part
            )

        except:
            pass

    try:

        os.remove(
            concat_txt
        )

    except:
        pass

    return str(
        final_video
    )


# =====================================================
# RENDER FINAL VIDEO
# =====================================================

def render_final_video(
    timeline,
    batch_size=50
):

    parts = render_parts(
        timeline,
        batch_size
    )

    return concat_parts(
        parts
    )


# =====================================================
# TEST
# =====================================================

if __name__ == "__main__":

    timeline = [

        {
            "path": "clips/1_A.mp4",
            "duration": 7.0
        },

        {
            "path": "clips/2_A.mp4",
            "duration": 7.0
        },

        {
            "path": "clips/3_A.mp4",
            "duration": 7.0
        },

    ]

    output = render_final_video(
        timeline,
        batch_size=2
    )

    print()

    print(
        "Output:",
        output
    )