Inference Support

This API Reference is based on DeGirum Tools version 0.16.6.

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

import degirum as dg
from degirum_tools.inference_support import (
    connect_model_zoo,
    attach_analyzers,
    annotate_video,
    model_time_profile,
)
from degirum_tools.result_analyzer_base import ResultAnalyzerBase

# Declaring model variable
# If you will use the DeGirum AI Hub model zoo, set the CLOUD_ZOO_URL environmental variable to a model zoo path such as degirum/degirum,
# and ensure the DEGIRUM_CLOUD_TOKEN environmental variable is set to your AI Hub token.
# CLOUD_ZOO_URL will default to degirum/public if left empty.
your_detection_model = "yolov8n_relu6_coco--640x640_quant_n2x_orca1_1"
your_video = "<path to your video>"
your_output_video_path = "<path to where the annotated output should be saved>"


# 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

# Connect to a model zoo.
# Set to 1 for CloudInference, 2 for AIServerInference, and 3 for LocalHWInference.
inference_manager = connect_model_zoo(1)

# Load a single detection model (example: YOLO-based detection)
model = inference_manager.load_model(
    model_name=your_detection_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=your_video,
    output_video_path=your_output_video_path,
    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}")

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).

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, *, 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
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

Returns:

Type	Description
	Union[InferenceResults, AnalyzingPostprocessor]: 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:

for res in predict_stream(my_model, "my_video.mp4", 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, *, 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
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:

annotate_video(my_model, "input.mp4", "output.mp4", analyzers=[MyAnalyzer()])

model_time_profile(model, ...)

model_time_profile(model, iterations=100)

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"])