luxonis_ml.data.parsers.BaseParser

class documentation

class BaseParser(ABC):

Constructor: BaseParser(dataset, dataset_type, task_name, full_warnings)

View In Hierarchy

Base class for dataset-format parsers.

Class Method	`discover_dir_splits`	Return present and valid split directories keyed by their canonical split names.
Class Method	`validate`	Validate whether the dataset directory has the expected format.
Static Method	`validate_split`	Validate whether a split directory has the expected format.
Method	`__init__`	Create a parser for a target dataset.
Method	`from_dir`	Parse all data in a source dataset directory.
Method	`from_split`	Parse data from one split subdirectory.
Method	`get_parser_issue_messages`	Return parser issue messages collected during the last parse.
Method	`parse_dir`	Parse an entire dataset directory into the target dataset.
Method	`parse_split`	Parse one split subdirectory into the target dataset.
Method	`reset_parser_issue_messages`	Clear collected parser issue messages.
Static Method	`_apply_counts_to_pool`	Distribute images across splits based on requested counts.
Static Method	`_apply_counts_to_splits`	Apply count-based split requests to existing splits.
Static Method	`_canonicalize_split_name`	All current parsers use `train` and `test` split names whereas validation splits can vary in name between `val` `valid` and `validation`.
Static Method	`_compare_stem_files`	Compare sets of files by stem.
Static Method	`_get_added_images`	Return unique images yielded by a dataset generator.
Static Method	`_list_images`	List OpenCV-supported images in a directory.
Static Method	`_sample_from_splits`	Sample from each original split independently.
Method	`_get_parser_issue_messages`	Return parser issue messages collected during the last parse.
Method	`_log_skipped_annotation_summary`	Undocumented
Method	`_parse_available_splits`	Undocumented
Method	`_parse_split`	Parse data in one split subdirectory.
Method	`_remove_unsplit_records`	Remove records that are not assigned to any split.
Method	`_reset_parser_issue_messages`	Clear collected parser issue messages and warning counters.
Method	`_warn_skipped_annotation`	Undocumented
Method	`_wrap_generator`	Add configured task names to generated records.
Constant	`_CANONICAL_SPLIT_NAMES`	Undocumented
Constant	`_SKIPPED_WARNING_LIMIT`	Undocumented
Constant	`_SPLIT_NAMES`	Undocumented
Instance Variable	`_dataset`	Undocumented
Instance Variable	`_dataset_type`	Undocumented
Instance Variable	`_full_warnings`	Undocumented
Instance Variable	`_logged_skipped_annotation_warnings`	Undocumented
Instance Variable	`_parser_issue_messages`	Undocumented
Instance Variable	`_seen_parser_issue_messages`	Undocumented
Instance Variable	`_skipped_annotation_counts_by_reason`	Undocumented
Instance Variable	`_suppressed_skipped_annotation_warnings`	Undocumented
Instance Variable	`_task_name`	Undocumented

@classmethod

def discover_dir_splits(cls, dataset_dir: Path) -> dict[str, dict[str, Any]]: ¶

overridden in luxonis_ml.data.parsers.COCOParser, luxonis_ml.data.parsers.YoloV6Parser, luxonis_ml.data.parsers.YOLOv8Parser

Return present and valid split directories keyed by their canonical split names.

@classmethod

def validate(cls, dataset_dir: Path) -> bool: ¶

overridden in luxonis_ml.data.parsers.COCOParser, luxonis_ml.data.parsers.UltralyticsNDJSONParser, luxonis_ml.data.parsers.YoloV6Parser, luxonis_ml.data.parsers.YOLOv8Parser

Validate whether the dataset directory has the expected format.

Parameters
dataset_dir:`Path`	Source dataset directory.

Returns
`bool`	Whether the dataset is in the expected format.

@staticmethod
@abstractmethod

def validate_split(split_path: Path) -> dict[str, Any] | None: ¶

overridden in luxonis_ml.data.parsers.ClassificationDirectoryParser, luxonis_ml.data.parsers.COCOParser, luxonis_ml.data.parsers.CreateMLParser, luxonis_ml.data.parsers.DarknetParser, luxonis_ml.data.parsers.FiftyOneClassificationParser, luxonis_ml.data.parsers.NativeParser, luxonis_ml.data.parsers.SegmentationMaskDirectoryParser, luxonis_ml.data.parsers.SOLOParser, luxonis_ml.data.parsers.TensorflowCSVParser, luxonis_ml.data.parsers.UltralyticsNDJSONParser, luxonis_ml.data.parsers.VOCParser, luxonis_ml.data.parsers.YoloV4Parser, luxonis_ml.data.parsers.YoloV6Parser, luxonis_ml.data.parsers.YOLOv8Parser

Validate whether a split directory has the expected format.

Parameters
split_path:`Path`	Path to a split directory.

Returns
`dict[str, Any] \| None`	Keyword arguments for `from_split`, or `None` if the split is not in the expected format.

def __init__(self, dataset: BaseDataset, dataset_type: DatasetType, task_name: str | dict[str, str] | None, full_warnings: bool = False): ¶

Create a parser for a target dataset.

Parameters
dataset:`BaseDataset`	Dataset to populate with parsed records.
dataset_type:`DatasetType`	Source dataset format.
task_name:`str \| dict[str, str] \| None`	Optional task naming rule. A string is used for all records. A mapping uses class names as keys and task names as values.
full_warnings:`bool`	Whether to log every skipped-annotation warning. When `False`, only the first 50 warnings are logged and the rest are summarized.

@abstractmethod

def from_dir(self, dataset_dir: Path, **kwargs) -> tuple[list[Path], list[Path], list[Path]]: ¶

Parse all data in a source dataset directory.

Parameters
dataset_dir:`Path`	Source dataset directory.
**kwargs	Additional parser-specific arguments.

Returns
`tuple[list[Path], list[Path], list[Path]]`	Added images for the train, validation, and test splits.

@abstractmethod

def from_split(self, **kwargs) -> ParserOutput: ¶

Parse data from one split subdirectory.

Example

split_kwargs = parser.validate_split(split_path)
if split_kwargs is not None:
    parser.from_split(**split_kwargs)

Parameters
**kwargs	Parser-specific arguments, usually produced by `validate_split`.

Returns
`ParserOutput`	LDF generator, skeleton metadata, and added images.

def get_parser_issue_messages(self) -> list[ParserIssueMessage]: ¶

Return parser issue messages collected during the last parse.

def parse_dir(self, dataset_dir: Path, **kwargs) -> BaseDataset: ¶

overridden in luxonis_ml.data.parsers.UltralyticsNDJSONParser

Parse an entire dataset directory into the target dataset.

Parameters
dataset_dir:`Path`	Source dataset directory.
**kwargs	Parser-specific arguments.

Returns
`BaseDataset`	Dataset with parsed images and annotations.

Raises
`ValueError`	If a parser that expects top-level splits cannot find a `train` directory.

def parse_split(self, split: str | None = None, random_split: bool = False, split_ratios: dict[str, float | int] | None = None, **kwargs) -> BaseDataset: ¶

overridden in luxonis_ml.data.parsers.UltralyticsNDJSONParser

Parse one split subdirectory into the target dataset.

Parameters
split:`str \| None`	Optional split name to assign to parsed data. When set, `split_ratios` and `random_split` are ignored.
random_split:`bool`	Whether to generate random splits using `split_ratios`.
split_ratios:`dict[str, float \| int] \| None`	Optional ratios or counts. Float values are treated as ratios; integer values are treated as counts.
**kwargs	Parser-specific arguments.

Returns
`BaseDataset`	Dataset with parsed images and annotations.

def reset_parser_issue_messages(self): ¶

Clear collected parser issue messages.

@staticmethod

def _apply_counts_to_pool(images: Sequence[PathType], split_ratios: dict[str, int]) -> dict[str, Sequence[PathType]]: ¶

Distribute images across splits based on requested counts.

When total requested exceeds available, fills splits by priority (most requested first).

Parameters
images:`Sequence[PathType]`	Images to distribute.
split_ratios:`dict[str, int]`	Requested counts for each split.

Returns
`dict[str, Sequence[PathType]]`	Split names mapped to assigned images.

@staticmethod

def _apply_counts_to_splits(original_splits: dict[str, Sequence[PathType]], split_ratios: dict[str, int]) -> dict[str, Sequence[PathType]]: ¶

Apply count-based split requests to existing splits.

Samples from each original split independently. If more samples are requested than available in a split, all available samples from that split are used.

Parameters
original_splits:`dict[str, Sequence[PathType]]`	Existing split assignments.
split_ratios:`dict[str, int]`	Requested counts for each split.

Returns
`dict[str, Sequence[PathType]]`	Split names mapped to assigned images.

@staticmethod

def _canonicalize_split_name(split_name: str) -> str: ¶

All current parsers use train and test split names whereas validation splits can vary in name between val valid and validation.

This maps valid -> val and validation -> val

@staticmethod

def _compare_stem_files(list1: Iterable[Path], list2: Iterable[Path]) -> bool: ¶

Compare sets of files by stem.

Example

>>> BaseParser._compare_stem_files([Path("a.jpg"), Path("b.jpg")],
...                                [Path("a.xml"), Path("b.xml")])
True
>>> BaseParser._compare_stem_files([Path("a.jpg")], [Path("b.txt")])
False

Parameters
list1:`Iterable[Path]`	First files to compare.
list2:`Iterable[Path]`	Second files to compare.

Returns
`bool`	Whether the non-empty file stem sets are equal.

@staticmethod

def _get_added_images(generator: DatasetIterator) -> list[Path]: ¶

Return unique images yielded by a dataset generator.

Parameters
generator:`DatasetIterator`	Dataset record generator.

Returns
`list[Path]`	Unique added image paths.

@staticmethod

def _list_images(image_dir: Path) -> list[Path]: ¶

List OpenCV-supported images in a directory.

Parameters
image_dir:`Path`	Directory with images.

Returns
`list[Path]`	Supported image paths.

@staticmethod

def _sample_from_splits(original_splits: dict[str, Sequence[PathType]], split_ratios: dict[str, int]) -> dict[str, Sequence[PathType]]: ¶

Sample from each original split independently.

Parameters
original_splits:`dict[str, Sequence[PathType]]`	Existing split assignments.
split_ratios:`dict[str, int]`	Requested counts for each split.

Returns
`dict[str, Sequence[PathType]]`	Split names mapped to sampled images.

def _get_parser_issue_messages(self) -> list[ParserIssueMessage]: ¶

Return parser issue messages collected during the last parse.

def _log_skipped_annotation_summary(self): ¶

Undocumented

def _parse_available_splits(self, dataset_dir: Path, **kwargs) -> dict[str, list[Path]]: ¶

overridden in luxonis_ml.data.parsers.COCOParser

Undocumented

def _parse_split(self, **kwargs) -> list[Path]: ¶

Parse data in one split subdirectory.

Parameters
**kwargs	Parser-specific arguments.

Returns
`list[Path]`	Added images.

def _remove_unsplit_records(self): ¶

Remove records that are not assigned to any split.

def _reset_parser_issue_messages(self): ¶

Clear collected parser issue messages and warning counters.

def _warn_skipped_annotation(self, parser_issue: ParserIssue, reason: str, *, source: PathType | None = None, image: PathType | None = None, annotation_id: str | int | None = None): ¶

Undocumented

def _wrap_generator(self, generator: DatasetIterator) -> DatasetIterator: ¶

Add configured task names to generated records.

Parameters
generator:`DatasetIterator`	Dataset record generator.

Returns
`DatasetIterator`	Generator that yields records with task names applied.

_CANONICAL_SPLIT_NAMES: tuple[str, ...] = ¶

Undocumented

Value

('train', 'val', 'test')

_SKIPPED_WARNING_LIMIT: int = ¶

Undocumented

Value

_SPLIT_NAMES: tuple[str, ...] = ¶

overridden in luxonis_ml.data.parsers.FiftyOneClassificationParser, luxonis_ml.data.parsers.NativeParser

Undocumented

Value

('train', 'valid', 'test')

_dataset = ¶

Undocumented

_dataset_type = ¶

Undocumented

_full_warnings = ¶

Undocumented

_logged_skipped_annotation_warnings: int = ¶

Undocumented

_parser_issue_messages: list[ParserIssueMessage] = ¶

Undocumented

_seen_parser_issue_messages: set[ParserIssueMessage] = ¶

Undocumented

_skipped_annotation_counts_by_reason: dict[str, int] = ¶

Undocumented

_suppressed_skipped_annotation_warnings: int = ¶

Undocumented

_task_name = ¶

Undocumented