luxonis_ml.data.utils

package documentation

Utility helpers shared by the data package.

This package collects public helper APIs used by dataset creation, parsing, loading, exporting, validation, and visualization. The utilities are grouped by the part of the data workflow they support:

Utility groups
Group	Public APIs
Task keys	`task_is_metadata`, `split_task`, `get_task_name`, `get_task_type`, and `task_type_iterator` parse and filter `"task_name/task_type"` labels.
Storage and parser enums	`BucketStorage`, `BucketType`, `MediaType`, `ImageType`, `UpdateMode`, `ParserIssue`, and `ParserIssueMessage`.
Dataframe and parquet helpers	`ParquetFileManager`, `ParquetRecord`, duplicate detection, class distributions, missing-annotation summaries, heatmaps, and UUID merging.
Remote media	`RemoteFileDownloader` and `download_remote_file` copy supported remote files to local paths and validate image inputs.
Visualization	`visualize`, color-map helpers, image concatenation, augmentation footers, and dataset-statistic plots.
Equivalence and augmentation inspection	LDF equivalence checks and `AugmentationsCollector` summaries for configured augmentation pipelines.

The task-key helpers follow the same convention as LuxonisLoader: labels are addressed by "task_name/task_type" and metadata labels use "task_name/metadata/key" or "metadata/key" when no task name is present.

Module	`augmentations_collector`	No module docstring; 1/1 class documented
Module	`cli_utils`	No module docstring; 2/5 functions documented
Module	`constants`	Undocumented
Module	`data_utils`	Undocumented
Module	`enums`	Undocumented
Module	`ldf_equivalence`	Undocumented
Module	`parquet`	Undocumented
Module	`plot_utils`	No module docstring; 3/3 functions documented
Module	`remote_file_downloader`	Undocumented
Module	`task_utils`	Undocumented
Module	`visualizations`	No module docstring; 16/16 functions, 1/1 class documented

From __init__.py:

Class	`AugmentationsCollector`	Collect augmentation names that were applied to a sample.
Class	`COCOFormat`	Supported COCO directory layouts.
Class	`ParquetFileManager`	Manage append-style writes across partitioned parquet files.
Class	`ParquetRecord`	Single annotation row written to parquet.
Class	`RemoteFileDownloader`	Download remote files with optional image validation.
Function	`download_remote_file`	Download a remote file to the requested destination.
Function	`find_duplicates`	Collect duplicate UUID and annotation information.
Function	`get_class_distributions`	Return class distribution information for non-classification tasks.
Function	`get_duplicates_info`	Return duplicate UUID and annotation information.
Function	`get_heatmaps`	Generate heatmaps for boxes, keypoints, and segmentation masks.
Function	`get_missing_annotations`	Return file paths that have no annotations.
Function	`get_task_name`	Return the task name from a task string.
Function	`get_task_type`	Return the task type from a task string.
Function	`infer_task`	Undocumented
Function	`merge_uuids`	Merge multiple UUIDs into a single deterministic UUID, independent of order.
Function	`plot_class_distribution`	Plot a class distribution bar chart.
Function	`plot_heatmap`	Plot a heatmap.
Function	`rgb_to_bool_masks`	Convert an RGB segmentation mask to boolean class masks.
Function	`split_task`	Split a task into task name and task type.
Function	`task_is_metadata`	Check whether a task is a metadata task.
Function	`task_type_iterator`	Iterate over labels of a specific task type.
Function	`warn_on_duplicates`	Log warnings for duplicate UUIDs and annotations.

def download_remote_file(url: str, destination: Path, *, timeout: float = 30.0, validate_image: bool = False) -> Path: ¶

Download a remote file to the requested destination.

Parameters
url:`str`	Source URL.
destination:`Path`	Local destination path.
timeout:`float`	HTTP request timeout in seconds.
validate_image:`bool`	Whether to validate image content and format compatibility after download.

Returns
`Path`	Local destination path.

Raises
`ValueError`	If the URL scheme is unsupported or image validation fails.

def find_duplicates(df: pl.LazyFrame) -> dict[str, list[dict[str, Any]]]: ¶

Collect duplicate UUID and annotation information.

Parameters
df:`pl.LazyFrame`	Dataset information.

Returns
`Dictionary with`	`"duplicate_uuids"` List of dictionaries containing a duplicated `"uuid"` and all `"files"` associated with it. `"duplicate_annotations"` List of dictionaries containing `"file_name"`, `"task_type"`, `"task_name"`, serialized `"annotation"` (or `"<binary mask>"` for segmentation), and duplicate `"count"`.

def get_class_distributions(df: pl.LazyFrame) -> dict[str, dict[str, list[dict[str, Any]]]]: ¶

Return class distribution information for non-classification tasks.

Parameters
df:`pl.LazyFrame`	Dataset information.

Returns
`dict[str, dict[str, list[dict[str, Any]]]]`	Class counts grouped by task name and task type.

def get_duplicates_info(df: pl.LazyFrame) -> dict[str, Any]: ¶

Return duplicate UUID and annotation information.

Parameters
df:`pl.LazyFrame`	Dataset information.

Returns
`Dictionary with duplicate UUID and annotation details`	`"duplicate_uuids"` List of `{"uuid": str, "files": list[str]}` entries. `"duplicate_annotations"` List of entries with `"file_name"`, `"task_name"`, `"task_type"`, `"annotation"`, and `"count"`.

def get_heatmaps(df: pl.LazyFrame, sample_size: int | None = None, downsample_factor: int = 5) -> dict[str, dict[str, list[list[int]]]]: ¶

Generate heatmaps for boxes, keypoints, and segmentation masks.

Heatmaps are accumulated on a fixed 15×15 grid over normalized image coordinates. Bounding boxes contribute their center points, keypoints contribute visible points, and segmentation masks contribute foreground pixels after optional downsampling.

Parameters
df:`pl.LazyFrame`	Dataset information.
sample_size:`int \| None`	Optional number of samples used to generate heatmaps.
downsample_factor:`int`	Factor used to downsample segmentation masks. A value of 5 keeps every fifth row and column.

Returns
`dict[str, dict[str, list[list[int]]]]`	Heatmaps grouped by task name and task type. Each heatmap is a 15×15 nested list of counts.

def get_missing_annotations(df: pl.LazyFrame) -> list[str]: ¶

Return file paths that have no annotations.

Parameters
df:`pl.LazyFrame`	Dataset information.

Returns
`list[str]`	File paths with no annotations.

@lru_cache

def get_task_name(task: str) -> str: ¶

Return the task name from a task string.

Examples

>>> get_task_name("detector/boundingbox")
'detector'
>>> get_task_name("classification")
'classification'

Parameters
task:`str`	Task string.

Returns
`str`	Task name.

@lru_cache

def get_task_type(task: str) -> str: ¶

Return the task type from a task string.

Example

>>> get_task_type("task_name/type")
'type'
>>> get_task_type("metadata/name")
'metadata/name'
>>> get_task_type("task_name/metadata/name")
'metadata/name'

Parameters
task:`str`	Task string, such as `"task_name/type"`.

Returns
`str`	Task type. Metadata tasks are returned as `"metadata/type"`.

def infer_task(old_task: str, class_name: str | None, current_classes: dict[str, dict[str, int]]) -> str: ¶

Undocumented

def merge_uuids(uuids: Iterable[str]) -> uuid.UUID: ¶

Merge multiple UUIDs into a single deterministic UUID, independent of order.

Examples

>>> first = "00000000-0000-0000-0000-000000000001"
>>> second = "00000000-0000-0000-0000-000000000002"
>>> merge_uuids([first, second]) == merge_uuids([second, first])
True
>>> isinstance(merge_uuids([first]), uuid.UUID)
True

Parameters
uuids:`Iterable[str]`	UUID strings to merge.

Returns
`uuid.UUID`	Deterministic merged UUID.

def plot_class_distribution(ax: plt.Axes, task_type: str, task_data: list[dict[str, Any]]): ¶

Plot a class distribution bar chart.

Parameters
ax:`plt.Axes`	Axis to plot on.
task_type:`str`	Task type.
task_data:`list[dict[str, Any]]`	Class distribution records.

def plot_heatmap(ax: plt.Axes, fig: plt.Figure, task_type: str, heatmap_data: list[list[float]] | None): ¶

Plot a heatmap.

Parameters
ax:`plt.Axes`	Axis to plot on.
fig:`plt.Figure`	Figure containing the axis.
task_type:`str`	Task type.
heatmap_data:`list[list[float]] \| None`	Heatmap values to plot.

def rgb_to_bool_masks(segmentation_mask: np.ndarray, class_colors: dict[str, RGB], add_background_class: bool = False) -> Iterator[tuple[str, np.ndarray]]: ¶

Convert an RGB segmentation mask to boolean class masks.

Example

>>> segmentation_mask = np.array([
...     [[0, 0, 0], [255, 0, 0], [0, 255, 0]],
...     [[0, 0, 0], [0, 255, 0], [0, 0, 255]],
...     ], dtype=np.uint8)
>>> class_colors = {
...     "red": (255, 0, 0),
...     "green": (0, 255, 0),
...     "blue": (0, 0, 255),
... }
>>> for class_name, mask in rgb_to_bool_masks(
...     segmentation_mask,
...     class_colors,
...     add_background_class=True,
... ):
...     print(class_name, np.array2string(mask, separator=", "))
background [[ True, False, False],
            [ True, False, False]]
red        [[False,  True, False],
            [False, False, False]]
green      [[False, False,  True],
            [False, True, False]]
blue       [[False, False, False],
            [False, False,  True]]

Parameters
segmentation_mask:`np.ndarray`	RGB segmentation mask where each pixel color identifies its class.
class_colors:`dict[str, RGB]`	Mapping from class names to RGB colors.
add_background_class:`bool`	Whether to add a `"background"` mask for pixels that do not belong to any class. The background mask is yielded first.

Returns
`Iterator[tuple[str, np.ndarray]]`	Iterator of class names and their boolean masks.

@lru_cache

def split_task(task: str) -> tuple[str, str]: ¶

Split a task into task name and task type.

Examples

>>> split_task("detector/boundingbox")
('detector', 'boundingbox')
>>> split_task("classification")
('', 'classification')

Parameters
task:`str`	Task to split.

Returns
`tuple[str, str]`	Task name and task type.

@lru_cache

def task_is_metadata(task: str) -> bool: ¶

Check whether a task is a metadata task.

Examples

>>> task_is_metadata("metadata/weather")
True
>>> task_is_metadata("camera/metadata/weather")
True
>>> task_is_metadata("camera/boundingbox")
False

Parameters
task:`str`	Task to check.

Returns
`bool`	Whether the task is a metadata task.

def task_type_iterator(labels: Labels, task_type: TaskType) -> Iterator[tuple[str, np.ndarray]]: ¶

Iterate over labels of a specific task type.

Examples

>>> labels = {
...     "detector/boundingbox": np.array([1]),
...     "pose/keypoints": np.array([2]),
... }
>>> [(task, arr.tolist()) for task, arr in task_type_iterator(labels, "keypoints")]
[('pose/keypoints', [2])]

Parameters
labels:`Labels`	Labels to iterate over.
task_type:`TaskType`	Label type to yield.

Returns
`Iterator[tuple[str, np.ndarray]]`	Iterator over matching labels.

def warn_on_duplicates(df: pl.LazyFrame): ¶

Log warnings for duplicate UUIDs and annotations.

Parameters
df:`pl.LazyFrame`	Dataset information.