Skip to content

video

Video creation module for Direktor.

direktor.core.video

Video creation module for Direktor.

This module handles combining audio and images into the final video.

create_video(audio_file, image_files, image_prompts, temp_dir, keywords=None)

Create a video from audio and images with optional keyword overlays.

Parameters:

Name Type Description Default
audio_file str | PathLike[str]

Path to the audio file.

required
image_files Sequence[str | PathLike[str]]

List of paths to image files.

required
image_prompts Sequence[dict[str, Any]]

List of image prompts with timestamps.

required
temp_dir str | PathLike[str]

Temporary directory for intermediate files.

required
keywords Sequence[tuple[str, float, float]] | None

Optional sequence of (keyword, start_time, end_time) tuples for overlays.

None

Returns:

Type Description
Path

Path to the output video file.

Raises:

Type Description
VideoCreationError

If FFmpeg fails or inputs are inconsistent.

Source code in direktor/core/video.py
def create_video(
    audio_file: str | os.PathLike[str],
    image_files: Sequence[str | os.PathLike[str]],
    image_prompts: Sequence[dict[str, Any]],
    temp_dir: str | os.PathLike[str],
    keywords: Sequence[tuple[str, float, float]] | None = None,
) -> Path:
    """Create a video from audio and images with optional keyword overlays.

    Args:
        audio_file: Path to the audio file.
        image_files: List of paths to image files.
        image_prompts: List of image prompts with timestamps.
        temp_dir: Temporary directory for intermediate files.
        keywords: Optional sequence of ``(keyword, start_time, end_time)``
            tuples for overlays.

    Returns:
        Path to the output video file.

    Raises:
        VideoCreationError: If FFmpeg fails or inputs are inconsistent.
    """
    temp_path = Path(temp_dir)
    audio_path = Path(audio_file)
    output_file = temp_path / "output.mp4"

    if output_file.exists():
        logger.info("Video already exists: %s", output_file)
        return output_file

    if len(image_files) != len(image_prompts):
        raise VideoCreationError(
            f"Number of images ({len(image_files)}) does not match number of "
            f"prompts ({len(image_prompts)})."
        )

    png_images = [_convert_to_png(Path(img), temp_path) for img in image_files]
    concat_file = temp_path / "concat.txt"
    temp_video = temp_path / "temp_video.mp4"

    try:
        _build_concat_file(png_images, image_prompts, concat_file)

        run_subprocess(
            [
                "ffmpeg",
                "-f",
                "concat",
                "-safe",
                "0",
                "-i",
                str(concat_file),
                "-vsync",
                "vfr",
                "-pix_fmt",
                "yuv420p",
                "-vf",
                "scale=1920:1080:force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2",
                str(temp_video),
            ],
            cwd=temp_path,
        )

        drawtext_filter = _build_drawtext_filter(keywords or [], Path(FONT_PATH))
        output_command = [
            "ffmpeg",
            "-i",
            str(temp_video),
            "-i",
            str(audio_path),
            "-c:a",
            "aac",
            "-shortest",
            str(output_file),
        ]
        if drawtext_filter:
            output_command[-4:-4] = ["-filter_complex", drawtext_filter]

        run_subprocess(output_command, cwd=temp_path)
    except Exception as e:
        raise VideoCreationError(f"Video creation failed: {e}") from e
    finally:
        for path in [concat_file, temp_video]:
            try:
                path.unlink()
            except OSError:
                logger.warning("Could not remove temporary file %s", path)
        for png_file in png_images:
            if png_file.suffix.lower() == ".png" and png_file not in [
                Path(img) for img in image_files
            ]:
                try:
                    png_file.unlink()
                except OSError:
                    logger.warning("Could not remove converted image %s", png_file)

    logger.info("Video created: %s", output_file)
    return output_file