Inference Support

DeGirum Tools API Reference Guide. Utilities to connect, run, annotate, and profile inferences.

This API Reference is based on DeGirum Tools version 0.25.1.

Inference Support Overview

This module provides utility functions and classes for integrating DeGirum PySDK models into various inference scenarios, including:

• Connecting to different model zoo endpoints (cloud, AI server, or local accelerators).

• Attaching custom result analyzers to models or compound models, enabling additional data processing or custom overlay annotation on inference results.

• Running inferences on video sources or streams (local camera, file, RTSP, YouTube links, etc.) with optional real-time annotation and saving to output video files.

• Measuring model performance using a profiling function that times inference runs under various conditions.

Key Concepts

1. Model Zoo Connections: Functions like connect_model_zoo provide a unified way to choose between different inference endpoints (cloud, server, local hardware).

2. Analyzer Attachment: By calling attach_analyzers or using specialized classes within the streaming toolkit, you can process each inference result through user-defined or library-provided analyzers (subclasses of ResultAnalyzerBase).

3. Video Inference and Annotation: Functions predict_stream and annotate_video demonstrate how to run inference on live or file-based video streams. They optionally include steps to create overlays (bounding boxes, labels, etc.) and even show a real-time display or write to an output video file.

4. Model Time Profiling: model_time_profile provides a convenient way to measure the performance (FPS, average inference time, etc.) of a given DeGirum PySDK model under test conditions.

Basic Usage Example

from degirum_tools import (
    ModelSpec,
    remote_assets,
    attach_analyzers,
    annotate_video,
    model_time_profile,
    ResultAnalyzerBase,
)

# Define a simple analyzer that draws text on each frame
class MyDummyAnalyzer(ResultAnalyzerBase):
    def analyze(self, result):
        # Optional: add custom logic here, e.g. track or filter detections
        pass

    def annotate(self, result, image):
        """
        Draws a simple "Dummy Analyzer" label in the top-left corner of each frame.
        """
        import cv2
        cv2.putText(
            image,
            "Dummy Analyzer",
            (10, 30),
            cv2.FONT_HERSHEY_SIMPLEX,
            1.0,
            (0, 255, 0),
            2
        )
        return image

# Describe the model once so the configuration is reusable.
model_spec = ModelSpec(
    model_name="<your_model_name>",
    zoo_url="degirum/degirum",
    inference_host_address="@cloud",
)

with model_spec.load_model() as model:
    # Attach dummy analyzer
    attach_analyzers(model, MyDummyAnalyzer())

    # Annotate a video with detection results + dummy analyzer and set a path to save the video
    annotate_video(
        model,
        video_source_id=remote_assets.store_short,
        output_video_path="annotated_output.mp4",
        show_progress=True,      # Show a progress bar in console
        visual_display=True,     # Open an OpenCV window to display frames
        show_ai_overlay=True,    # Use model's overlay with bounding boxes
        fps=None                 # Use the source's native frame rate
    )

    # Time-profile the model
    profile = model_time_profile(model, iterations=100)
    print("Time profiling results:")
    print(f"  Elapsed time: {profile.elapsed:.3f} s")
    print(f"  Observed FPS: {profile.observed_fps:.2f}")
    print(f"  Max possible FPS: {profile.max_possible_fps:.2f}")

Functions

connect_model_zoo(inference_option=CloudInference)

connect_model_zoo(inference_option=CloudInference)

Connect to a model zoo endpoint based on the specified inference option.

This function provides a convenient way to switch between

• Cloud-based inference (CloudInference),

• AI server on LAN/VPN (AIServerInference),

• Local hardware accelerator (LocalHWInference).

It uses environment variables (see degirum_tools.environment) to resolve the zoo address/URL and token as needed.

Parameters:

Name	Type	Description	Default
inference_option	int	One of CloudInference, AIServerInference, or LocalHWInference	CloudInference

Raises:

Type	Description
Exception	If an invalid inference_option is provided.

Returns:

Type	Description
ZooManager	dg.zoo_manager.ZooManager: A model zoo manager connected to the requested endpoint.

attach_analyzers(model, ...)

attach_analyzers(model, analyzers)

Attach or detach analyzer(s) to a model or compound model.

For single or compound models, analyzers can augment the inference results with extra metadata and/or custom overlay. If attaching analyzers to a compound model (e.g., compound_models.CompoundModelBase), the analyzers are invoked at the final stage of each inference result.

Parameters:

Name	Type	Description	Default
model	Union[Model, CompoundModelBase]	The model or compound model to which analyzers will be attached.	required
analyzers	Union[ResultAnalyzerBase, List[ResultAnalyzerBase], None]	One or more analyzer objects. Passing None will detach any previously attached analyzers.	required

Usage

attach_analyzers(my_model, MyAnalyzerSubclass())

Notes

• If model is a compound model, the call is forwarded to model.attach_analyzers().

• If model is a standard PySDK model, this function subclasses the model's current result class with a new class that additionally calls each analyzer in turn for analyze() and annotate() steps. This subclass is assigned to model._custom_postprocessor property.

predict_stream(model, ...)

predict_stream(model, video_source_id, source_type=VideoSourceType.AUTO, *, fps=None, analyzers=None)

Run a PySDK model on a live or file-based video source, yielding inference results.

This function is a generator that continuously

1. Reads frames from the specified video source.

2. Runs inference on each frame via model.predict_batch.

3. If analyzers are provided, each result is wrapped in a dynamic postprocessor that calls analyzers' analyze() and annotate() methods.

Parameters:

Name	Type	Description	Default
model	Model	Model to run on each incoming frame.	required
video_source_id	Union[int, str, Path, None]	Identifier for the video source. Possible types include: - An integer camera index (e.g., 0 for default webcam). - A local file path or string/Path, e.g., "video.mp4". - A streaming URL (RTSP/YouTube link). - None if no source is available (not typical).	required
source_type	Union[str, VideoSourceType]	Video backend to use. Options: - VideoSourceType.AUTO or "auto": Automatically choose best backend - VideoSourceType.GSTREAMER or "gstream": Force GStreamer backend - VideoSourceType.OPENCV or "opencv": Force OpenCV backend	AUTO
fps	Optional[float]	If provided, caps the effective reading/processing rate to the given FPS. If the input source is slower, this has no effect. If faster, frames are decimated.	None
analyzers	Union[ResultAnalyzerBase, List[ResultAnalyzerBase], None]	One or more analyzers to apply to each inference result. If None, no additional analysis or annotation is performed beyond the model's standard postprocessing.	None

Yields:

Name	Type	Description
InferenceResults	InferenceResults	The inference result for each processed frame. If analyzers are present, the result object is wrapped to allow custom annotation in its .image_overlay.

Example:

# Using enum (recommended)
for res in predict_stream(my_model, "my_video.mp4", VideoSourceType.GSTREAMER, fps=15, analyzers=MyAnalyzer()):
    annotated_img = res.image_overlay  # includes custom overlay
    # do something with annotated_img
# Using string (backward compatible)
for res in predict_stream(my_model, "my_video.mp4", "gstream", fps=15, analyzers=MyAnalyzer()):
    annotated_img = res.image_overlay  # includes custom overlay
    # do something with annotated_img

annotate_video(model, ...)

annotate_video(model, video_source_id, output_video_path, source_type=VideoSourceType.AUTO, *, show_progress=True, visual_display=True, show_ai_overlay=True, fps=None, analyzers=None)

Run a model on a video source and save the annotated output to a video file.

This function

1. Opens the input video source.

2. Processes each frame with the specified model.

3. (Optionally) calls any analyzers to modify or annotate the inference result.

4. If show_ai_overlay is True, retrieves the image_overlay from the result (which includes bounding boxes, labels, etc.). Otherwise, uses the original frame.

5. Writes the annotated frame to output_video_path.

6. (Optionally) displays the annotated frames in a GUI window and shows progress.

Parameters:

Name	Type	Description	Default
model	Model	The model to run on each frame of the video.	required
video_source_id	Union[int, str, Path, None, VideoCapture]	The video source, which can be: - A cv2.VideoCapture object already opened by open_video_stream, - An integer camera index (e.g., 0), - A file path or a URL (RTSP/YouTube).	required
output_video_path	str	Path to the output video file. The file is created or overwritten as needed.	required
source_type	Union[str, VideoSourceType]	Video backend to use. Options: - VideoSourceType.AUTO or "auto": Automatically choose best backend - VideoSourceType.GSTREAMER or "gstream": Force GStreamer backend - VideoSourceType.OPENCV or "opencv": Force OpenCV backend	AUTO
show_progress	bool	If True, displays a textual progress bar or frame counter (for local file streams).	True
visual_display	bool	If True, opens an OpenCV window to show the annotated frames in real time.	True
show_ai_overlay	bool	If True, uses the result's image_overlay. If False, uses the original unannotated frame.	True
fps	Optional[float]	If provided, caps the effective processing rate to the given FPS. Otherwise, uses the native FPS of the video source if known.	None
analyzers	Union[ResultAnalyzerBase, List[ResultAnalyzerBase], None]	One or more analyzers to apply. Each analyzer's analyze_and_annotate() is called on the result before writing the frame.	None

Example:

# Using enum (recommended)
annotate_video(my_model, "input.mp4", "output.mp4", VideoSourceType.GSTREAMER, analyzers=[MyAnalyzer()])
# Using string (backward compatible)
annotate_video(my_model, "input.mp4", "output.mp4", "gstream", analyzers=[MyAnalyzer()])

model_time_profile(model, ...)

model_time_profile(model, iterations=100, input_image_format='JPEG')

Profile the inference performance of a DeGirum PySDK model by running a specified number of iterations on a synthetic (zero-pixel) image.

This function

1. Adjusts the model's settings to measure time (model.measure_time = True).

2. Warms up the model by performing one inference.

3. Resets time statistics and runs the specified number of iterations.

4. Restores original model parameters after profiling.

Parameters:

Name	Type	Description	Default
model	Model	A PySDK model to profile. Must accept images as input.	required
iterations	int	Number of inference iterations to run (excluding the warm-up iteration).	100

Raises:

Type	Description
NotImplementedError	If the model does not accept images (e.g., audio/text).

Returns:

Name	Type	Description
ModelTimeProfile	ModelTimeProfile	An object containing timing details, measured FPS, and other stats.

Example:

profile = model_time_profile(my_model, iterations=50)
print("Elapsed seconds:", profile.elapsed)
print("Observed FPS:", profile.observed_fps)
print("Core Inference Stats:", profile.time_stats["CoreInferenceDuration_ms"])

warmup_device(model, ...)

warmup_device(model, chosen_device)

Warm up given model on the given device in devices_available

warmup_model(model)

warmup_model(model)

Warms up a model by performing one inference on a dummy input frame on each device selected.

This is useful for initializing the model internals and hardware, which can reduce latency on the first real inference.

Parameters:

Name	Type	Description	Default
model	Model	The DeGirum PySDK model object to warm up.	required

Classes

ModelTimeProfile

ModelTimeProfile

dataclass

Container for model time profiling results.

Attributes:

Name	Type	Description
elapsed	float	Total elapsed time in seconds for the profiling run.
iterations	int	Number of inference iterations performed.
observed_fps	float	The measured frames per second (iterations / elapsed).
max_possible_fps	float	Estimated maximum possible FPS based on the model's core inference duration, ignoring overhead.
parameters	dict	A copy of the model's metadata or parameters for reference.
time_stats	dict	A dictionary containing detailed timing statistics from the model's built-in time measurement (if available).