# luxonis_dataset Python API: `luxonis_ml.data.datasets.luxonis_dataset` ## Classes ### LuxonisDataset Luxonis Dataset Format (LDF) dataset handle. LDF is a flexible and feature-rich dataset format designed for use within the Luxonis MLOps ecosystem. #### Methods ##### init ```python def __init__(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` (`str`): Dataset name. * `team_id` (`str | None`): Optional 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` (`bool`): Whether to delete a local dataset with the same name before initialization. * `delete_remote` (`bool`): Whether to delete the remote dataset as well. Raises * `ValueError`: If the dataset exists and deletion flags are not set. * `ValueError`: If the dataset is remote but no bucket is configured. * `NotImplementedError`: If Azure Blob Storage is selected as the bucket storage. ##### add ```python def add(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:```python 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` (`int`): The 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 * `Self` Raises * `ValueError`: If the records yielded by the generator are not in the expected format. * `ValueError`: If the dataset contains metadata annotations with conflicting types. ##### clone ```python def clone(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` (`str`): Name of the cloned dataset. * `push_to_cloud` (`bool`): Whether to push the cloned dataset to the cloud when the current dataset is remote. * `splits_to_clone` (`list[str] | None`): Optional split names to clone. If omitted, all data is cloned. * `team_id` (`str | None`): Optional team identifier for the cloned dataset. Returns * `LuxonisDataset`: Cloned dataset handle. Raises * `FileNotFoundError`: If the current dataset is empty and `splits_to_clone` is specified. ##### delete_dataset ```python def delete_dataset(*, delete_remote: bool = False, delete_local: bool = False): ``` Delete the dataset from local storage and optionally the cloud. Parameters * `delete_remote` (`bool`): Whether to delete the remote dataset. * `delete_local` (`bool`): Whether to delete the local dataset files. Raises * `ValueError`: If neither `delete_remote` nor `delete_local` is set to `True`. ##### exists ```python 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` (`str`): Dataset name to check. * `team_id` (`str | None`): Optional team identifier. * `bucket_storage` (`BucketStorage`): Storage backend to inspect. * `bucket` (`str | None`): Optional bucket name for remote storage. Returns * `bool`: `True` if the dataset exists, `False` otherwise. Raises * `ValueError`: If bucket storage is remote but no bucket name is provided. ##### export ```python def export(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` (`PathType`): Directory where the dataset should be exported. * `dataset_type` (`DatasetType`): Export format. * `max_partition_size_gb` (`float | None`): Optional maximum partition size. If the dataset exceeds this size, it is split into partitions named `{dataset_name}_part{partition_number}`. * `zip_output` (`bool`): Whether 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 * `NotImplementedError`: If the specified export format is not supported. * `ValueError`: If the output path already exists. ##### get_categorical_encodings ```python def get_categorical_encodings(self) -> dict[str, dict[str, int]]: ``` Get the categorical encodings for the dataset grouped by task. Example output: ```python { "vehicles": { "color": {"red": 0, "green": 1, "blue": 2}, "brand": {"audi": 0, "bmw": 1, "mercedes": 2}, } } ``` ##### get_classes ```python 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`:```pycon { "color": {"red": 0, "green": 1, "blue": 2}, "brand": {"audi": 0, "bmw": 1, "mercedes": 2}, } ``` ##### get_metadata_types ```python 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: ```python { "id": "int", "time_of_day": "Category", "temperature": "float", } ``` ##### get_skeletons ```python 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. ##### get_source_names ```python def get_source_names(self) -> list[str]: ``` Return input source names for the dataset. Returns * `list[str]`: Source names used to identify input data. ##### get_splits ```python def get_splits(self) -> dict[str, list[str]] | None: ``` Get the dataset splits definitions. Returns * `dict[str, list[str]] | None`: A mapping of split names to list of UUIDs, or `None` if no splits are defined. ##### get_statistics ```python def get_statistics(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 | None`): Optional number of samples used for heatmap generation. * `view` (`str | None`): Optional split name to analyze. If omitted, the entire dataset is analyzed. Returns * `dict[str, Any]`: Dataset statistics. ##### get_tasks ```python 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. ##### list_datasets ```python 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 | None`): Optional team identifier. * `bucket_storage` (`BucketStorage`): Storage backend to inspect. * `bucket` (`str | None`): Optional bucket name for remote storage. Returns * `list[str]`: List of dataset names. Raises * `ValueError`: If bucket storage is remote but no bucket name is provided. * `ValueError`: If the dataset is stored remotely but no `bucket` parameter is provided or no `LUXONISML_BUCKET` environment variable is set. ##### make_splits ```python def make_splits(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** > 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** > Deprecated since version 0.4.0:Use `splits` instead. * `replace_old_splits` (`bool`): Whether 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 * `ValueError`: If both `ratios` and `definitions` are provided. * `ValueError`: If neither `splits`, `ratios`, nor `definitions` is provided. * `ValueError`: If both `splits` and `ratios`/`definitions` are provided. * `ValueError`: If `splits` is provided but is empty. * `ValueError`: If `ratios` is provided but does not sum to 1. * `ValueError`: If `definitions` is provided but the total number of files in definitions exceeds the dataset size. * `ValueError`: If `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`. * `FileNotFoundError`: If the dataset is empty. * `TypeError`: If the splits definitions are not in the expected format. ##### merge_with ```python def merge_with(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` (`LuxonisDataset`): Dataset to merge into this dataset. * `inplace` (`bool`): Whether to merge into this dataset. If `False`, a new dataset is created. * `new_dataset_name` (`str | None`): Name of the new dataset when `inplace` is `False`. * `splits_to_merge` (`list[str] | None`): Optional split names to merge. * `team_id` (`str | None`): Optional team identifier for a newly created dataset. Returns * `LuxonisDataset`: Dataset containing the merged data. Raises * `ValueError`: If the datasets have different bucket storage types. * `ValueError`: If `inplace` is `False` but no name for the new dataset is provided. ##### pull_from_cloud ```python def pull_from_cloud(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` (`UpdateMode`): Media synchronization mode. ##### push_to_cloud ```python def push_to_cloud(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 | None`): Remote storage backend to push to. If unset, uses the dataset's current bucket storage type. * `update_mode` (`UpdateMode`): Media synchronization mode. Raises * `ValueError`: If the dataset is empty or not initialized. * `FileNotFoundError`: If any media files are missing locally when attempting to push. ##### remove_duplicates ```python def remove_duplicates(self): ``` Remove duplicate files and annotations from the dataset. Raises * `FileNotFoundError`: If the dataset is empty. ##### set_class_order_per_task ```python def set_class_order_per_task(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 * `ValueError`: If a task is missing or the provided class names do not match the dataset classes for that task. ##### set_classes ```python def set_classes(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 | None`): Optional task to update. If omitted, all tasks are updated. * `rewrite_metadata` (`bool`) ##### set_skeletons ```python def set_skeletons(labels: list[str] | None = None, edges: list[tuple[int, int]] | None = None, task: str | None = None): ``` Set keypoint skeleton metadata. Parameters * `labels` (`list[str] | None`): Optional keypoint names. * `edges` (`list[tuple[int, int]] | None`): Optional keypoint edges as 0-based index pairs. * `task` (`str | None`): Optional task to update. If omitted, all tasks are updated. Raises * `ValueError`: If neither `labels` nor `edges` is provided. ##### set_tasks ```python def set_tasks(tasks: Mapping[str, Iterable[str]]): ``` Set dataset tasks. Parameters * `tasks` (`Mapping[str, Iterable[str]]`): Mapping from task names to task types. ##### update_source ```python def update_source(source: LuxonisSource): ``` Update the dataset source definition. Parameters * `source` (`LuxonisSource`): Source definition to store. #### Attributes ##### bucket_storage Underlying storage backend for the dataset. ##### bucket_type Whether the dataset uses internal or external buckets. ##### dataset_name Name of the dataset. ##### identifier The unique identifier for the dataset. ##### is_remote Whether the dataset is stored remotely (in a cloud bucket) or locally. ##### metadata Get the dataset metadata. Returns * Deep copy of the dataset metadata. ##### source Get the source information for the dataset. Returns * Dataset source metadata. Raises * `ValueError`: If source metadata is missing. ##### team_id Optional cloud team identifier. ##### version The version of the underlying LDF that the dataset adheres to.