Detection

Detection module for object detection using YOLO.

Detector

Detector(
    model: DetectorModel | str = DetectorModel.YOLO26x,
    weights: str | None = None,
    conf: float = 0.25,
    nms: float = 0.7,
    max_det: int = 300,
    device: str = "auto",
    half: bool = False,
    imgsz: int = 640,
    classes: list[int] | None = None,
    class_names: list[str] | None = None,
    rect: bool = False,
    agnostic_nms: bool = False,
    embed: list[int] | None = None,
    fast: bool = True,
    batch: int = 8,
)

A wrapper around Ultralytics detection models for running object detection on videos and selected frames.

This class loads a YOLO (v8, v11, 26) or RT-DETR model from a local models/ directory (or from a user-supplied .pt file) and provides convenience methods to:

• detect objects frame by frame in a video and return results as a pandas DataFrame,
• run detection only on specified frame indices,
• process a batch of videos and save per-video detection text files, and
• query basic video properties (FPS, frame count).

The detector automatically chooses an inference device (cuda, xpu, mps, or cpu) when device="auto", and it can optionally enable half-precision inference on GPU.

Parameters:

Name	Type	Description	Default
model	DetectorModel or str	Built-in model weights to use. Accepts a DetectorModel member (DetectorModel.YOLO26x), its value / file name ("yolo26x.pt"), its stem ("yolo26x"), or its enum name ("YOLO26x"). Default is DetectorModel.YOLO26x.	YOLO26x
weights	str	Optional custom model weights to load. If relative, path is resolved under <module_dir>/models/. Default is None.	None
conf	float	Confidence threshold for detections. Default is 0.25.	0.25
nms	float	IoU / non-maximum suppression threshold. Default is 0.7.	0.7
max_det	int	Maximum number of detections per frame. Default is 300.	300
device	(auto, cuda, xpu, cpu, mps)	Inference device to use. If "auto", the detector will pick an available accelerator first (cuda → xpu → mps) and fall back to CPU. Default is "auto".	"auto"
half	bool	Whether to enable half-precision inference. This is only effective on GPU (CUDA). Default is False.	False
imgsz	int	Inference image size (longest side, pixels), forwarded to Ultralytics model.predict(imgsz=...). Default is 640, Ultralytics' own default.	640
classes	list of int	Restrict detection to these class indices, forwarded to Ultralytics model.predict(classes=...). Default is None (all classes), Ultralytics' own default.	None
class_names	list of str	Restrict detection to these class names (e.g. ["car", "truck", "bus"]), resolved against dnt.shared.util.load_class_dict() (COCO names, matching the stock YOLO/RT-DETR weights) and merged into classes. Default is None.	None
rect	bool	Whether to use rectangular (non-square) inference, forwarded to Ultralytics model.predict(rect=...). Default is False, Ultralytics' own default.	False
agnostic_nms	bool	Whether to use class-agnostic NMS, forwarded to Ultralytics model.predict(agnostic_nms=...). Default is False, Ultralytics' own default.	False
embed	list of int	Layer indices to return feature embeddings from, forwarded to Ultralytics model.predict(embed=...). Default is None (no embeddings), Ultralytics' own default.	None
fast	bool	Whether detect() redirects to detect_fast(batch=self.batch) (True, the default) or runs the traditional one-frame-at-a-time pipeline (False). Mutate detector.fast after construction to change it for later calls. See detect_fast() for the speed/behavior trade-off fast=True makes.	True
batch	int	Frames per forward pass, used when fast is True: by detect()'s redirect, and as the default for direct detect_fast() calls that don't pass their own batch=. Default is 8.	8

Notes

• The class expects model weight files to be located under <module_dir>/models/ when using the built-in weight names.
• Returned detection tables typically contain the columns: frame, res, x, y, w, h, conf, class.

Initialize a Detector for Ultralytics YOLO/RT-DETR models.

Parameters:

Name	Type	Description	Default
model	DetectorModel or str	Built-in model to use. Accepts a DetectorModel member, its value / file name ("yolo26x.pt"), its stem ("yolo26x"), or its enum name ("YOLO26x"). Default is "yolo26x".	YOLO26x
weights	str	Customized model weights to load. Default is None, which means using the built-in weights in model choice.	None
conf	float	Confidence threshold. Default is 0.25.	0.25
nms	float	IoU/NMS threshold. Default is 0.7.	0.7
max_det	int	Maximum detections per frame. Default is 300. In crowded scenes, you may want to increase this.	300
device	(auto, cuda, xpu, cpu, mps)	Inference device. Default is "auto".	"auto"
half	bool	Whether to use half precision (GPU only). Default is False.	False
imgsz	int	Inference image size, as passed to Ultralytics model.predict(). Default is 640 (Ultralytics' own default).	640
classes	list of int	Class indices to restrict detection to, as passed to Ultralytics model.predict(). Default is None, i.e. all classes (Ultralytics' own default).	None
class_names	list of str	Class names to restrict detection to (e.g. ["car", "truck"]), resolved via dnt.shared.util.load_class_dict() (COCO names) and merged into classes. Raises ValueError for an unknown name. Default is None.	None
rect	bool	Rectangular inference, as passed to Ultralytics model.predict(). Default is False (Ultralytics' own default).	False
agnostic_nms	bool	Class-agnostic NMS, as passed to Ultralytics model.predict(). Default is False (Ultralytics' own default).	False
embed	list of int	Layer indices to return feature embeddings from, as passed to Ultralytics model.predict(). Default is None, i.e. no embeddings (Ultralytics' own default).	None
fast	bool	Whether detect() redirects to detect_fast(batch=self.batch) by default. Default is True. See detect_fast() for the trade-off this makes; set detector.fast = False after construction (or per instance here) to keep detect() on the traditional per-frame pipeline.	True
batch	int	Default frames per forward pass when fast is True. Default is 8.	8

Source code in src/dnt/detect/yolo/detector.py

def __init__(
    self,
    model: DetectorModel | str = DetectorModel.YOLO26x,
    weights: str | None = None,
    conf: float = 0.25,
    nms: float = 0.7,
    max_det: int = 300,
    device: str = "auto",
    half: bool = False,
    imgsz: int = 640,
    classes: list[int] | None = None,
    class_names: list[str] | None = None,
    rect: bool = False,
    agnostic_nms: bool = False,
    embed: list[int] | None = None,
    fast: bool = True,
    batch: int = 8,
):
    """Initialize a Detector for Ultralytics YOLO/RT-DETR models.

    Parameters
    ----------
    model : DetectorModel or str, optional
        Built-in model to use. Accepts a `DetectorModel` member, its value /
        file name (`"yolo26x.pt"`), its stem (`"yolo26x"`), or its enum name
        (`"YOLO26x"`). Default is "yolo26x".
    weights : str, optional
        Customized model weights to load.
        Default is None, which means using the built-in weights in `model` choice.
    conf : float, optional
        Confidence threshold. Default is 0.25.
    nms : float, optional
        IoU/NMS threshold. Default is 0.7.
    max_det : int, optional
        Maximum detections per frame.
        Default is 300. In crowded scenes, you may want to increase this.
    device : {"auto", "cuda", "xpu", "cpu", "mps"}, optional
        Inference device. Default is "auto".
    half : bool, optional
        Whether to use half precision (GPU only). Default is False.
    imgsz : int, optional
        Inference image size, as passed to Ultralytics `model.predict()`.
        Default is 640 (Ultralytics' own default).
    classes : list of int, optional
        Class indices to restrict detection to, as passed to Ultralytics
        `model.predict()`. Default is None, i.e. all classes (Ultralytics' own
        default).
    class_names : list of str, optional
        Class names to restrict detection to (e.g. `["car", "truck"]`), resolved
        via `dnt.shared.util.load_class_dict()` (COCO names) and merged into
        `classes`. Raises `ValueError` for an unknown name. Default is None.
    rect : bool, optional
        Rectangular inference, as passed to Ultralytics `model.predict()`.
        Default is False (Ultralytics' own default).
    agnostic_nms : bool, optional
        Class-agnostic NMS, as passed to Ultralytics `model.predict()`.
        Default is False (Ultralytics' own default).
    embed : list of int, optional
        Layer indices to return feature embeddings from, as passed to
        Ultralytics `model.predict()`. Default is None, i.e. no embeddings
        (Ultralytics' own default).
    fast : bool, optional
        Whether `detect()` redirects to `detect_fast(batch=self.batch)` by default.
        Default is True. See `detect_fast()` for the trade-off this makes; set
        `detector.fast = False` after construction (or per instance here) to keep
        `detect()` on the traditional per-frame pipeline.
    batch : int, optional
        Default frames per forward pass when `fast` is True. Default is 8.

    """
    model = _coerce_model(model)

    # Load model
    cwd = Path(__file__).parent.absolute()
    model_dir = cwd / "models"
    if not model_dir.exists():
        os.makedirs(model_dir)

    if weights:
        model_path = Path(weights) if os.path.isabs(weights) else model_dir / weights
    else:
        model_path = model_dir / f"{model.value}"

    # actually load model
    if ("yolo" in str(weights).lower()) or (model in YOLO_MODELS):
        self.model = YOLO(str(model_path))
    elif ("rtdetr" in str(weights).lower()) or (model in RTDETR_MODELS):
        self.model = RTDETR(str(model_path))
    else:
        raise ValueError(
            f"Cannot infer model family from model={model} and weights={weights!r}. "
            "Use a known DetectorModel or provide weights containing 'yolo' or 'rtdetr'."
        )
    self.conf = conf
    self.nms = nms
    self.max_det = max_det
    self.imgsz = imgsz
    if class_names:
        # deferred: dnt.shared imports dnt.detect (Synchronizer), so importing at module level would cycle
        from ...shared.util import load_class_dict

        name_to_idx = load_class_dict()
        try:
            resolved = {name_to_idx[name] for name in class_names}
        except KeyError as e:
            raise ValueError(
                f"Unknown class name {e.args[0]!r}. "
                "See dnt.shared.util.load_class_dict() for valid names."
            ) from e
        classes = sorted({*(classes or []), *resolved})
    self.classes = classes
    self.rect = rect
    self.agnostic_nms = agnostic_nms
    self.embed = embed
    self.fast = fast
    self.batch = batch

    self.device = resolve_device(device)
    # half precision only makes sense on CUDA (any index)
    self.half = half_allowed(self.device, half)
    # translates to quantize= on Ultralytics versions that deprecated half= (avoids a
    # per-frame deprecation warning); falls back to half= on older versions
    self._precision_kwargs = predict_precision_kwargs(self.half)

detect

detect(
    input_video: str,
    iou_file: str | None = None,
    video_index: int | None = None,
    video_tot: int | None = None,
    start_frame: int | None = None,
    end_frame: int | None = None,
    verbose: bool = True,
    show: bool = False,
    message: str | None = None,
    return_df: bool = True,
) -> pd.DataFrame | None

Run object detection on a video and return per-frame detections.

Parameters:

Name	Type	Description	Default
input_video	str	Path to the input video file.	required
iou_file	str	If provided, detection results are written to this file (CSV without header).	None
video_index	int	Index of this video in a batch, used only for progress display.	None
video_tot	int	Total number of videos in the batch, used only for progress display.	None
start_frame	int	Frame index to start detection from. If None or out of range, starts at 0.	None
end_frame	int	Frame index to stop detection at. If None or out of range, uses the last frame.	None
verbose	bool	Whether to show a progress bar. Default is True.	True
show	bool	Whether to display the video frames with detections. Default is False. Not supported when self.fast is True (detect_fast() has no preview); set detector.fast = False to use show.	False
message	str | None	Optional message shown in the progress bar description. Default is None.	None
return_df	bool	Whether to return the detections. Default is True. With False, detections are only written to iou_file (which is then required) and None is returned; for long videos this avoids holding every detection in memory, since the fast path streams rows to iou_file as it goes.	True

Returns:

Type	Description
DataFrame or None	DataFrame with columns: frame, res, x, y, w, h, conf, class. If the video cannot be opened or no detections are found, an empty DataFrame with those columns is returned. None if return_df is False.

Notes

With self.fast, iou_file is written incrementally as <iou_file>.part and renamed when complete; an interrupted run resumes from that file on the next call (see detect_fast()).

Source code in src/dnt/detect/yolo/detector.py

def detect(
    self,
    input_video: str,
    iou_file: str | None = None,
    video_index: int | None = None,
    video_tot: int | None = None,
    start_frame: int | None = None,
    end_frame: int | None = None,
    verbose: bool = True,
    show: bool = False,
    message: str | None = None,
    return_df: bool = True,
) -> pd.DataFrame | None:
    """Run object detection on a video and return per-frame detections.

    Parameters
    ----------
    input_video : str
        Path to the input video file.
    iou_file : str, optional
        If provided, detection results are written to this file (CSV without header).
    video_index : int, optional
        Index of this video in a batch, used only for progress display.
    video_tot : int, optional
        Total number of videos in the batch, used only for progress display.
    start_frame : int, optional
        Frame index to start detection from. If None or out of range, starts at 0.
    end_frame : int, optional
        Frame index to stop detection at. If None or out of range, uses the last frame.
    verbose : bool, optional
        Whether to show a progress bar. Default is True.
    show : bool, optional
        Whether to display the video frames with detections. Default is False.
        Not supported when `self.fast` is True (`detect_fast()` has no preview); set
        `detector.fast = False` to use `show`.
    message : str | None, optional
        Optional message shown in the progress bar description.
        Default is None.
    return_df : bool, optional
        Whether to return the detections. Default is True. With False, detections are
        only written to `iou_file` (which is then required) and None is returned; for
        long videos this avoids holding every detection in memory, since the `fast`
        path streams rows to `iou_file` as it goes.

    Returns
    -------
    pandas.DataFrame or None
        DataFrame with columns:
        `frame, res, x, y, w, h, conf, class`.
        If the video cannot be opened or no detections are found, an empty DataFrame
        with those columns is returned. None if `return_df` is False.

    Notes
    -----
    With `self.fast`, `iou_file` is written incrementally as
    `<iou_file>.part` and renamed when complete; an interrupted run resumes from that
    file on the next call (see `detect_fast()`).

    """
    if not return_df and not iou_file:
        msg = "return_df=False needs an iou_file; otherwise the detections would be discarded."
        raise ValueError(msg)
    if self.fast:
        if show:
            msg = (
                "detect(show=True) is not supported while self.fast is True: "
                "detect_fast() has no preview. Set detector.fast = False for a live preview."
            )
            raise ValueError(msg)
        return self.detect_fast(
            input_video,
            iou_file=iou_file,
            video_index=video_index,
            video_tot=video_tot,
            start_frame=start_frame,
            end_frame=end_frame,
            verbose=verbose,
            message=message,
            batch=self.batch,
            return_df=return_df,
        )

    # validate path
    cap = cv2.VideoCapture(input_video)
    if not cap.isOpened():
        if verbose:
            print(f"Cannot open video: {input_video}")
        return pd.DataFrame(columns=self.DET_FIELDS) if return_df else None

    tot_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    # normalize start_frame
    if start_frame is None or start_frame < 0 or start_frame >= tot_frames:
        start_frame = 0
    # normalize end_frame
    if end_frame is None or end_frame < 0 or end_frame >= tot_frames:
        end_frame = tot_frames - 1
    if start_frame > end_frame:
        cap.release()
        raise ValueError("start_frame must be less than or equal to end_frame.")

    frame_total = end_frame - start_frame + 1

    # Some codecs return 0 or -1 for frame count
    if verbose:
        if (video_index is not None) and (video_tot is not None):
            desc = f"Detecting {video_index} of {video_tot}"
        else:
            desc = "Detecting"

        if message is not None:
            desc += f" {message}"

        if tot_frames <= 0:
            pbar = tqdm(desc=desc, unit="frame")
        else:
            pbar = tqdm(total=frame_total, desc=desc, unit="frame")

    results: list[dict] = []
    frame_idx = start_frame
    cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)

    win_name = "Detection (press q/ESC to quit)"
    if show:
        cv2.namedWindow(win_name, cv2.WINDOW_NORMAL)

    # optional FPS calc
    t0 = time()
    n_show = 0

    while cap.isOpened():
        pos_frame = int(cap.get(cv2.CAP_PROP_POS_FRAMES))
        ret, frame = cap.read()
        if not ret:
            break

        if end_frame is not None and frame_idx > end_frame:
            break

        preds = self.model.predict(
            source=frame,
            conf=self.conf,
            iou=self.nms,
            max_det=self.max_det,
            device=self.device,
            imgsz=self.imgsz,
            classes=self.classes,
            rect=self.rect,
            agnostic_nms=self.agnostic_nms,
            embed=self.embed,
            **self._precision_kwargs,
            verbose=False,
        )

        det = preds[0]
        boxes = det.boxes
        if boxes is not None and len(boxes) > 0:
            xyxy = boxes.xyxy.cpu().numpy()  # (N,4)
            confs = boxes.conf.cpu().numpy()  # (N,)
            clss = boxes.cls.cpu().numpy().astype(int)  # (N,)

            for (x1, y1, x2, y2), cf, c in zip(xyxy, confs, clss, strict=True):
                results.append({
                    "frame": pos_frame,
                    "res": -1,
                    "x": float(x1),
                    "y": float(y1),
                    "x2": float(x2),
                    "y2": float(y2),
                    "conf": float(cf),
                    "class": int(c),
                })

        if show:
            # Ultralytics built-in drawing (fast & clean)
            vis = det.plot()  # returns BGR image with boxes/labels

            # add simple overlay: frame index + FPS
            n_show += 1
            dt = time() - t0
            fps = n_show / dt if dt > 0 else 0.0
            cv2.putText(
                vis,
                f"frame={pos_frame}/{frame_total}  fps={fps:.1f}",
                (10, 25),
                cv2.FONT_HERSHEY_SIMPLEX,
                0.7,
                (0, 255, 255),
                2,
            )

            cv2.imshow(win_name, vis)
            key = cv2.waitKey(1) & 0xFF
            if key == ord("q") or key == 27:  # q or ESC
                break

        if verbose and pbar is not None:
            pbar.update(1)

        frame_idx += 1

    cap.release()
    if verbose and pbar is not None:
        pbar.close()

    if not results:
        empty_df = pd.DataFrame(columns=self.DET_FIELDS)
        if iou_file:
            empty_df.to_csv(iou_file, index=False, header=False)
        return empty_df if return_df else None

    results_df = self._results_to_df(results)

    if iou_file:
        folder = Path(iou_file).parent
        if not folder.exists():
            Path(folder).mkdir(parents=True, exist_ok=True)

        results_df.to_csv(iou_file, index=False, header=False)

    return results_df if return_df else None

detect_fast

detect_fast(
    input_video: str,
    iou_file: str | None = None,
    video_index: int | None = None,
    video_tot: int | None = None,
    start_frame: int | None = None,
    end_frame: int | None = None,
    verbose: bool = True,
    message: str | None = None,
    batch: int | None = None,
    return_df: bool = True,
) -> pd.DataFrame | None

Like detect(), but batches frames through the Ultralytics AutoBackend directly.

detect() calls Ultralytics' high-level model.predict() once per frame. That call rebuilds an inference dataset around the single frame and forces several CUDA synchronizations on every call, regardless of batch size, which is roughly 2x slower than running several frames through the backend in one forward pass. This method does that instead: it reuses Ultralytics' own model-family-correct preprocess()/inference()/postprocess() (so decoding, e.g. RT-DETR's vs YOLO's NMS, is unchanged), just without model.predict()'s per-frame dataset/callback setup.

batch here means frames within this one video, unlike detect_batch()'s batch, which means a list of videos.

Parameters:

Name	Type	Description	Default
input_video	str	Path to the input video file.	required
iou_file	str	If provided, detection results are written to this file (CSV without header), batch by batch (see Notes).	None
video_index	int	Index of this video in a batch, used only for progress display.	None
video_tot	int	Total number of videos in the batch, used only for progress display.	None
start_frame	int	Frame index to start detection from. If None or out of range, starts at 0.	None
end_frame	int	Frame index to stop detection at. If None or out of range, uses the last frame.	None
verbose	bool	Whether to show a progress bar. Default is True.	True
message	str | None	Optional message shown in the progress bar description. Default is None.	None
batch	int	Number of frames sent through the model in one forward pass. Default is None: use self.batch (see Detector). Larger values usually raise GPU throughput at the cost of more GPU memory and higher latency per flush; tune it to your GPU and frame resolution.	None
return_df	bool	Whether to return the detections. Default is True. With False, rows are only written to iou_file (then required) and None is returned, so memory stays flat however long the video is.	True

Returns:

Type	Description
DataFrame or None	Same columns and semantics as detect(): frame, res, x, y, w, h, conf, class. If the video cannot be opened or no detections are found, an empty DataFrame with those columns is returned. None if return_df is False.

Notes

• No show= preview: batching trades per-frame latency for throughput, so a live preview does not fit this path. Use detect() for that.
• batch=1 reproduces detect() exactly (verified frame-for-frame, CPU and CUDA, YOLO and RT-DETR): the speed gain at batch=1 comes purely from skipping model.predict()'s per-frame dataset/callback setup (see _ensure_predictor()), not from any change in what gets computed.

• batch > 1 on CUDA with half=True can shift a small fraction of results: running the same image alone vs. as part of a larger batch takes a different path through cuDNN/cuBLAS's FP16 kernels, so reduction order (and, rarely, which of two near-duplicate boxes NMS keeps) can differ slightly. Measured on an RTX 5070 Ti, RT-DETR-x, 200 frames, batch=8: half=True shifted results on 6% of frames (up to ~0.17 confidence, ~67 px on the frames it did shift); half=False shifted only 2.5% of frames (up to 0.02 confidence, 1 px, i.e. rounding noise). This is inherent to batched half-precision GPU inference, not specific to this bypass; detect() called once per frame never hits it, since it never batches. If exact reproducibility with detect() matters more than throughput, use half=False and/or batch=1.

• With iou_file, each batch's rows are appended to <iou_file>.part as soon as they are detected, and the file is renamed to iou_file only once the whole range is done, so an existing iou_file is always complete (and detect_batch() can keep skipping existing files). If <iou_file>.part already exists, e.g. after a crash or Ctrl-C, detection resumes from it: the rows of its last frame are dropped and that frame is detected again. Delete the .part file to start over instead. With batch > 1 and half=True, frames after the resume point are grouped into batches differently than in an uninterrupted run, so they can differ slightly in the way described above.

Source code in src/dnt/detect/yolo/detector.py

def detect_fast(
    self,
    input_video: str,
    iou_file: str | None = None,
    video_index: int | None = None,
    video_tot: int | None = None,
    start_frame: int | None = None,
    end_frame: int | None = None,
    verbose: bool = True,
    message: str | None = None,
    batch: int | None = None,
    return_df: bool = True,
) -> pd.DataFrame | None:
    """Like `detect()`, but batches frames through the Ultralytics `AutoBackend` directly.

    `detect()` calls Ultralytics' high-level `model.predict()` once per frame. That call
    rebuilds an inference dataset around the single frame and forces several CUDA
    synchronizations on every call, regardless of batch size, which is roughly 2x slower
    than running several frames through the backend in one forward pass. This method
    does that instead: it reuses Ultralytics' own model-family-correct
    `preprocess()`/`inference()`/`postprocess()` (so decoding, e.g. RT-DETR's vs YOLO's
    NMS, is unchanged), just without `model.predict()`'s per-frame dataset/callback setup.

    `batch` here means frames within this one video, unlike `detect_batch()`'s `batch`,
    which means a list of videos.

    Parameters
    ----------
    input_video : str
        Path to the input video file.
    iou_file : str, optional
        If provided, detection results are written to this file (CSV without header),
        batch by batch (see Notes).
    video_index : int, optional
        Index of this video in a batch, used only for progress display.
    video_tot : int, optional
        Total number of videos in the batch, used only for progress display.
    start_frame : int, optional
        Frame index to start detection from. If None or out of range, starts at 0.
    end_frame : int, optional
        Frame index to stop detection at. If None or out of range, uses the last frame.
    verbose : bool, optional
        Whether to show a progress bar. Default is True.
    message : str | None, optional
        Optional message shown in the progress bar description.
        Default is None.
    batch : int, optional
        Number of frames sent through the model in one forward pass. Default is None:
        use `self.batch` (see `Detector`). Larger values usually raise GPU throughput
        at the cost of more GPU memory and higher latency per flush; tune it to your
        GPU and frame resolution.
    return_df : bool, optional
        Whether to return the detections. Default is True. With False, rows are only
        written to `iou_file` (then required) and None is returned, so memory stays
        flat however long the video is.

    Returns
    -------
    pandas.DataFrame or None
        Same columns and semantics as `detect()`: `frame, res, x, y, w, h, conf, class`.
        If the video cannot be opened or no detections are found, an empty DataFrame
        with those columns is returned. None if `return_df` is False.

    Notes
    -----
    - No `show=` preview: batching trades per-frame latency for throughput, so a live
      preview does not fit this path. Use `detect()` for that.
    - `batch=1` reproduces `detect()` exactly (verified frame-for-frame, CPU and CUDA,
      YOLO and RT-DETR): the speed gain at `batch=1` comes purely from skipping
      `model.predict()`'s per-frame dataset/callback setup (see `_ensure_predictor()`),
      not from any change in what gets computed.
    - `batch > 1` on CUDA with `half=True` can shift a small fraction of results:
      running the same image alone vs. as part of a larger batch takes a different
      path through cuDNN/cuBLAS's FP16 kernels, so reduction order (and, rarely, which
      of two near-duplicate boxes NMS keeps) can differ slightly. Measured on an
      RTX 5070 Ti, RT-DETR-x, 200 frames, `batch=8`: `half=True` shifted results on 6%
      of frames (up to ~0.17 confidence, ~67 px on the frames it did shift); `half=False`
      shifted only 2.5% of frames (up to 0.02 confidence, 1 px, i.e. rounding noise).
      This is inherent to batched half-precision GPU inference, not specific to this
      bypass; `detect()` called once per frame never hits it, since it never batches.
      If exact reproducibility with `detect()` matters more than throughput, use
      `half=False` and/or `batch=1`.

    - With `iou_file`, each batch's rows are appended to `<iou_file>.part` as soon as
      they are detected, and the file is renamed to `iou_file` only once the whole range
      is done, so an existing `iou_file` is always complete (and `detect_batch()` can
      keep skipping existing files). If `<iou_file>.part` already exists, e.g. after a
      crash or Ctrl-C, detection resumes from it: the rows of its last frame are dropped
      and that frame is detected again. Delete the `.part` file to start over instead.
      With `batch > 1` and `half=True`, frames after the resume point are grouped into
      batches differently than in an uninterrupted run, so they can differ slightly in
      the way described above.

    """
    batch = self.batch if batch is None else batch

    def prepare():
        predictor = self._ensure_predictor()
        return lambda im0s: self._predict_arrays(predictor, im0s, input_video)

    return self._run_batched(
        input_video,
        prepare,
        iou_file=iou_file,
        video_index=video_index,
        video_tot=video_tot,
        start_frame=start_frame,
        end_frame=end_frame,
        verbose=verbose,
        message=message,
        batch=batch,
        return_df=return_df,
    )

detect_frames

detect_frames(
    input_video: str,
    frames: list[int],
    verbose: bool = True,
) -> pd.DataFrame

Run object detection on specific frames of a video.

This method is useful when you don't need to process the entire video and only want detections for selected frame indices.

Parameters:

Name	Type	Description	Default
input_video	str	Path to the input video file.	required
frames	list of int	List of frame indices to process.	required
verbose	bool	Whether to show a progress bar. Default is True.	True

Returns:

Type	Description
DataFrame	DataFrame with columns frame, res, x, y, w, h, conf, class. If the video cannot be opened or no detections are found, an empty DataFrame with those columns is returned.

Source code in src/dnt/detect/yolo/detector.py

def detect_frames(
    self,
    input_video: str,
    frames: list[int],
    verbose: bool = True,
) -> pd.DataFrame:
    """Run object detection on specific frames of a video.

    This method is useful when you don't need to process the entire video and
    only want detections for selected frame indices.

    Parameters
    ----------
    input_video : str
        Path to the input video file.
    frames : list of int
        List of frame indices to process.
    verbose : bool, optional
        Whether to show a progress bar. Default is True.

    Returns
    -------
    pandas.DataFrame
        DataFrame with columns
        `frame, res, x, y, w, h, conf, class`.
        If the video cannot be opened or no detections are found, an empty
        DataFrame with those columns is returned.

    """
    # validate path
    if not os.path.exists(input_video):
        # return an empty, well-shaped DataFrame instead of None
        if verbose:
            print(f"{input_video} does not exist!")
        return pd.DataFrame(columns=self.DET_FIELDS)

    cap = cv2.VideoCapture(input_video)
    if not cap.isOpened():
        if verbose:
            print(f"Cannot open {input_video}")
        return pd.DataFrame(columns=self.DET_FIELDS)

    results: list[dict] = []

    pbar = tqdm(total=len(frames), unit=" frames") if verbose else None

    for pos_frame in frames:
        # move to target frame
        cap.set(cv2.CAP_PROP_POS_FRAMES, pos_frame)
        ret, frame = cap.read()
        if not ret:
            # e.g. frame index out of range
            continue

        preds = self.model.predict(
            frame,
            verbose=False,
            conf=self.conf,
            iou=self.nms,
            max_det=self.max_det,
            device=self.device,
            imgsz=self.imgsz,
            classes=self.classes,
            rect=self.rect,
            agnostic_nms=self.agnostic_nms,
            **self._precision_kwargs,
            embed=self.embed,
        )
        det = preds[0]
        boxes = det.boxes
        if boxes is not None and len(boxes) > 0:
            xyxy = boxes.xyxy.cpu().numpy()  # (N,4)
            confs = boxes.conf.cpu().numpy()  # (N,)
            clss = boxes.cls.cpu().numpy().astype(int)  # (N,)

            for (x1, y1, x2, y2), cf, c in zip(xyxy, confs, clss, strict=True):
                results.append({
                    "frame": pos_frame,
                    "res": -1,
                    "x": float(x1),
                    "y": float(y1),
                    "x2": float(x2),
                    "y2": float(y2),
                    "conf": float(cf),
                    "class": int(c),
                })

        if pbar is not None:
            pbar.update()

    if pbar is not None:
        pbar.close()
    cap.release()

    return self._results_to_df(results)

detect_batch

detect_batch(
    input_videos: list[str],
    output_path: str | None = None,
    is_overwrite: bool = False,
    is_report: bool = True,
    verbose: bool = True,
    message: str | None = None,
) -> list[str]

Run detection on multiple videos and optionally write per-video output files.

Parameters:

Name	Type	Description	Default
input_videos	list of str	Paths to the input video files to be processed.	required
output_path	str	Directory where per-video detection files will be written. If None, detections are not written to disk and the returned list will be empty.	None
is_overwrite	bool	If False (default), existing detection files with the same name will be skipped. If True, they will be regenerated.	False
is_report	bool	If True (default), existing detection files (that were skipped) are still included in the returned list.	True
verbose	bool	If True, prints progress messages. Default is True.	True
message	str | None	Optional message shown in each progress bar description. Default is None.	None

Returns:

Type	Description
list of str	A list of paths to detection files that were created or already existed. If output_path is None, this will be an empty list.

Notes

• Each video's detect() call uses this Detector instance's fast/batch (see Detector), the same as calling detect() directly; there is no separate argument for them here.
• A video whose _iou.txt doesn't exist yet but whose _iou.txt.part does (an interrupted fast run) resumes from the .part file; with is_overwrite=True the .part file is deleted and the video starts over.

Source code in src/dnt/detect/yolo/detector.py

def detect_batch(
    self,
    input_videos: list[str],
    output_path: str | None = None,
    is_overwrite: bool = False,
    is_report: bool = True,
    verbose: bool = True,
    message: str | None = None,
) -> list[str]:
    """Run detection on multiple videos and optionally write per-video output files.

    Parameters
    ----------
    input_videos : list of str
        Paths to the input video files to be processed.
    output_path : str, optional
        Directory where per-video detection files will be written. If None,
        detections are not written to disk and the returned list will be empty.
    is_overwrite : bool, optional
        If False (default), existing detection files with the same name will be
        skipped. If True, they will be regenerated.
    is_report : bool, optional
        If True (default), existing detection files (that were skipped) are still
        included in the returned list.
    verbose : bool, optional
        If True, prints progress messages. Default is True.
    message : str | None, optional
        Optional message shown in each progress bar description.
        Default is None.

    Returns
    -------
    list of str
        A list of paths to detection files that were created or already existed.
        If `output_path` is None, this will be an empty list.

    Notes
    -----
    - Each video's `detect()` call uses this `Detector` instance's `fast`/`batch` (see
      `Detector`), the same as calling `detect()` directly; there is no separate
      argument for them here.
    - A video whose `_iou.txt` doesn't exist yet but whose `_iou.txt.part` does (an
      interrupted `fast` run) resumes from the `.part` file; with
      `is_overwrite=True` the `.part` file is deleted and the video starts over.

    """
    results: list[str] = []
    total_videos = len(input_videos)

    for idx, input_video in enumerate(input_videos, start=1):
        # default: no output file
        iou_file = None

        # build output path / file name if requested
        if output_path is not None:
            Path(output_path).mkdir(parents=True, exist_ok=True)
            base_filename = os.path.splitext(os.path.basename(input_video))[0]
            iou_file = os.path.join(output_path, f"{base_filename}_iou.txt")

        # if we have an output file name, check overwrite logic
        if (iou_file is not None) and (not is_overwrite) and os.path.exists(iou_file):
            if is_report:
                results.append(iou_file)
            # skip processing this video
            continue

        # regenerating: don't resume from an earlier, possibly different, partial run
        if iou_file is not None and is_overwrite:
            Path(f"{iou_file}.part").unlink(missing_ok=True)

        # run detection (may write to iou_file if not None); the returned table is
        # discarded here, so don't build it when the rows go to a file anyway
        self.detect(
            input_video=input_video,
            iou_file=iou_file,
            video_index=idx,
            video_tot=total_videos,
            verbose=verbose,
            message=message,
            return_df=iou_file is None,
        )

        if iou_file is not None:
            results.append(iou_file)

    return results

get_fps staticmethod

get_fps(video: str) -> float

Return the frames-per-second (FPS) value of a video file.

Parameters:

Name	Type	Description	Default
video	str	Path to the video file.	required

Returns:

Type	Description
float	FPS of the video. Returns 0.0 if the video cannot be opened.

Source code in src/dnt/detect/yolo/detector.py

@staticmethod
def get_fps(video: str) -> float:
    """Return the frames-per-second (FPS) value of a video file.

    Parameters
    ----------
    video : str
        Path to the video file.

    Returns
    -------
    float
        FPS of the video. Returns 0.0 if the video cannot be opened.

    """
    if not Path(video).exists():
        print(f"{video} does not exist!")
        return 0.0
    cap = cv2.VideoCapture(video)
    if not cap.isOpened():
        print(f"Failed to open the video: {video}")
        return 0.0

    fps = float(cap.get(cv2.CAP_PROP_FPS))
    cap.release()
    return fps

get_frames staticmethod

get_frames(video: str) -> int

Return the total number of frames in a video file.

Parameters:

Name	Type	Description	Default
video	str	Path to the video file.	required

Returns:

Type	Description
int	Total frame count. Returns 0 if the video cannot be opened.

Source code in src/dnt/detect/yolo/detector.py

@staticmethod
def get_frames(video: str) -> int:
    """Return the total number of frames in a video file.

    Parameters
    ----------
    video : str
        Path to the video file.

    Returns
    -------
    int
        Total frame count. Returns 0 if the video cannot be opened.

    """
    if not Path(video).exists():
        print(f"{video} does not exist!")
        return 0
    cap = cv2.VideoCapture(video)
    if not cap.isOpened():
        print(f"Failed to open the video: {video}")
        return 0

    frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    cap.release()
    return frames

DetectorModel

Bases: str, Enum

Enum of available YOLO and RT-DETR model weights.

Each member represents a different model variant with its corresponding weight file name.