Skip to content

FFdecoder API

FFdecoder API Functional Diagram

FFdecoder API compiles and executes the FFmpeg pipeline inside a subprocess pipe for generating real-time, low-overhead, lightning fast video frames with robust error-handling in python 🎞️⚡

FFdecoder API implements a standalone highly-extensible wrapper around FFmpeg multimedia framework that provides complete control over the underline pipeline including access to almost any FFmpeg specification thinkable such as framerate, resolution, hardware decoder(s), complex filter(s), and pixel format(s) that are readily supported by all well known Computer Vision libraries.

FFdecoder API compiles its FFmpeg pipeline by processing input Video Source metadata and User-defined options, and runs it inside a subprocess pipe concurrently with the main thread, while extracting output dataframes(1D arrays) into a Numpy buffer. These dataframes are consecutively grabbed from the buffer and decoded into 24-bit RGB (default) ndarray 3D frames that are readily available through its generateFrame() method.

FFdecoder API employs Sourcer API at its backend for gathering, processing, and validating metadata of all multimedia streams available in the given source for formulating/compiling its default FFmpeg pipeline. This metadata information is also available as a JSON string with its metadata property object and can be updated as desired.

FFdecoder API supports a wide-ranging media stream as input source such as USB/Virtual/IP Camera Feed, Multimedia video file, Screen Capture, Image Sequence, Network protocols (such as HTTP(s), RTP/RSTP, etc.), so on and so forth.

Furthermore, FFdecoder API maintains the standard OpenCV-Python (Python API for OpenCV) coding syntax, thereby making it even easier to integrate this API in any Computer Vision application.

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

FFdecoder API parameters are explained here ➶

Source code in deffcode/ffdecoder.py
class FFdecoder:
    """
    > FFdecoder API compiles and executes the FFmpeg pipeline inside a subprocess pipe for generating real-time, low-overhead, lightning fast video frames
    with robust error-handling in python 🎞️⚡

    FFdecoder API implements a **standalone highly-extensible wrapper around [FFmpeg](https://ffmpeg.org/)** multimedia framework that provides complete
    control over the underline pipeline including **access to almost any FFmpeg specification thinkable** such as framerate, resolution, hardware decoder(s),
    complex filter(s), and pixel format(s) that are readily supported by all well known Computer Vision libraries.

    FFdecoder API **compiles its FFmpeg pipeline** by processing input Video Source metadata and User-defined options, and **runs it inside a
    [`subprocess`](https://docs.python.org/3/library/subprocess.html) pipe** concurrently with the main thread, while extracting output dataframes(1D arrays)
    into a Numpy buffer. These dataframes are consecutively grabbed from the buffer and decoded into ==[24-bit RGB](https://en.wikipedia.org/wiki/List_of_monochrome_and_RGB_color_formats#24-bit_RGB) _(default)_
    [`ndarray`](https://numpy.org/doc/stable/reference/arrays.ndarray.html#the-n-dimensional-array-ndarray) 3D frames== that are readily available
    through its [`generateFrame()`](#deffcode.ffdecoder.FFdecoder.generateFrame) method.

    FFdecoder API **employs [Sourcer API](../../reference/sourcer) at its backend** for gathering, processing, and validating metadata of all
    multimedia streams available in the given source for formulating/compiling its default FFmpeg pipeline. This metadata information is also
    available as a JSON string with its [`metadata`](#deffcode.ffdecoder.FFdecoder.metadata) property object and can be updated as desired.

    FFdecoder API **supports a wide-ranging media stream** as input source such as USB/Virtual/IP Camera Feed, Multimedia video file,
    Screen Capture, Image Sequence, Network protocols _(such as HTTP(s), RTP/RSTP, etc.)_, so on and so forth.

    Furthermore, FFdecoder API maintains the **standard [OpenCV-Python](https://docs.opencv.org/4.x/d6/d00/tutorial_py_root.html) _(Python API for OpenCV)_ coding syntax**, thereby making it even easier to
    integrate this API in any Computer Vision application.

    !!! example "For usage examples, kindly refer our **[Basic Recipes :cake:](../../recipes/basic)** and **[Advanced Recipes :croissant:](../../recipes/advanced)**"

    !!! info "FFdecoder API parameters are explained [here ➶](params/)"
    """

    def __init__(
        self,
        source,
        source_demuxer=None,
        frame_format=None,
        custom_ffmpeg="",
        verbose=False,
        **ffparams
    ):
        """
        This constructor method initializes the object state and attributes of the FFdecoder Class.

        Parameters:
            source (str): defines the input(`-i`) source filename/URL/device-name/device-path.
            source_demuxer (str): specifies the demuxer(`-f`) for the input source.
            frame_format (str): sets pixel format(`-pix_fmt`) of the decoded frames.
            custom_ffmpeg (str): assigns the location of custom path/directory for custom FFmpeg executable.
            verbose (bool): enables/disables verbose.
            ffparams (dict): provides the flexibility to control supported internal and FFmpeg parameters.
        """

        # enable verbose if specified
        self.__verbose_logs = (
            verbose if (verbose and isinstance(verbose, bool)) else False
        )

        # define whether initializing
        self.__initializing = True

        # define frame pixel-format for decoded frames
        self.__frame_format = (
            frame_format.lower().strip() if isinstance(frame_format, str) else None
        )

        # handles user-defined parameters
        self.__extra_params = {}

        # handles process to be frames written
        self.__process = None

        # handles disabling window for ffmpeg subprocess on Windows
        self.__ffmpeg_window_disabler_patch = False

        # checks machine OS
        self.__machine_OS = platform.system()

        # handles exclusive metadata
        self.__ff_pixfmt_metadata = None  # metadata
        self.__raw_frame_num = None  # raw-frame number
        self.__raw_frame_pixfmt = None  # raw-frame pixformat
        self.__raw_frame_dtype = None  # raw-frame dtype
        self.__raw_frame_depth = None  # raw-frame depth
        self.__raw_frame_resolution = None  # raw-frame resolution/dimension

        # define supported mode of operation
        self.__supported_opmodes = {
            "av": "Audio-Video",  # audio is only for pass-through, not really for audio decoding yet.
            "vo": "Video-Only",
            "imgseq": "Image-Sequence",
            # "ao":"Audio-Only", # reserved for future
        }
        # operation mode variable
        self.__opmode = None

        # handles termination
        self.__terminate_stream = False

        # cleans and reformat user-defined parameters
        self.__extra_params = {
            str(k).strip(): (
                str(v).strip()
                if not (v is None)
                and not isinstance(v, (dict, list, int, float, tuple))
                else v
            )
            for k, v in ffparams.items()
        }

        # handles custom Sourcer API params
        sourcer_params = self.__extra_params.pop("-custom_sourcer_params", {})
        # reset improper values
        sourcer_params = {} if not isinstance(sourcer_params, dict) else sourcer_params

        # handles user ffmpeg pre-headers(parameters such as `-re`) parameters (must be a list)
        self.__ffmpeg_prefixes = self.__extra_params.pop("-ffprefixes", [])
        # check if not valid type
        if not isinstance(self.__ffmpeg_prefixes, list):
            # log it
            logger.warning(
                "Discarding invalid `-ffprefixes` value of wrong type: `{}`!".format(
                    type(self.__ffmpeg_prefixes).__name__
                )
            )
            # reset improper values
            self.__ffmpeg_prefixes = []
        else:
            # also pass valid ffmpeg pre-headers to Sourcer API
            sourcer_params["-ffprefixes"] = self.__ffmpeg_prefixes

        # pass parameter(if specified) to Sourcer API, specifying where to save the downloaded FFmpeg Static
        # assets on Windows(if specified)
        sourcer_params["-ffmpeg_download_path"] = self.__extra_params.pop(
            "-ffmpeg_download_path", ""
        )

        # handle video and audio stream indexes in case of multiple ones.
        default_stream_indexes = self.__extra_params.pop(
            "-default_stream_indexes", (0, 0)
        )
        # reset improper values
        default_stream_indexes = (
            (0, 0)
            if not isinstance(default_stream_indexes, (list, tuple))
            else default_stream_indexes
        )

        # pass FFmpeg filter to Sourcer API params for processing
        if set(["-vf", "-filter_complex"]).intersection(self.__extra_params.keys()):
            key = "-vf" if "-vf" in self.__extra_params else "-filter_complex"
            sourcer_params[key] = self.__extra_params[key]

        # define dict to store user-defined parameters
        self.__user_metadata = {}
        # extract and assign source metadata as dict
        (self.__sourcer_metadata, self.__missing_prop) = (
            Sourcer(
                source=source,
                source_demuxer=source_demuxer,
                verbose=verbose,
                custom_ffmpeg=custom_ffmpeg if isinstance(custom_ffmpeg, str) else "",
                **sourcer_params
            )
            .probe_stream(default_stream_indexes=default_stream_indexes)
            .retrieve_metadata(force_retrieve_missing=True)
        )

        # handle valid FFmpeg assets location
        self.__ffmpeg = self.__sourcer_metadata["ffmpeg_binary_path"]

        # handle YUV pixel formats(such as `yuv420p`, `yuv444p`, `nv12`, `nv21` etc.)
        # patch for compatibility with OpenCV APIs.
        self.__cv_patch = self.__extra_params.pop("-enforce_cv_patch", False)
        if not (isinstance(self.__cv_patch, bool)):
            self.__cv_patch = False
            self.__verbose_logs and logger.critical(
                "Enforcing OpenCV compatibility patch for YUV/NV video frames."
            )

        # handle disabling window for ffmpeg subprocess on Windows OS
        # this patch prevents ffmpeg creation window from opening when
        # building exe files
        ffmpeg_window_disabler_patch = self.__extra_params.pop(
            "-disable_ffmpeg_window", False
        )
        if ffmpeg_window_disabler_patch and isinstance(
            ffmpeg_window_disabler_patch, bool
        ):
            # check if value is valid
            if self.__machine_OS != "Windows" or self.__verbose_logs:
                logger.warning(
                    "Optional `-disable_ffmpeg_window` flag is only available on Windows OS with `verbose=False`. Discarding!"
                )
            else:
                self.__ffmpeg_window_disabler_patch = ffmpeg_window_disabler_patch
        else:
            # handle improper values
            self.__ffmpeg_window_disabler_patch = False

        # handle pass-through audio mode works in conjunction with WriteGear [TODO]
        self.__passthrough_mode = self.__extra_params.pop("-passthrough_audio", False)
        if not (isinstance(self.__passthrough_mode, bool)):
            self.__passthrough_mode = False

        # handle mode of operation
        if self.__sourcer_metadata["source_has_image_sequence"]:
            # image-sequence mode
            self.__opmode = "imgseq"
        elif (
            self.__sourcer_metadata[
                "source_has_video"
            ]  # audio is only for pass-through, not really for audio decoding yet.
            and self.__sourcer_metadata["source_has_audio"]
            and self.__passthrough_mode  # [TODO]
        ):
            self.__opmode = "av"
        # elif __defop_mode == "ao" and self.__sourcer_metadata.contains_audio: # [TODO]
        #    self.__opmode = "ao"
        elif self.__sourcer_metadata["source_has_video"]:
            # video-only mode
            self.__opmode = "vo"
        else:
            # raise if unknown mode
            raise ValueError(
                "Unable to find any usable video stream in the given source!"
            )
        # store as metadata
        self.__missing_prop["ffdecoder_operational_mode"] = self.__supported_opmodes[
            self.__opmode
        ]

        # handle user-defined output framerate
        __framerate = self.__extra_params.pop("-framerate", None)
        if (
            isinstance(__framerate, str)
            and __framerate
            == "null"  # special mode to discard `-framerate/-r` parameter
        ):
            self.__inputframerate = __framerate
        elif isinstance(__framerate, (float, int)):
            self.__inputframerate = float(__framerate) if __framerate > 0.0 else 0.0
        else:
            # warn if wrong type
            not (__framerate is None) and logger.warning(
                "Discarding invalid `-framerate` value of wrong type `{}`!".format(
                    type(__framerate).__name__
                )
            )
            # reset to default
            self.__inputframerate = 0.0

        # handle user defined decoded frame resolution
        self.__custom_resolution = self.__extra_params.pop("-custom_resolution", None)
        if (
            isinstance(self.__custom_resolution, str)
            and self.__custom_resolution
            == "null"  # special mode to discard `-size/-s` parameter
        ) or (
            isinstance(self.__custom_resolution, (list, tuple))
            and len(self.__custom_resolution)
            == 2  # valid resolution(must be a tuple or list)
        ):
            # log it
            self.__verbose_logs and not isinstance(
                self.__custom_resolution, str
            ) and logger.debug(
                "Setting raw frames size: `{}`.".format(self.__custom_resolution)
            )
        else:
            # log it
            not (self.__custom_resolution is None) and logger.warning(
                "Discarding invalid `-custom_resolution` value: `{}`!".format(
                    self.__custom_resolution
                )
            )
            # reset improper values
            self.__custom_resolution = None

    def formulate(self):
        """
        This method formulates all necessary FFmpeg pipeline arguments and executes it inside the FFmpeg `subprocess` pipe.

        **Returns:** A reference to the FFdecoder class object.
        """
        # assign values to class variables on first run
        if self.__initializing:
            # prepare parameter dict
            input_params = OrderedDict()
            output_params = OrderedDict()

            # dynamically pre-assign a default video-decoder (if not assigned by user).
            supported_vdecodecs = get_supported_vdecoders(self.__ffmpeg)
            default_vdecodec = (
                self.__sourcer_metadata["source_video_decoder"]
                if self.__sourcer_metadata["source_video_decoder"]
                in supported_vdecodecs
                else "unknown"
            )
            if "-c:v" in self.__extra_params:
                self.__extra_params["-vcodec"] = self.__extra_params.pop(
                    "-c:v", default_vdecodec
                )
            # handle image sequence separately
            if self.__opmode == "imgseq":
                # -vcodec is discarded by default
                # (This is correct or maybe -vcodec required in some unknown case) [TODO]
                self.__extra_params.pop("-vcodec", None)
            elif (
                "-vcodec" in self.__extra_params
                and self.__extra_params["-vcodec"] is None
            ):
                # special case when -vcodec is not needed intentionally
                self.__extra_params.pop("-vcodec", None)
            else:
                # assign video decoder selected here.
                if not "-vcodec" in self.__extra_params:
                    input_params["-vcodec"] = default_vdecodec
                else:
                    input_params["-vcodec"] = self.__extra_params.pop(
                        "-vcodec", default_vdecodec
                    )
                if (
                    default_vdecodec != "unknown"
                    and not input_params["-vcodec"] in supported_vdecodecs
                ):
                    # reset to default if not supported
                    logger.warning(
                        "Provided FFmpeg does not support `{}` video decoder. Switching to default supported `{}` decoder!".format(
                            input_params["-vcodec"], default_vdecodec
                        )
                    )
                    input_params["-vcodec"] = default_vdecodec
                # raise error if not valid decoder found
                if not input_params["-vcodec"] in supported_vdecodecs:
                    raise RuntimeError(
                        "Provided FFmpeg does not support any known usable video-decoders."
                        " Either define your own manually or switch to another FFmpeg binaries(if available)."
                    )

            # handle user-defined number of frames.
            if "-vframes" in self.__extra_params:
                self.__extra_params["-frames:v"] = self.__extra_params.pop(
                    "-vframes", None
                )
            if "-frames:v" in self.__extra_params:
                value = self.__extra_params.pop("-frames:v", None)
                if not (value is None) and value > 0:
                    output_params["-frames:v"] = value

            # dynamically calculate default raw-frames pixel format(if not assigned by user).
            # notify FFmpeg `-pix_fmt` parameter cannot be assigned directly
            if "-pix_fmt" in self.__extra_params:
                logger.warning(
                    "Discarding user-defined `-pix_fmt` value as it can only be assigned with `frame_format` parameter!"
                )
                self.__extra_params.pop("-pix_fmt", None)
            # get supported FFmpeg pixfmt data with depth and bpp(bits-per-pixel)
            self.__ff_pixfmt_metadata = get_supported_pixfmts(self.__ffmpeg)
            supported_pixfmts = [fmts[0] for fmts in self.__ff_pixfmt_metadata]

            # calculate default pixel-format
            # Check special case  - `frame_format`(or `-pix_fmt`) parameter discarded from pipeline
            self.__frame_format == "null" and logger.critical(
                "Manually discarding `frame_format`(or `-pix_fmt`) parameter from this pipeline."
            )
            # choose between rgb24(if available) or source pixel-format
            # otherwise, only source pixel-format for special case
            default_pixfmt = (
                "rgb24"
                if "rgb24" in supported_pixfmts and self.__frame_format != "null"
                else self.__sourcer_metadata["source_video_pixfmt"]
            )
            # assign output raw-frames pixel format
            rawframe_pixfmt = None
            if (
                not (self.__frame_format is None)
                and self.__frame_format in supported_pixfmts
            ):
                # check if valid and supported `frame_format` parameter assigned
                rawframe_pixfmt = self.__frame_format.strip()
                self.__verbose_logs and logger.info(
                    "User-defined `{}` frame pixel-format will be used for this pipeline.".format(
                        rawframe_pixfmt
                    )
                )
            elif (
                "output_frames_pixfmt"
                in self.__sourcer_metadata  # means `format` filter is defined
                and self.__sourcer_metadata["output_frames_pixfmt"] in supported_pixfmts
            ):
                # assign if valid and supported
                rawframe_pixfmt = self.__sourcer_metadata[
                    "output_frames_pixfmt"
                ].strip()
                self.__verbose_logs and logger.info(
                    "FFmpeg filter values will be used for this pipeline for defining output pixel-format."
                )
            else:
                # reset to default if not supported
                rawframe_pixfmt = default_pixfmt
                # log it accordingly
                if self.__frame_format is None:
                    logger.info(
                        "Using default `{}` pixel-format for this pipeline.".format(
                            default_pixfmt
                        )
                    )
                else:
                    logger.warning(
                        "{} Switching to default `{}` pixel-format!".format(
                            (
                                "Provided FFmpeg does not supports `{}` pixel-format.".format(
                                    self.__sourcer_metadata["output_frames_pixfmt"]
                                    if "output_frames_pixfmt" in self.__sourcer_metadata
                                    else self.__frame_format
                                )
                                if self.__frame_format != "null"
                                else "No usable pixel-format defined."
                            ),
                            default_pixfmt,
                        )
                    )

            # dynamically calculate raw-frame datatype based on pixel-format selected
            (self.__raw_frame_depth, rawframesbpp) = [
                (int(x[1]), int(x[2]))
                for x in self.__ff_pixfmt_metadata
                if x[0] == rawframe_pixfmt
            ][0]
            raw_bit_per_component = (
                rawframesbpp // self.__raw_frame_depth if self.__raw_frame_depth else 0
            )
            if 4 <= raw_bit_per_component <= 8:
                self.__raw_frame_dtype = np.dtype("u1")
            elif 8 < raw_bit_per_component <= 16 and rawframe_pixfmt.endswith(
                ("le", "be")
            ):
                if rawframe_pixfmt.endswith("le"):
                    self.__raw_frame_dtype = np.dtype("<u2")
                else:
                    self.__raw_frame_dtype = np.dtype(">u2")
            else:
                # reset to both pixel-format and datatype to default if not supported
                not (self.__frame_format is None) and logger.warning(
                    "Selected pixel-format `{}` dtype is not supported by FFdecoder API. Switching to default `rgb24` pixel-format!".format(
                        rawframe_pixfmt
                    )
                )
                rawframe_pixfmt = "rgb24"
                self.__raw_frame_dtype = np.dtype("u1")

            # Check if not special case
            if self.__frame_format != "null":
                # assign to FFmpeg pipeline otherwise
                output_params["-pix_fmt"] = rawframe_pixfmt
            # assign to global parameter further usage
            self.__raw_frame_pixfmt = rawframe_pixfmt
            # also override as metadata(if available)
            if "output_frames_pixfmt" in self.__sourcer_metadata:
                self.__sourcer_metadata["output_frames_pixfmt"] = (
                    self.__raw_frame_pixfmt
                )

            # handle raw-frame resolution
            # notify FFmpeg `-s` parameter cannot be assigned directly
            if "-s" in self.__extra_params:
                logger.warning(
                    "Discarding user-defined `-s` FFmpeg parameter as it can only be assigned with `-custom_resolution` attribute! Read docs for more details."
                )
                self.__extra_params.pop("-s", None)
            # assign output rawframe resolution
            if not (self.__custom_resolution is None) and not isinstance(
                self.__custom_resolution, str
            ):
                # assign if assigned by user and not "null"(str)
                self.__raw_frame_resolution = self.__custom_resolution
                self.__verbose_logs and logger.info(
                    "User-defined `{}` frame resolution will be used for this pipeline.".format(
                        self.__raw_frame_resolution
                    )
                )
            elif (
                "output_frames_resolution"
                in self.__sourcer_metadata  # means `scale` filter is defined
                and self.__sourcer_metadata["output_frames_resolution"]
                and len(self.__sourcer_metadata["output_frames_resolution"]) == 2
            ):
                # calculate raw-frame resolution/dimensions based on output.
                self.__raw_frame_resolution = self.__sourcer_metadata[
                    "output_frames_resolution"
                ]
            elif (
                self.__sourcer_metadata["source_video_resolution"]
                and len(self.__sourcer_metadata["source_video_resolution"]) == 2
            ):
                # calculate raw-frame resolution/dimensions based on source.
                self.__raw_frame_resolution = self.__sourcer_metadata[
                    "source_video_resolution"
                ]
            else:
                # otherwise raise error
                raise RuntimeError(
                    "Both source and output metadata values found Invalid with {} `-custom_resolution` attribute. Aborting!".format(
                        "null"
                        if isinstance(self.__inputframerate, str)
                        else "undefined"
                    )
                )
            # special mode to discard `-size/-s` FFmpeg parameter completely
            if isinstance(self.__custom_resolution, str):
                logger.critical(
                    "Manually discarding `-size/-s` FFmpeg parameter from this pipeline."
                )
            else:
                # add to pipeline
                dimensions = "{}x{}".format(
                    self.__raw_frame_resolution[0], self.__raw_frame_resolution[1]
                )
                output_params["-s"] = str(dimensions)
            # log if filters or default source is used
            self.__verbose_logs and (
                self.__custom_resolution is None
                or isinstance(self.__custom_resolution, str)
            ) and logger.info(
                "{} for this pipeline for defining output resolution.".format(
                    "FFmpeg filter values will be used"
                    if "output_frames_resolution" in self.__sourcer_metadata
                    else "Default source resolution will be used"
                )
            )

            # dynamically calculate raw-frame framerate based on source (if not assigned by user).
            if (
                not isinstance(self.__inputframerate, str)
                and self.__inputframerate > 0.0
            ):
                # assign if assigned by user and not "null"(str)
                output_params["-framerate"] = str(self.__inputframerate)
                self.__verbose_logs and logger.info(
                    "User-defined `{}` output framerate will be used for this pipeline.".format(
                        str(self.__inputframerate)
                    )
                )
            elif (
                "output_framerate"
                in self.__sourcer_metadata  # means `fps` filter is defined
                and self.__sourcer_metadata["output_framerate"] > 0.0
            ):
                # special mode to discard `-framerate/-r` FFmpeg parameter completely
                if self.__inputframerate == "null":
                    logger.critical(
                        "Manually discarding `-framerate/-r` FFmpeg parameter from this pipeline."
                    )
                else:
                    # calculate raw-frame framerate based on output
                    output_params["-framerate"] = str(
                        self.__sourcer_metadata["output_framerate"]
                    )
                self.__verbose_logs and logger.info(
                    "FFmpeg filter values will be used for this pipeline for defining output framerate."
                )
            elif self.__sourcer_metadata["source_video_framerate"] > 0.0:
                # special mode to discard `-framerate/-r` FFmpeg parameter completely
                if self.__inputframerate == "null":
                    logger.critical(
                        "Manually disabling `-framerate/-r` FFmpeg parameter for this pipeline."
                    )
                else:
                    # calculate raw-frame framerate based on source
                    output_params["-framerate"] = str(
                        self.__sourcer_metadata["source_video_framerate"]
                    )
                self.__verbose_logs and logger.info(
                    "Default source framerate will be used for this pipeline for defining output framerate."
                )
            else:
                # otherwise raise error
                raise RuntimeError(
                    "Both source and output metadata values found Invalid with {} `-framerate` attribute. Aborting!".format(
                        "null"
                        if isinstance(self.__inputframerate, str)
                        else "undefined"
                    )
                )

            # add rest to output parameters
            output_params.update(self.__extra_params)

            # dynamically calculate raw-frame numbers based on source (if not assigned by user).
            # TODO Added support for `-re -stream_loop` and `-loop`
            if "-frames:v" in input_params:
                self.__raw_frame_num = input_params["-frames:v"]
            elif (
                not (self.__sourcer_metadata["approx_video_nframes"] is None)
                and self.__sourcer_metadata["approx_video_nframes"] > 0
            ):
                self.__raw_frame_num = self.__sourcer_metadata["approx_video_nframes"]
            else:
                self.__raw_frame_num = None
                # log that number of frames are unknown
                self.__verbose_logs and logger.info(
                    "Number of frames in given source are unknown. Live/Network/Looping stream detected!"
                )

            # log Mode of Operation
            self.__verbose_logs and logger.critical(
                "Activating {} Mode of Operation.".format(
                    self.__supported_opmodes[self.__opmode]
                )
            )

            # compose the Pipeline using formulated FFmpeg parameters
            self.__launch_FFdecoderline(input_params, output_params)

            # inform the initialization is completed
            self.__initializing = False
        else:
            # warn if pipeline is recreated
            logger.error("This pipeline is already created and running!")
        return self

    def __fetchNextfromPipeline(self):
        """
        This Internal method to fetch next dataframes(1D arrays) from `subprocess` pipe's standard output(`stdout`) into a Numpy buffer.
        """
        assert not (
            self.__process is None
        ), "Pipeline is not running! You must call `formulate()` method first."

        # formulated raw frame size and apply YUV pixel formats patch(if applicable)
        raw_frame_size = (
            (self.__raw_frame_resolution[0] * (self.__raw_frame_resolution[1] * 3 // 2))
            if self.__raw_frame_pixfmt.startswith(("yuv", "nv")) and self.__cv_patch
            else (
                self.__raw_frame_depth
                * self.__raw_frame_resolution[0]
                * self.__raw_frame_resolution[1]
            )
        )
        # next dataframe as numpy ndarray
        nparray = None
        try:
            # read bytes frames from buffer
            nparray = np.frombuffer(
                self.__process.stdout.read(
                    raw_frame_size * self.__raw_frame_dtype.itemsize
                ),
                dtype=self.__raw_frame_dtype,
            )
        except Exception as e:
            raise RuntimeError("Frame buffering failed with error: {}".format(str(e)))
        return (
            nparray
            if not (nparray is None) and len(nparray) == raw_frame_size
            else None
        )

    def __fetchNextFrame(self):
        """
        This Internal method grabs and decodes next 3D `ndarray` video-frame from the buffer.
        """
        # Read next and reconstruct as numpy array
        frame = self.__fetchNextfromPipeline()
        # check if empty
        if frame is None:
            return frame
        elif self.__raw_frame_pixfmt.startswith("gray"):
            # reconstruct exclusive `gray` frames
            frame = frame.reshape(
                (
                    self.__raw_frame_resolution[1],
                    self.__raw_frame_resolution[0],
                    self.__raw_frame_depth,
                )
            )[:, :, 0]
        elif self.__raw_frame_pixfmt.startswith(("yuv", "nv")) and self.__cv_patch:
            # reconstruct exclusive YUV formats frames for OpenCV APIs
            frame = frame.reshape(
                self.__raw_frame_resolution[1] * 3 // 2,
                self.__raw_frame_resolution[0],
            )
        else:
            # reconstruct default frames
            frame = frame.reshape(
                (
                    self.__raw_frame_resolution[1],
                    self.__raw_frame_resolution[0],
                    self.__raw_frame_depth,
                )
            )
        # return frame
        return frame

    def generateFrame(self):
        """
        This method returns a [Generator function](https://wiki.python.org/moin/Generators)
        _(also an Iterator using `next()`)_ of video frames, grabbed continuously from the buffer.
        """
        if self.__raw_frame_num is None or not self.__raw_frame_num:
            while not self.__terminate_stream:  # infinite raw frames
                frame = self.__fetchNextFrame()
                if frame is None:
                    self.__terminate_stream = True
                    break
                yield frame
        else:
            for _ in range(self.__raw_frame_num):  # finite raw frames
                frame = self.__fetchNextFrame()
                if frame is None:
                    self.__terminate_stream = True
                    break
                yield frame

    def __enter__(self):
        """
        Handles entry with the `with` statement. See [PEP343 -- The 'with' statement'](https://peps.python.org/pep-0343/).

        **Returns:** Output of `formulate()` method.
        """
        return self.formulate()

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Handles exit with the `with` statement. See [PEP343 -- The 'with' statement'](https://peps.python.org/pep-0343/).
        """
        self.terminate()

    @property
    def metadata(self):
        """
        A property object that dumps metadata information as JSON string.

        **Returns:** Metadata as JSON string.
        """
        # import dependency
        import json

        # return complete metadata information as JSON string
        return json.dumps(
            {
                **self.__sourcer_metadata,  # source video
                **self.__missing_prop,  # missing properties
                **self.__user_metadata,  # user-defined
            },
            indent=2,
        )

    @metadata.setter
    def metadata(self, value):
        """
        A property object that updates metadata information with user-defined dictionary.

        Parameters:
            value (dict): User-defined dictionary.
        """
        # check if value dict type
        if value and isinstance(value, dict):
            # log it
            self.__verbose_logs and logger.info("Updating Metadata...")
            # extract any source and output internal metadata keys
            default_keys = set(value).intersection(
                {**self.__sourcer_metadata, **self.__missing_prop}
            )
            # counterpart source properties for each output properties
            counterpart_prop = {
                "output_frames_resolution": "source_video_resolution",
                "output_frames_pixfmt": "source_video_pixfmt",
                "output_framerate": "source_video_framerate",
            }
            # iterate over source metadata keys and sanitize it
            for key in default_keys or []:
                if key == "source":
                    # metadata properties that cannot be altered
                    logger.warning(
                        "`{}` metadata property value cannot be altered. Discarding!".format(
                            key
                        )
                    )
                elif key in self.__missing_prop:
                    # missing metadata properties are unavailable and read-only
                    # notify user about alternative counterpart property (if available)
                    logger.warning(
                        "`{}` metadata property is read-only".format(key)
                        + (
                            ". Try updating `{}` property instead!".format(
                                counterpart_prop[key]
                            )
                            if key in counterpart_prop.keys()
                            else " and cannot be updated!"
                        )
                    )
                elif isinstance(value[key], type(self.__sourcer_metadata[key])):
                    # check if correct datatype as original
                    self.__verbose_logs and logger.info(
                        "Updating `{}`{} metadata property to `{}`.".format(
                            key,
                            (
                                " and its counterpart"
                                if key in counterpart_prop.values()
                                else ""
                            ),
                            value[key],
                        )
                    )
                    # update source metadata if valid
                    self.__sourcer_metadata[key] = value[key]
                    # also update missing counterpart property (if available)
                    counter_key = next(
                        (k for k, v in counterpart_prop.items() if v == key), ""
                    )
                    if counter_key:
                        self.__missing_prop[counter_key] = value[key]
                else:
                    # otherwise discard and log it
                    logger.warning(
                        "Manually assigned `{}` metadata property value is of invalid type. Discarding!"
                    ).format(key)
                # delete invalid key
                del value[key]
            # There is no concept of a tuple in the JSON format.
            # Python's `json` module converts Python tuples to JSON lists
            # because that's the closest thing in JSON to a tuple.
            any(isinstance(value[x], tuple) for x in value) and logger.warning(
                "All TUPLE metadata properties will be converted to LIST datatype. Read docs for more details."
            )
            # update user-defined metadata
            self.__user_metadata.update(value)
        else:
            # otherwise raise error
            raise ValueError("Invalid datatype metadata assigned. Aborting!")

    def __launch_FFdecoderline(self, input_params, output_params):
        """
        This Internal method executes FFmpeg pipeline arguments inside a `subprocess` pipe in a new process.

        Parameters:
            input_params (dict): Input FFmpeg parameters
            output_params (dict): Output FFmpeg parameters
        """
        # convert input parameters to list
        input_parameters = dict2Args(input_params)

        # convert output parameters to list
        output_parameters = dict2Args(output_params)

        # format command
        cmd = (
            [self.__ffmpeg]
            + (["-hide_banner"] if not self.__verbose_logs else [])
            + self.__ffmpeg_prefixes
            + input_parameters
            + (
                ["-f", self.__sourcer_metadata["source_demuxer"]]
                if ("source_demuxer" in self.__sourcer_metadata.keys())
                else []
            )
            + ["-i", self.__sourcer_metadata["source"]]
            + output_parameters
            + ["-f", "rawvideo", "-"]
        )
        # compose the FFmpeg process
        if self.__verbose_logs:
            logger.debug("Executing FFmpeg command: `{}`".format(" ".join(cmd)))
            # In debugging mode
            self.__process = sp.Popen(
                cmd, stdin=sp.DEVNULL, stdout=sp.PIPE, stderr=None
            )
        else:
            # In silent mode
            self.__process = sp.Popen(
                cmd,
                stdin=sp.DEVNULL,
                stdout=sp.PIPE,
                stderr=sp.DEVNULL,
                creationflags=(  # this prevents ffmpeg creation window from opening when building exe files on Windows
                    sp.DETACHED_PROCESS if self.__ffmpeg_window_disabler_patch else 0
                ),
            )

    def terminate(self):
        """
        Safely terminates all processes.
        """

        # signal we are closing
        self.__verbose_logs and logger.debug("Terminating FFdecoder Pipeline...")
        self.__terminate_stream = True
        # check if no process was initiated at first place
        if self.__process is None or not (self.__process.poll() is None):
            logger.info("Pipeline already terminated.")
            return
        # Attempt to close pipeline.
        # close `stdin` output
        self.__process.stdin and self.__process.stdin.close()
        # close `stdout` output
        self.__process.stdout and self.__process.stdout.close()
        # terminate/kill process if still processing
        self.__process.poll() is None and self.__process.terminate()
        # wait if not exiting
        self.__process.wait()
        self.__process = None
        logger.info("Pipeline terminated successfully.")

metadata property writable

A property object that dumps metadata information as JSON string.

Returns: Metadata as JSON string.

__enter__(self) special

Handles entry with the with statement. See PEP343 -- The 'with' statement'.

Returns: Output of formulate() method.

Source code in deffcode/ffdecoder.py
def __enter__(self):
    """
    Handles entry with the `with` statement. See [PEP343 -- The 'with' statement'](https://peps.python.org/pep-0343/).

    **Returns:** Output of `formulate()` method.
    """
    return self.formulate()

__exit__(self, exc_type, exc_val, exc_tb) special

Handles exit with the with statement. See PEP343 -- The 'with' statement'.

Source code in deffcode/ffdecoder.py
def __exit__(self, exc_type, exc_val, exc_tb):
    """
    Handles exit with the `with` statement. See [PEP343 -- The 'with' statement'](https://peps.python.org/pep-0343/).
    """
    self.terminate()

__init__(self, source, source_demuxer=None, frame_format=None, custom_ffmpeg='', verbose=False, **ffparams) special

This constructor method initializes the object state and attributes of the FFdecoder Class.

Parameters:

Name Type Description Default
source str

defines the input(-i) source filename/URL/device-name/device-path.

required
source_demuxer str

specifies the demuxer(-f) for the input source.

None
frame_format str

sets pixel format(-pix_fmt) of the decoded frames.

None
custom_ffmpeg str

assigns the location of custom path/directory for custom FFmpeg executable.

''
verbose bool

enables/disables verbose.

False
ffparams dict

provides the flexibility to control supported internal and FFmpeg parameters.

{}
Source code in deffcode/ffdecoder.py
def __init__(
    self,
    source,
    source_demuxer=None,
    frame_format=None,
    custom_ffmpeg="",
    verbose=False,
    **ffparams
):
    """
    This constructor method initializes the object state and attributes of the FFdecoder Class.

    Parameters:
        source (str): defines the input(`-i`) source filename/URL/device-name/device-path.
        source_demuxer (str): specifies the demuxer(`-f`) for the input source.
        frame_format (str): sets pixel format(`-pix_fmt`) of the decoded frames.
        custom_ffmpeg (str): assigns the location of custom path/directory for custom FFmpeg executable.
        verbose (bool): enables/disables verbose.
        ffparams (dict): provides the flexibility to control supported internal and FFmpeg parameters.
    """

    # enable verbose if specified
    self.__verbose_logs = (
        verbose if (verbose and isinstance(verbose, bool)) else False
    )

    # define whether initializing
    self.__initializing = True

    # define frame pixel-format for decoded frames
    self.__frame_format = (
        frame_format.lower().strip() if isinstance(frame_format, str) else None
    )

    # handles user-defined parameters
    self.__extra_params = {}

    # handles process to be frames written
    self.__process = None

    # handles disabling window for ffmpeg subprocess on Windows
    self.__ffmpeg_window_disabler_patch = False

    # checks machine OS
    self.__machine_OS = platform.system()

    # handles exclusive metadata
    self.__ff_pixfmt_metadata = None  # metadata
    self.__raw_frame_num = None  # raw-frame number
    self.__raw_frame_pixfmt = None  # raw-frame pixformat
    self.__raw_frame_dtype = None  # raw-frame dtype
    self.__raw_frame_depth = None  # raw-frame depth
    self.__raw_frame_resolution = None  # raw-frame resolution/dimension

    # define supported mode of operation
    self.__supported_opmodes = {
        "av": "Audio-Video",  # audio is only for pass-through, not really for audio decoding yet.
        "vo": "Video-Only",
        "imgseq": "Image-Sequence",
        # "ao":"Audio-Only", # reserved for future
    }
    # operation mode variable
    self.__opmode = None

    # handles termination
    self.__terminate_stream = False

    # cleans and reformat user-defined parameters
    self.__extra_params = {
        str(k).strip(): (
            str(v).strip()
            if not (v is None)
            and not isinstance(v, (dict, list, int, float, tuple))
            else v
        )
        for k, v in ffparams.items()
    }

    # handles custom Sourcer API params
    sourcer_params = self.__extra_params.pop("-custom_sourcer_params", {})
    # reset improper values
    sourcer_params = {} if not isinstance(sourcer_params, dict) else sourcer_params

    # handles user ffmpeg pre-headers(parameters such as `-re`) parameters (must be a list)
    self.__ffmpeg_prefixes = self.__extra_params.pop("-ffprefixes", [])
    # check if not valid type
    if not isinstance(self.__ffmpeg_prefixes, list):
        # log it
        logger.warning(
            "Discarding invalid `-ffprefixes` value of wrong type: `{}`!".format(
                type(self.__ffmpeg_prefixes).__name__
            )
        )
        # reset improper values
        self.__ffmpeg_prefixes = []
    else:
        # also pass valid ffmpeg pre-headers to Sourcer API
        sourcer_params["-ffprefixes"] = self.__ffmpeg_prefixes

    # pass parameter(if specified) to Sourcer API, specifying where to save the downloaded FFmpeg Static
    # assets on Windows(if specified)
    sourcer_params["-ffmpeg_download_path"] = self.__extra_params.pop(
        "-ffmpeg_download_path", ""
    )

    # handle video and audio stream indexes in case of multiple ones.
    default_stream_indexes = self.__extra_params.pop(
        "-default_stream_indexes", (0, 0)
    )
    # reset improper values
    default_stream_indexes = (
        (0, 0)
        if not isinstance(default_stream_indexes, (list, tuple))
        else default_stream_indexes
    )

    # pass FFmpeg filter to Sourcer API params for processing
    if set(["-vf", "-filter_complex"]).intersection(self.__extra_params.keys()):
        key = "-vf" if "-vf" in self.__extra_params else "-filter_complex"
        sourcer_params[key] = self.__extra_params[key]

    # define dict to store user-defined parameters
    self.__user_metadata = {}
    # extract and assign source metadata as dict
    (self.__sourcer_metadata, self.__missing_prop) = (
        Sourcer(
            source=source,
            source_demuxer=source_demuxer,
            verbose=verbose,
            custom_ffmpeg=custom_ffmpeg if isinstance(custom_ffmpeg, str) else "",
            **sourcer_params
        )
        .probe_stream(default_stream_indexes=default_stream_indexes)
        .retrieve_metadata(force_retrieve_missing=True)
    )

    # handle valid FFmpeg assets location
    self.__ffmpeg = self.__sourcer_metadata["ffmpeg_binary_path"]

    # handle YUV pixel formats(such as `yuv420p`, `yuv444p`, `nv12`, `nv21` etc.)
    # patch for compatibility with OpenCV APIs.
    self.__cv_patch = self.__extra_params.pop("-enforce_cv_patch", False)
    if not (isinstance(self.__cv_patch, bool)):
        self.__cv_patch = False
        self.__verbose_logs and logger.critical(
            "Enforcing OpenCV compatibility patch for YUV/NV video frames."
        )

    # handle disabling window for ffmpeg subprocess on Windows OS
    # this patch prevents ffmpeg creation window from opening when
    # building exe files
    ffmpeg_window_disabler_patch = self.__extra_params.pop(
        "-disable_ffmpeg_window", False
    )
    if ffmpeg_window_disabler_patch and isinstance(
        ffmpeg_window_disabler_patch, bool
    ):
        # check if value is valid
        if self.__machine_OS != "Windows" or self.__verbose_logs:
            logger.warning(
                "Optional `-disable_ffmpeg_window` flag is only available on Windows OS with `verbose=False`. Discarding!"
            )
        else:
            self.__ffmpeg_window_disabler_patch = ffmpeg_window_disabler_patch
    else:
        # handle improper values
        self.__ffmpeg_window_disabler_patch = False

    # handle pass-through audio mode works in conjunction with WriteGear [TODO]
    self.__passthrough_mode = self.__extra_params.pop("-passthrough_audio", False)
    if not (isinstance(self.__passthrough_mode, bool)):
        self.__passthrough_mode = False

    # handle mode of operation
    if self.__sourcer_metadata["source_has_image_sequence"]:
        # image-sequence mode
        self.__opmode = "imgseq"
    elif (
        self.__sourcer_metadata[
            "source_has_video"
        ]  # audio is only for pass-through, not really for audio decoding yet.
        and self.__sourcer_metadata["source_has_audio"]
        and self.__passthrough_mode  # [TODO]
    ):
        self.__opmode = "av"
    # elif __defop_mode == "ao" and self.__sourcer_metadata.contains_audio: # [TODO]
    #    self.__opmode = "ao"
    elif self.__sourcer_metadata["source_has_video"]:
        # video-only mode
        self.__opmode = "vo"
    else:
        # raise if unknown mode
        raise ValueError(
            "Unable to find any usable video stream in the given source!"
        )
    # store as metadata
    self.__missing_prop["ffdecoder_operational_mode"] = self.__supported_opmodes[
        self.__opmode
    ]

    # handle user-defined output framerate
    __framerate = self.__extra_params.pop("-framerate", None)
    if (
        isinstance(__framerate, str)
        and __framerate
        == "null"  # special mode to discard `-framerate/-r` parameter
    ):
        self.__inputframerate = __framerate
    elif isinstance(__framerate, (float, int)):
        self.__inputframerate = float(__framerate) if __framerate > 0.0 else 0.0
    else:
        # warn if wrong type
        not (__framerate is None) and logger.warning(
            "Discarding invalid `-framerate` value of wrong type `{}`!".format(
                type(__framerate).__name__
            )
        )
        # reset to default
        self.__inputframerate = 0.0

    # handle user defined decoded frame resolution
    self.__custom_resolution = self.__extra_params.pop("-custom_resolution", None)
    if (
        isinstance(self.__custom_resolution, str)
        and self.__custom_resolution
        == "null"  # special mode to discard `-size/-s` parameter
    ) or (
        isinstance(self.__custom_resolution, (list, tuple))
        and len(self.__custom_resolution)
        == 2  # valid resolution(must be a tuple or list)
    ):
        # log it
        self.__verbose_logs and not isinstance(
            self.__custom_resolution, str
        ) and logger.debug(
            "Setting raw frames size: `{}`.".format(self.__custom_resolution)
        )
    else:
        # log it
        not (self.__custom_resolution is None) and logger.warning(
            "Discarding invalid `-custom_resolution` value: `{}`!".format(
                self.__custom_resolution
            )
        )
        # reset improper values
        self.__custom_resolution = None

formulate(self)

This method formulates all necessary FFmpeg pipeline arguments and executes it inside the FFmpeg subprocess pipe.

Returns: A reference to the FFdecoder class object.

Source code in deffcode/ffdecoder.py
def formulate(self):
    """
    This method formulates all necessary FFmpeg pipeline arguments and executes it inside the FFmpeg `subprocess` pipe.

    **Returns:** A reference to the FFdecoder class object.
    """
    # assign values to class variables on first run
    if self.__initializing:
        # prepare parameter dict
        input_params = OrderedDict()
        output_params = OrderedDict()

        # dynamically pre-assign a default video-decoder (if not assigned by user).
        supported_vdecodecs = get_supported_vdecoders(self.__ffmpeg)
        default_vdecodec = (
            self.__sourcer_metadata["source_video_decoder"]
            if self.__sourcer_metadata["source_video_decoder"]
            in supported_vdecodecs
            else "unknown"
        )
        if "-c:v" in self.__extra_params:
            self.__extra_params["-vcodec"] = self.__extra_params.pop(
                "-c:v", default_vdecodec
            )
        # handle image sequence separately
        if self.__opmode == "imgseq":
            # -vcodec is discarded by default
            # (This is correct or maybe -vcodec required in some unknown case) [TODO]
            self.__extra_params.pop("-vcodec", None)
        elif (
            "-vcodec" in self.__extra_params
            and self.__extra_params["-vcodec"] is None
        ):
            # special case when -vcodec is not needed intentionally
            self.__extra_params.pop("-vcodec", None)
        else:
            # assign video decoder selected here.
            if not "-vcodec" in self.__extra_params:
                input_params["-vcodec"] = default_vdecodec
            else:
                input_params["-vcodec"] = self.__extra_params.pop(
                    "-vcodec", default_vdecodec
                )
            if (
                default_vdecodec != "unknown"
                and not input_params["-vcodec"] in supported_vdecodecs
            ):
                # reset to default if not supported
                logger.warning(
                    "Provided FFmpeg does not support `{}` video decoder. Switching to default supported `{}` decoder!".format(
                        input_params["-vcodec"], default_vdecodec
                    )
                )
                input_params["-vcodec"] = default_vdecodec
            # raise error if not valid decoder found
            if not input_params["-vcodec"] in supported_vdecodecs:
                raise RuntimeError(
                    "Provided FFmpeg does not support any known usable video-decoders."
                    " Either define your own manually or switch to another FFmpeg binaries(if available)."
                )

        # handle user-defined number of frames.
        if "-vframes" in self.__extra_params:
            self.__extra_params["-frames:v"] = self.__extra_params.pop(
                "-vframes", None
            )
        if "-frames:v" in self.__extra_params:
            value = self.__extra_params.pop("-frames:v", None)
            if not (value is None) and value > 0:
                output_params["-frames:v"] = value

        # dynamically calculate default raw-frames pixel format(if not assigned by user).
        # notify FFmpeg `-pix_fmt` parameter cannot be assigned directly
        if "-pix_fmt" in self.__extra_params:
            logger.warning(
                "Discarding user-defined `-pix_fmt` value as it can only be assigned with `frame_format` parameter!"
            )
            self.__extra_params.pop("-pix_fmt", None)
        # get supported FFmpeg pixfmt data with depth and bpp(bits-per-pixel)
        self.__ff_pixfmt_metadata = get_supported_pixfmts(self.__ffmpeg)
        supported_pixfmts = [fmts[0] for fmts in self.__ff_pixfmt_metadata]

        # calculate default pixel-format
        # Check special case  - `frame_format`(or `-pix_fmt`) parameter discarded from pipeline
        self.__frame_format == "null" and logger.critical(
            "Manually discarding `frame_format`(or `-pix_fmt`) parameter from this pipeline."
        )
        # choose between rgb24(if available) or source pixel-format
        # otherwise, only source pixel-format for special case
        default_pixfmt = (
            "rgb24"
            if "rgb24" in supported_pixfmts and self.__frame_format != "null"
            else self.__sourcer_metadata["source_video_pixfmt"]
        )
        # assign output raw-frames pixel format
        rawframe_pixfmt = None
        if (
            not (self.__frame_format is None)
            and self.__frame_format in supported_pixfmts
        ):
            # check if valid and supported `frame_format` parameter assigned
            rawframe_pixfmt = self.__frame_format.strip()
            self.__verbose_logs and logger.info(
                "User-defined `{}` frame pixel-format will be used for this pipeline.".format(
                    rawframe_pixfmt
                )
            )
        elif (
            "output_frames_pixfmt"
            in self.__sourcer_metadata  # means `format` filter is defined
            and self.__sourcer_metadata["output_frames_pixfmt"] in supported_pixfmts
        ):
            # assign if valid and supported
            rawframe_pixfmt = self.__sourcer_metadata[
                "output_frames_pixfmt"
            ].strip()
            self.__verbose_logs and logger.info(
                "FFmpeg filter values will be used for this pipeline for defining output pixel-format."
            )
        else:
            # reset to default if not supported
            rawframe_pixfmt = default_pixfmt
            # log it accordingly
            if self.__frame_format is None:
                logger.info(
                    "Using default `{}` pixel-format for this pipeline.".format(
                        default_pixfmt
                    )
                )
            else:
                logger.warning(
                    "{} Switching to default `{}` pixel-format!".format(
                        (
                            "Provided FFmpeg does not supports `{}` pixel-format.".format(
                                self.__sourcer_metadata["output_frames_pixfmt"]
                                if "output_frames_pixfmt" in self.__sourcer_metadata
                                else self.__frame_format
                            )
                            if self.__frame_format != "null"
                            else "No usable pixel-format defined."
                        ),
                        default_pixfmt,
                    )
                )

        # dynamically calculate raw-frame datatype based on pixel-format selected
        (self.__raw_frame_depth, rawframesbpp) = [
            (int(x[1]), int(x[2]))
            for x in self.__ff_pixfmt_metadata
            if x[0] == rawframe_pixfmt
        ][0]
        raw_bit_per_component = (
            rawframesbpp // self.__raw_frame_depth if self.__raw_frame_depth else 0
        )
        if 4 <= raw_bit_per_component <= 8:
            self.__raw_frame_dtype = np.dtype("u1")
        elif 8 < raw_bit_per_component <= 16 and rawframe_pixfmt.endswith(
            ("le", "be")
        ):
            if rawframe_pixfmt.endswith("le"):
                self.__raw_frame_dtype = np.dtype("<u2")
            else:
                self.__raw_frame_dtype = np.dtype(">u2")
        else:
            # reset to both pixel-format and datatype to default if not supported
            not (self.__frame_format is None) and logger.warning(
                "Selected pixel-format `{}` dtype is not supported by FFdecoder API. Switching to default `rgb24` pixel-format!".format(
                    rawframe_pixfmt
                )
            )
            rawframe_pixfmt = "rgb24"
            self.__raw_frame_dtype = np.dtype("u1")

        # Check if not special case
        if self.__frame_format != "null":
            # assign to FFmpeg pipeline otherwise
            output_params["-pix_fmt"] = rawframe_pixfmt
        # assign to global parameter further usage
        self.__raw_frame_pixfmt = rawframe_pixfmt
        # also override as metadata(if available)
        if "output_frames_pixfmt" in self.__sourcer_metadata:
            self.__sourcer_metadata["output_frames_pixfmt"] = (
                self.__raw_frame_pixfmt
            )

        # handle raw-frame resolution
        # notify FFmpeg `-s` parameter cannot be assigned directly
        if "-s" in self.__extra_params:
            logger.warning(
                "Discarding user-defined `-s` FFmpeg parameter as it can only be assigned with `-custom_resolution` attribute! Read docs for more details."
            )
            self.__extra_params.pop("-s", None)
        # assign output rawframe resolution
        if not (self.__custom_resolution is None) and not isinstance(
            self.__custom_resolution, str
        ):
            # assign if assigned by user and not "null"(str)
            self.__raw_frame_resolution = self.__custom_resolution
            self.__verbose_logs and logger.info(
                "User-defined `{}` frame resolution will be used for this pipeline.".format(
                    self.__raw_frame_resolution
                )
            )
        elif (
            "output_frames_resolution"
            in self.__sourcer_metadata  # means `scale` filter is defined
            and self.__sourcer_metadata["output_frames_resolution"]
            and len(self.__sourcer_metadata["output_frames_resolution"]) == 2
        ):
            # calculate raw-frame resolution/dimensions based on output.
            self.__raw_frame_resolution = self.__sourcer_metadata[
                "output_frames_resolution"
            ]
        elif (
            self.__sourcer_metadata["source_video_resolution"]
            and len(self.__sourcer_metadata["source_video_resolution"]) == 2
        ):
            # calculate raw-frame resolution/dimensions based on source.
            self.__raw_frame_resolution = self.__sourcer_metadata[
                "source_video_resolution"
            ]
        else:
            # otherwise raise error
            raise RuntimeError(
                "Both source and output metadata values found Invalid with {} `-custom_resolution` attribute. Aborting!".format(
                    "null"
                    if isinstance(self.__inputframerate, str)
                    else "undefined"
                )
            )
        # special mode to discard `-size/-s` FFmpeg parameter completely
        if isinstance(self.__custom_resolution, str):
            logger.critical(
                "Manually discarding `-size/-s` FFmpeg parameter from this pipeline."
            )
        else:
            # add to pipeline
            dimensions = "{}x{}".format(
                self.__raw_frame_resolution[0], self.__raw_frame_resolution[1]
            )
            output_params["-s"] = str(dimensions)
        # log if filters or default source is used
        self.__verbose_logs and (
            self.__custom_resolution is None
            or isinstance(self.__custom_resolution, str)
        ) and logger.info(
            "{} for this pipeline for defining output resolution.".format(
                "FFmpeg filter values will be used"
                if "output_frames_resolution" in self.__sourcer_metadata
                else "Default source resolution will be used"
            )
        )

        # dynamically calculate raw-frame framerate based on source (if not assigned by user).
        if (
            not isinstance(self.__inputframerate, str)
            and self.__inputframerate > 0.0
        ):
            # assign if assigned by user and not "null"(str)
            output_params["-framerate"] = str(self.__inputframerate)
            self.__verbose_logs and logger.info(
                "User-defined `{}` output framerate will be used for this pipeline.".format(
                    str(self.__inputframerate)
                )
            )
        elif (
            "output_framerate"
            in self.__sourcer_metadata  # means `fps` filter is defined
            and self.__sourcer_metadata["output_framerate"] > 0.0
        ):
            # special mode to discard `-framerate/-r` FFmpeg parameter completely
            if self.__inputframerate == "null":
                logger.critical(
                    "Manually discarding `-framerate/-r` FFmpeg parameter from this pipeline."
                )
            else:
                # calculate raw-frame framerate based on output
                output_params["-framerate"] = str(
                    self.__sourcer_metadata["output_framerate"]
                )
            self.__verbose_logs and logger.info(
                "FFmpeg filter values will be used for this pipeline for defining output framerate."
            )
        elif self.__sourcer_metadata["source_video_framerate"] > 0.0:
            # special mode to discard `-framerate/-r` FFmpeg parameter completely
            if self.__inputframerate == "null":
                logger.critical(
                    "Manually disabling `-framerate/-r` FFmpeg parameter for this pipeline."
                )
            else:
                # calculate raw-frame framerate based on source
                output_params["-framerate"] = str(
                    self.__sourcer_metadata["source_video_framerate"]
                )
            self.__verbose_logs and logger.info(
                "Default source framerate will be used for this pipeline for defining output framerate."
            )
        else:
            # otherwise raise error
            raise RuntimeError(
                "Both source and output metadata values found Invalid with {} `-framerate` attribute. Aborting!".format(
                    "null"
                    if isinstance(self.__inputframerate, str)
                    else "undefined"
                )
            )

        # add rest to output parameters
        output_params.update(self.__extra_params)

        # dynamically calculate raw-frame numbers based on source (if not assigned by user).
        # TODO Added support for `-re -stream_loop` and `-loop`
        if "-frames:v" in input_params:
            self.__raw_frame_num = input_params["-frames:v"]
        elif (
            not (self.__sourcer_metadata["approx_video_nframes"] is None)
            and self.__sourcer_metadata["approx_video_nframes"] > 0
        ):
            self.__raw_frame_num = self.__sourcer_metadata["approx_video_nframes"]
        else:
            self.__raw_frame_num = None
            # log that number of frames are unknown
            self.__verbose_logs and logger.info(
                "Number of frames in given source are unknown. Live/Network/Looping stream detected!"
            )

        # log Mode of Operation
        self.__verbose_logs and logger.critical(
            "Activating {} Mode of Operation.".format(
                self.__supported_opmodes[self.__opmode]
            )
        )

        # compose the Pipeline using formulated FFmpeg parameters
        self.__launch_FFdecoderline(input_params, output_params)

        # inform the initialization is completed
        self.__initializing = False
    else:
        # warn if pipeline is recreated
        logger.error("This pipeline is already created and running!")
    return self

generateFrame(self)

This method returns a Generator function (also an Iterator using next()) of video frames, grabbed continuously from the buffer.

Source code in deffcode/ffdecoder.py
def generateFrame(self):
    """
    This method returns a [Generator function](https://wiki.python.org/moin/Generators)
    _(also an Iterator using `next()`)_ of video frames, grabbed continuously from the buffer.
    """
    if self.__raw_frame_num is None or not self.__raw_frame_num:
        while not self.__terminate_stream:  # infinite raw frames
            frame = self.__fetchNextFrame()
            if frame is None:
                self.__terminate_stream = True
                break
            yield frame
    else:
        for _ in range(self.__raw_frame_num):  # finite raw frames
            frame = self.__fetchNextFrame()
            if frame is None:
                self.__terminate_stream = True
                break
            yield frame

terminate(self)

Safely terminates all processes.

Source code in deffcode/ffdecoder.py
def terminate(self):
    """
    Safely terminates all processes.
    """

    # signal we are closing
    self.__verbose_logs and logger.debug("Terminating FFdecoder Pipeline...")
    self.__terminate_stream = True
    # check if no process was initiated at first place
    if self.__process is None or not (self.__process.poll() is None):
        logger.info("Pipeline already terminated.")
        return
    # Attempt to close pipeline.
    # close `stdin` output
    self.__process.stdin and self.__process.stdin.close()
    # close `stdout` output
    self.__process.stdout and self.__process.stdout.close()
    # terminate/kill process if still processing
    self.__process.poll() is None and self.__process.terminate()
    # wait if not exiting
    self.__process.wait()
    self.__process = None
    logger.info("Pipeline terminated successfully.")