FFGear API References

class FFGear:
    """
    FFGear API is a multi-threaded, high-performance wrapper around
    [**DeFFcode's FFdecoder API**](https://abhitronix.github.io/deffcode/latest/reference/ffdecoder/)
    that compiles and executes an FFmpeg pipeline inside a subprocess pipe for generating
    real-time, low-overhead, lightning-fast decoded video frames in Python.

    FFGear API provides direct, transparent access to the full FFdecoder feature-set, including:

    - **Hardware-Accelerated Decoding** — CUDA/CUVID and other `-hwaccel` backends.
    - **Flexible Pixel Formats** — any FFmpeg-supported `-pix_fmt` (e.g. `bgr24`, `yuv420p`, `gray`), with an optional OpenCV-compatibility patch (`-enforce_cv_patch`) for YUV/NV layouts.
    - **Per-Frame Metadata Extraction** — asynchronous `showinfo` filter integration via `-extract_metadata`, yielding `(frame, metadata)` tuples with `frame_num`, `pts_time`, `is_keyframe`, and `frame_type`.
    - **Complex Filtergraphs** — live simple (`-vf`) and complex (`-filter_complex`) FFmpeg filter pipelines.
    - **Multi-Input Sources** — multiple simultaneous inputs routed via `-map` or `-filter_complex`.
    - **Wide Source Support** — USB/Virtual/IP camera feeds, multimedia files, image sequences, screen recordings, and network protocols (HTTP(s), RTP/RTSP, etc.).

    Internally, FFGear employs a Producer-Consumer threaded queue (configurable via `QUEUE_SIZE`,
    `THREADED_QUEUE_MODE`, `THREAD_TIMEOUT`) for zero-bottleneck asynchronous frame delivery, and
    maintains the standard OpenCV-Python coding syntax for drop-in integration.

    Similar to CamGear, FFGear also supports the `yt_dlp` backend via `stream_mode=True` for
    seamlessly pipelining live video-frames and metadata from streaming services.
    """

    def __init__(
        self,
        source: Any = None,
        stream_mode: bool = False,
        source_demuxer: str | None = None,
        frame_format: str = "bgr24",
        custom_ffmpeg: str = "",
        logging: bool = False,
        **options: dict,
    ):
        """
        This constructor method initializes the object state and attributes of the FFGear class.

        Parameters:
            source (Any): defines the source for the input stream (URL, file path, or device index).
            stream_mode (bool): controls the exclusive **Stream Mode** for handling streaming URLs.
            source_demuxer (str): specifies the demuxer for the source (e.g., 'v4l2', 'dshow').
            frame_format (str): specifies the pixel layout for decoding (e.g., 'bgr24', 'rgb24', 'yuv420p', 'gray').
            custom_ffmpeg (str): specifies a custom FFmpeg executable path.
            logging (bool): enables/disables console logging.
            options (dict): FFdecoder configuration (like -hwaccel, -enforce_cv_patch, -extract_metadata)
                            and FFGear queue tweaks.
        """
        # raise error for critical import
        import_dependency_safe(
            "deffcode" if deffcode is None else "", min_version="0.2.7"
        )

        self.__logging = logging if isinstance(logging, bool) else False
        logcurr_vidgear_ver(logging=self.__logging)

        # initialize global
        self.ytv_metadata = {}

        # check if Stream-Mode is ON (True)
        if stream_mode:
            stream_resolution = get_supported_resolution(
                options.pop("STREAM_RESOLUTION", "best"), logging=self.__logging
            )
            if yt_dlp is not None:
                yt_stream_params = options.pop("STREAM_PARAMS", {})
                if isinstance(yt_stream_params, dict):
                    yt_stream_params = {
                        str(k).strip(): v for k, v in yt_stream_params.items()
                    }
                else:
                    yt_stream_params = {}
                try:
                    logger.info(
                        "Verifying Streaming URL using yt-dlp backend. Please wait..."
                    )
                    ytbackend = YT_backend(
                        source_url=source, logging=self.__logging, **yt_stream_params
                    )
                    if ytbackend:
                        self.ytv_metadata = ytbackend.meta_data
                        ytbackend.is_livestream and logger.warning(
                            "Livestream URL detected. Consider using a GStreamer-based pipeline for best results."
                        )
                        if stream_resolution not in ytbackend.streams:
                            logger.warning(
                                "Specified stream-resolution `{}` is not available. Reverting to `best`!".format(
                                    stream_resolution
                                )
                            )
                            stream_resolution = "best"
                        else:
                            self.__logging and logger.debug(
                                "Using `{}` resolution for streaming.".format(
                                    stream_resolution
                                )
                            )
                        source = ytbackend.streams[stream_resolution]
                        self.__logging and logger.debug(
                            "YouTube source ID: `{}`, Title: `{}`, Quality: `{}`".format(
                                self.ytv_metadata["id"],
                                self.ytv_metadata["title"],
                                stream_resolution,
                            )
                        )
                except Exception:
                    raise ValueError(
                        "[FFGear:ERROR] :: Stream Mode is enabled but Input URL is invalid!"
                    )
            else:
                import_dependency_safe("yt_dlp")

        self.source = source
        self.frame_format = frame_format.strip().lower() if frame_format else "bgr24"

        # -------------------------------------------------------------
        # Extract FFGear-specific properties from **options
        # -------------------------------------------------------------

        # Extract Threaded Queue Mode constraints
        self.__threaded_queue_mode = options.pop("THREADED_QUEUE_MODE", True)
        if not isinstance(self.__threaded_queue_mode, bool):
            self.__threaded_queue_mode = True

        self.__thread_timeout = options.pop("THREAD_TIMEOUT", None)
        if self.__thread_timeout and isinstance(self.__thread_timeout, (int, float)):
            self.__thread_timeout = float(self.__thread_timeout)
        else:
            self.__thread_timeout = None

        # Parse queue size (similar to WriteGear parameter parsing)
        q_size = options.pop("QUEUE_SIZE", 96)
        q_size = q_size if isinstance(q_size, int) else 96
        self.__queue = (
            queue.Queue(maxsize=q_size) if self.__threaded_queue_mode else None
        )

        if self.__threaded_queue_mode:
            self.__logging and logger.debug(
                f"Threaded Queue Mode enabled with Queue Size: {q_size}."
            )
        else:
            self.__logging and logger.warning(
                "Threaded Queue Mode disabled. High performance may be bottlenecked!"
            )

        # -------------------------------------------------------------
        # Parse specialized FFdecoder features / Frame format handling
        # -------------------------------------------------------------

        # Support metadata extraction recipe natively
        self.__extract_metadata = options.get("-extract_metadata", False)
        self.frame_metadata = {}

        # Handle OpenCV patch for YUV420p frames
        # (This yields a H*1.5, W frame layout ready for cv2.cvtColor)
        self.__enforce_cv_patch = options.get("-enforce_cv_patch", False)
        if self.frame_format == "yuv420p" and self.__enforce_cv_patch:
            self.__logging and logger.debug(
                "Enforcing CV patch for yuv420p format. Frames will require OpenCV conversion."
            )

        # -------------------------------------------------------------
        # Formulate FFdecoder Stream Pipeline
        # -------------------------------------------------------------
        try:
            self.stream = FFdecoder(
                source=self.source,
                source_demuxer=source_demuxer,
                frame_format=self.frame_format,
                custom_ffmpeg=custom_ffmpeg,
                verbose=self.__logging,
                **options,  # Pass remaining FFmpeg params (-hwaccel, -vcodec, etc.)
            ).formulate()
        except Exception as e:
            raise RuntimeError(
                f"[FFGear:ERROR] :: Failed to initialize FFdecoder pipeline: {e!s}"
            )

        # Initialize the generator
        self.__generator = self.stream.generateFrame()

        # Pull first frame to confirm stream validity
        self.frame = None
        self.__read_first_frame()

        if self.frame is not None:
            if self.__threaded_queue_mode:
                # When metadata mode is active enqueue the full (frame, meta) tuple
                if self.__extract_metadata:
                    self.__queue.put((self.frame, self.frame_metadata))
                else:
                    self.__queue.put(self.frame)
        else:
            raise RuntimeError("[FFGear:ERROR] :: Source is invalid or unreadable!")

        # Synchronization primitives
        self.__thread = None
        self.__terminate = Event()
        self.__stream_read = Event()

    def __read_first_frame(self):
        """Helper to pull the first frame and process formatting/metadata."""
        try:
            out = next(self.__generator)
            # Route based on the '-extract_metadata' recipe output
            if self.__extract_metadata and isinstance(out, tuple):
                self.frame, self.frame_metadata = out[0], out[1]
            else:
                self.frame = out

            self.__process_frame_format()
        except StopIteration:
            self.frame = None

    def __process_frame_format(self):
        """
        Handles OpenCV conversion routines based on the user's chosen frame_format
        and patching parameters.
        """
        if self.frame is None:
            return

        # Automatically translate patched YUV420p data into standard BGR space via OpenCV
        # if the user requested the patch but wants a ready-to-display frame output
        if self.frame_format == "yuv420p" and self.__enforce_cv_patch:
            try:
                self.frame = cv2.cvtColor(self.frame, cv2.COLOR_YUV2BGR_I420)
            except Exception as e:
                self.__logging and logger.warning(
                    f"Failed to process patched YUV420p frame via OpenCV: {e!s}"
                )

    def start(self):
        """
        Starts the Threaded Video Streaming.
        """
        self.__thread = Thread(target=self.__update, name="FFGear", args=())
        self.__thread.daemon = True
        self.__thread.start()
        return self

    def __update(self):
        """
        Producer Thread: Iterates frames continuously from the FFdecoder generator,
        processes metadata/colorspace routines, and yields them to the bounded queue.
        """
        while not self.__terminate.is_set():
            self.__stream_read.clear()

            try:
                out = next(self.__generator)

                # Separate frame and metadata if recipe is activated
                if self.__extract_metadata and isinstance(out, tuple):
                    frame, self.frame_metadata = out[0], out[1]
                else:
                    frame = out

                grabbed = True
            except StopIteration:
                frame = None
                grabbed = False

            self.__stream_read.set()

            if not grabbed or frame is None:
                if self.__threaded_queue_mode:
                    if self.__queue.empty():
                        break
                    else:
                        continue
                else:
                    break

            self.frame = frame

            # Apply any OpenCV colorspace logic based on format choices
            self.__process_frame_format()

            # push frame (or frame+meta tuple) onto the producer queue
            if self.__threaded_queue_mode:
                if self.__extract_metadata:
                    self.__queue.put((self.frame, self.frame_metadata))
                else:
                    self.__queue.put(self.frame)

        # Poison pill for the queue signaling termination
        self.__threaded_queue_mode and self.__queue.put(None)
        self.__threaded_queue_mode = False

        self.__terminate.set()
        self.__stream_read.set()

        # Cleanup deffcode pipeline (similar to WriteGear close process)
        if self.stream is not None:
            self.stream.terminate()
            self.stream = None

    def read(self) -> NDArray | tuple[NDArray, dict] | None:
        """
        Consumer Thread: Pops frames symmetrically from the queue block.

        **Returns:**
            NDArray | tuple[NDArray, dict] | None: A decoded ``numpy.ndarray``
            frame, or a ``(frame, metadata)`` tuple when ``-extract_metadata``
            is enabled (metadata keys: ``frame_num``, ``pts_time``,
            ``is_keyframe``, ``frame_type``). Returns ``None`` when the
            stream has ended.
        """
        while self.__threaded_queue_mode and not self.__terminate.is_set():
            try:
                return self.__queue.get(timeout=self.__thread_timeout)
            except queue.Empty:
                continue

        if not self.__terminate.is_set() and self.__stream_read.wait(
            timeout=self.__thread_timeout
        ):
            if self.__extract_metadata:
                return (self.frame, self.frame_metadata)
            return self.frame
        return None

    def stop(self) -> None:
        """
        Safely halts processes, unblocks memory, and flushes pipelines.
        """
        self.__logging and logger.debug("Terminating processes.")
        self.__threaded_queue_mode = False

        self.__stream_read.set()
        self.__terminate.set()

        # Flush the remainder of the queue to unblock the producer
        if self.__thread is not None:
            if self.__queue is not None:
                while not self.__queue.empty():
                    try:
                        self.__queue.get_nowait()
                    except queue.Empty:
                        continue
                    self.__queue.task_done()
            self.__thread.join()

        # Forcibly teardown FFdecoder process safely
        if self.stream is not None:
            self.stream.terminate()
            self.stream = None

class YT_backend:
    """
    Internal YT-DLP Backend Class for extracting metadata from Streaming URLs.

    Shared across CamGear and FFGear APIs.

    Parameters:
        source_url (string): defines the URL of source stream
        logging (bool): enables/disables logging.
        stream_params (dict): provides ability to alter yt-dlp backend params.
    """

    def __init__(
        self, source_url: str, logging: bool = False, **stream_params: dict
    ):
        # initialize global params
        self.__logging = logging
        self.is_livestream = False
        self.streams_metadata = {}
        self.streams = {}

        # define supported resolution values
        self.supported_resolutions : dict[str, str] = {
            "256x144": "144p",
            "426x240": "240p",
            "640x360": "360p",
            "854x480": "480p",
            "1280x720": "720p",
            "1920x1080": "1080p",
            "2560x1440": "1440p",
            "3840x2160": "2160p",
            "7680x4320": "4320p",
        }

        # assign source_url
        self.source_url = source_url

        # define default options for yt-dlp backend
        self.ydl_opts = {
            "format": "best*[vcodec!=none]",
            "quiet": True,
            "prefer_insecure": False,
            "no_warnings": not logging,
            "dump_single_json": True,
            "extract_flat": True,
            "skip_download": True,
        }
        # remove any attribute from user dict
        # that can cause API to fail
        stream_params.pop("format", None)
        stream_params.pop("dump_single_json", None)
        stream_params.pop("extract_flat", None)

        # extract exclusive params
        std_hdrs = stream_params.pop("std_headers", None)
        if std_hdrs is not None and isinstance(std_hdrs, dict):
            yt_dlp.utils.std_headers.update(std_hdrs)

        # update with user defined options
        self.ydl_opts.update(stream_params)

        # extract metadata
        self.meta_data = self.__extract_meta()

        # check if source url is supported
        if (
            self.meta_data is not None  # meta-data is valid
            and "entries" not in self.meta_data  # playlists are not supported
            and len(self.meta_data.get("formats", {}))
            > 0  # video formats must exist
        ):
            self.is_livestream = self.meta_data.get("is_live", False)
            self.streams_metadata = self.meta_data.get("formats", {})
            self.streams = self.__extract_streams()
            if self.streams:
                logger.info(
                    "[Backend] :: Streaming URL is fully supported. Available Streams are: [{}]".format(
                        ", ".join(list(self.streams.keys()))
                    )
                )
            else:
                raise ValueError(
                    "[Backend] :: Streaming URL isn't supported. No usable video streams found!"
                )
        else:
            # otherwise notify user
            raise ValueError(
                "[Backend] :: Streaming URL isn't valid{}".format(
                    ". Playlists aren't supported yet!"
                    if self.meta_data is not None and "entries" in self.meta_data
                    else "!"
                )
            )

    def __extract_meta(self):
        extracted_data = None
        # run parser
        with YoutubeDL(self.ydl_opts) as ydl:
            try:
                # parse data
                extracted_data = ydl.extract_info(self.source_url, download=False)
            except yt_dlp.utils.DownloadError as e:
                # raise errors
                raise RuntimeError(" [Backend] : " + str(e))
        # return data
        return extracted_data

    def __extract_streams(self):
        # extract streams
        streams = {}
        streams_copy = {}
        for stream in self.streams_metadata:
            # extract useable metadata
            stream_dim = stream.get("resolution", "")
            stream_url = stream.get("url", "")
            stream_protocol = stream.get("protocol", "")
            stream_with_video = stream.get("vcodec", "none") != "none"
            stream_with_audio = stream.get("acodec", "none") != "none"
            # streams must contain video
            if (
                stream_with_video
                and stream_dim
                and stream_url
                and stream_protocol != "http_dash_segments"
            ):
                # check if stream resolution is supported
                if stream_dim in self.supported_resolutions:
                    stream_res = self.supported_resolutions[stream_dim]
                    if (
                        not stream_with_audio  # prefer audioless
                        or stream_protocol in ["https", "http"]  # prefer http/https
                        or stream_res not in streams  # check if already not in dict
                    ):
                        streams[stream_res] = stream_url
                # otherwise make a copy
                if (
                    not stream_with_audio  # prefer audioless
                    or stream_protocol in ["https", "http"]  # prefer http/https
                    or stream_dim
                    not in streams_copy  # check if already not in dict
                ):
                    streams_copy[stream_dim] = stream_url
        # use copy to decide best or worst
        if not streams_copy:
            return {}
        streams["best"] = streams_copy[list(streams_copy.keys())[-1]]
        streams["worst"] = streams_copy[next(iter(streams_copy.keys()))]
        return streams