luxonis_ml.data.__main_

module documentation

Undocumented

Function	`clone`	Clone an existing dataset with a new name.
Function	`delete`	Delete a dataset from local storage, remote storage, or both.
Function	`export`	Export a Luxonis dataset to disk.
Function	`health`	Plot class distributions and heatmaps for every task type and corresponding task name in the dataset.
Function	`info`	Print information about a dataset.
Function	`inspect`	Inspect images and annotations in a dataset.
Function	`ls`	List datasets.
Function	`merge`	Merge two datasets stored in the same type of bucket.
Function	`parse`	Parse a directory with data and create a Luxonis dataset.
Function	`pull`	Pull a remote dataset to local storage.
Function	`push`	Push a local dataset to cloud storage.
Function	`sanitize`	Remove duplicate annotations and duplicate files from the dataset.
Type Alias	`BucketStorageT`	Undocumented
Variable	`app`	Undocumented

@app.command

def clone(name: str, new_name: str, *, push: Annotated[bool, Parameter(alias=-p, negative='')] = True, bucket_storage: BucketStorageT = BucketStorage.LOCAL, split: Annotated[list[str] | None, Parameter(alias=(-s))] = None, team_id: Annotated[str | None, Parameter(alias=(-t))] = None): ¶

Clone an existing dataset with a new name.

Optionally push it to cloud storage if it is a remote dataset.

Parameters
name:`str`	Name of the source dataset to clone.
new_name:`str`	Name of the new cloned dataset.
push:`Annotated[bool, Parameter(alias=-p, negative='')]`	If True, upload the newly cloned dataset to cloud storage.
bucket_storage:`BucketStorageT`	Storage type of the source dataset.
split:`Annotated[list[str] \| None, Parameter(alias=(-s))]`	List of split names to clone. If not provided, all splits will be cloned. Example: `--split train --split val` to clone only the "train" and "val" splits.
team_id:`Annotated[str \| None, Parameter(alias=(-t))]`	Team ID to use for the new dataset.

@app.command(alias=['rm', 'remove'])

def delete(*names: str, bucket_storage: BucketStorageT = BucketStorage.LOCAL, local: Annotated[bool, Parameter(alias=-l, negative='')] = False, remote: Annotated[bool, Parameter(alias=-r, negative='')] = False, yes: Annotated[bool, Parameter(alias=-y, negative='')] = False): ¶

Delete a dataset from local storage, remote storage, or both.

Parameters
*names:`str`	Name(s) of the dataset to delete.
bucket_storage:`BucketStorageT`	Storage type of the dataset.
local:`Annotated[bool, Parameter(alias=-l, negative='')]`	If True, delete the dataset from local storage.
remote:`Annotated[bool, Parameter(alias=-r, negative='')]`	If True, delete the dataset from remote storage.
yes:`Annotated[bool, Parameter(alias=-y, negative='')]`	If True, skip confirmation prompt and delete immediately.

@app.command

def export(name: str, *, save_dir: Annotated[str | None, Parameter(alias=(-s))] = None, dataset_type: Annotated[DatasetType, Parameter(name=(--type), alias=(-t))] = DatasetType.NATIVE, delete_existing: Annotated[bool, Parameter(name=--delete, alias=-d, negative='')] = False, max_partition_size_gb: Annotated[float | None, Parameter(alias=(-m))] = None, zip: bool = True, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Export a Luxonis dataset to disk.

Parameters
name:`str`	Name of the dataset to export.
save_dir:`Annotated[str \| None, Parameter(alias=(-s))]`	Directory where the exported dataset will be saved. If not provided, a directory with the same name as the dataset will be created in the current working directory.
dataset_type:`Annotated[DatasetType, Parameter(name=(--type), alias=(-t))]`	Format of the exported dataset.
delete_existing:`Annotated[bool, Parameter(name=--delete, alias=-d, negative='')]`	If True, delete any existing directory at the save location before exporting.
max_partition_size_gb:`Annotated[float \| None, Parameter(alias=(-m))]`	Maximum size of each partition in GB. If not provided, no partitioning will be done.
zip:`bool`	If `True`, the exported dataset will be zipped into a single archive. If `False`, the dataset will be exported as a directory with the specified structure.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

@app.command

def health(name: str, *, view: Annotated[str | None, Parameter(alias=(-v))] = None, sample_size: Annotated[int | None, Parameter(alias=(-n))] = None, save_dir: Annotated[str | None, Parameter(alias=(-s))] = None, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Plot class distributions and heatmaps for every task type and corresponding task name in the dataset.

Also checks for files with missing annotations, files that share the same UUIDs, and files with duplicate annotations.

Parameters
name:`str`	Name of the dataset to inspect.
view:`Annotated[str \| None, Parameter(alias=(-v))]`	Which split of the dataset to inspect. If not provided, all splits will be inspected.
sample_size:`Annotated[int \| None, Parameter(alias=(-n))]`	Number of annotation rows to sample from the dataset for calculating statistics and plots. If not provided, all annotations will be used.
save_dir:`Annotated[str \| None, Parameter(alias=(-s))]`	Directory where the generated plots will be saved. If not provided, the plots will be displayed interactively instead of being saved.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

@app.command

def info(name: str, *, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Print information about a dataset.

Parameters
name:`str`	Name of the dataset.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

@app.command

def inspect(name: str, *, view: Annotated[list[str] | None, Parameter(alias=(-v))] = None, aug_config: Annotated[Path | None, Parameter(alias=-a, validator=validators.Path(exists=True, ext=set(['.json', '.yaml', '.yml'])))] = None, size_multiplier: Annotated[float, Parameter(alias=(-s))] = 1.0, ignore_aspect_ratio: Annotated[bool, Parameter(alias=-i, negative='')] = False, deterministic: Annotated[bool, Parameter(alias=-d, negative='')] = False, force_update: Annotated[bool, Parameter(alias=-f, negative='')] = False, blend_all: Annotated[bool, Parameter(alias=-bl, negative='')] = False, per_instance: Annotated[bool, Parameter(alias=-pi, negative='')] = False, list_augmentations: Annotated[bool, Parameter(negative='')] = False, print_sample_metadata: Annotated[bool, Parameter(negative='')] = True, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Inspect images and annotations in a dataset.

Parameters
name:`str`	Name of the dataset to inspect.
view:`Annotated[list[str] \| None, Parameter(alias=(-v))]`	Which splits of the dataset to inspect. If not provided, the "train" split will be inspected by default.
aug_config:`Annotated[Path \| None, Parameter(alias=-a, validator=validators.Path(exists=True, ext=set(['.json', '.yaml', '.yml'])))]`	Path to a JSON or YAML config defining augmentations to apply when inspecting the dataset. If not provided, no augmentations will be applied.
size_multiplier:`Annotated[float, Parameter(alias=(-s))]`	Multiplier for the displayed image size.
ignore_aspect_ratio:`Annotated[bool, Parameter(alias=-i, negative='')]`	Do not keep the aspect ratio when resizing images.
deterministic:`Annotated[bool, Parameter(alias=-d, negative='')]`	Use deterministic augmentation mode.
force_update:`Annotated[bool, Parameter(alias=-f, negative='')]`	Force synchronization with remote storage first.
blend_all:`Annotated[bool, Parameter(alias=-bl, negative='')]`	Draw labels belonging to different tasks on the same image.
per_instance:`Annotated[bool, Parameter(alias=-pi, negative='')]`	Show each label instance in a separate window.
list_augmentations:`Annotated[bool, Parameter(negative='')]`	Show the augmentations applied to each displayed image. Requires '--aug-config' to be set.
print_sample_metadata:`Annotated[bool, Parameter(negative='')]`	Print sample metadata for each displayed sample.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

@app.command

def ls(*, full: Annotated[bool, Parameter(alias=(-f))] = False, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

List datasets.

Parameters
full:`Annotated[bool, Parameter(alias=(-f))]`	If True, show full information about each dataset, including classes and tasks.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

@app.command

def merge(source_name: str, target_name: str, new_name: Annotated[str | None, Parameter(alias=(-n))] = None, splits_to_merge: Annotated[str | None, Parameter(name=(--split), alias=(-s))] = None, bucket_storage: BucketStorageT = BucketStorage.LOCAL, team_id: Annotated[str | None, Parameter(alias=(-t))] = None): ¶

Merge two datasets stored in the same type of bucket.

Parameters
source_name:`str`	Name of the source dataset to merge from.
target_name:`str`	Name of the target dataset to merge into.
new_name:`Annotated[str \| None, Parameter(alias=(-n))]`	If provided, the name of the new merged dataset. If not provided, the source dataset will be merged into the target dataset in place.
splits_to_merge:`Annotated[str \| None, Parameter(name=(--split), alias=(-s))]`	Comma-separated list of split names to merge. If not provided, all splits will be merged.
bucket_storage:`BucketStorageT`	Storage type for both datasets.
team_id:`Annotated[str \| None, Parameter(alias=(-t))]`	Team ID to use for the new dataset. If not provided, the team ID of the target dataset will be used.

@app.command

def parse(dataset: Annotated[str, Parameter(alias=(--dataset - dir))], *, name: Annotated[str | None, Parameter(alias=(-n))] = None, dataset_type: Annotated[DatasetType | None, Parameter(name=(--type), alias=(-t))] = None, bucket_storage: BucketStorageT = BucketStorage.LOCAL, delete_local: Annotated[bool, Parameter(name=--delete, alias=-d, negative='')] = False, save_dir: Annotated[Path | None, Parameter(alias=(-s))] = None, task_name: Annotated[str | None, Parameter(alias=(-tn))] = None, log_all_warnings: bool = False, split_ratio: Annotated[str | None, Parameter(alias=(-sr))] = None, train: float | None = None, val: float | None = None, test: float | None = None): ¶

Parse a directory with data and create a Luxonis dataset.

Parameters
dataset:`Annotated[str, Parameter(alias=(--dataset - dir))]`	Path or URL to the dataset.
name:`Annotated[str \| None, Parameter(alias=(-n))]`	Name of the dataset. If not provided, the directory name will be used.
dataset_type:`Annotated[DatasetType \| None, Parameter(name=(--type), alias=(-t))]`	Type of the dataset. If not provided, the parser will attempt to detect it.
bucket_storage:`BucketStorageT`	Storage type of the dataset.
delete_local:`Annotated[bool, Parameter(name=--delete, alias=-d, negative='')]`	If True, delete any existing local dataset with the same name before parsing.
save_dir:`Annotated[Path \| None, Parameter(alias=(-s))]`	If dataset_dir is a remote URL, this is the local directory where the dataset will be downloaded before parsing. If not provided, the dataset will be downloaded to the current working directory.
task_name:`Annotated[str \| None, Parameter(alias=(-tn))]`	Task name to use for all records parsed from this dataset.
log_all_warnings:`bool`	Log all skipped annotation warnings instead of capping the output at 50.
split_ratio:`Annotated[str \| None, Parameter(alias=(-sr))]`	A string representation of a Python list specifying the split ratios for train, val, and test sets. Deprecated in favor of `--train`, `--val`, and `--test`.
train:`float \| None`	Ratio or count of records to assign to the training set. Can be used together with `--val` and `--test`. If only some of these options are provided, the remaining split(s) receive an equal share of the leftover records (only supported for ratios, not counts).
val:`float \| None`	Ratio or count of records to assign to the validation set.
test:`float \| None`	Ratio or count of records to assign to the test set.

@app.command

def pull(name: str, *, force: Annotated[bool, Parameter(alias=-f, negative='')] = False, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Pull a remote dataset to local storage.

Parameters
name:`str`	Name of the dataset to pull.
force:`Annotated[bool, Parameter(alias=-f, negative='')]`	If True, pull all media files even if they already exist locally.
bucket_storage:`BucketStorageT`	Cloud storage type to pull from. Cannot be LOCAL.

@app.command

def push(name: str, *, bucket_storage: BucketStorage, force: Annotated[bool, Parameter(alias=-f, negative='')] = False): ¶

Push a local dataset to cloud storage.

Parameters
name:`str`	Name of the dataset to push.
bucket_storage:`BucketStorage`	Cloud storage type to push to. Cannot be LOCAL.
force:`Annotated[bool, Parameter(alias=-f, negative='')]`	If True, push all media files even if they already exist in the target cloud storage.

@app.command

def sanitize(name: str, bucket_storage: BucketStorageT = BucketStorage.LOCAL): ¶

Remove duplicate annotations and duplicate files from the dataset.

Parameters
name:`str`	Name of the dataset to sanitize.
bucket_storage:`BucketStorageT`	Storage type of the dataset.

BucketStorageT: TypeAlias = ¶

Undocumented

Value

Annotated[BucketStorage, Parameter(alias=(-b))]

app = ¶

Undocumented

luxonis_ml.data.__main__

`luxonis_ml.data.main`