Inference

Predictor(model, *, preprocessing=_DEFAULT_PREPROCESSING, device=None, use_amp=None, pred_params=None, strategy=None, nms=None, tile_hw=None, stride_hw=None, tiler=None)

Main class for running inference with DeepVisionTools models.

Predictor wraps a DeepVisionModel or a saved deepvisiontools model artifact and exposes a single high-level predict(...) entry point. It supports tensor inputs, single image paths, lists of image paths, optional file streaming, optional tiled inference, automatic task-specific prediction strategies, NMS, visualisation export, device selection and AMP control.

Parameters:

Name	Type	Description	Default
model	`DeepVisionModel | str | Path`	Model used for inference. It can be an already-instantiated DeepVisionModel, or a path to a model artifact saved with deepvisiontools and restored through ModelFactory.load.	required
preprocessing	`Callable`	Callable applied to image tensors before prediction. It should usually match the preprocessing used during training. Defaults to imagenet_preprocessing().	deepvisiontools.inference.predictor._DEFAULT_PREPROCESSING
device	`str | torch.device | None`	Device used for inference. If None, the device falls back to the global current() configuration.	None
use_amp	`bool | None`	Whether to use automatic mixed precision during inference. If None, the model's own use_amp setting is preserved.	None
pred_params	`BasePredParams | None`	Optional prediction parameters used only by this Predictor, such as confidence threshold or NMS-related thresholds depending on the model wrapper. If None, the model's own pred_params are used.	None
strategy	`TaskPredictorStrategy | None`	Optional task-specific strategy used to post-process, merge and finalize predictions. If None, the strategy is automatically selected from the model task.	None
nms	`NMSProvider | None`	Optional NMS implementation used by tiling and detection/instance merging strategies. If None, DefaultNMS() is used.	None
tile_hw	`tuple[int, int] | None`	Optional tiled inference patch size as (height, width). If provided, a default SlidingWindowTiler is built.	None
stride_hw	`tuple[int, int] | None`	Optional tiled inference stride as (height, width). Smaller stride than tile_hw creates overlapping patches. Used only when tile_hw builds the default tiler.	None
tiler	`PatchTiler | None`	Optional custom tiler. If provided, it takes precedence over the default tiler built from tile_hw and stride_hw.	None

Accepted prediction inputs

predict(...) accepts:

• a single torch.Tensor image with shape C, H, W;
• a batch torch.Tensor with shape B, C, H, W;
• a single image path (str or Path);
• a list of image paths.

Tensor inputs are already resident in memory, so they are processed as one batch by default. Path inputs are streamed one image at a time by default. Use max_images_per_batch to batch several file paths together during inference.

Tiled inference

Tiled inference is enabled when at least one tiling source is available:

• tile_hw=(height, width) builds a default SlidingWindowTiler;
• stride_hw=(height, width) controls overlap for the default tiler;
• tiler=... can be passed at construction time;
• tiler=... can also be passed directly to predict(...).

During tiled inference, the batch_size argument of predict(...) controls the number of patches sent to the model at once. It is not the same thing as max_images_per_batch, which controls how many full images are loaded or processed together.

Visualisations

predict(..., generate_visu=True) saves one visualisation per predicted image. If visu_names is not provided and the input is a path, output files are saved next to the source images with names like visu__image_name.png.

If the input is a tensor, default visualisation names are generated as visu_0.png, visu_1.png, etc. For batches, pass one output path per image through visu_names.

Usage

from pathlib import Path

import torch

from deepvisiontools.inference import Predictor
from deepvisiontools.inference.tiling import SlidingWindowTiler
from deepvisiontools.models import ModelFactory

# ------------------------------------------------------------------
# 1. Build or load a model, then create a Predictor
# ------------------------------------------------------------------

model = ModelFactory(
    "smp",
    arch="Unet",
    encoder_name="resnet18",
    encoder_weights="imagenet",
    in_channels=3,
    num_classes=2,
)

predictor = Predictor(
    model,
    device="cuda",
    use_amp=True,
)

# You can also create a Predictor directly from a saved model artifact.
predictor = Predictor(
    "/path/to/deepvisiontools_model_artifact",
    device="cuda",
    use_amp=True,
)

# ------------------------------------------------------------------
# 2. Predict from a single tensor image
# ------------------------------------------------------------------

image = torch.rand(3, 512, 512)
preds = predictor.predict(image)

# preds is a BatchData object, even for a single image.
first_prediction = preds[0]

# ------------------------------------------------------------------
# 3. Predict from a tensor batch
# ------------------------------------------------------------------

images = torch.rand(4, 3, 512, 512)
preds = predictor.predict(images)

# For tensor batches, the full batch is processed by default.
# You can force smaller full-image chunks with max_images_per_batch.
preds = predictor.predict(
    images,
    max_images_per_batch=2,
)

# ------------------------------------------------------------------
# 4. Predict from a single image path
# ------------------------------------------------------------------

preds = predictor.predict("images/image_001.png")

# ------------------------------------------------------------------
# 5. Predict from several image paths
# ------------------------------------------------------------------

image_paths = [
    "images/image_001.png",
    "images/image_002.png",
    "images/image_003.png",
]

# Default behavior for paths is true streaming: one image at a time.
preds = predictor.predict(image_paths)

# To load and predict several files together, set max_images_per_batch.
# Images with different spatial sizes are center-padded internally and
# predictions are cropped back to their original sizes.
preds = predictor.predict(
    image_paths,
    max_images_per_batch=2,
)

# ------------------------------------------------------------------
# 6. Save prediction visualisations
# ------------------------------------------------------------------

preds = predictor.predict(
    image_paths,
    generate_visu=True,
)

# Or provide explicit output names.
preds = predictor.predict(
    image_paths,
    generate_visu=True,
    visu_names=[
        "outputs/visu_image_001.png",
        "outputs/visu_image_002.png",
        "outputs/visu_image_003.png",
    ],
)

# ------------------------------------------------------------------
# 7. Use tiled inference from tile_hw / stride_hw
# ------------------------------------------------------------------

tiled_predictor = Predictor(
    model,
    device="cuda",
    use_amp=True,
    tile_hw=(512, 512),
    stride_hw=(384, 384),
)

# Here, batch_size is the number of patches processed at once.
preds = tiled_predictor.predict(
    "large_images/large_image.png",
    batch_size=4,
)

# The tiling parameters can also be changed after construction.
tiled_predictor.tile_hw = (768, 768)
tiled_predictor.stride_hw = (512, 512)

preds = tiled_predictor.predict(
    "large_images/large_image.png",
    batch_size=2,
)

# ------------------------------------------------------------------
# 8. Use an explicit tiler
# ------------------------------------------------------------------

tiler = SlidingWindowTiler(
    patch_hw=(512, 512),
    stride_hw=(256, 256),
)

predictor = Predictor(
    model,
    device="cuda",
    tiler=tiler,
)

preds = predictor.predict(
    "large_images/large_image.png",
    batch_size=4,
)

# You can also pass a one-off tiler directly to predict(...).
other_tiler = SlidingWindowTiler(
    patch_hw=(1024, 1024),
    stride_hw=(768, 768),
)

preds = predictor.predict(
    "large_images/large_image.png",
    tiler=other_tiler,
    batch_size=1,
)

# ------------------------------------------------------------------
# 9. Override prediction parameters for this Predictor
# ------------------------------------------------------------------

# The exact pred_params class depends on the model wrapper.
# For wrappers exposing a pred_params object, copy or instantiate the
# relevant params class and modify fields such as confidence threshold.
pred_params = model.pred_params
pred_params.conf = 0.35

predictor = Predictor(
    model,
    pred_params=pred_params,
)

preds = predictor.predict("images/image_001.png")

# ------------------------------------------------------------------
# 10. Forward extra keyword arguments to model.predict(...)
# ------------------------------------------------------------------

# Extra keyword arguments are forwarded through Predictor to the model
# as extra_args. Their meaning depends on the model wrapper.
preds = predictor.predict(
    "images/image_001.png",
    some_model_specific_option=True,
)

Source code in src/deepvisiontools/inference/predictor.py

def __init__(
    self,
    model: DeepVisionModel | str | Path,
    *,
    preprocessing: Callable = _DEFAULT_PREPROCESSING,
    device: str | torch.device | None = None,
    use_amp: bool | None = None,
    pred_params: BasePredParams | None = None,
    strategy: TaskPredictorStrategy | None = None,
    nms: NMSProvider | None = None,
    tile_hw: tuple[int, int] | None = None,
    stride_hw: tuple[int, int] | None = None,
    tiler: PatchTiler | None = None,
) -> None:

    self.model = self._load_model(model, device=device, use_amp=use_amp)
    self.preprocessing = preprocessing
    self._pred_params_override = pred_params

    # If the default strategy needs pred_params-derived defaults, make the override
    # visible only while the strategy is chosen.
    if strategy is not None:
        self.strategy = strategy
    else:
        if pred_params is None:
            self.strategy = default_strategy_for_model(self.model)
        else:
            old_pred_params = getattr(self.model, "pred_params", None)
            self.model.pred_params = pred_params
            try:
                self.strategy = default_strategy_for_model(self.model)
            finally:
                self.model.pred_params = old_pred_params

    self.nms = nms if nms is not None else DefaultNMS()

    # Tiling configuration
    self._tile_hw: tuple[int, int] | None = None
    self._stride_hw: tuple[int, int] | None = None
    self._explicit_tiler: PatchTiler | None = None
    self._default_tiler: PatchTiler | None = None

    self._tile_hw = self._normalize_hw(tile_hw)
    self._stride_hw = self._normalize_hw(stride_hw)
    self._explicit_tiler = tiler
    self._rebuild_default_tiler_if_possible()

predict(images, *, tiler=None, batch_size=1, generate_visu=False, visu_names=None, max_images_per_batch=None, _display_progress=True, **kwargs)

Main prediction function.

Notes: - For tensor input, the tensor is already resident in memory, so the default is to process the full tensor batch unless max_images_per_batch is set. - For path/list-of-path input, the default is true streaming: one image at a time. Set max_images_per_batch > 1 to batch file loading/inference. - batch_size controls patch batch size during tiled inference.

Source code in src/deepvisiontools/inference/predictor.py

@torch.no_grad()
def predict(
    self,
    images: torch.Tensor | str | Path | list[str] | list[Path],
    *,
    tiler: PatchTiler | None = None,
    batch_size: int = 1,
    generate_visu: bool = False,
    visu_names: str | Path | list[str] | list[Path] | None = None,
    max_images_per_batch: int | None = None,
    _display_progress: bool = True,
    **kwargs: Any,
) -> BatchData:
    """
    Main prediction function.

    Notes:
    - For tensor input, the tensor is already resident in memory, so the default
        is to process the full tensor batch unless max_images_per_batch is set.
    - For path/list-of-path input, the default is true streaming: one image at a time.
        Set max_images_per_batch > 1 to batch file loading/inference.
    - batch_size controls patch batch size during tiled inference.
    """
    patchify_now = (
        tiler is not None
        or self._explicit_tiler is not None
        or self._default_tiler is not None
    )
    resolved_tiler = self._resolve_tiler(tiler) if patchify_now else None

    # ---------------------------- #
    # Case A: images is a Tensor
    # ---------------------------- #
    if isinstance(images, torch.Tensor):
        images_batch = self._ensure_batched(images)
        B = int(images_batch.shape[0])

        limit = B if max_images_per_batch is None else int(max_images_per_batch)
        if limit <= 0:
            limit = B

        out_paths_full: list[Path] | None = None
        if generate_visu and B >= 1:
            out_paths_full = self._resolve_visu_output_paths(
                batch_size=B,
                src_paths=None,
                visu_names=visu_names,
            )

        outputs: list[BatchData] = []
        rng = range(0, B, limit)
        progresser = (
            tqdm(
                rng,
                total=(B + limit - 1) // limit,
                desc="Predict image batches: ",
            )
            if _display_progress and B > 1
            else rng
        )

        for start in progresser:
            end = min(start + limit, B)
            img_chunk = images_batch[start:end]

            images_t = (
                self.preprocessing(img_chunk)
                if self.preprocessing is not None
                else img_chunk
            )

            if not patchify_now:
                preds = self._run_model_predict(images_t, **kwargs)
            else:
                if resolved_tiler is None:
                    raise RuntimeError(
                        "Patchified prediction expected a resolved tiler."
                    )
                preds = self._predict_patchified(
                    images_t,
                    tiler=resolved_tiler,
                    batch_size=batch_size,
                    _display_progress=False,
                    **kwargs,
                )

            outputs.append(preds)

            if generate_visu and out_paths_full is not None:
                self._save_visualizations(
                    images=img_chunk,
                    preds=preds,
                    out_paths=out_paths_full[start:end],
                )

            del preds, images_t, img_chunk

        return self._concat_batchdatas(
            outputs,
            device=outputs[0].device if outputs else str(images_batch.device),
        )

    # ------------------------------------ #
    # Case B: images is path(s), streamed
    # ------------------------------------ #
    src_paths = self._as_path_list(images)
    B = len(src_paths)
    if B == 0:
        return BatchData([], device=str(self._model_device()))

    # Default to real streaming for file paths.
    limit = 1 if max_images_per_batch is None else int(max_images_per_batch)
    if limit <= 0:
        limit = B

    out_paths_full = None
    if generate_visu and B >= 1:
        out_paths_full = self._resolve_visu_output_paths(
            batch_size=B,
            src_paths=src_paths,
            visu_names=visu_names,
        )

    outputs = []
    rng = range(0, B, limit)
    progresser = (
        tqdm(rng, total=(B + limit - 1) // limit, desc="Predict image batches: ")
        if _display_progress and B > 1
        else rng
    )

    for start in progresser:
        end = min(start + limit, B)
        chunk_paths = src_paths[start:end]

        original_tensors: list[torch.Tensor] = [
            self._load_image_from_path(p) for p in chunk_paths
        ]
        original_hws: list[tuple[int, int]] = [
            (int(t.shape[1]), int(t.shape[2])) for t in original_tensors
        ]

        max_h = max(h for h, _ in original_hws)
        max_w = max(w for _, w in original_hws)

        padded = [self._center_pad_chw(t, (max_h, max_w)) for t in original_tensors]
        images_batch = torch.stack(padded, dim=0).to(self._model_device())

        images_t = (
            self.preprocessing(images_batch)
            if self.preprocessing is not None
            else images_batch
        )

        if not patchify_now:
            preds = self._run_model_predict(images_t, **kwargs)
        else:
            if resolved_tiler is None:
                raise RuntimeError(
                    "Patchified prediction expected a resolved tiler."
                )
            preds = self._predict_patchified(
                images_t,
                tiler=resolved_tiler,
                batch_size=batch_size,
                _display_progress=False,
                **kwargs,
            )

        preds = self._undo_batch_center_padding(
            preds,
            original_hws=original_hws,
            padded_hw=(max_h, max_w),
        )

        outputs.append(preds)

        if generate_visu and out_paths_full is not None:
            self._save_visualizations(
                images=original_tensors,
                preds=preds,
                out_paths=out_paths_full[start:end],
            )

        del preds, images_t, images_batch, padded, original_tensors

    return self._concat_batchdatas(
        outputs,
        device=outputs[0].device if outputs else str(self._model_device()),
    )