Reference for ultralytics/data/loaders.py

def __init__(self, sources: str = "file.streams", vid_stride: int = 1, buffer: bool = False, channels: int = 3):
    """
    Initialize stream loader for multiple video sources, supporting various stream types.

    Args:
        sources (str): Path to streams file or single stream URL.
        vid_stride (int): Video frame-rate stride.
        buffer (bool): Whether to buffer input streams.
        channels (int): Number of image channels (1 for grayscale, 3 for RGB).
    """
    torch.backends.cudnn.benchmark = True  # faster for fixed-size inference
    self.buffer = buffer  # buffer input streams
    self.running = True  # running flag for Thread
    self.mode = "stream"
    self.vid_stride = vid_stride  # video frame-rate stride
    self.cv2_flag = cv2.IMREAD_GRAYSCALE if channels == 1 else cv2.IMREAD_COLOR  # grayscale or RGB

    sources = Path(sources).read_text().rsplit() if os.path.isfile(sources) else [sources]
    n = len(sources)
    self.bs = n
    self.fps = [0] * n  # frames per second
    self.frames = [0] * n
    self.threads = [None] * n
    self.caps = [None] * n  # video capture objects
    self.imgs = [[] for _ in range(n)]  # images
    self.shape = [[] for _ in range(n)]  # image shapes
    self.sources = [ops.clean_str(x).replace(os.sep, "_") for x in sources]  # clean source names for later
    for i, s in enumerate(sources):  # index, source
        # Start thread to read frames from video stream
        st = f"{i + 1}/{n}: {s}... "
        if urllib.parse.urlparse(s).hostname in {"www.youtube.com", "youtube.com", "youtu.be"}:  # YouTube video
            # YouTube format i.e. 'https://www.youtube.com/watch?v=Jsn8D3aC840' or 'https://youtu.be/Jsn8D3aC840'
            s = get_best_youtube_url(s)
        s = eval(s) if s.isnumeric() else s  # i.e. s = '0' local webcam
        if s == 0 and (IS_COLAB or IS_KAGGLE):
            raise NotImplementedError(
                "'source=0' webcam not supported in Colab and Kaggle notebooks. "
                "Try running 'source=0' in a local environment."
            )
        self.caps[i] = cv2.VideoCapture(s)  # store video capture object
        if not self.caps[i].isOpened():
            raise ConnectionError(f"{st}Failed to open {s}")
        w = int(self.caps[i].get(cv2.CAP_PROP_FRAME_WIDTH))
        h = int(self.caps[i].get(cv2.CAP_PROP_FRAME_HEIGHT))
        fps = self.caps[i].get(cv2.CAP_PROP_FPS)  # warning: may return 0 or nan
        self.frames[i] = max(int(self.caps[i].get(cv2.CAP_PROP_FRAME_COUNT)), 0) or float(
            "inf"
        )  # infinite stream fallback
        self.fps[i] = max((fps if math.isfinite(fps) else 0) % 100, 0) or 30  # 30 FPS fallback

        success, im = self.caps[i].read()  # guarantee first frame
        im = cv2.cvtColor(im, cv2.COLOR_BGR2GRAY)[..., None] if self.cv2_flag == cv2.IMREAD_GRAYSCALE else im
        if not success or im is None:
            raise ConnectionError(f"{st}Failed to read images from {s}")
        self.imgs[i].append(im)
        self.shape[i] = im.shape
        self.threads[i] = Thread(target=self.update, args=([i, self.caps[i], s]), daemon=True)
        LOGGER.info(f"{st}Success ✅ ({self.frames[i]} frames of shape {w}x{h} at {self.fps[i]:.2f} FPS)")
        self.threads[i].start()
    LOGGER.info("")  # newline

def __init__(self, source: str, channels: int = 3):
    """
    Initialize screenshot capture with specified screen and region parameters.

    Args:
        source (str): Screen capture source string in format "screen_num left top width height".
        channels (int): Number of image channels (1 for grayscale, 3 for RGB).
    """
    check_requirements("mss")
    import mss  # noqa

    source, *params = source.split()
    self.screen, left, top, width, height = 0, None, None, None, None  # default to full screen 0
    if len(params) == 1:
        self.screen = int(params[0])
    elif len(params) == 4:
        left, top, width, height = (int(x) for x in params)
    elif len(params) == 5:
        self.screen, left, top, width, height = (int(x) for x in params)
    self.mode = "stream"
    self.frame = 0
    self.sct = mss.mss()
    self.bs = 1
    self.fps = 30
    self.cv2_flag = cv2.IMREAD_GRAYSCALE if channels == 1 else cv2.IMREAD_COLOR  # grayscale or RGB

    # Parse monitor shape
    monitor = self.sct.monitors[self.screen]
    self.top = monitor["top"] if top is None else (monitor["top"] + top)
    self.left = monitor["left"] if left is None else (monitor["left"] + left)
    self.width = width or monitor["width"]
    self.height = height or monitor["height"]
    self.monitor = {"left": self.left, "top": self.top, "width": self.width, "height": self.height}

def __init__(self, path: Union[str, Path, List], batch: int = 1, vid_stride: int = 1, channels: int = 3):
    """
    Initialize dataloader for images and videos, supporting various input formats.

    Args:
        path (str | Path | List): Path to images/videos, directory, or list of paths.
        batch (int): Batch size for processing.
        vid_stride (int): Video frame-rate stride.
        channels (int): Number of image channels (1 for grayscale, 3 for RGB).
    """
    parent = None
    if isinstance(path, str) and Path(path).suffix == ".txt":  # *.txt file with img/vid/dir on each line
        parent = Path(path).parent
        path = Path(path).read_text().splitlines()  # list of sources
    files = []
    for p in sorted(path) if isinstance(path, (list, tuple)) else [path]:
        a = str(Path(p).absolute())  # do not use .resolve() https://github.com/ultralytics/ultralytics/issues/2912
        if "*" in a:
            files.extend(sorted(glob.glob(a, recursive=True)))  # glob
        elif os.path.isdir(a):
            files.extend(sorted(glob.glob(os.path.join(a, "*.*"))))  # dir
        elif os.path.isfile(a):
            files.append(a)  # files (absolute or relative to CWD)
        elif parent and (parent / p).is_file():
            files.append(str((parent / p).absolute()))  # files (relative to *.txt file parent)
        else:
            raise FileNotFoundError(f"{p} does not exist")

    # Define files as images or videos
    images, videos = [], []
    for f in files:
        suffix = f.rpartition(".")[-1].lower()  # Get file extension without the dot and lowercase
        if suffix in IMG_FORMATS:
            images.append(f)
        elif suffix in VID_FORMATS:
            videos.append(f)
    ni, nv = len(images), len(videos)

    self.files = images + videos
    self.nf = ni + nv  # number of files
    self.ni = ni  # number of images
    self.video_flag = [False] * ni + [True] * nv
    self.mode = "video" if ni == 0 else "image"  # default to video if no images
    self.vid_stride = vid_stride  # video frame-rate stride
    self.bs = batch
    self.cv2_flag = cv2.IMREAD_GRAYSCALE if channels == 1 else cv2.IMREAD_COLOR  # grayscale or RGB
    if any(videos):
        self._new_video(videos[0])  # new video
    else:
        self.cap = None
    if self.nf == 0:
        raise FileNotFoundError(f"No images or videos found in {p}. {FORMATS_HELP_MSG}")

def __init__(self, im0: Union[Image.Image, np.ndarray, List], channels: int = 3):
    """
    Initialize a loader for PIL and Numpy images, converting inputs to a standardized format.

    Args:
        im0 (PIL.Image.Image | np.ndarray | List): Single image or list of images in PIL or numpy format.
        channels (int): Number of image channels (1 for grayscale, 3 for RGB).
    """
    if not isinstance(im0, list):
        im0 = [im0]
    # use `image{i}.jpg` when Image.filename returns an empty path.
    self.paths = [getattr(im, "filename", "") or f"image{i}.jpg" for i, im in enumerate(im0)]
    pil_flag = "L" if channels == 1 else "RGB"  # grayscale or RGB
    self.im0 = [self._single_check(im, pil_flag) for im in im0]
    self.mode = "image"
    self.bs = len(self.im0)

def __init__(self, im0: torch.Tensor) -> None:
    """
    Initialize LoadTensor object for processing torch.Tensor image data.

    Args:
        im0 (torch.Tensor): Input tensor with shape (B, C, H, W).
    """
    self.im0 = self._single_check(im0)
    self.bs = self.im0.shape[0]
    self.mode = "image"
    self.paths = [getattr(im, "filename", f"image{i}.jpg") for i, im in enumerate(im0)]