oakutils.optimizer package

Submodules

oakutils.optimizer.objective module

Module containing objective functions for choosing the best set of pipeline arguments.

Functions

highest_fps: Use to get the set of arguments with the highest fps
lowest_avg_latency: Use to get the set of arguments with the lowest avg latency
lowest_latency: Use to get the set of arguments with the lowest latency for a specific stream

oakutils.optimizer.objective.highest_fps(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the highest fps.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.

Returns:

dict[str, Any] – The arguments with the highest fps
tuple[float, float, dict[str, float]] – The measurement results

oakutils.optimizer.objective.lowest_avg_latency(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the lowest avg latency.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.

Returns:

dict[str, Any] – The arguments with the lowest avg latency
tuple[float, float, dict[str, float]] – The measurement results

oakutils.optimizer.objective.lowest_latency(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]], stream: str) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the lowest latency for a specific stream.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.
stream (str) – The name of the stream to get the lowest latency for

Returns:

dict[str, Any] – The arguments with the lowest latency for the specified stream
tuple[float, float, dict[str, float]] – The measurement results for the specified stream

Module contents

Module for optimizing pipelines.

Submodules

objective: Module containing objective functions for choosing the best set of pipeline arguments.

Classes

Optimizer: Class for optimizing pipelines.

Functions

highest_fps: Use to get the set of arguments with the highest fps
lowest_avg_latency: Use to get the set of arguments with the lowest avg latency
lowest_latency: Use to get the set of arguments with the lowest latency for a specific stream

class oakutils.optimizer.Optimizer(algorithm: str = 'grid', max_measure_time: float = 10.0, measure_trials: int = 1, warmup_cycles: int = 10, stability_threshold: float = 0.002, stability_length: int = 10)

Bases: object

Class for optimizing a pipeline onboard an OAK camera.

measure(pipeline_func: Callable[[dai.Pipeline, dict[str, Any]], list[Callable[[dai.DeviceBase], None]]], pipeline_args: dict[str, Any]) → tuple[float, float, dict[str, float]]

Use to measure the FPS of a pipeline.

Parameters:

pipeline_func (Callable[[dai.Pipeline, dict[str, Any]], list[Callable[[dai.Device], None]]]) – The function to generate a pipeline
pipeline_args (dict[str, Any]) – The arguments to measure

Notes

This function will set XLinkChunkSize to 0 for the pipeline.

Returns:

float – The average cycle time of the whole pipeline (FPS)
float – The average latency of all data packets in the pipeline
dict[str, float] – The average latency of each data packet in the pipeline Key is the name of the data stream

optimize(pipeline_func: Callable[[dai.Pipeline, dict[str, Any]], list[Callable[[dai.DeviceBase], None]]], pipeline_args: dict[str, list[Any]], objective_func: Callable[[list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]], tuple[dict[str, Any], tuple[float, float, dict[str, float]]]]) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to generate optimized arguments for a pipeline.

Parameters:

pipeline_func (Callable[[dai.Pipeline, dict[str, Any]], list[Callable[[dai.DeviceBase], None]]]) – The function to generate a pipeline
pipeline_args (dict[str, list[Any]]) – The arguments to optimize
objective_func (Callable[[list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]], dict[str, Any]]) – The function to use to choose the best arguments

Returns:

dict[str, Any] – The optimized arguments
tuple[float, float, dict[str, float]] – The best measurement results

oakutils.optimizer.highest_fps(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the highest fps.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.

Returns:

dict[str, Any] – The arguments with the highest fps
tuple[float, float, dict[str, float]] – The measurement results

oakutils.optimizer.lowest_avg_latency(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the lowest avg latency.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.

Returns:

dict[str, Any] – The arguments with the lowest avg latency
tuple[float, float, dict[str, float]] – The measurement results

oakutils.optimizer.lowest_latency(options: list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]], stream: str) → tuple[dict[str, Any], tuple[float, float, dict[str, float]]]

Use to get the set of arguments with the lowest latency for a specific stream.

Parameters:

options (list[tuple[tuple[float, float, dict[str, float]], dict[str, Any]]]) – The list of options to choose from. A list of tuples, where the first element is the result of Optimizer.measure and the second element is the arguments used to generate the pipeline.
stream (str) – The name of the stream to get the lowest latency for

Returns:

dict[str, Any] – The arguments with the lowest latency for the specified stream
tuple[float, float, dict[str, float]] – The measurement results for the specified stream