Skip to content

deffcode.ffhelper

Following methods are exclusively design to handle FFmpeg related tasks. These tasks includes validation of installed FFmpeg binaries, downloading of FFmpeg binaries(on Windows), and parsing of FFmpeg metadata into useful information using various pattern matching methods.

For usage examples, kindly refer our Basic Recipes 🍰 and Advanced Recipes 🥐

 

get_valid_ffmpeg_path

Validate the given FFmpeg path/binaries, and returns a valid FFmpeg executable path.

Parameters:

Name Type Description Default
custom_ffmpeg string

path to custom FFmpeg executables

''
is_windows boolean

is running on Windows OS?

False
ffmpeg_download_path string

FFmpeg static binaries download location (Windows only)

''
verbose bool

enables verbose for its operations

False

Returns: A valid FFmpeg executable path string.

Source code in deffcode/ffhelper.py
def get_valid_ffmpeg_path(
    custom_ffmpeg="", is_windows=False, ffmpeg_download_path="", verbose=False
):
    """
    ## get_valid_ffmpeg_path

    Validate the given FFmpeg path/binaries, and returns a valid FFmpeg executable path.

    Parameters:
        custom_ffmpeg (string): path to custom FFmpeg executables
        is_windows (boolean): is running on Windows OS?
        ffmpeg_download_path (string): FFmpeg static binaries download location _(Windows only)_
        verbose (bool): enables verbose for its operations

    **Returns:** A valid FFmpeg executable path string.
    """
    final_path = ""
    if is_windows:
        # checks if current os is windows
        if custom_ffmpeg:
            # if custom FFmpeg path is given assign to local variable
            final_path += custom_ffmpeg
        else:
            # otherwise auto-download them
            try:
                if not (ffmpeg_download_path):
                    # otherwise save to Temp Directory
                    import tempfile

                    ffmpeg_download_path = tempfile.gettempdir()

                verbose and logger.debug(
                    "FFmpeg Windows Download Path: {}".format(ffmpeg_download_path)
                )

                # download Binaries
                os_bit = (
                    ("win64" if platform.machine().endswith("64") else "win32")
                    if is_windows
                    else ""
                )
                _path = download_ffmpeg_binaries(
                    path=ffmpeg_download_path, os_windows=is_windows, os_bit=os_bit
                )
                # assign to local variable
                final_path += _path

            except Exception as e:
                # log if any error occurred
                logger.exception(str(e))
                logger.error(
                    "Error in downloading FFmpeg binaries, Check your network and Try again!"
                )
                return False

        if os.path.isfile(final_path):
            # check if valid FFmpeg file exist
            pass
        elif os.path.isfile(os.path.join(final_path, "ffmpeg.exe")):
            # check if FFmpeg directory exists, if does, then check for valid file
            final_path = os.path.join(final_path, "ffmpeg.exe")
        else:
            # else return False
            verbose and logger.debug(
                "No valid FFmpeg executables found at Custom FFmpeg path!"
            )
            return False
    else:
        # otherwise perform test for Unix
        if custom_ffmpeg:
            # if custom FFmpeg path is given assign to local variable
            if os.path.isfile(custom_ffmpeg):
                # check if valid FFmpeg file exist
                final_path += custom_ffmpeg
            elif os.path.isfile(os.path.join(custom_ffmpeg, "ffmpeg")):
                # check if FFmpeg directory exists, if does, then check for valid file
                final_path = os.path.join(custom_ffmpeg, "ffmpeg")
            else:
                # else return False
                verbose and logger.debug(
                    "No valid FFmpeg executables found at Custom FFmpeg path!"
                )
                return False
        else:
            # otherwise assign ffmpeg binaries from system
            final_path += "ffmpeg"

    verbose and logger.debug("Final FFmpeg Path: {}".format(final_path))

    # Final Auto-Validation for FFmeg Binaries. returns final path if test is passed
    return final_path if validate_ffmpeg(final_path, verbose=verbose) else False

 

get_valid_ffmpeg_path

Validate the given FFmpeg path/binaries, and returns a valid FFmpeg executable path.

Parameters:

Name Type Description Default
custom_ffmpeg string

path to custom FFmpeg executables

''
is_windows boolean

is running on Windows OS?

False
ffmpeg_download_path string

FFmpeg static binaries download location (Windows only)

''
verbose bool

enables verbose for its operations

False

Returns: A valid FFmpeg executable path string.

Source code in deffcode/ffhelper.py
def get_valid_ffmpeg_path(
    custom_ffmpeg="", is_windows=False, ffmpeg_download_path="", verbose=False
):
    """
    ## get_valid_ffmpeg_path

    Validate the given FFmpeg path/binaries, and returns a valid FFmpeg executable path.

    Parameters:
        custom_ffmpeg (string): path to custom FFmpeg executables
        is_windows (boolean): is running on Windows OS?
        ffmpeg_download_path (string): FFmpeg static binaries download location _(Windows only)_
        verbose (bool): enables verbose for its operations

    **Returns:** A valid FFmpeg executable path string.
    """
    final_path = ""
    if is_windows:
        # checks if current os is windows
        if custom_ffmpeg:
            # if custom FFmpeg path is given assign to local variable
            final_path += custom_ffmpeg
        else:
            # otherwise auto-download them
            try:
                if not (ffmpeg_download_path):
                    # otherwise save to Temp Directory
                    import tempfile

                    ffmpeg_download_path = tempfile.gettempdir()

                verbose and logger.debug(
                    "FFmpeg Windows Download Path: {}".format(ffmpeg_download_path)
                )

                # download Binaries
                os_bit = (
                    ("win64" if platform.machine().endswith("64") else "win32")
                    if is_windows
                    else ""
                )
                _path = download_ffmpeg_binaries(
                    path=ffmpeg_download_path, os_windows=is_windows, os_bit=os_bit
                )
                # assign to local variable
                final_path += _path

            except Exception as e:
                # log if any error occurred
                logger.exception(str(e))
                logger.error(
                    "Error in downloading FFmpeg binaries, Check your network and Try again!"
                )
                return False

        if os.path.isfile(final_path):
            # check if valid FFmpeg file exist
            pass
        elif os.path.isfile(os.path.join(final_path, "ffmpeg.exe")):
            # check if FFmpeg directory exists, if does, then check for valid file
            final_path = os.path.join(final_path, "ffmpeg.exe")
        else:
            # else return False
            verbose and logger.debug(
                "No valid FFmpeg executables found at Custom FFmpeg path!"
            )
            return False
    else:
        # otherwise perform test for Unix
        if custom_ffmpeg:
            # if custom FFmpeg path is given assign to local variable
            if os.path.isfile(custom_ffmpeg):
                # check if valid FFmpeg file exist
                final_path += custom_ffmpeg
            elif os.path.isfile(os.path.join(custom_ffmpeg, "ffmpeg")):
                # check if FFmpeg directory exists, if does, then check for valid file
                final_path = os.path.join(custom_ffmpeg, "ffmpeg")
            else:
                # else return False
                verbose and logger.debug(
                    "No valid FFmpeg executables found at Custom FFmpeg path!"
                )
                return False
        else:
            # otherwise assign ffmpeg binaries from system
            final_path += "ffmpeg"

    verbose and logger.debug("Final FFmpeg Path: {}".format(final_path))

    # Final Auto-Validation for FFmeg Binaries. returns final path if test is passed
    return final_path if validate_ffmpeg(final_path, verbose=verbose) else False

 

download_ffmpeg_binaries

Generates FFmpeg Static Binaries for windows(if not available)

Parameters:

Name Type Description Default
path string

path for downloading custom FFmpeg executables

required
os_windows boolean

is running on Windows OS?

False
os_bit string

32-bit or 64-bit OS?

''

Returns: A valid FFmpeg executable path string.

Source code in deffcode/ffhelper.py
def download_ffmpeg_binaries(path, os_windows=False, os_bit=""):
    """
    ## download_ffmpeg_binaries

    Generates FFmpeg Static Binaries for windows(if not available)

    Parameters:
        path (string): path for downloading custom FFmpeg executables
        os_windows (boolean): is running on Windows OS?
        os_bit (string): 32-bit or 64-bit OS?

    **Returns:** A valid FFmpeg executable path string.
    """
    final_path = ""
    if os_windows and os_bit:
        # initialize with available FFmpeg Static Binaries GitHub Server
        file_url = "https://github.com/abhiTronix/FFmpeg-Builds/releases/latest/download/ffmpeg-static-{}-gpl.zip".format(
            os_bit
        )

        file_name = os.path.join(
            os.path.abspath(path), "ffmpeg-static-{}-gpl.zip".format(os_bit)
        )
        file_path = os.path.join(
            os.path.abspath(path),
            "ffmpeg-static-{}-gpl/bin/ffmpeg.exe".format(os_bit),
        )
        base_path, _ = os.path.split(file_name)  # extract file base path
        # check if file already exists
        if os.path.isfile(file_path):
            final_path += file_path  # skip download if does
        else:
            # import libs
            import zipfile

            # check if given path has write access
            assert os.access(path, os.W_OK), (
                "[Helper:ERROR] :: Permission Denied, Cannot write binaries to directory = "
                + path
            )
            # remove leftovers if exists
            os.path.isfile(file_name) and delete_file_safe(file_name)
            # download and write file to the given path
            with open(file_name, "wb") as f:
                logger.debug(
                    "No Custom FFmpeg path provided. Auto-Installing FFmpeg static binaries from GitHub Mirror now. Please wait..."
                )
                # create session
                with requests.Session() as http:
                    # setup retry strategy
                    retries = Retry(
                        total=3,
                        backoff_factor=1,
                        status_forcelist=[429, 500, 502, 503, 504],
                    )
                    # Mount it for https usage
                    adapter = TimeoutHTTPAdapter(timeout=2.0, max_retries=retries)
                    http.mount("https://", adapter)
                    response = http.get(file_url, stream=True)
                    response.raise_for_status()
                    total_length = (
                        response.headers.get("content-length")
                        if "content-length" in response.headers
                        else len(response.content)
                    )
                    assert not (
                        total_length is None
                    ), "[Helper:ERROR] :: Failed to retrieve files, check your Internet connectivity!"
                    bar = tqdm(total=int(total_length), unit="B", unit_scale=True)
                    for data in response.iter_content(chunk_size=4096):
                        f.write(data)
                        len(data) > 0 and bar.update(len(data))
                    bar.close()
            logger.debug("Extracting executables.")
            with zipfile.ZipFile(file_name, "r") as zip_ref:
                zip_fname, _ = os.path.split(zip_ref.infolist()[0].filename)
                zip_ref.extractall(base_path)
            # perform cleaning
            delete_file_safe(file_name)
            logger.debug("FFmpeg binaries for Windows configured successfully!")
            final_path += file_path
    # return final path
    return final_path

 

validate_ffmpeg

Validate FFmpeg Binaries. Returns True if validity test passes successfully.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required
verbose bool

enables verbose for its operations

False

Returns: A boolean value, confirming whether tests passed, or not?.

Source code in deffcode/ffhelper.py
def validate_ffmpeg(path, verbose=False):
    """
    ## validate_ffmpeg

    Validate FFmpeg Binaries. Returns `True` if validity test passes successfully.

    Parameters:
        path (string): absolute path of FFmpeg binaries
        verbose (bool): enables verbose for its operations

    **Returns:** A boolean value, confirming whether tests passed, or not?.
    """
    try:
        # get the FFmpeg version
        version = check_sp_output([path, "-version"])
        firstline = version.split(b"\n")[0]
        version = firstline.split(b" ")[2].strip()
        if verbose:  # log if test are passed
            logger.debug("FFmpeg validity Test Passed!")
            logger.debug(
                "Found valid FFmpeg Version: `{}` installed on this system".format(
                    version
                )
            )
    except Exception as e:
        # log if test are failed
        if verbose:
            logger.exception(str(e))
            logger.warning("FFmpeg validity Test Failed!")
        return False
    return True

 

get_supported_pixfmts

Find and returns all FFmpeg's supported pixel formats.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required

Returns: List of supported pixel formats as (PIXEL FORMAT, NB_COMPONENTS, BITS_PER_PIXEL).

Source code in deffcode/ffhelper.py
def get_supported_pixfmts(path):
    """
    ## get_supported_pixfmts

    Find and returns all FFmpeg's supported pixel formats.

    Parameters:
        path (string): absolute path of FFmpeg binaries

    **Returns:** List of supported pixel formats as (PIXEL FORMAT, NB_COMPONENTS, BITS_PER_PIXEL).
    """
    pxfmts = check_sp_output([path, "-hide_banner", "-pix_fmts"])
    splitted = pxfmts.split(b"\n")
    srtindex = [i for i, s in enumerate(splitted) if b"-----" in s]
    # extract video encoders
    supported_pxfmts = [
        x.decode("utf-8").strip()
        for x in splitted[srtindex[0] + 1 :]
        if x.decode("utf-8").strip()
    ]
    # compile regex
    finder = re.compile(r"([A-Z]*[\.]+[A-Z]*\s[a-z0-9_-]*)(\s+[0-4])(\s+[0-9]+)")
    # find all outputs
    outputs = finder.findall("\n".join(supported_pxfmts))
    # return output findings
    return [
        ([s for s in o[0].split(" ")][-1], o[1].strip(), o[2].strip())
        for o in outputs
        if len(o) == 3
    ]

 

get_supported_vdecoders

Find and returns all FFmpeg's supported video decoders.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required

Returns: List of supported decoders.

Source code in deffcode/ffhelper.py
def get_supported_vdecoders(path):
    """
    ## get_supported_vdecoders

    Find and returns all FFmpeg's supported video decoders.

    Parameters:
        path (string): absolute path of FFmpeg binaries

    **Returns:** List of supported decoders.
    """
    decoders = check_sp_output([path, "-hide_banner", "-decoders"])
    splitted = decoders.split(b"\n")
    # extract video encoders
    supported_vdecoders = [
        x.decode("utf-8").strip()
        for x in splitted[2 : len(splitted) - 1]
        if x.decode("utf-8").strip().startswith("V")
    ]
    # compile regex
    finder = re.compile(r"[A-Z]*[\.]+[A-Z]*\s[a-z0-9_-]*")
    # find all outputs
    outputs = finder.findall("\n".join(supported_vdecoders))
    # return output findings
    return [[s for s in o.split(" ")][-1] for o in outputs]

 

get_supported_demuxers

Find and returns all FFmpeg's supported demuxers.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required

Returns: List of supported demuxers.

Source code in deffcode/ffhelper.py
def get_supported_demuxers(path):
    """
    ## get_supported_demuxers

    Find and returns all FFmpeg's supported demuxers.

    Parameters:
        path (string): absolute path of FFmpeg binaries

    **Returns:** List of supported demuxers.
    """
    # extract and clean FFmpeg output
    demuxers = check_sp_output([path, "-hide_banner", "-demuxers"])
    splitted = [x.decode("utf-8").strip() for x in demuxers.split(b"\n")]
    split_index = [idx for idx, s in enumerate(splitted) if "--" in s][0]
    supported_demuxers = splitted[split_index + 1 : len(splitted) - 1]
    # search all demuxers
    outputs = [re.search(r"\s[a-z0-9_,-]{2,}\s", d) for d in supported_demuxers]
    outputs = [o.group(0) for o in outputs if o]
    # return demuxers output
    return [o.strip() if not ("," in o) else o.split(",")[-1].strip() for o in outputs]

 

validate_imgseqdir

Validates Image Sequence by counting number of Image files.

Parameters:

Name Type Description Default
source string

video source to be validated

required
extension string

extension of image sequence.

'jpg'

Returns: A boolean value, confirming whether tests passed, or not?.

Source code in deffcode/ffhelper.py
def validate_imgseqdir(source, extension="jpg", verbose=False):
    """
    ## validate_imgseqdir

    Validates Image Sequence by counting number of Image files.

    Parameters:
        source (string): video source to be validated
        extension (string): extension of image sequence.

    **Returns:** A boolean value, confirming whether tests passed, or not?.
    """
    # check if path exists
    dirpath = Path(source).parent
    try:
        if not (dirpath.exists() and dirpath.is_dir()):
            verbose and logger.warning(
                "Specified path `{}` doesn't exists or valid.".format(dirpath)
            )
            return False
        else:
            return (
                True if len(list(dirpath.glob("*.{}".format(extension)))) > 2 else False
            )
    except:
        return False

 

is_valid_image_seq

Checks Image sequence validity by testing its extension against FFmpeg's supported pipe formats and number of Image files.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required
source string

video source to be validated

None
verbose bool

enables verbose for its operations

False

Returns: A boolean value, confirming whether tests passed, or not?.

Source code in deffcode/ffhelper.py
def is_valid_image_seq(path, source=None, verbose=False):
    """
    ## is_valid_image_seq

    Checks Image sequence validity by testing its extension against
    FFmpeg's supported pipe formats and number of Image files.

    Parameters:
        path (string): absolute path of FFmpeg binaries
        source (string): video source to be validated
        verbose (bool): enables verbose for its operations

    **Returns:** A boolean value, confirming whether tests passed, or not?.
    """
    if source is None or not (source):
        logger.error("Source is empty!")
        return False
    # extract all FFmpeg supported protocols
    formats = check_sp_output([path, "-hide_banner", "-formats"])
    extract_formats = re.findall(r"\w+_pipe", formats.decode("utf-8").strip())
    supported_image_formats = [
        x.split("_")[0] for x in extract_formats if x.endswith("_pipe")
    ]
    filename, extension = os.path.splitext(source)
    # Test and return result whether scheme is supported
    if extension and source.endswith(tuple(supported_image_formats)):
        if validate_imgseqdir(source, extension=extension[1:], verbose=verbose):
            verbose and logger.debug(
                "A valid Image Sequence source of format `{}` found.".format(extension)
            )
            return True
        else:
            ValueError(
                "Given Image Sequence source of format `{}` contains insignificant(invalid) sample size, Check the `source` parameter value again!".format(
                    source.split(".")[1]
                )
            )
    else:
        verbose and logger.warning("Source isn't a valid Image Sequence")
        return False

 

is_valid_url

Checks URL validity by testing its scheme against FFmpeg's supported protocols.

Parameters:

Name Type Description Default
path string

absolute path of FFmpeg binaries

required
url string

URL to be validated

None
verbose bool

enables verbose for its operations

False

Returns: A boolean value, confirming whether tests passed, or not?.

Source code in deffcode/ffhelper.py
def is_valid_url(path, url=None, verbose=False):
    """
    ## is_valid_url

    Checks URL validity by testing its scheme against
    FFmpeg's supported protocols.

    Parameters:
        path (string): absolute path of FFmpeg binaries
        url (string): URL to be validated
        verbose (bool): enables verbose for its operations

    **Returns:** A boolean value, confirming whether tests passed, or not?.
    """
    if url is None or not (url):
        logger.warning("URL is empty!")
        return False
    # extract URL scheme
    extracted_scheme_url = url.split("://", 1)[0]
    # extract all FFmpeg supported protocols
    protocols = check_sp_output([path, "-hide_banner", "-protocols"])
    splitted = [x.decode("utf-8").strip() for x in protocols.split(b"\n")]
    supported_protocols = splitted[splitted.index("Output:") + 1 : len(splitted) - 1]
    # RTSP is a demuxer somehow
    # support both RTSP and RTSPS(over SSL)
    supported_protocols += (
        ["rtsp", "rtsps"] if "rtsp" in get_supported_demuxers(path) else []
    )
    # Test and return result whether scheme is supported
    if extracted_scheme_url and extracted_scheme_url in supported_protocols:
        verbose and logger.debug(
            "URL scheme `{}` is supported by FFmpeg.".format(extracted_scheme_url)
        )
        return True
    else:
        verbose and logger.warning(
            "URL scheme `{}` isn't supported by FFmpeg!".format(extracted_scheme_url)
        )
        return False

 

check_sp_output

Returns FFmpeg stdout output from subprocess module.

Parameters:

Name Type Description Default
args based on input

Non Keyword Arguments

()
kwargs based on input

Keyword Arguments

{}

Returns: A string value.

Source code in deffcode/ffhelper.py
def check_sp_output(*args, **kwargs):
    """
    ## check_sp_output

    Returns FFmpeg `stdout` output from subprocess module.

    Parameters:
        args (based on input): Non Keyword Arguments
        kwargs (based on input): Keyword Arguments

    **Returns:** A string value.
    """
    # workaround for python bug: https://bugs.python.org/issue37380
    if platform.system() == "Windows":
        # see comment https://bugs.python.org/msg370334
        sp._cleanup = lambda: None
    # handle additional params
    retrieve_stderr = kwargs.pop("force_retrieve_stderr", False)
    # execute command in subprocess
    process = sp.Popen(
        stdout=sp.PIPE,
        stderr=sp.DEVNULL if not (retrieve_stderr) else sp.PIPE,
        *args,
        **kwargs,
    )
    # communicate and poll process
    output, stderr = process.communicate()
    retcode = process.poll()
    # handle return code
    if retcode and not (retrieve_stderr):
        logger.error("[Pipeline-Error] :: {}".format(output.decode("utf-8")))
        cmd = kwargs.get("args")
        if cmd is None:
            cmd = args[0]
        error = sp.CalledProcessError(retcode, cmd)
        error.output = output
        raise error
    # raise error if no output
    bool(output) or bool(stderr) or logger.error(
        "[Pipeline-Error] :: Pipeline failed to exact any data from command: {}!".format(
            args[0] if args else []
        )
    )
    # return output otherwise
    return stderr if retrieve_stderr and stderr else output