class documentation

Luxonis Dataset Format (LDF) dataset handle.

LDF is a flexible and feature-rich dataset format designed for use within the Luxonis MLOps ecosystem.

Static Method exists Check whether a dataset exists.
Static Method list_datasets List available datasets.
Method __eq__ Compare datasets for equivalence.
Method __init__ Create a Luxonis Dataset Format dataset handle.
Method __len__ Return the number of records in the dataset.
Method add Add data to the dataset from a generator of records.
Method clone Create a local copy of the current dataset.
Method delete_dataset Delete the dataset from local storage and optionally the cloud.
Method export Export the dataset into one of the supported formats.
Method get_categorical_encodings Get the categorical encodings for the dataset grouped by task.
Method get_classes Get class names and IDs per task.
Method get_metadata_types Get the metadata types for each metadata annotation in the dataset.
Method get_skeletons Return keypoint skeletons for each task.
Method get_source_names Return input source names for the dataset.
Method get_splits Get the dataset splits definitions.
Method get_statistics Return dataset statistics for a view or the full dataset.
Method get_tasks Return task names and task types.
Method make_splits Create dataset splits for training, validation, and testing.
Method merge_with Merge another dataset into this or a new dataset.
Method pull_from_cloud Synchronize the dataset from a remote bucket to a local storage.
Method push_to_cloud Push the local dataset to a remote bucket.
Method remove_duplicates Remove duplicate files and annotations from the dataset.
Method set_class_order_per_task Set class order for specific tasks.
Method set_classes Set classes for one or more tasks.
Method set_skeletons Set keypoint skeleton metadata.
Method set_tasks Set dataset tasks.
Method update_source Update the dataset source definition.
Instance Variable bucket_storage Underlying storage backend for the dataset.
Instance Variable bucket_type Whether the dataset uses internal or external buckets.
Instance Variable dataset_name Name of the dataset.
Instance Variable team_id Optional cloud team identifier.
Property identifier The unique identifier for the dataset.
Property is_remote Whether the dataset is stored remotely (in a cloud bucket) or locally.
Property metadata Get the dataset metadata.
Property source Get the source information for the dataset.
Property version The version of the underlying LDF that the dataset adheres to.
Static Method _construct_url Undocumented
Static Method _scan_annotation_files Undocumented
Method _add_process_batch Undocumented
Method _get_credential Get secret credentials from the credentials file or environment.
Method _get_index Load unique file entries from annotation data.
Method _get_metadata Load metadata from local storage or cloud.
Method _ignore_files_not_in_uuid_set Undocumented
Method _init_credentials Undocumented
Method _init_paths Configure local paths or a bucket directory.
Method _load_df_offline Load the dataset DataFrame from local storage.
Method _load_splits Undocumented
Method _merge_metadata_with Merge relevant metadata from other into self.
Method _merge_splits Undocumented
Method _process_arrays Undocumented
Method _save_df_offline Save annotations DataFrame into parquet files.
Method _save_splits Undocumented
Method _warn_on_duplicates Undocumented
Method _write_metadata Undocumented
Instance Variable _annotations_path Undocumented
Instance Variable _arrays_path Undocumented
Instance Variable _base_path Undocumented
Instance Variable _bucket Undocumented
Instance Variable _bucket_storage Undocumented
Instance Variable _bucket_type Undocumented
Instance Variable _credentials Undocumented
Instance Variable _dataset_name Undocumented
Instance Variable _fs Undocumented
Instance Variable _is_synced Undocumented
Instance Variable _local_path Undocumented
Instance Variable _media_path Undocumented
Instance Variable _metadata Undocumented
Instance Variable _metadata_path Undocumented
Instance Variable _path Undocumented
Instance Variable _team_id Undocumented
Property _progress Undocumented

Inherited from BaseDataset:

Method get_class_names Return class names per task.
Method get_n_classes Return number of classes per task.
Method get_n_keypoints Return the number of keypoints for each task.
Method get_task_names Return task names for the dataset.
@staticmethod
@override
def exists(dataset_name: str, team_id: str | None = None, bucket_storage: BucketStorage = BucketStorage.LOCAL, bucket: str | None = None) -> bool:

Check whether a dataset exists.

Parameters
dataset_name:strDataset name to check.
team_id:str | NoneOptional team identifier.
bucket_storage:BucketStorageStorage backend to inspect.
bucket:str | NoneOptional bucket name for remote storage.
Returns
boolTrue if the dataset exists, False otherwise.
Raises
ValueErrorIf bucket storage is remote but no bucket name is provided.
@staticmethod
def list_datasets(team_id: str | None = None, bucket_storage: BucketStorage = BucketStorage.LOCAL, bucket: str | None = None) -> list[str]:

List available datasets.

Parameters
team_id:str | NoneOptional team identifier.
bucket_storage:BucketStorageStorage backend to inspect.
bucket:str | NoneOptional bucket name for remote storage.
Returns
list[str]List of dataset names.
Raises
ValueErrorIf bucket storage is remote but no bucket name is provided.
ValueErrorIf the dataset is stored remotely but no bucket parameter is provided or no LUXONISML_BUCKET environment variable is set.
def __eq__(self, other: object) -> bool | NotImplementedType:

Compare datasets for equivalence.

def __init__(self, dataset_name: str, team_id: str | None = None, bucket_type: BucketType | Literal['internal', 'external'] = BucketType.INTERNAL, bucket_storage: BucketStorage | Literal['local', 'gcs', 's3', 'azure'] = BucketStorage.LOCAL, *, delete_local: bool = False, delete_remote: bool = False):

Create a Luxonis Dataset Format dataset handle.

Parameters
dataset_name:strDataset name.
team_id:str | NoneOptional cloud team identifier.
bucket_type:BucketType | Literal['internal', 'external']Whether the dataset uses internal or external buckets.
bucket_storage:BucketStorage | Literal['local', 'gcs', 's3', 'azure']Underlying storage backend.
delete_local:boolWhether to delete a local dataset with the same name before initialization.
delete_remote:boolWhether to delete the remote dataset as well.
Raises
ValueErrorIf the dataset exists and deletion flags are not set.
ValueErrorIf the dataset is remote but no bucket is configured.
NotImplementedErrorIf Azure Blob Storage is selected as the bucket storage.
def __len__(self) -> int:

Return the number of records in the dataset.

@override
def add(self, generator: DatasetIterator, batch_size: int = 1000000) -> Self:

Add data to the dataset from a generator of records.

Parameters
generator:DatasetIterator

The generator should yield either dictionaries that can be converted to DatasetRecord objects or actual DatasetRecord instances. Each record must contain at least a file path and can optionally include an annotation and a task name.

For example:

def record_generator():
    yield {
        "file": f"/path/to/image.jpg",
        "task_name": "animals",
        "annotation": {
            "instance_id": 1,
            "class_name": "cat",
            "boundingbox": {
                "x": 10,
                "y": 20,
                "w": 100,
                "h": 150,
            }
            "keypoints": {
                "keypoints": [
                    (15, 25, 1),
                    (50, 60, 1),
                    (70, 80, 0),
                ],
            },
            "instance_segmentation": {
                "mask": "/path/to/mask.png",
            }
        },
    }
batch_size:intThe number of records to process in a batch before writing to storage. Larger batch sizes may be more efficient but will use more memory.
Returns
SelfUndocumented
Raises
ValueErrorIf the records yielded by the generator are not in the expected format.
ValueErrorIf the dataset contains metadata annotations with conflicting types.
def clone(self, new_dataset_name: str, push_to_cloud: bool = True, splits_to_clone: list[str] | None = None, team_id: str | None = None) -> LuxonisDataset:

Create a local copy of the current dataset.

Warning

The cloned dataset overwrites any existing dataset with the same name.

Parameters
new_dataset_name:strName of the cloned dataset.
push_to_cloud:boolWhether to push the cloned dataset to the cloud when the current dataset is remote.
splits_to_clone:list[str] | NoneOptional split names to clone. If omitted, all data is cloned.
team_id:str | NoneOptional team identifier for the cloned dataset.
Returns
LuxonisDatasetCloned dataset handle.
Raises
FileNotFoundErrorIf the current dataset is empty and splits_to_clone is specified.
@override
def delete_dataset(self, *, delete_remote: bool = False, delete_local: bool = False):

Delete the dataset from local storage and optionally the cloud.

Parameters
delete_remote:boolWhether to delete the remote dataset.
delete_local:boolWhether to delete the local dataset files.
Raises
ValueErrorIf neither delete_remote nor delete_local is set to True.
def export(self, output_path: PathType, dataset_type: DatasetType = DatasetType.NATIVE, max_partition_size_gb: float | None = None, zip_output: bool = False) -> Path | list[Path]:

Export the dataset into one of the supported formats.

Parameters
output_path:PathTypeDirectory where the dataset should be exported.
dataset_type:DatasetTypeExport format.
max_partition_size_gb:float | NoneOptional maximum partition size. If the dataset exceeds this size, it is split into partitions named {dataset_name}_part{partition_number}.
zip_output:boolWhether to zip the exported dataset or each partition after export.
Returns
Path | list[Path]Export directory, or ZIP archive paths when zip_output is enabled.
Raises
NotImplementedErrorIf the specified export format is not supported.
ValueErrorIf the output path already exists.
def get_categorical_encodings(self) -> dict[str, dict[str, int]]:

Get the categorical encodings for the dataset grouped by task.

Example output:

{
    "vehicles": {
        "color": {"red": 0, "green": 1, "blue": 2},
        "brand": {"audi": 0, "bmw": 1, "mercedes": 2},
    }
}
@override
def get_classes(self) -> dict[str, dict[str, int]]:

Get class names and IDs per task.

Returns
Mapping from class names to class IDs grouped by task name
{
    "color": {"red": 0, "green": 1, "blue": 2},
    "brand": {"audi": 0, "bmw": 1, "mercedes": 2},
}
def get_metadata_types(self) -> dict[str, Literal['float', 'int', 'str', 'Category']]:

Get the metadata types for each metadata annotation in the dataset.

Example output:

{
    "id": "int",
    "time_of_day": "Category",
    "temperature": "float",
}
@override
def get_skeletons(self) -> dict[str, tuple[list[str], list[tuple[int, int]]]]:

Return keypoint skeletons for each task.

Returns
dict[str, tuple[list[str], list[tuple[int, int]]]]Keypoint labels and edges keyed by task name.
@override
def get_source_names(self) -> list[str]:

Return input source names for the dataset.

Returns
list[str]Source names used to identify input data.
def get_splits(self) -> dict[str, list[str]] | None:

Get the dataset splits definitions.

Returns
dict[str, list[str]] | NoneA mapping of split names to list of UUIDs, or None if no splits are defined.
def get_statistics(self, sample_size: int | None = None, view: str | None = None) -> dict[str, Any]:

Return dataset statistics for a view or the full dataset.

The returned statistics include:

  • "duplicates": Analysis of duplicated content.
  • "class_distributions": Class frequencies organized by task name and task type. Classification tasks are excluded.
  • "missing_annotations": File paths that lack annotations.
  • "heatmaps": Spatial annotation distributions.
Parameters
sample_size:int | NoneOptional number of samples used for heatmap generation.
view:str | NoneOptional split name to analyze. If omitted, the entire dataset is analyzed.
Returns
dict[str, Any]Dataset statistics.
@override
def get_tasks(self) -> dict[str, list[str]]:

Return task names and task types.

Returns
dict[str, list[str]]Task types keyed by task name.
@deprecated('ratios', 'definitions', suggest={'ratios': 'splits', 'definitions': 'splits'})
@override
def make_splits(self, splits: Mapping[str, Sequence[PathType]] | Mapping[str, float] | tuple[float, float, float] | None = None, *, ratios: dict[str, float] | tuple[float, float, float] | None = None, definitions: dict[str, list[PathType]] | None = None, replace_old_splits: bool = False):

Create dataset splits for training, validation, and testing.

Note

Although "train", "val", and "test" are the conventional split names, you can use any split names you want by providing a mapping to the splits argument. This can be useful for combining records from multiple sources ("train_real", "train_synth") or for creating fully custom splits.

Parameters
splits:Mapping[str, Sequence[PathType]] | Mapping[str, float] | tuple[float, float, float] | None

A mapping defining the splits. Can be one of the following:

  • A mapping of split names to lists of file paths.
  • A mapping of split names to float ratios.
  • A tuple of three float ratios for train, val, and test splits.
ratios:dict[str, float] | tuple[float, float, float] | None

A mapping of split names to float ratios or a tuple of three float ratios for train, val, and test splits.

Deprecated since version 0.4.0: Use splits instead.
definitions:dict[str, list[PathType]] | None

A mapping of split names to lists of file paths.

Deprecated since version 0.4.0: Use splits instead.
replace_old_splits:boolWhether to replace old splits with new ones. If False` (default), new splits will be added to old splits, and duplicate group IDs will be filtered out. If ``True, old splits will be replaced with new splits.
Raises
ValueErrorIf both ratios and definitions are provided.
ValueErrorIf neither splits, ratios, nor definitions is provided.
ValueErrorIf both splits and ratios/definitions are provided.
ValueErrorIf splits is provided but is empty.
ValueErrorIf ratios is provided but does not sum to 1.
ValueErrorIf definitions is provided but the total number of files in definitions exceeds the dataset size.
ValueErrorIf definitions are provided but all of them are already included in old splits, resulting in no new files to add to splits while replace_old_splits is False.
FileNotFoundErrorIf the dataset is empty.
TypeErrorIf the splits definitions are not in the expected format.
def merge_with(self, other: LuxonisDataset, inplace: bool = True, new_dataset_name: str | None = None, splits_to_merge: list[str] | None = None, team_id: str | None = None) -> LuxonisDataset:

Merge another dataset into this or a new dataset.

Parameters
other:LuxonisDatasetDataset to merge into this dataset.
inplace:boolWhether to merge into this dataset. If False, a new dataset is created.
new_dataset_name:str | NoneName of the new dataset when inplace is False.
splits_to_merge:list[str] | NoneOptional split names to merge.
team_id:str | NoneOptional team identifier for a newly created dataset.
Returns
LuxonisDatasetDataset containing the merged data.
Raises
ValueErrorIf the datasets have different bucket storage types.
ValueErrorIf inplace is False but no name for the new dataset is provided.
def pull_from_cloud(self, update_mode: UpdateMode = UpdateMode.MISSING):

Synchronize the dataset from a remote bucket to a local storage.

Annotations and metadata are always pulled. Media files are pulled either when missing locally or always, depending on update_mode.

Parameters
update_mode:UpdateModeMedia synchronization mode.
def push_to_cloud(self, bucket_storage: BucketStorage | None, update_mode: UpdateMode = UpdateMode.MISSING):

Push the local dataset to a remote bucket.

Annotations and metadata are always pushed. Media files are pushed either when missing remotely or always, depending on update_mode.

Parameters
bucket_storage:BucketStorage | NoneRemote storage backend to push to. If unset, uses the dataset's current bucket storage type.
update_mode:UpdateModeMedia synchronization mode.
Raises
ValueErrorIf the dataset is empty or not initialized.
FileNotFoundErrorIf any media files are missing locally when attempting to push.
def remove_duplicates(self):

Remove duplicate files and annotations from the dataset.

Raises
FileNotFoundErrorIf the dataset is empty.
def set_class_order_per_task(self, class_order_per_task: dict[str, list[str]]):

Set class order for specific tasks.

Parameters
class_order_per_task:dict[str, list[str]]Mapping from task names to class names in the desired order.
Raises
ValueErrorIf a task is missing or the provided class names do not match the dataset classes for that task.
@override
def set_classes(self, classes: list[str] | dict[str, int], task: str | None = None, rewrite_metadata: bool = True):

Set classes for one or more tasks.

Parameters
classes:list[str] | dict[str, int]Class names, or class IDs keyed by class name. If class names are provided, IDs are assigned alphabetically starting from 0. A class named "background" is always assigned ID 0.
task:str | NoneOptional task to update. If omitted, all tasks are updated.
rewrite_metadata:boolUndocumented
@override
def set_skeletons(self, labels: list[str] | None = None, edges: list[tuple[int, int]] | None = None, task: str | None = None):

Set keypoint skeleton metadata.

Parameters
labels:list[str] | NoneOptional keypoint names.
edges:list[tuple[int, int]] | NoneOptional keypoint edges as 0-based index pairs.
task:str | NoneOptional task to update. If omitted, all tasks are updated.
Raises
ValueErrorIf neither labels nor edges is provided.
@override
def set_tasks(self, tasks: Mapping[str, Iterable[str]]):

Set dataset tasks.

Parameters
tasks:Mapping[str, Iterable[str]]Mapping from task names to task types.
@override
def update_source(self, source: LuxonisSource):

Update the dataset source definition.

Parameters
source:LuxonisSourceSource definition to store.
bucket_storage =

Underlying storage backend for the dataset.

bucket_type =

Whether the dataset uses internal or external buckets.

dataset_name =

Name of the dataset.

team_id =

Optional cloud team identifier.

@property
@override
identifier: str =

The unique identifier for the dataset.

@property
is_remote: bool =

Whether the dataset is stored remotely (in a cloud bucket) or locally.

@property
metadata: Metadata =

Get the dataset metadata.

Returns
Deep copy of the dataset metadata.
@property
source: LuxonisSource =

Get the source information for the dataset.

Returns
Dataset source metadata.
Raises
ValueErrorIf source metadata is missing.
@cached_property
version: Version =

The version of the underlying LDF that the dataset adheres to.

@staticmethod
def _construct_url(bucket_storage: BucketStorage, bucket: str, team_id: str, dataset_name: str) -> str:

Undocumented

@staticmethod
def _scan_annotation_files(files: list[Path]) -> pl.LazyFrame:

Undocumented

def _add_process_batch(self, data_batch: list[DatasetRecord], pfm: ParquetFileManager, index: pl.DataFrame | None):

Undocumented

def _get_credential(self, key: str) -> str:

Get secret credentials from the credentials file or environment.

@overload
def _get_index(self, lazy: Literal[False] = ..., raise_when_empty: Literal[False] = ...) -> pl.DataFrame | None:
@overload
def _get_index(self, lazy: Literal[False] = ..., raise_when_empty: Literal[True] = ...) -> pl.DataFrame:
@overload
def _get_index(self, lazy: Literal[True] = ..., raise_when_empty: Literal[False] = ...) -> pl.LazyFrame | None:
@overload
def _get_index(self, lazy: Literal[True] = ..., raise_when_empty: Literal[True] = ...) -> pl.LazyFrame:

Load unique file entries from annotation data.

Parameters
lazy:boolWhether to return a LazyFrame that can be further processed before collecting.
raise_when_empty:boolWhether to raise an error if the dataset is empty. If False, returns None when the dataset is empty.
Returns
pl.DataFrame | pl.LazyFrame | NoneDataFrame or LazyFrame containing the dataset records.
Raises
FileNotFoundErrorIf the dataset is empty and raise_when_empty is True.
def _get_metadata(self) -> Metadata:

Load metadata from local storage or cloud.

Cloud metadata is always downloaded before loading.

def _ignore_files_not_in_uuid_set(self, dir_path: PathLike[str] | str, names: list[str], uuids_to_keep: set[str]) -> set[str]:

Undocumented

def _init_credentials(self) -> dict[str, Any]:

Undocumented

def _init_paths(self):

Configure local paths or a bucket directory.

@overload
def _load_df_offline(self, lazy: Literal[False] = ..., raise_when_empty: Literal[False] = ..., attempt_migration: bool = ...) -> pl.DataFrame | None:
@overload
def _load_df_offline(self, lazy: Literal[False] = ..., raise_when_empty: Literal[True] = ..., attempt_migration: bool = ...) -> pl.DataFrame:
@overload
def _load_df_offline(self, lazy: Literal[True] = ..., raise_when_empty: Literal[False] = ..., attempt_migration: bool = ...) -> pl.LazyFrame | None:
@overload
def _load_df_offline(self, lazy: Literal[True] = ..., raise_when_empty: Literal[True] = ..., attempt_migration: bool = ...) -> pl.LazyFrame:

Load the dataset DataFrame from local storage.

Parameters
lazy:boolWhether to return a LazyFrame that can be further processed before collecting.
raise_when_empty:boolWhether to raise an error if the dataset is empty. If False, returns None when the dataset is empty.
attempt_migration:boolWhether to attempt internal migration of the DataFrame if the LDF version of the dataset does not match the supported LDF version of the current luxonis-ml installation. If True, will perform internal migration using migrate_dataframe. If False, will return the DataFrame as-is without migration, which may lead to errors if the LDF versions do not match.
Returns
pl.DataFrame | pl.LazyFrame | NoneThe dataset annotations as a Polars DataFrame or LazyFrame, or None if the dataset is empty and raise_when_empty is False.
Raises
FileNotFoundErrorIf the dataset is empty and raise_when_empty is True.
def _load_splits(self, path: Path) -> dict[str, list[str]]:

Undocumented

def _merge_metadata_with(self, other: LuxonisDataset):

Merge relevant metadata from other into self.

def _merge_splits(self, splits_self: dict[str, list[str]], splits_other: dict[str, list[str]]):

Undocumented

def _process_arrays(self, data_batch: list[DatasetRecord]):

Undocumented

def _save_df_offline(self, pl_df: pl.DataFrame):

Save annotations DataFrame into parquet files.

Uses ParquetFileManager to preserve the same structure as the original dataset.

Parameters
pl_df:pl.DataFrameDataFrame to save.
Raises
ValueErrorIf any row in the DataFrame is missing a 'uuid' value.
def _save_splits(self, splits: dict[str, list[str]]):

Undocumented

def _warn_on_duplicates(self):

Undocumented

def _write_metadata(self):

Undocumented

_annotations_path =

Undocumented

_arrays_path =

Undocumented

_base_path =

Undocumented

_bucket =

Undocumented

_bucket_storage =

Undocumented

_bucket_type =

Undocumented

_credentials =

Undocumented

_dataset_name =

Undocumented

_fs =

Undocumented

_is_synced: bool =

Undocumented

_local_path =

Undocumented

_media_path =

Undocumented

_metadata =

Undocumented

_metadata_path =

Undocumented

_path =

Undocumented

_team_id =

Undocumented

@cached_property
_progress: Progress =

Undocumented