Source code for src.converters.ffmpeg

"""
Author: Ismael Seidel (ismael.seidel@ufsc.br)
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)

Description:
    This module contains the implementation of PPM->YUV conversion. It is a wrapper class for calling FFMPEG.
    FFMPEG will perform the conversion considering the aspects of the LF, such as width, height, number of views, encoding order.
    Also, more importantly, it will deal with the colour conversion.
"""

import os
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Union

from lfc_toolkit.src.data_handlers.encoding_orders import (
    ScanFunctionType, get_serpentine_scan_list)
from lfc_toolkit.src.data_handlers.formatters import \
    get_view_name_with_leading_zeros
from lfc_toolkit.src.data_handlers.lightfield import \
    RAW_BT709_FR_YUV444p10le_LightField_Data

if TYPE_CHECKING:
    from lfc_toolkit.src.data_handlers.lightfield import RAWLightFieldData


FFMPEG_COMMAND_TEMPLATE = """
ffmpeg 
-y
-r 30
-f concat
-safe 0
-i {list_filename}
-s {view_width}x{view_height}
-framerate 30
-c:v rawvideo
-vf scale=out_color_matrix=bt709:out_range=full,format={pix_fmt} 
-colorspace bt709
-color_primaries bt709
{output_filename}
"""



[docs]
def generate_list_for_ffmpeg(
    n_views_width: int,
    n_views_height: int,
    initial_width: int = 0,
    initial_height: int = 0,
    step_width: int = 1,
    step_height: int = 1,
    views_path: Union[str, Path] = ".",
    scan_order: ScanFunctionType = get_serpentine_scan_list,
) -> str:
    """
    Generate a list of views for FFMPEG conversion.

    :param n_views_width: Number of views in width
    :type n_views_width: int
    :param n_views_height: Number of views in height
    :type n_views_height: int
    :param initial_width: Initial width offset
    :type initial_width: int
    :param initial_height: Initial height offset
    :type initial_height: int
    :param step_width: Step size in width
    :type step_width: int
    :param step_height: Step size in height
    :type step_height: int
    :param views_path: Path to the views
    :type views_path: Union[str, Path]
    :param scan_order: Function defining the scan order
    :type scan_order: ScanFunctionType
    :return: List string for FFMPEG
    :rtype: str
    """
    scan_list = scan_order(
        n_views_width * step_width,
        n_views_height * step_height,
        initial_width,
        initial_height,
        step_width,
        step_height,
    )

    list_string = ""
    for j, i in scan_list:
        view_name = get_view_name_with_leading_zeros(j, i)
        list_string += f"file '{views_path}/{view_name}.ppm'\nduration 1\n"

    return list_string




[docs]
def create_list_file_for_ffmpeg(
    list_filename: Union[str, Path],
    n_views_width: int,
    n_views_height: int,
    initial_width: int = 0,
    initial_height: int = 0,
    step_width: int = 1,
    step_height: int = 1,
    views_path: Union[str, Path] = ".",
    scan_order: ScanFunctionType = get_serpentine_scan_list,
) -> None:
    """
    Create a list file for FFMPEG conversion.

    :param list_filename: Filename for the list file
    :type list_filename: Union[str, Path]
    :param n_views_width: Number of views in width
    :type n_views_width: int
    :param n_views_height: Number of views in height
    :type n_views_height: int
    :param initial_width: Initial width offset
    :type initial_width: int
    :param initial_height: Initial height offset
    :type initial_height: int
    :param step_width: Step size in width
    :type step_width: int
    :param step_height: Step size in height
    :type step_height: int
    :param views_path: Path to the views
    :type views_path: Union[str, Path]
    :param scan_order: Function defining the scan order
    :type scan_order: ScanFunctionType
    :return: None
    :rtype: None
    """
    list_string = generate_list_for_ffmpeg(
        n_views_width,
        n_views_height,
        initial_width,
        initial_height,
        step_width,
        step_height,
        views_path,
        scan_order,
    )

    Path(list_filename).write_text(list_string)

    with open(list_filename, "w") as text_file:
        text_file.write(list_string)




[docs]
class FFMPEGLightFieldFormatConverter:

[docs]
    def convert_rgb_ppm_lf_to_bt709fr_yuv(
        raw_light_field_rgb_ppm: "RAWLightFieldData",
        yuv_path: Union[str, Path],
        remove_after_using: bool = False,
    ) -> "RAW_BT709_FR_YUV444p10le_LightField_Data":
        """
        Convert RAW RGB PPM light field to BT.709 Full Range YUV.

        :param raw_light_field_rgb_ppm: RAW RGB PPM light field data
        :type raw_light_field_rgb_ppm: RAWLightFieldData
        :param yuv_path: Path to save the YUV files
        :type yuv_path: Union[str, Path]
        :param remove_after_using: Whether to remove intermediate files
        :type remove_after_using: bool
        :return: Converted YUV light field data
        :rtype: RAW_BT709_FR_YUV444p10le_LightField_Data
        """
        assert raw_light_field_rgb_ppm.colour_space == "sRGB"
        assert raw_light_field_rgb_ppm.pix_fmt == "PPM"

        yuv_path = Path(yuv_path)
        yuv_path.mkdir(parents=True, exist_ok=True)
        lightfield = raw_light_field_rgb_ppm.copy()

        output_raw_lf = RAW_BT709_FR_YUV444p10le_LightField_Data(
            lightfield=lightfield,
            yuv_path=yuv_path,
            bpp_for_naming=raw_light_field_rgb_ppm.bpp_for_naming,
            remove_after_using=remove_after_using,
        )

        # todo define a path to include the list file
        list_filename = yuv_path / f"list_{raw_light_field_rgb_ppm.name}.txt"
        create_list_file_for_ffmpeg(
            list_filename,
            raw_light_field_rgb_ppm.n_views_width,
            raw_light_field_rgb_ppm.n_views_height,
            raw_light_field_rgb_ppm.initial_width,
            raw_light_field_rgb_ppm.initial_height,
            raw_light_field_rgb_ppm.step_width,
            raw_light_field_rgb_ppm.step_height,
            raw_light_field_rgb_ppm.raw_path,
            raw_light_field_rgb_ppm.scan_order,
        )

        command = FFMPEG_COMMAND_TEMPLATE.format(
            list_filename=list_filename,
            view_width=output_raw_lf.view_width,
            view_height=output_raw_lf.view_height,
            pix_fmt=output_raw_lf.pix_fmt,
            output_filename=output_raw_lf.raw_path,
        ).split()

        res = subprocess.run(command, capture_output=True)

        if res.returncode != 0:
            print(command)
            raise Exception(res.stderr)

        if remove_after_using:
            os.remove(list_filename)

        return output_raw_lf