class documentation

class BaseDataset(ABC):

Known subclasses: luxonis_ml.data.datasets.LuxonisDataset

View In Hierarchy

Base class for datasets in the Luxonis MLOps ecosystem.

Static Method exists Check whether a dataset exists.
Method add Write annotations to parquet files.
Method delete_dataset Delete local files belonging to the dataset.
Method get_class_names Return class names per task.
Method get_classes Get class names and IDs 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_skeletons Return keypoint skeletons for each task.
Method get_source_names Return input source names for the dataset.
Method get_task_names Return task names for the dataset.
Method get_tasks Return task names and task types.
Method make_splits Generate dataset splits.
Method set_classes Set classes for one or more tasks.
Method set_skeletons Set keypoint skeleton semantics for tasks that use keypoints.
Method set_tasks Set dataset tasks.
Method update_source Update the dataset source definition.
Property identifier The unique identifier for the dataset.
Property version The version of the underlying LDF.
@staticmethod
@abstractmethod
def exists(dataset_name: str) -> bool:

Check whether a dataset exists.

Parameters
dataset_name:strDataset name to check.
Returns
boolTrue if the dataset exists, False otherwise.
@abstractmethod
def add(self, generator: DatasetIterator, batch_size: int = 1000000):

Write annotations to parquet files.

Parameters
generator:DatasetIteratorIterator yielding DatasetRecord instances or dictionaries that can be converted to DatasetRecord.
batch_size:intNumber of records to buffer before processing. Lower values reduce peak memory usage.
@abstractmethod
def delete_dataset(self):

Delete local files belonging to the dataset.

def get_class_names(self) -> dict[str, list[str]]:

Return class names per task.

Returns
Class names keyed by task name
{
    "vehicles": ["red", "green", "blue"],
    "brands": ["audi", "bmw", "mercedes"],
}
@abstractmethod
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_n_classes(self) -> dict[str, int]:

Return number of classes per task.

Returns
dict[str, int]Mapping from task names to class counts.
def get_n_keypoints(self) -> dict[str, int]:

Return the number of keypoints for each task.

Returns
dict[str, int]Number of keypoints keyed by task name.
@abstractmethod
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.
@abstractmethod
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_task_names(self) -> list[str]:

Return task names for the dataset.

This is equivalent to get_tasks but returns only the task names.

Returns
list[str]Task names.
@abstractmethod
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.
@abstractmethod
def make_splits(self, splits: dict[str, Sequence[PathType]] | dict[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):

Generate dataset splits.

Parameters
splits:dict[str, Sequence[PathType]] | dict[str, float] | tuple[float, float, float] | NoneSplit definitions or ratios. Accepts explicit filepath lists, split ratios keyed by split name, or a (train, val, test) ratio tuple.
ratios:dict[str, float] | tuple[float, float, float] | NoneOptional deprecated split ratios. Use splits instead.
definitions:dict[str, list[PathType]] | NoneOptional deprecated filepath split definitions. Use splits instead.
replace_old_splits:boolWhether to replace existing split assignments instead of adding only new files.
@abstractmethod
def set_classes(self, classes: list[str] | dict[str, int], task: str | None = None):

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.
@abstractmethod
def set_skeletons(self, labels: list[str] | None = None, edges: list[tuple[int, int]] | None = None, task: str | None = None):

Set keypoint skeleton semantics for tasks that use keypoints.

For example:

dataset.set_skeletons(
    labels=["right hand", "right shoulder", ...],
    edges=[[0, 1], [4, 5], ...]
)
Parameters
labels:list[str] | NoneOptional keypoint names.
edges:list[tuple[int, int]] | NoneOptional edges between keypoints.
task:str | NoneOptional task to update. If omitted, all keypoint tasks are updated.
Raises
ValueErrorIf neither labels nor edges are provided.
@abstractmethod
def set_tasks(self, tasks: dict[str, list[str]]):

Set dataset tasks.

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

Update the dataset source definition.

Parameters
source:LuxonisSourceSource definition to store.
@property
@abstractmethod
identifier: str =

The unique identifier for the dataset.

@property
@abstractmethod
version: Version =

The version of the underlying LDF.