CloudSoftwareHardware
Software Stack
DepthAI
OAK Apps
AI Models

ON THIS PAGE

  • API reference

API reference

0.6.5
package

luxonis_ml

package
data
package
embeddings
package
enums
module
guard_extras
package
nn_archive
package
tracker
module
typing
package
utils
variable
__version__: str
package

luxonis_ml.data

package
augmentations
package
datasets
package
loaders
package
parsers
package
utils
class
BucketStorage
Underlying object storage for a bucket.
class
BucketType
Whether bucket storage is internal to Luxonis or not.
class
ImageType
Image type for IMAGE HType.
class
MediaType
Individual file type.
function
load_dataset_plugins()
Registers any external dataset BaseDataset class plugins.
function
load_loader_plugins()
Registers any external dataset BaseLoader class plugins.
package

luxonis_ml.data.augmentations

module
albumentations_engine
module
module
module
batch_transform
package
custom
module
utils
class
AlbumentationsEngine
Augmentation engine using the Albumentations library under the hood.  Configuration Format  The configuration is a list of dictionaries, where the dictionaries contain the name of the transformation and optionally its parameters. It can also contain a boolean flag use_for_resizing that indicates whether the transformation should be used for resizing. If no resizing augmentation is provided, the engine will use either A.Resize or LetterboxResize depending on the keep_aspect_ratio parameter.  The name must be either a valid name of an Albumentations transformation (accessible under the albumentations namespace), or a name of a custom transformation registered in the TRANSFORMATIONS registry.  Example:      [         {             "name": "Affine",             "params": {                 "rotate": 30,                 "scale": 0.5,                 "p": 0.3,             },         },         {             "name": "MixUp",             "params": {                 "alpha": [0.3, 0.7],                 "p": 0.5,             },         },         {             "name": "CustomResize",             "use_for_resizing": True,         },     ]  Transformation Order  The order of transformations provided in the configuration is not guaranteed to be preserved. The transformations are divided into the following groups and are applied in the same order:  batch transformations: Subclasses of BatchTransform.  spatial transformations: Subclasses of `A.DualTransform`.  custom transformations: Subclasses of `A.BasicTransform`, but not subclasses of any of more specific base classes above.  pixel transformations: Subclasses of `A.ImageOnlyTransform`. These transformations act only on the image.  Supported Augmentations  Official Augmentations  All augmentations provided by the Albumentations library are supported.  Supported Batch Augmentations  MixUp  MixUp is a data augmentation technique that blends 2 source images into a single image using a weight coefficient alpha.  Mosaic4  Mosaic4 transformation combines 4 images into a single image by placing them in a 2x2 grid.  Augmenting Unsupported Tasks  Albumentations do not natively support all the tasks supported by Luxonis Data Format. This sections describes how unsupported tasks are handled.  Note that the following applies only to officially supported augmentations. Custom augmentations can be implemented to handle arbitrary tasks.  Classification  Classification tasks can be properly augmented only for multi-label tasks, where each class is tied to a bounding box. In such cases, the classes belonging to bboxes falling outside the image are removed. In other cases, the classification annotation is kept as is.  Metadata  Metadata tasks can contain arbitrary data and their semantics are unknown to the augmentation engine. Therefore, the only transformation applied to metadata is discarding metadata associated with boxes falling outside the image.  Arrays  Arrays are dealt with in the same way as metadata. The only transformation applied to arrays is discarding arrays associated with bboxes falling outside the image.  Oriented Bounding Boxes  (Not yet implemented)  Oriented bounding boxes are of shape (n_boxes, 5) where the last dimension contains the angle of the box. This format is not supported by Albumentations, however, Albumentations support angle to be part of the keypoints. So, the oriented bounding boxes are split into regular bounding boxes and a set of keypoints that represent the center of the bbox and contain the angle as the third coordinate.  Both the keypoints and the bboxes are augmented separately. At the end, the angle is extracted from the keypoints and added back to the bounding boxes. The keypoints are discarded.  Custom Augmentations  Custom augmentations can be implemented by creating a subclass of A.BasicTransform and registering it in the TRANSFORMATIONS registry.  Possible target types that the augmentation can receive are:  'image': The image. All augmentations should usually support this target. For subclasses of A.ImageOnlyTransform or A.DualTransform this means overriding the apply method.  'bboxes': Bounding boxes. For subclasses of A.DualTransform, this means overriding the apply_to_bboxes method.  'keypoints': Keypoints. For subclasses of A.DualTransform, this means overriding the apply_to_keypoints method.  'mask': Segmentation masks. For subclasses of A.DualTransform, this means overriding the apply_to_mask method.  'instance_mask': Instance segmentation masks. For subclasses of BatchTransform, this means overriding the apply_to_instance_mask method.  Subclasses of A.DualTransform do not support this target, instance masks are treated as regular masks instead.  Custom augmentations can support instance masks by implementing their own logic for handling them and overriding the targets property to include the instance_mask target.  'array': Arbitrary arrays. Can only be supported by custom augmentations by implementing their own logic and adding the array target to the targets property.  'metadata': Metadata labels. Same situation as with the 'array' type.  'classification': One-hot encoded multi-task classification labels. Same situation as with the 'array' type.  Example:      class CustomArrayAugmentation(A.BasicTransform):          @property         @override         def targets(self) -> Dict[str, Any]:             return {                 "image": self.apply,                 "array": self.apply_to_array,             }          def apply(self, image: np.ndarray, **kwargs) -> np.ndarray:             ...          def apply_to_array(             self, array: np.ndarray, **kwargs         ) -> np.ndarray:             ...
constant
class
AugmentationEngine
class
BatchCompose
module

luxonis_ml.data.augmentations.albumentations_engine

type alias
type alias
class
AlbumentationConfigItem
function
class

luxonis_ml.data.augmentations.albumentations_engine.AlbumentationConfigItem(luxonis_ml.typing.ConfigItem)

variable
module

luxonis_ml.data.augmentations.batch_transform

class
BatchTransform
class

luxonis_ml.data.augmentations.batch_transform.BatchTransform(abc.ABC, albumentations.DualTransform)

method
__init__(self, batch_size: int, kwargs)
Batch transformation that combines multiple images and associated labels into one.  @param batch_size: Batch size needed for augmentation to work @type batch_size: int
variable
property
method
method
method
method
method
method
method
method
method
method
package

luxonis_ml.data.augmentations.custom

module
module
module
mosaic
class
LetterboxResize
class
MixUp
class
Mosaic4
constant
module

luxonis_ml.data.augmentations.custom.mosaic

function
function
function
apply_mosaic4_to_images(image_batch: List [ np.ndarray ], out_height: int, out_width: int, x_crop: int, y_crop: int, padding: Optional [ Union [ int , float , List [ int ] , List [ float ] ] ] = None) -> np.ndarray: np.ndarray
Arrange the images in a 2x2 grid layout.  The input images should have the same number of channels but can have different widths and heights. The gaps are filled by the padding value.
function
apply_mosaic4_to_bboxes(bbox: np.ndarray, in_height: int, in_width: int, position_index: int, out_height: int, out_width: int, x_crop: int, y_crop: int) -> np.ndarray: np.ndarray
Adjust bounding box coordinates to account for mosaic grid position.  This function modifies bounding boxes according to their placement in a 2x2 grid mosaic, shifting their coordinates based on the tile's relative position within the mosaic.
function
apply_mosaic4_to_keypoints(keypoints: np.ndarray, in_height: int, in_width: int, position_index: int, out_height: int, out_width: int, x_crop: int, y_crop: int) -> np.ndarray: np.ndarray
Adjust keypoint coordinates based on mosaic grid position.  This function adjusts the keypoint coordinates by placing them in one of the 2x2 mosaic grid cells, with shifts relative to the mosaic center.
class

luxonis_ml.data.augmentations.custom.LetterboxResize(albumentations.DualTransform)

variable
method
__init__(self, height: int, width: int, interpolation: int = cv2.INTER_LINEAR, image_fill_value: Color = 'black', mask_fill_value: int = 0, p: float = 1.0)
Augmentation to apply letterbox resizing to images. Also transforms masks, bboxes and keypoints to correct shape.  @type height: int @param height: Desired height of the output. @type width: int @param width: Desired width of the output. @type interpolation: int @param interpolation: cv2 flag to specify interpolation used     when resizing. Defaults to C{cv2.INTER_LINEAR}. @type image_fill_value: int @param image_fill_value: Padding value for images. Defaults to     "black". @type mask_fill_value: int @param mask_fill_value: Padding value for masks. Must be an     integer representing the class label. Defaults to C{0}     (background class). @type p: float @param p: Probability of applying the transform. Defaults to     C{1.0}.
variable
variable
variable
variable
property
method
update_params(self, params: Dict [ str , Any ], kwargs) -> Dict[str, Any]: Dict[str, Any]
Updates augmentation parameters with the necessary metadata.  @param params: The existing augmentation parameters dictionary. @type params: Dict[str, Any] @param kwargs: Additional keyword arguments to add the     parameters. @type kwargs: Any @return: Updated dictionary containing the merged parameters. @rtype: Dict[str, Any]
static method
LetterboxResize.compute_padding(orig_height: int, orig_width: int, out_height: int, out_width: int) -> Tuple[int, int, int, int]: Tuple[int, int, int, int]
Computes the padding required to resize an image to a letterbox format.  @type orig_height: int @param orig_height: Original height of the image. @type orig_width: int @param orig_width: Original width of the image. @type out_height: int @param out_height: Desired height of the output. @type out_width: int @param out_width: Desired width of the output. @rtype: Tuple[int, int, int, int] @return: Padding values for the top, bottom, left and right     sides of the image.
method
apply(self, img: np.ndarray, pad_top: int, pad_bottom: int, pad_left: int, pad_right: int, _) -> np.ndarray: np.ndarray
Applies the letterbox augmentation to an image.  @type img: np.ndarray @param img: Input image to which resize is applied. @type pad_top: int @param pad_top: Number of pixels to pad at the top. @type pad_bottom: int @param pad_bottom: Number of pixels to pad at the bottom. @type pad_left: int @param pad_left: Number of pixels to pad on the left. @type pad_right: int @param pad_right: Number of pixels to pad on the right. @rtype: np.ndarray @return: Image with applied letterbox resize.
method
method
method
class

luxonis_ml.data.augmentations.custom.MixUp(luxonis_ml.data.augmentations.batch_transform.BatchTransform)

method
__init__(self, alpha: Union [ float , Tuple [ float , float ] ] = 0.5, keep_aspect_ratio: bool = True, p: float = 0.5)
MixUp augmentation that merges two images and their annotations into one. If images are not of same size then second one is first resized to match the first one.  @type alpha: Union[float, Tuple[float, float]] @param alpha: Mixing coefficient, either a single float or a     tuple representing the range. Defaults to C{0.5}. @type keep_aspect_ratio: bool @param keep_aspect_ratio: Whether to keep the aspect ratio of     the second image when resizing. Defaults to C{True}. @type p: float, optional @param p: Probability of applying the transform. Defaults to     C{0.5}.
variable
variable
method
apply(self, image_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], alpha: float, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of images.  @type image_batch: List[np.ndarray] @param image_batch: Batch of input images to which the     transformation is applied. @type image_shapes: List[Tuple[int, int]] @param image_shapes: Shapes of the input images in the batch. @rtype: List[np.ndarray] @return: List of transformed images.
method
apply_to_mask(self, mask_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], alpha: float, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of masks.  Blends masks together. In case of a conflict, the class from the mask associated with higher alpha is chosen.  @type mask_batch: List[np.ndarray] @param mask_batch: Batch of input masks to which the     transformation is applied. @type image_shapes: List[Tuple[int, int]] @param image_shapes: Shapes of the input images in the batch. @type alpha: float @param alpha: Mixing coefficient. @rtype: List[np.ndarray] @return: List of transformed masks.
method
apply_to_instance_mask(self, mask_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of instance masks.  @type masks_batch: List[np.ndarray] @param masks_batch: Batch of input instance masks to which the     transformation is applied. @rtype: np.ndarray @return: Transformed instance masks.
method
apply_to_bboxes(self, bboxes_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], rows: int, cols: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of bboxes.  @type bboxes_batch: List[np.ndarray] @param bboxes_batch: Batch of input bboxes to which the     transformation is applied. @rtype: np.ndarray @return: Transformed bboxes.
method
apply_to_keypoints(self, keypoints_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of keypoints.  @type keypoints_batch: List[np.ndarray] @param keypoints_batch: Batch of input keypoints to which the     transformation is applied. @rtype: np.ndarray @return: Transformed keypoints.
method
get_params(self) -> Dict[str, Any]: Dict[str, Any]
Update parameters.  @param params: Dictionary containing parameters. @type params: Dict[str, Any] @return: Dictionary containing updated parameters. @rtype: Dict[str, Any]
method
get_params_dependent_on_data(self, params: Dict [ str , Any ], data: Dict [ str , Any ]) -> Dict[str, Any]: Dict[str, Any]
Get parameters dependent on the targets.  @param params: Dictionary containing parameters. @type params: Dict[str, Any] @return: Dictionary containing parameters dependent on the     targets. @rtype: Dict[str, Any]
method
class

luxonis_ml.data.augmentations.custom.Mosaic4(luxonis_ml.data.augmentations.batch_transform.BatchTransform)

method
__init__(self, out_height: int, out_width: int, value: Optional [ Union [ int , float , List [ int ] , List [ float ] ] ] = None, mask_value: Optional [ Union [ int , float , List [ int ] , List [ float ] ] ] = None, p: float = 0.5)
Mosaic augmentation arranges selected four images into single image in a 2x2 grid layout. This is done in deterministic way meaning first image in the batch will always be in top left. The input images should have the same number of channels but can have different widths and heights. The output is cropped around the intersection point of the four images with the size (out_with x out_height). If the mosaic image is smaller than width x height, the gap is filled by the C{fill_value}.  @type out_height: int @param out_height: Output image height. The mosaic image is     cropped by this height around the mosaic center. If the size     of the mosaic image is smaller than this value the gap is     filled by the C{value}. @type out_width: int @param out_width: Output image width. The mosaic image is     cropped by this height around the mosaic center. If the size     of the mosaic image is smaller than this value the gap is     filled by the C{value}. @type value: Optional[Union[int, float, List[int], List[float]]] @param value: Padding value. Defaults to C{None}. @type mask_value: Optional[Union[int, float, List[int],     List[float]]] @param mask_value: Padding value for masks. Defaults to C{None}. @type p: float @param p: Probability of applying the transform. Defaults to     C{0.5}.
variable
variable
variable
variable
method
generate_random_crop_center(self) -> Tuple[int, int]: Tuple[int, int]
Generate a random crop center within the bounds of the mosaic image size.
method
apply(self, image_batch: List [ np.ndarray ], x_crop: int, y_crop: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of images.  @type image_batch: List[np.ndarray] @param image_batch: Batch of input images to which the     transformation is applied. @type x_crop: int @param x_crop: x-coordinate of the croping start point @type y_crop: int @param y_crop: y-coordinate of the croping start point @rtype: np.ndarray @return: Transformed images.
method
apply_to_mask(self, mask_batch: List [ np.ndarray ], x_crop: int, y_crop: int, cols: int, rows: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of masks.  @type mask_batch: List[np.ndarray] @param mask_batch: Batch of input masks to which the     transformation is applied. @type x_crop: int @param x_crop: x-coordinate of the croping start point @type y_crop: int @param y_crop: y-coordinate of the croping start point @rtype: np.ndarray @return: Transformed masks.
method
apply_to_instance_mask(self, masks_batch: List [ np.ndarray ], x_crop: int, y_crop: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of instance masks.  @type mask_batch: List[np.ndarray] @param mask_batch: Batch of input masks to which the     transformation is applied. @type x_crop: int @param x_crop: x-coordinate of the croping start point @type y_crop: int @param y_crop: y-coordinate of the croping start point @rtype: np.ndarray @return: Transformed masks.
method
apply_to_bboxes(self, bboxes_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], x_crop: int, y_crop: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of bboxes.  @type bboxes_batch: List[np.ndarray] @param bboxes_batch: Batch of input bboxes to which the     transformation is applied. @type indices: List[Tuple[int, int]] @param indices: Indices of images in the batch. @type image_shapes: List[Tuple[int, int]] @param image_shapes: Shapes of the input images in the batch. @type params: Any @param params: Additional parameters for the transformation. @type x_crop: int @param x_crop: x-coordinate of the croping start point @type y_crop: int @param y_crop: y-coordinate of the croping start point @rtype: List[np.ndarray] @return: List of transformed bboxes.
method
apply_to_keypoints(self, keypoints_batch: List [ np.ndarray ], image_shapes: List [ Tuple [ int , int ] ], x_crop: int, y_crop: int, _) -> np.ndarray: np.ndarray
Applies the transformation to a batch of keypoints.  @type keypoints_batch: List[KeypointType] @param keypoints_batch: Batch of input keypoints to which the     transformation is applied. @type indices: List[Tuple[int, int]] @param indices: Indices of images in the batch. @type image_shapes: List[Tuple[int, int]] @param image_shapes: Shapes of the input images in the batch. @type params: Any @param params: Additional parameters for the transformation. @type x_crop: int @param x_crop: x-coordinate of the croping start point @type y_crop: int @param y_crop: y-coordinate of the croping start point @rtype: List[KeypointType] @return: List of transformed keypoints.
method
get_params_dependent_on_data(self, params: Dict [ str , Any ], data: Dict [ str , Any ]) -> Dict[str, Any]: Dict[str, Any]
Get parameters dependent on the targets.  @type params: Dict[str, Any] @param params: Dictionary containing parameters. @rtype: Dict[str, Any] @return: Dictionary containing parameters dependent on the     targets.
module

luxonis_ml.data.augmentations.utils

function
function
function
function
function
function
type variable
function
yield_batches(data_batch: List [ Dict [ str , T ] ], batch_size: int) -> Iterator[Dict[str, List[T]]]: Iterator[Dict[str, List[T]]]
Yield batches of data.  @type data_batch: List[Dict[str, Any]] @param data_batch: List of dictionaries containing data. @type batch_size: int @param batch_size: Size of the batch. @rtype: Iterator[Dict[str, List[Any]]] @return: Generator of batches of data.
class

luxonis_ml.data.augmentations.AlbumentationsEngine(luxonis_ml.data.augmentations.AugmentationEngine)

method
__init__(self, height: int, width: int, targets: Dict [ str , str ], n_classes: Dict [ str , int ], config: Iterable [ Params ], keep_aspect_ratio: bool = True, is_validation_pipeline: bool = False, min_bbox_visibility: float = 0.0)
variable
variable
variable
variable
variable
variable
variable
variable
variable
property
method
method
preprocess_batch(self, labels_batch: List [ LoaderOutput ]) -> Tuple[List[Data], Dict[str, int]]: Tuple[List[Data], Dict[str, int]]
Preprocess a batch of labels.  @type labels_batch: List[Data] @param labels_batch: List of dictionaries mapping task names to     the annotations as C{np.ndarray} @rtype: Tuple[List[Data], Dict[str, int]] @return: Tuple containing the preprocessed data and a dictionary     mapping task names to the number of keypoints for that task.
method
postprocess(self, data: Data, n_keypoints: Dict [ str , int ]) -> LoaderOutput: LoaderOutput
Postprocess the augmented data back to LDF format.  Discards labels associated with bboxes that are outside the image.  @type data: Data @param data: Dictionary mapping task names to the annotations as     C{np.ndarray} @type n_keypoints: Dict[str, int] @param n_keypoints: Dictionary mapping task names to the number     of keypoints for that task. @rtype: LoaderOutput @return: Tuple containing the augmented image and the labels.
static method
static method
class

luxonis_ml.data.augmentations.AugmentationEngine(abc.ABC)

method
__init__(self, height: int, width: int, targets: Mapping [ str , str ], n_classes: Mapping [ str , int ], config: Iterable [ Params ], keep_aspect_ratio: bool, is_validation_pipeline: bool, min_bbox_visibility: float = 0.0)
Initialize augmentation pipeline from configuration.  @type height: int @param height: Target image height @type width: int @param width: Target image width  @type targets: Dict[str, str] @param targets: Dictionary mapping task names to task types.     Example::         {             "detection/boundingbox": "bbox",             "detection/segmentation": "mask",         }  @type n_classes: Dict[str, int] @param n_classes: Dictionary mapping task names to the number     of associated classes. Example::         {             "cars/boundingbox": 2,             "cars/segmentation": 2,             "motorbikes/boundingbox": 1,         }  @type config: List[Params] @param config: List of dictionaries with configuration for each     augmentation. It is up to the augmentation engine to parse     and interpret this configuration.  @type keep_aspect_ratio: bool @param keep_aspect_ratio: Whether to keep aspect ratio @type is_validation_pipeline: bool @param is_validation_pipeline: Whether this is a validation     pipeline (in which case some augmentations are skipped) @type min_bbox_visibility: float @param min_bbox_visibility: Minimum fraction of the original bounding box that must remain visible after augmentation.
method
apply(self, data: List [ LoaderOutput ]) -> LoaderOutput: LoaderOutput
Apply the augmentation pipeline to the data.  @type data: List[LuxonisLoaderOutput] @param data: List of data to augment. The length of the list     must be equal to the batch size. @rtype: LuxonisLoaderOutput @return: Augmented data
property
batch_size
Getter for the batch size.  The batch size is the number of images necessary for the augmentation pipeline to work in case of batch-based augmentations.  For example, if the augmentation pipeline contains the MixUp augmentation, the batch size should be 2.  If the pipeline requires MixUp and also Mosaic4 augmentations, the batch size should be 8 (2 * 4).
class

luxonis_ml.data.augmentations.BatchCompose(albumentations.Compose)

variable
method
__init__(self, transforms: TransformsSeqType, kwargs)
Compose transforms and handle all transformations regarding bounding boxes.  @param transforms: List of transformations to compose @type transforms: TransformsSeqType @param kwargs: Additional arguments to pass to A.Compose
variable
method
__call__(self, data_batch: List [ Dict [ str , np.ndarray ] ]) -> Dict[str, np.ndarray]: Dict[str, np.ndarray]
static method
package

luxonis_ml.data.datasets

module
annotation
module
module
module
metadata
module
migration
module
module
utils
class
Annotation
Base class for an annotation.
class
ArrayAnnotation
A custom unspecified annotation that is an arbitrary numpy array.  All instances of this annotation must have the same shape.
class
BBoxAnnotation
Bounding box annotation.  Values are normalized based on the image size.
class
Category
class
DatasetRecord
class
Detection
class
KeypointAnnotation
Keypoint annotation.  Values are normalized to [0, 1] based on the image size.
function
constant
class
BaseDataset
Base abstract dataset class for managing datasets in the Luxonis MLOps ecosystem.
type alias
class
LuxonisDataset
class
UpdateMode
Update mode for the dataset.
class
Metadata
class
LuxonisComponent
Abstraction for a piece of media within a source. Most commonly,  this abstracts an image sensor.
class
LuxonisSource
Abstracts the structure of a dataset and which components/media are included.  For example, with an OAK-D, you can have a source with 4 image components: rgb (color), left (mono), right (mono), and depth.
module

luxonis_ml.data.datasets.annotation

type alias
type alias
class
ClassificationAnnotation
class
SegmentationAnnotation
Run-length encoded segmentation mask.
class
InstanceSegmentationAnnotation
function
check_valid_identifier(name: str, label: str)
Check if a name is a valid Python identifier after converting dashes to underscores.  Albumentations requires that the names of the targets passed as `additional_targets` are valid Python identifiers.
class

luxonis_ml.data.datasets.annotation.ClassificationAnnotation(luxonis_ml.data.datasets.Annotation)

static method
class

luxonis_ml.data.datasets.annotation.SegmentationAnnotation(luxonis_ml.data.datasets.Annotation)

variable
height
The height of the segmentation mask.
variable
width
The width of the segmentation mask.
variable
counts
The run-length encoded mask. This can be a list of integers or a byte string.
method
static method
method
CLASS_METHOD
CLASS_METHOD
CLASS_METHOD
class

luxonis_ml.data.datasets.annotation.InstanceSegmentationAnnotation(luxonis_ml.data.datasets.annotation.SegmentationAnnotation)

static method
module

luxonis_ml.data.datasets.metadata

class
Skeletons
class

luxonis_ml.data.datasets.metadata.Skeletons(typing_extensions.TypedDict)

variable
variable
module

luxonis_ml.data.datasets.migration

constant
constant
class
LDF_1_0_0_MetadataDict
function
function
module

luxonis_ml.data.datasets.utils

function
function
function
class

luxonis_ml.data.datasets.Annotation(abc.ABC, luxonis_ml.utils.BaseModelExtraForbid)

static method
class

luxonis_ml.data.datasets.ArrayAnnotation(luxonis_ml.data.datasets.Annotation)

variable
path
The path to the numpy array saved as a .npy file.
method
static method
method
CLASS_METHOD
class

luxonis_ml.data.datasets.BBoxAnnotation(luxonis_ml.data.datasets.Annotation)

variable
x
The top-left x coordinate of the bounding box. Normalized to [0, 1].
variable
y
The top-left y coordinate of the bounding box. Normalized to [0, 1].
variable
w
The width of the bounding box. Normalized to [0, 1].
variable
h
The height of the bounding box. Normalized to [0, 1].
method
static method
CLASS_METHOD
class

luxonis_ml.data.datasets.Category(str)

CLASS_METHOD
__get_pydantic_core_schema__
class

luxonis_ml.data.datasets.DatasetRecord(luxonis_ml.utils.BaseModelExtraForbid)

variable
variable
variable
property
method
CLASS_METHOD
CLASS_METHOD
method
to_parquet_rows(self) -> Iterable[ParquetRecord]: Iterable[ParquetRecord]
Converts an annotation to a dictionary for writing to a parquet file.  @rtype: L{ParquetDict} @return: A dictionary of annotation data.
class

luxonis_ml.data.datasets.Detection(luxonis_ml.utils.BaseModelExtraForbid)

variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
method
method
method
class

luxonis_ml.data.datasets.KeypointAnnotation(luxonis_ml.data.datasets.Annotation)

variable
keypoints
List of keypoints. Each keypoint is a tuple of (x, y, visibility). x and y are normalized to [0, 1]. visibility is one of 0, 1, or 2 where:  0: Not visible / not labeled  1: Occluded  2: Visible
method
static method
CLASS_METHOD
class

luxonis_ml.data.datasets.BaseDataset(abc.ABC)

property
identifier
The unique identifier for the dataset.
property
version
The version of the underlying LDF.
method
set_tasks(self, tasks: Dict [ str , List [ str ] ])
Sets the tasks for the dataset.  @type tasks: Dict[str, List[str]] @param tasks: A dictionary mapping task names to task types.
method
get_tasks(self) -> Dict[str, str]: Dict[str, str]
Returns a dictionary mapping task names to task types.  @rtype: Dict[str, str] @return: A dictionary mapping task names to task types.
method
update_source(self, source: LuxonisSource)
Updates underlying source of the dataset with a new LuxonisSource.  @type source: L{LuxonisSource} @param source: The new C{LuxonisSource} to replace the old one.
method
set_classes(self, classes: Union [ List [ str ] , Dict [ str , int ] ], task: Optional [ str ] = None)
Sets the classes for the dataset. This can be across all CV tasks or certain tasks.  @type classes: Union[List[str], Dict[str, int]] @param classes: Either a list of class names or a dictionary     mapping class names to class IDs. If list is provided, the     class IDs will be assigned I{alphabetically} starting from     C{0}. If the class names contain the class C{"background"},     it will be assigned the class ID C{0}. @type task: Optional[str] @param task: Optionally specify the task where these classes     apply.
method
get_classes(self) -> Dict[str, List[str]]: Dict[str, List[str]]
Get classes according to computer vision tasks.  @rtype: Dict[str, List[str]] @return: A dictionary mapping tasks to the classes used in each     task.
method
set_skeletons(self, labels: Optional [ List [ str ] ] = None, edges: Optional [ List [ Tuple [ int , int ] ] ] = None, task: Optional [ str ] = None)
Sets the semantic structure of keypoint skeletons for the classes that use keypoints.  Example::      dataset.set_skeletons(         labels=["right hand", "right shoulder", ...],         edges=[[0, 1], [4, 5], ...]     )  @type labels: Optional[List[str]] @param labels: List of keypoint names. @type edges: Optional[List[Tuple[int, int]]] @param edges: List of edges between keypoints. @type task: Optional[str] @param task: Optionally specify the task where these skeletons apply.     If not specified, the skeletons are set for all tasks that use keypoints.
method
get_skeletons(self) -> Dict[str, Tuple[List[str], List[Tuple[int, int]]]]: Dict[str, Tuple[List[str], List[Tuple[int, int]]]]
Returns the dictionary defining the semantic skeleton for each class using keypoints.  @rtype: Dict[str, Tuple[List[str], List[Tuple[int, int]]]] @return: For each task, a tuple containing a list of keypoint     names and a list of edges between the keypoints.
method
add(self, generator: DatasetIterator, batch_size: int = 1000000)
Write annotations to parquet files.  @type generator: L{DatasetIterator} @param generator: A Python iterator that yields either instances     of C{DatasetRecord} or a dictionary that can be converted to     C{DatasetRecord}. @type batch_size: int @param batch_size: The number of annotations generated before     processing. This can be set to a lower value to reduce     memory usage.
method
make_splits(self, splits: Optional [ Union [ Dict [ str , Sequence [ PathType ] ] , Dict [ str , float ] , Tuple [ float , float , float ] ] ] = None, ratios: Optional [ Union [ Dict [ str , float ] , Tuple [ float , float , float ] ] ] = None, definitions: Optional [ Dict [ str , List [ PathType ] ] ] = None, replace_old_splits: bool = False)
Generates splits for the dataset.  @type splits: Optional[Union[Dict[str, Sequence[PathType]], Dict[str, float], Tuple[float, float, float]]] @param splits: A dictionary of splits or a tuple of ratios for train, val, and test splits. Can be one of:     - A dictionary of splits with keys as split names and values as lists of filepaths     - A dictionary of splits with keys as split names and values as ratios     - A 3-tuple of ratios for train, val, and test splits @type ratios: Optional[Union[Dict[str, float], Tuple[float, float, float]]] @param ratios: Deprecated! A dictionary of splits with keys as split names and values as ratios. @type definitions: Optional[Dict[str, List[PathType]]] @param definitions: Deprecated! A dictionary of splits with keys as split names and values as lists of filepaths. @type replace_old_splits: bool @param replace_old_splits: Whether to remove old splits and generate new ones. If set to False, only new files will be added to the splits. Default is False.
method
delete_dataset(self)
Deletes all local files belonging to the dataset.
static method
BaseDataset.exists(dataset_name: str) -> bool: bool
Checks whether a dataset exists.  @warning: For offline mode only. @type dataset_name: str @param dataset_name: Name of the dataset @rtype: bool @return: Whether the dataset exists
method
get_task_names(self) -> List[str]: List[str]
Get the task names for the dataset.  Like `get_tasks`, but returns only the task names instead of the entire names.  @rtype: List[str] @return: List of task names.
method
class

luxonis_ml.data.datasets.LuxonisDataset(luxonis_ml.data.datasets.BaseDataset)

method
__init__(self, dataset_name: str, team_id: Optional [ str ] = None, bucket_type: Union [ BucketType , Literal [ ' internal ' , ' external ' ] ] = BucketType.INTERNAL, bucket_storage: Union [ BucketStorage , Literal [ ' local ' , ' gcs ' , ' s3 ' , ' azure ' ] ] = BucketStorage.LOCAL, delete_existing: bool = False, delete_remote: bool = False)
Luxonis Dataset Format (LDF) is used to define datasets in the Luxonis MLOps ecosystem.  @type dataset_name: str @param dataset_name: Name of the dataset @type team_id: Optional[str] @param team_id: Optional unique team identifier for the cloud @type bucket_type: BucketType @param bucket_type: Whether to use external cloud buckets @type bucket_storage: BucketStorage @param bucket_storage: Underlying bucket storage. Can be one of     C{local}, C{S3}, or C{GCS}. @type delete_existing: bool @param delete_existing: Whether to delete a dataset with the     same name if it exists @type delete_remote: bool @param delete_remote: Whether to delete the dataset from the     cloud as well
variable
variable
variable
variable
variable
variable
variable
variable
property
metadata
Returns a copy of the dataset metadata.  The metadata is a pydantic model with the following fields:  source: LuxonisSource  ldf_version: str  classes: Dict[task_name, Dict[class_name, class_id]]  tasks: Dict[task_name, List[task_type]]  skeletons: Dict[task_name, Skeletons]  Skeletons is a dictionary with keys 'labels' and 'edges'  labels: List[str]  edges: List[Tuple[int, int]]  categorical_encodings: Dict[task_name, Dict[metadata_name, Dict[metadata_value, int]]]  Encodings for string metadata values  Example:        {           "vehicle": {               "color": {"red": 0, "green": 1, "blue": 2},               "brand": {"audi": 0, "bmw": 1, "mercedes": 2},           }       }
property
property
property
method
__len__(self) -> int: int
Returns the number of instances in the dataset.
variable
variable
variable
variable
variable
variable
method
clone(self, new_dataset_name: str, push_to_cloud: bool = True) -> LuxonisDataset: LuxonisDataset
Create a new LuxonisDataset that is a local copy of the current dataset. Cloned dataset will overwrite the existing dataset with the same name.  @type new_dataset_name: str @param new_dataset_name: Name of the newly created dataset. @type push_to_cloud: bool @param push_to_cloud: Whether to push the new dataset to the     cloud. Only if the current dataset is remote.
method
sync_to_cloud(self)
Uploads data to a remote cloud bucket.
method
merge_with(self, other: LuxonisDataset, inplace: bool = True, new_dataset_name: Optional [ str ] = None) -> LuxonisDataset: LuxonisDataset
Merge all data from `other` LuxonisDataset into the current dataset (in-place or in a new dataset).  @type other: LuxonisDataset @param other: The dataset to merge into the current dataset. @type inplace: bool @param inplace: Whether to merge into the current dataset (True)     or create a new dataset (False). @type new_dataset_name: str @param new_dataset_name: The name of the new dataset to create     if inplace is False.
property
method
method
method
method
get_n_classes(self) -> Dict[str, int]: Dict[str, int]
Returns a mapping of task names to number of classes.  @rtype: Dict[str, int] @return: A mapping from task names to number of classes.
method
method
method
method
method
method
method
sync_from_cloud(self, update_mode: UpdateMode = UpdateMode.IF_EMPTY)
Synchronizes the dataset from a remote cloud bucket to the local directory.  This method performs the download only if local data is empty, or always downloads depending on the provided update_mode.  @type update_mode: UpdateMode @param update_mode: Specifies the update behavior.     - UpdateMode.IF_EMPTY: Downloads data only if the local dataset is empty.     - UpdateMode.ALWAYS: Always downloads and overwrites the local dataset.
method
delete_dataset(self, delete_remote: bool = False)
Deletes the dataset from local storage and optionally from the cloud.  @type delete_remote: bool @param delete_remote: Whether to delete the dataset from the     cloud.
method
method
method
static method
LuxonisDataset.exists(dataset_name: str, team_id: Optional [ str ] = None, bucket_storage: BucketStorage = BucketStorage.LOCAL, bucket: Optional [ str ] = None) -> bool: bool
Checks if a dataset exists.  @type dataset_name: str @param dataset_name: Name of the dataset to check @type team_id: Optional[str] @param team_id: Optional team identifier @type bucket_storage: BucketStorage @param bucket_storage: Underlying bucket storage from C{local},     C{S3}, or C{GCS}. Default is C{local}. @type bucket: Optional[str] @param bucket: Name of the bucket. Default is C{None}. @rtype: bool @return: Whether the dataset exists.
static method
LuxonisDataset.list_datasets(team_id: Optional [ str ] = None, bucket_storage: BucketStorage = BucketStorage.LOCAL, bucket: Optional [ str ] = None) -> List[str]: List[str]
Returns a list of all datasets.  @type team_id: Optional[str] @param team_id: Optional team identifier @type bucket_storage: BucketStorage @param bucket_storage: Underlying bucket storage (local, S3, or     GCS). Default is local. @type bucket: Optional[str] @param bucket: Name of the bucket. Default is None. @rtype: List[str] @return: List of all dataset names.
method
export(self, output_path: PathType, dataset_type: DatasetType = DatasetType.NATIVE) -> Path: Path
Exportes the dataset into on of the supported formats.  @type output_path: PathType @param output_path: Path to the directory where the dataset will     be exported. @type dataset_type: DatasetType @param dataset_type: To what format to export the dataset.     Currently only DatasetType.COCO is supported. @rtype: Path @return: Path to the zip file containing the exported dataset.
method
get_statistics(self, sample_size: Optional [ int ] = None, view: Optional [ str ] = None) -> Dict[str, Any]: Dict[str, Any]
Returns comprehensive dataset statistics as a structured dictionary for the given view or the entire dataset.  The returned dictionary contains:      - "duplicates": Analysis of duplicated content         - "duplicate_uuids": List of {"uuid": str, "files": List[str]} for images with same UUID         - "duplicate_annotations": List of repeated annotations with file_name, task_name, task_type,             annotation content, and count      - "class_distributions": Nested dictionary of class frequencies organized by task_name and task_type     (excludes classification tasks)      - "missing_annotations": List of file paths that exist in the dataset but lack annotations      - "heatmaps": Spatial distribution of annotations as 15x15 grid matrices organized by task_name and task_type  @type sample_size: Optional[int] @param sample_size: Number of samples to use for heatmap generation @type view: Optional[str] @param view: Name of the view to analyze. If None, the entire dataset is analyzed. @rtype: Dict[str, Any] @return: Dataset statistics dictionary as described above
class

luxonis_ml.data.datasets.UpdateMode(enum.Enum)

constant
constant
class

luxonis_ml.data.datasets.Metadata(luxonis_ml.utils.pydantic_utils.BaseModelExtraForbid)

variable
variable
variable
variable
variable
variable
variable
variable
method
method
class

luxonis_ml.data.datasets.LuxonisComponent(luxonis_ml.utils.BaseModelExtraForbid)

variable
name
A recognizable name for the component.
variable
media_type
Enum for the type of media for the component.
variable
image_type
Enum for the image type. Only used if media_type==MediaType.IMAGE.
class

luxonis_ml.data.datasets.LuxonisSource(luxonis_ml.utils.BaseModelExtraForbid)

variable
name
A recognizable name for the source. Defaults to "default".
variable
components
If not using the default configuration, a list of LuxonisComponent to group together in the source.
variable
main_component
The name of the component that should be primarily visualized.
method
merge_with(self, other: LuxonisSource) -> LuxonisSource: LuxonisSource
Merge two sources together.  @type other: LuxonisSource @param other: The other source to merge with. @rtype: LuxonisSource @return: A new source with the components of both sources.
CLASS_METHOD
package

luxonis_ml.data.loaders

module
module
constant
class
BaseLoader
Base abstract loader class.  Enforces the LuxonisLoaderOutput output label structure.
class
LuxonisLoader
class

luxonis_ml.data.loaders.BaseLoader(abc.ABC)

method
__len__(self) -> int: int
Returns the length of the dataset.  @rtype: int @return: Length of the dataset.
method
__getitem__(self, idx: int) -> LoaderOutput: LoaderOutput
Loads sample from dataset.  @type idx: int @param idx: Index of the sample to load. @rtype: L{LuxonisLoaderOutput} @return: Sample's data in C{LuxonisLoaderOutput} format.
method
__iter__(self) -> Iterator[LoaderOutput]: Iterator[LoaderOutput]
Iterates over the dataset.  @rtype: Iterator[L{LuxonisLoaderOutput}] @return: Iterator over the dataset.
class

luxonis_ml.data.loaders.LuxonisLoader(luxonis_ml.data.loaders.BaseLoader)

method
__init__(self, dataset: LuxonisDataset, view: Union [ str , List [ str ] ] = 'train', augmentation_engine: Union [ Literal [ ' albumentations ' ] , str ] = 'albumentations', augmentation_config: Optional [ Union [ List [ Params ] , PathType ] ] = None, height: Optional [ int ] = None, width: Optional [ int ] = None, keep_aspect_ratio: bool = True, exclude_empty_annotations: bool = False, color_space: Literal [ ' RGB ' , ' BGR ' ] = 'RGB', keep_categorical_as_strings: bool = False, update_mode: Union [ UpdateMode , Literal [ ' always ' , ' if_empty ' ] ] = UpdateMode.ALWAYS)
A loader class used for loading data from L{LuxonisDataset}.  @type dataset: LuxonisDataset @param dataset: Instance of C{LuxonisDataset} to use. @type view: Union[str, List[str]] @param view: What splits to use. Can be either a single split or     a list of splits. Defaults to C{"train"}. @type augmentation_engine: Union[Literal["albumentations"], str] @param augmentation_engine: The augmentation engine to use.     Defaults to C{"albumentations"}. @type augmentation_config: Optional[Union[List[Params],     PathType]] @param augmentation_config: The configuration for the     augmentations. This can be either a list of C{Dict[str, JsonValue]} or     a path to a configuration file.     The config member is a dictionary with two keys: C{name} and     C{params}. C{name} is the name of the augmentation to     instantiate and C{params} is an optional dictionary     of parameters to pass to the augmentation.      Example::          [             {"name": "HorizontalFlip", "params": {"p": 0.5}},             {"name": "RandomBrightnessContrast", "params": {"p": 0.1}},             {"name": "Defocus"}         ]  @type height: Optional[int] @param height: The height of the output images. Defaults to     C{None}. @type width: Optional[int] @param width: The width of the output images. Defaults to     C{None}. @type keep_aspect_ratio: bool @param keep_aspect_ratio: Whether to keep the aspect ratio of the     images. Defaults to C{True}. @type color_space: Literal["RGB", "BGR"] @param color_space: The color space of the output images. Defaults     to C{"RGB"}. @type exclude_empty_annotations: bool @param exclude_empty_annotations: Whether to exclude     empty annotations from the final label dictionary.     Defaults to C{False} (i.e. include empty annotations).  @type keep_categorical_as_strings: bool @param keep_categorical_as_strings: Whether to keep categorical     metadata labels as strings.     Defaults to C{False} (i.e. convert categorical labels to integers).  @type update_mode: UpdateMode @param update_mode: Enum that determines the sync mode:     - UpdateMode.ALWAYS: Force a fresh download     - UpdateMode.IF_EMPTY: Skip downloading if local data exists
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
method
__len__(self) -> int: int
Returns length of the dataset.  @rtype: int @return: Length of the loader.
method
__getitem__(self, idx: int) -> LoaderOutput: LoaderOutput
Function to load a sample consisting of an image and its annotations.  @type idx: int @param idx: The integer index of the sample to retrieve. @rtype: L{LuxonisLoaderOutput} @return: The loader ouput consisting of the image and a     dictionary defining its annotations.
package

luxonis_ml.data.parsers

module
base_parser
module
module
coco_parser
module
module
module
luxonis_parser
module
native_parser
module
module
module
module
module
module
class
BaseParser
class
ClassificationDirectoryParser
Parses directory with ClassificationDirectory annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── class1/     │   │   ├── img1.jpg     │   │   ├── img2.jpg     │   │   └── ...     │   ├── class2/     │   └── ...     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
COCOParser
Parses directory with COCO annotations to LDF.  Expected formats:      dataset_dir/     ├── train/     │   ├── data/     │   │   ├── img1.jpg     │   │   ├── img2.jpg     │   │   └── ...     │   └── labels.json     ├── validation/     │   ├── data/     │   └── labels.json     └── test/         ├── data/         └── labels.json      This is default format returned when using FiftyOne package.  or:      dataset_dir/         ├── train/         │   ├── img1.jpg         │   ├── img2.jpg         │   └── ...         │   └── _annotations.coco.json         ├── valid/         └── test/      This is one of the formats that can be generated by     U{Roboflow <https://roboflow.com/>}.
class
CreateMLParser
Parses directory with CreateML annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img2.jpg     │   └── ...     │   └── _annotations.createml.json     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
DarknetParser
Parses directory with DarkNet annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img1.txt     │   ├── ...     │   └── _darknet.labels     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
LuxonisParser
class
SegmentationMaskDirectoryParser
Parses directory with SegmentationMask annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img1_mask.png     │   ├── ...     │   └── _classes.csv     ├── valid/     └── test/  _classes.csv contains mappings between pixel value and class name.  This is one of the formats that can be generated by Roboflow.
class
SOLOParser
Parses directory with SOLO annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── metadata.json     │   ├── sensor_definitions.json     │   ├── annotation_definitions.json     │   ├── metric_definitions.json     │   └── sequence.<SequenceNUM>/     │       ├── step<StepNUM>.camera.jpg     │       ├── step<StepNUM>.frame_data.json     │       └── (OPTIONAL: step<StepNUM>.camera.semantic segmentation.jpg)     ├── valid/     └── test/  This is the default format returned by Unity simulation engine.
class
TensorflowCSVParser
Parses directory with TensorflowCSV annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img2.jpg     │   ├── ...     │   └── _annotations.csv     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
VOCParser
Parses directory with VOC annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img1.xml     │   └── ...     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
YoloV4Parser
Parses directory with YoloV4 annotations to LDF.  Expected format:      dataset_dir/     ├── train/     │   ├── img1.jpg     │   ├── img2.jpg     │   ├── ...     │   ├── _annotations.txt     │   └── _classes.txt     ├── valid/     └── test/  This is one of the formats that can be generated by Roboflow.
class
YoloV6Parser
Parses annotations from YoloV6 annotations to LDF.  Expected format:      dataset_dir/     ├── images/     │   ├── train/     │   │   ├── img1.jpg     │   │   ├── img2.jpg     │   │   └── ...     │   ├── valid/     │   └── test/     ├── labels/     │   ├── train/     │   │   ├── img1.txt     │   │   ├── img2.txt     │   │   └── ...     │   ├── valid/     │   └── test/     └── data.yaml  data.yaml contains names of all present classes.  This is one of the formats that can be generated by Roboflow.
module

luxonis_ml.data.parsers.base_parser

type alias
module

luxonis_ml.data.parsers.coco_parser

class
Format
function
clean_annotations(annotation_path: Path) -> Path: Path
Cleans annotations by removing images that are known to cause issues.  @type annotation_path: Path @param annotation_path: Path to the annotation JSON file. @rtype: Path @return: Path to the cleaned annotation JSON file     ("labels_fixed.json").
class

luxonis_ml.data.parsers.coco_parser.Format(str, enum.Enum)

constant
constant
module

luxonis_ml.data.parsers.luxonis_parser

class
ParserType
type variable
class

luxonis_ml.data.parsers.luxonis_parser.ParserType(enum.Enum)

constant
constant
module

luxonis_ml.data.parsers.native_parser

class
NativeParser
Parses directory with native LDF annotations.  Expected format:      dataset_dir/     ├── train/     │   └── annotations.json     ├── valid/     └── test/  The annotations are stored in a single JSON file as a list of dictionaries in the same format as the output of the generator function used in BaseDataset.add method.
class

luxonis_ml.data.parsers.native_parser.NativeParser(luxonis_ml.data.parsers.base_parser.BaseParser)

static method
static method
method
method
from_split(self, annotation_path: Path) -> ParserOutput: ParserOutput
Parses annotations from LDF Format.  @type annotation_path: C{Path} @param annotation_dir: Path to the JSON file with annotations. @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.BaseParser(abc.ABC)

method
__init__(self, dataset: BaseDataset, dataset_type: DatasetType, task_name: Optional [ Union [ str , Dict [ str , str ] ] ])
@type dataset: BaseDataset @param dataset: Dataset to add the parsed data to. @type dataset_type: DatasetType @param dataset_type: Type of the dataset. @type task_name: Optional[Union[str, Dict[str, str]]] @param task_name: Optional task name(s) for the dataset.     Can be either a single string, in which case all the records     added to the dataset will use this value as `task_name`, or     a dictionary with class names as keys and task names as values.     In the latter case, the task name for a record with a given     class name will be taken from the dictionary.
variable
variable
variable
static method
BaseParser.validate_split(split_path: Path) -> Optional[Dict[str, Any]]: Optional[Dict[str, Any]]
Validates if a split subdirectory is in an expected format. If so, returns kwargs to pass to L{from_split} method.  @type split_path: Path @param split_path: Path to split directory. @rtype: Optional[Dict[str, Any]] @return: Dictionary with kwargs to pass to L{from_split} method     or C{None} if the split is not in the expected format.
static method
BaseParser.validate(dataset_dir: Path) -> bool: bool
Validates if the dataset is in an expected format.  @type dataset_dir: Path @param dataset_dir: Path to source dataset directory. @rtype: bool @return: If the dataset is in the expected format.
method
from_dir(self, dataset_dir: Path, kwargs) -> Tuple[List[str], List[str], List[str]]: Tuple[List[str], List[str], List[str]]
Parses all present data to L{LuxonisDataset} format.  @type dataset_dir: str @param dataset_dir: Path to source dataset directory. @type kwargs: Any @param kwargs: Additional arguments for a specific parser     implementation. @rtype: Tuple[List[str], List[str], List[str]] @return: Tuple with added images for C{train}, C{val} and     C{test} splits.
method
from_split(self, kwargs) -> ParserOutput: ParserOutput
Parses a data in a split subdirectory to L{LuxonisDataset} format.  @type kwargs: Dict[str, Any] @param kwargs: Additional kwargs for specific parser implementation.     Should work together with L{validate_split} method like:          >>> from_split(**validate_split(split_path))  @rtype: ParserOutput @return: C{LDF} generator, list of class names,     skeleton dictionary for keypoints and list of added images.
method
parse_split(self, split: Optional [ str ] = None, random_split: bool = False, split_ratios: Optional [ Dict [ str , float ] ] = None, kwargs) -> BaseDataset: BaseDataset
Parses data in a split subdirectory to L{LuxonisDataset} format.  @type split: Optional[str] @param split: As what split the data will be added to LDF. If     set, C{split_ratios} and C{random_split} are ignored. @type random_split: bool @param random_split: If random splits should be made. If     C{True}, C{split_ratios} are used. @type split_ratios: Optional[Tuple[float, float, float]] @param split_ratios: Ratios for random splits. Only used if     C{random_split} is C{True}. Defaults to C{(0.8, 0.1, 0.1)}. @type kwargs: Dict[str, Any] @param kwargs: Additional C{kwargs} for specific parser     implementation. @rtype: LuxonisDataset @return: C{LDF} with all the images and annotations parsed.
method
parse_dir(self, dataset_dir: Path, kwargs) -> BaseDataset: BaseDataset
Parses entire dataset directory to L{LuxonisDataset} format.  @type dataset_dir: str @param dataset_dir: Path to source dataset directory. @type kwargs: Dict[str, Any] @param kwargs: Additional C{kwargs} for specific parser     implementation. @rtype: LuxonisDataset @return: C{LDF} with all the images and annotations parsed.
class

luxonis_ml.data.parsers.ClassificationDirectoryParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, class_dir: Path) -> ParserOutput: ParserOutput
Parses annotations from classification directory format to LDF. Annotations include classification.  @type class_dir: Path @param class_dir: Path to top level directory @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.COCOParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_path: Path) -> ParserOutput: ParserOutput
Parses annotations from COCO format to LDF. Annotations include classification, segmentation, object detection and keypoints if present.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_path: Path @param annotation_path: Path to annotation json file @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.CreateMLParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_path: Path) -> ParserOutput: ParserOutput
Parses annotations from CreateML format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_path: Path @param annotation_path: Path to annotation json file @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.DarknetParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, classes_path: Path) -> ParserOutput: ParserOutput
Parses annotations from Darknet format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type classes_path: Path @param classes_path: Path to file with class names @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.LuxonisParser(typing.Generic)

variable
method
__init__(self, dataset_dir: str, dataset_name: Optional [ str ] = None, save_dir: Optional [ Union [ Path , str ] ] = None, dataset_plugin: T = None, dataset_type: Optional [ DatasetType ] = None, task_name: Optional [ Union [ str , Dict [ str , str ] ] ] = None, kwargs)
High-level abstraction over various parsers.  Automatically recognizes the dataset format and uses the appropriate parser.  @type dataset_dir: str @param dataset_dir: Identifier of the dataset directory.     Can be one of:         - Local path to the dataset directory.         - Remote URL supported by L{LuxonisFileSystem}.           - C{gcs://} for Google Cloud Storage           - C{s3://} for Amazon S3         - C{roboflow://} for Roboflow datasets.           - Expected format: C{roboflow://workspace/project/version/format}.     Can be a remote URL supported by L{LuxonisFileSystem}. @type dataset_name: Optional[str] @param dataset_name: Name of the dataset. If C{None}, the name     is derived from the name of the dataset directory. @type save_dir: Optional[Union[Path, str]] @param save_dir: If a remote URL is provided in C{dataset_dir},     the dataset will be downloaded to this directory. If     C{None}, the dataset will be downloaded to the current     working directory. @type dataset_plugin: Optional[str] @param dataset_plugin: Name of the dataset plugin to use. If     C{None}, C{LuxonisDataset} is used. @type dataset_type: Optional[DatasetType] @param dataset_type: If provided, the parser will use this     dataset type instead of trying to recognize it     automatically. @type task_name: Optional[Union[str, Dict[str, str]]] @param task_name: Optional task name(s) for the dataset.     Can be either a single string, in which case all the records     added to the dataset will use this value as `task_name`, or     a dictionary with class names as keys and task names as values.     In the latter case, the task name for a record with a given     class name will be taken from the dictionary. @type kwargs: Dict[str, Any] @param kwargs: Additional C{kwargs} to be passed to the     constructor of specific L{BaseDataset} implementation.
variable
variable
variable
variable
variable
variable
method
parse(self, kwargs) -> BaseDataset: BaseDataset
Parses the dataset and returns it in LuxonisDataset format.  If the dataset already exists, parsing will be skipped and the existing dataset will be returned instead.  @type kwargs: Dict[str, Any] @param kwargs: Additional C{kwargs} for specific parser     implementation. @rtype: LuxonisDataset @return: Parsed dataset in L{LuxonisDataset} format.
class

luxonis_ml.data.parsers.SegmentationMaskDirectoryParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, seg_dir: Path, classes_path: Path) -> ParserOutput: ParserOutput
Parses annotations with SegmentationMask format to LDF.  Annotations include classification and segmentation.  @type image_dir: Path @param image_dir: Path to directory with images @type seg_dir: Path @param seg_dir: Path to directory with segmentation mask @type classes_path: Path @param classes_path: Path to CSV file with class names @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images
class

luxonis_ml.data.parsers.SOLOParser(luxonis_ml.data.parsers.BaseParser)

static method
SOLOParser.validate_split(split_path: Path) -> Optional[Dict[str, Any]]: Optional[Dict[str, Any]]
Validates if a split subdirectory is in an expected format.  @type split_path: Path @param split_path: Path to split directory. @rtype: Optional[Dict[str, Any]] @return: Dictionary with kwargs to pass to L{from_split} method     or C{None} if the split is not in the expected format.
static method
SOLOParser.validate(dataset_dir: Path) -> bool: bool
Validates if the dataset is in an expected format.  @type dataset_dir: Path @param dataset_dir: Path to source dataset directory. @rtype: bool @return: True if the dataset is in the expected format.
method
from_dir(self, dataset_dir: Path) -> Tuple[List[Path], List[Path], List[Path]]: Tuple[List[Path], List[Path], List[Path]]
Parses all present data to L{LuxonisDataset} format.  @type dataset_dir: str @param dataset_dir: Path to source dataset directory. @rtype: Tuple[List[Path], List[Path], List[Path]] @return: Tuple with added images for train, valid and test     splits.
method
from_split(self, split_path: Path) -> ParserOutput: ParserOutput
Parses data in a split subdirectory from SOLO format to L{LuxonisDataset} format.  @type split_path: Path @param split_path: Path to directory with sequences of images     and annotations. @rtype: L{ParserOutput} @return: C{LuxonisDataset} generator, list of class names,     skeleton dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.TensorflowCSVParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_path: Path) -> ParserOutput: ParserOutput
Parses annotations from TensorflowCSV format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_path: Path @param annotation_path: Path to annotation CSV file @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for
class

luxonis_ml.data.parsers.VOCParser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_dir: Path) -> ParserOutput: ParserOutput
Parses annotations from VOC format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_dir: Path @param annotation_dir: Path to directory with C{.xml}     annotations @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.YoloV4Parser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_path: Path, classes_path: Path) -> ParserOutput: ParserOutput
Parses annotations from YoloV4 format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_path: Path @param annotation_path: Path to annotation file @type classes_path: Path @param classes_path: Path to file with class names @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
class

luxonis_ml.data.parsers.YoloV6Parser(luxonis_ml.data.parsers.BaseParser)

static method
static method
method
method
from_split(self, image_dir: Path, annotation_dir: Path, classes_path: Path) -> ParserOutput: ParserOutput
Parses annotations from YoloV6 format to LDF. Annotations include classification and object detection.  @type image_dir: Path @param image_dir: Path to directory with images @type annotation_dir: Path @param annotation_dir: Path to directory with annotations @type classes_path: Path @param classes_path: Path to yaml file with classes names @rtype: L{ParserOutput} @return: Annotation generator, list of classes names, skeleton     dictionary for keypoints and list of added images.
package

luxonis_ml.data.utils

module
constants
module
module
module
module
plot_utils
module
module
visualizations
function
find_duplicates(df: pl.LazyFrame) -> Dict[str, List[Dict[str, Any]]]: Dict[str, List[Dict[str, Any]]]
Collects information about duplicate UUIDs and duplicate annotations in the dataset.  @type df: pl.LazyFrame @param df: Polars lazy frame containing dataset information. @rtype: Dict[str, List[Dict[str, Any]]] @return: A dictionary with two keys:     - "duplicate_uuids": list of dicts with "uuid" as key and "files" as value     - "duplicate_annotations": list of dicts with "file_name", "task_type",       "task_name", "annotation", and "count"
function
get_class_distributions(df: pl.LazyFrame) -> Dict[str, Dict[str, List[Dict[str, Any]]]]: Dict[str, Dict[str, List[Dict[str, Any]]]]
Gets class distribution info for non-classification tasks.  @type df: pl.LazyFrame @param df: Polars lazy frame containing dataset information. @rtype: Dict[str, Dict[str, List[Dict[str, Any]]]] @return: A dictionary with task names as keys, and dictionaries with     task types as keys and lists of dictionaries with class names     and counts as values.
function
get_duplicates_info(df: pl.LazyFrame) -> Dict[str, Any]: Dict[str, Any]
Collects and returns information about duplicate UUIDs and annotations.  @type df: pl.DataFrame @param df: Polars DataFrame containing dataset information. @rtype: Dict[str, Any] @return: A dictionary with two keys:     - "duplicate_uuids": list of dicts with "uuid" as key and "files" as value     - "duplicate_annotations": list of dicts with "file_name", "task_name",       "task_type", "annotation", and "count"
function
get_heatmaps(df: pl.LazyFrame, sample_size: Optional [ int ] = None, downsample_factor: int = 5) -> Dict[str, Dict[str, List[List[int]]]]: Dict[str, Dict[str, List[List[int]]]]
Generates heatmaps for bounding boxes, keypoints, and segmentations.  @type df: pl.LazyFrame @param df: Polars lazy frame containing dataset information. @type sample_size: Optional[int] @param sample_size: Number of samples to take from the dataset.     Default is None. @type downsample_factor: int @param downsample_factor: Factor to downsample the segmentation     masks. @rtype: Dict[str, Dict[str, List[List[int]]]] @return: A dictionary with task names as keys, and dictionaries with     task types as keys and lists of lists of integers representing     the heatmaps as values
function
get_missing_annotations(df: pl.LazyFrame) -> List[str]: List[str]
Returns file paths that exist but have no annotations.  @type df: pl.LazyFrame @param df: Polars lazy frame containing dataset information. @rtype: List[str] @return: A list of file paths that exist but have no annotations.
function
function
rgb_to_bool_masks(segmentation_mask: np.ndarray, class_colors: Dict [ str , RGB ], add_background_class: bool = False) -> Iterator[Tuple[str, np.ndarray]]: Iterator[Tuple[str, np.ndarray]]
Helper function to convert an RGB segmentation mask to boolean masks for each class.  Example:      >>> segmentation_mask = np.array([     ...     [[0, 0, 0], [255, 0, 0], [0, 255, 0]],     ...     [[0, 0, 0], [0, 255, 0], [0, 0, 255]],     ...     ], dtype=np.uint8)     >>> class_colors = {     ...     "red": (255, 0, 0),     ...     "green": (0, 255, 0),     ...     "blue": (0, 0, 255),     ... }     >>> for class_name, mask in rgb_to_bool_masks(     ...     segmentation_mask,     ...     class_colors,     ...     add_background_class=True,     ... ):     ...     print(class_name, np.array2string(mask, separator=", "))     background [[ True, False, False],                 [ True, False, False]]     red        [[False,  True, False],                 [False, False, False]]     green      [[False, False,  True],                 [False, True, False]]     blue       [[False, False, False],                 [False, False,  True]]  @type segmentation_mask: npt.NDArray[np.uint8] @param segmentation_mask: An RGB segmentation mask where each pixel     is colored according to the class it belongs to. @type class_colors: Dict[str, Tuple[int, int, int]] @param class_colors: A dictionary mapping class names to RGB colors. @type add_background_class: bool @param add_background_class: Whether to add a background class with a mask for all pixels     that do not belong to any class. The class name will be set to "background".     The background class will be yielded first. Default is False. @rtype: Iterator[Tuple[str, npt.NDArray[np.bool_]]] @return: An iterator of tuples where the first element is the class name and     the second element is a boolean mask for that class.
function
warn_on_duplicates(df: pl.LazyFrame)
Logs warnings for duplicate UUIDs and annotations in the dataset.  @type df: pl.LazyFrame @param df: Polars lazy frame containing dataset information.
class
ParquetFileManager
class
ParquetRecord
function
plot_class_distribution(ax: plt.Axes, task_type: str, task_data: List [ Dict [ str , Any ] ])
Plots a bar chart of class distribution.  @type ax: plt.Axes @param ax: The axis to plot on. @type task_type: str @param task_type: The type of task. @type task_data: List[Dict[str, Any]] @param task_data: The task data to plot.
function
plot_heatmap(ax: plt.Axes, fig: plt.Figure, task_type: str, heatmap_data: Optional [ List [ List [ float ] ] ])
" Plots a heatmap of heatmap_data.  @type ax: plt.Axes @param ax: The axis to plot on. @type fig: plt.Figure @param fig: The figure to plot on. @type task_type: str @param task_type: The type of task. @type heatmap_data: Optional[List[List[float]]] @param heatmap_data: The heatmap data to plot.
function
get_task_name(task: str) -> str: str
Returns the task name from a task.  @type task: str @param task: The task. @rtype: str @return: The task name.
function
get_task_type(task: str) -> str: str
Returns the task type from a task.  Example:      >>> get_task_type("task_name/type")     'type'     >>> get_task_type("metadata/name")     'metadata/name'     >>> get_task_type("task_name/metadata/name")     'metadata/name'  @type task: str @param task: The task in a format like "task_name/type". @rtype: str @return: The task type. If the task is a metadata task,     the type will be "metadata/type".
function
split_task(task: str) -> Tuple[str, str]: Tuple[str, str]
Splits a task into its task name and type.  @type task: str @param task: The task to split. @rtype: Tuple[str, str] @return: A tuple containing the task name and type.
function
task_is_metadata(task: str) -> bool: bool
Returns whether a task is a metadata task.  @type task: str @param task: The task to check. @rtype: bool @return: Whether the task is a metadata task.
function
task_type_iterator(labels: Labels, task_type: TaskType) -> Iterator[Tuple[str, np.ndarray]]: Iterator[Tuple[str, np.ndarray]]
Iterates over labels of a specific type.  @type labels: Labels @param labels: The labels to iterate over. @type task_type: str @param task_type: The type of label to iterate over. @rtype: Iterator[Tuple[str, np.ndarray]] @return: An iterator over the labels of the specified type.
module

luxonis_ml.data.utils.constants

constant
module

luxonis_ml.data.utils.plot_utils

module

luxonis_ml.data.utils.visualizations

class
ColorMap
A mapping that assigns distinct RGB colors to hashable labels.  The ColorMap class generates and stores distinct colors for any hashable labels. Colors are lazily assigned upon request using a distinct_color_generator.
function
distinct_color_generator(stop: int = -1) -> Generator[RGB, None, None]: Generator[RGB, None, None]
Generate distinct RGB colors using the golden ratio.  This generator produces a sequence of distinct colors in RGB format. The colors are generated by incrementing the hue by the golden ratio and keeping saturation and value fixed. This ensures a wide distribution of visually distinct colors.  @param stop: Optional. The maximum number of colors to generate. If     set to -1 (default), the generator will continue indefinitely. @type stop: int @yield: A tuple representing an RGB color, where each component (R,     G, B) is an integer in the range [0, 255]. @rtype: Generator[tuple[int, int, int], None, None]
function
resolve_color(color: Color) -> RGB: RGB
Resolves a color to an RGB tuple.  @type color: Color @param color: The color to resolve. Can be a string, an integer or a     tuple. @rtype: Tuple[int, int, int] @return: The RGB tuple.
function
rgb_to_hsv(color: Color) -> HSV: HSV
Converts an RGB color to HSV.  @type color: Color @param color: The color to convert. @rtype: Tuple[float, float, float] @return: The HSV tuple.
function
hsv_to_rgb(color: HSV) -> RGB: RGB
Converts an HSV color to RGB.  @type color: Tuple[float, float, float] @param color: The color to convert as an HSV tuple. @rtype: Tuple[int, int, int] @return: The RGB tuple.
function
get_contrast_color(color: Color) -> RGB: RGB
Returns a contrasting color for the given RGB color.  @type color: Color @param color: The color to contrast. @rtype: Tuple[int, int, int] @return: The contrasting color.
function
str_to_rgb(string: str) -> RGB: RGB
Converts a string to its unique RGB color.  @type string: str @param string: The string to convert. @rtype: Tuple[int, int, int] @return: The RGB tuple.
function
draw_dashed_rectangle(image: np.ndarray, pt1: Tuple [ int , int ], pt2: Tuple [ int , int ], color: Color, thickness: int = 1, dash_length: int = 10)
Draws a dashed rectangle on the image.  Adheres to OpenCV's rectangle drawing convention.  @type image: np.ndarray @param image: The image to draw on. @type pt1: Tuple[int, int] @param pt1: The top-left corner of the rectangle. @type pt2: Tuple[int, int] @param pt2: The bottom-right corner of the rectangle. @type color: Color @param color: The color of the rectangle. @type thickness: int @param thickness: The thickness of the rectangle. Default is 1. @type dash_length: int @param dash_length: The length of the dashes. Default is 10.
function
draw_cross(img: np.ndarray, center: Tuple [ int , int ], size: int = 5, color: Color = 0, thickness: int = 1)
Draws a cross on the image.  @type img: np.ndarray @param img: The image to draw on. @type center: Tuple[int, int] @param center: The center of the cross. @type size: int @param size: The size of the cross. Default is 5. @type color: Color @param color: The color of the cross. Default is black. @type thickness: int @param thickness: The thickness of the cross. Default is 1.
function
create_text_image(text: str, width: int, height: int, font_size: float = 0.7, bg_color: Color = 255, text_color: Color = 0) -> np.ndarray: np.ndarray
Creates an image with the given text centered in the image.  @type text: str @param text: The text to display. @type width: int @param width: The width of the image. @type height: int @param height: The height of the image. @type font_size: float @param font_size: The font size of the text. Default is 0.7. @type bg_color: Tuple[int, int, int] @param bg_color: The background color of the image. Default is     white. @type text_color: Tuple[int, int, int] @param text_color: The color of the text. Default is black.
function
concat_images(image_dict: Dict [ str , np.ndarray ], padding: int = 10, label_height: int = 30) -> np.ndarray: np.ndarray
Concatenates images into a single image with labels.  It will attempt to create a square grid of images.  @type image_dict: Dict[str, np.ndarray] @param image_dict: A dictionary mapping image names to images. @type padding: int @param padding: The padding between images. Default is 10. @type label_height: int @param label_height: The height of the label. Default @rtype: np.ndarray @return: The concatenated image.
function
visualize(image: np.ndarray, labels: Labels, classes: Dict [ str , Dict [ str , int ] ], blend_all: bool = False) -> np.ndarray: np.ndarray
Visualizes the labels on the image.  @type image: np.ndarray @param image: The image to visualize. @type labels: Labels @param labels: The labels to visualize. @type class_names: Dict[str, List[str]] @param class_names: A dictionary mapping task names to a list of     class names. @type blend_all: bool @param blend_all: Whether to blend all labels (apart from semantic     segmentations) into a single image. This means mixing labels     belonging to different tasks. Default is False. @rtype: np.ndarray @return: The visualized image.
class

luxonis_ml.data.utils.ParquetFileManager

method
__init__(self, directory: PathType, num_rows: int = 100000)
Manages the insertion of data into parquet files.  @type directory: str @param directory: The local directory in which parquet files are     stored. @type num_rows: int @param num_rows: The maximum number of rows permitted in a     parquet file before another file is created.
variable
variable
variable
variable
variable
variable
variable
method
write(self, uuid: str, data: ParquetRecord)
Writes a row to the current working parquet file.  @type data: Dict @param data: A dictionary representing annotations, mapping     annotation types to values.
method
__enter__(self) -> ParquetFileManager: ParquetFileManager
method
__exit__(self, _)
class

luxonis_ml.data.utils.ParquetRecord(typing.TypedDict)

variable
variable
variable
variable
variable
variable
variable
class

luxonis_ml.data.BucketStorage(enum.Enum)

constant
constant
constant
constant
class

luxonis_ml.data.BucketType(enum.Enum)

constant
constant
class

luxonis_ml.data.ImageType(enum.Enum)

constant
constant
constant
class

luxonis_ml.data.MediaType(enum.Enum)

constant
constant
constant
package

luxonis_ml.embeddings

package
methods
package
utils
package

luxonis_ml.embeddings.methods

module
OOD
Out-of-Distribution Detection for Embeddings.  This module provides two primary methods for detecting out-of-distribution (OOD) samples in embeddings. OOD samples can be crucial to identify as they represent anomalies or novel patterns that don't conform to the expected distribution of the dataset.  Methods available:  Isolation Forests: A tree-based model that partitions the space in such a manner that anomalies are isolated from the rest.  Leverage with Linear Regression: Leverages (or hat values) represent the distance between the predicted values and the true values. Higher leverages indicate potential OOD points.  Typical use cases include:  Anomaly Detection: Identifying rare patterns or outliers.  Dataset Reduction: By removing or studying OOD samples, we can have a more homogeneous dataset.  Expanding Datasets: Recognizing valuable data points that are distinct from the current distribution can be helpful when we're looking to diversify the dataset, especially in iterative learning scenarios.  Dependencies:  numpy  scikit-learn
module
duplicate
Near-duplicate Search with Qdrant and Weaviate.  Overview: This module provides utilities to detect and remove near-duplicate data points within a given set of embeddings. It leverages vector databases (Qdrant or Weaviate) for efficient search and retrieval, and employs Kernel Density Estimation (KDE) for optimal split based on embeddings' cosine similarity. This approach is particularly well-suited for handling high-dimensional embeddings.  Key Features:  Vector Database Integration: Supports both Qdrant and Weaviate for flexible deployment options.  KDE-Based Near-Duplicate Detection: Uses KDE to identify clusters of near-duplicates, ensuring accuracy in high-dimensional spaces.  Visualization: Allows plotting KDE results using matplotlib for intuitive understanding.  Dynamic KDE Peak Selection: Automatically determines the best candidates for removal based on KDE peaks, minimizing manual thresholding.  Dependencies:  KDEpy  Qdrant (optional, for Qdrant-specific features)  Weaviate (optional, for Weaviate-specific features)  Functions:  search_vectordb(vectordb_api, query_vector, property_name, top_k): Searches for similar embeddings within the specified vector database.  _plot_kde(xs, s, density, maxima, minima): Plots a KDE distribution.  kde_peaks(data, bandwidth="scott", plot=False): Identifies peaks in a KDE distribution.  find_similar(reference_embeddings, vectordb_api, k=100, n=1000, method="first", k_method=None, kde_bw="scott", plot=False): Finds the most similar embeddings to the given reference embeddings.  Examples (using Qdrant):  Initialize a Qdrant client and retrieve embeddings: from luxonis_ml.embeddings.utils.qdrant import QdrantAPI qdrant_api = QdrantAPI(host="localhost", port=6333) qdrant_api.create_collection("images", ["image_path", "embedding"]) id_X, X = qdrant_api.get_all_embeddings()  Find similar embeddings using various methods: # By instance ID: ix, paths = find_similar_qdrant(id_X[i], qdrant_api, "image_path", 5, 100, "first")  # Using KDE Peaks method: ix, paths = find_similar_qdrant(X[i], qdrant_api, "image_path", 5, 100, "first", "kde_peaks", "silverman", plot=False)  # Based on average of multiple embeddings: dark_ix = np.array([10, 123, 333, 405]) emb_dark = X[dark_ix] remove_dark_ix, paths = find_similar_qdrant(emb_dark, qdrant_api, "image_path", 25, 5000, "average", "kde_basic", "scott", plot=True)  Additional Notes:  For Weaviate-specific examples, refer to the provided code examples.  The search_vectordb function can be used with either Qdrant or Weaviate, depending on the provided vectordb_api object.  Adjust parameters like k, n, and kde_bw based on your dataset and requirements.
module
mistakes
Mismatch Detection in Labelled Data.  This module provides functionalities to detect mismatches or potential mislabelling in a dataset based on various strategies. This is crucial in supervised machine learning tasks where the quality of labels significantly affects model performance.  Methods implemented  Centroids: This method identifies mismatches by comparing the distance of data points to the centroid of their own class against the distances to centroids of other classes.  KNN (k-Nearest Neighbors): This approach leverages the idea that if the majority of data is correctly labelled, then mislabelled data will be corrected by its nearest neighbors.  [Note: DBSCAN was considered but not implemented due to underperformance.]  Usage  To use this module, import the desired methods and provide the embeddings and labels:  >>> from mismatch_detection import find_mismatches_centroids, find_mismatches_knn >>> # Detect mismatches using centroids >>> mismatches, new_labels = find_mismatches_centroids(X_train, y_train) >>> # Detect mismatches using KNN >>> mismatches, new_labels = find_mismatches_knn(X_train, y_train)
module
representative
Find Representative Images from Embeddings.  This module offers techniques to identify representative images or embeddings within a dataset. This aids in achieving a condensed yet expressive view of your data.  Methods  Greedy Search: Aims to find a diverse subset of images by maximizing the minimum similarity to any image outside the set.  K-Medoids: An adaptation of the k-means clustering algorithm, it partitions data into k clusters, each associated with a medoid.  Main Applications  Dataset Reduction: Helps in representing large datasets with a minimal subset while retaining the essence.  Validation Set Creation: Identifies diverse samples for a robust validation set.  Dependencies  numpy  scikit-learn  kmedoids  luxonis_ml  Example  Greedy Search:  # Assuming you have 'embeddings' as a numpy array of shape (num_images, embedding_dim) similarity_matrix = calculate_similarity_matrix(embeddings) desired_size = int(len(embeddings) * 0.1) selected_image_indices = find_representative_greedy(1-similarity_matrix, desired_size)  K-Medoids:  # to get all embeddings from qdrant: ids, embeddings = get_all_embeddings(qdrant_client, collection_name="mnist") # Assuming you have 'embeddings' as a numpy array of shape (num_images, embedding_dim) similarity_matrix = calculate_similarity_matrix(embeddings) desired_size = int(len(embeddings) * 0.1) selected_image_indices = find_representative_kmedoids(similarity_matrix, desired_size)
function
find_similar(reference_embeddings: Union [ str , List [ str ] , List [ List [ float ] ] , np.ndarray ], vectordb_api: VectorDBAPI, k: int = 100, n: int = 1000, method: str = 'first', k_method: Union [ str , None ] = None, kde_bw: Union [ str , float ] = 'scott', plot: bool = False) -> np.ndarray: np.ndarray
Find the most similar embeddings to the reference embeddings.  @type reference_embeddings: Union[str, List[str], List[List[float]],     np.ndarray] @param reference_embeddings: The embeddings to compare against. Or a     list of of embedding instance_ids that reside in VectorDB. @type vectordb_api: VectorDBAPI @param vectordb_api: The VectorDBAPI instance to use. @type k: int @param k: The number of embeddings to return. Default is 100. @type n: int @param n: The number of embeddings to compare against. Default is     1000. (This is the number of embeddings that are returned by the     VectorDB search. It matters for the KDE, as it can be slow for     large n. Your choice of n depends on the amount of duplicates in     your dataset, the more duplicates, the larger n should be. If     you have 2-10 duplicates per image, n=100 should be ok. If you     have 50-300 duplicates per image, n=1000 should work good     enough. @type method: str @param method: The method to use to find the most similar     embeddings. If 'first' use the first of the reference     embeddings. If 'average', use the average of the reference     embeddings. @type k_method: str @param k_method: The method to select the best k. If None, use k as     is. If 'kde_basic', use the minimum of the KDE. If 'kde_peaks',     use the minimum of the KDE peaks, according to a specific     hardcoded hevristics/thresholds. @type kde_bw: Union[str, float] @param kde_bw: The bandwidth to use for the KDE. Default is 'scott'. @type plot: bool @param plot: Whether to plot the KDE. @rtype: np.array @return: The instance_ids of the most similar embeddings.
function
find_mismatches_centroids(X: np.array, y: np.array) -> Tuple[np.array, np.array]: Tuple[np.array, np.array]
Find mismatches in the dataset. A mismatch is defined as a sample that is closer to another centroid than to its own centroid.  @type X: np.array @param X: The embeddings to use. @type y: np.array @param y: The targets to use. @rtype: Tuple[np.array, np.array] @return: The indices of the mismatches and the new labels.
function
find_mismatches_knn(X: np.array, y: np.array, n_neighbors: int = 5) -> Tuple[np.array, np.array]: Tuple[np.array, np.array]
Find mismatches in the dataset. Single Algorithm Filter (see Figure 1 in Brodley, Carla E., and Mark A. Friedl. "Identifying mislabeled training data."). Idea: if the vast majority of the data is correctly labeled and you do knn prediction, the minority of mislabeled data will be engulfed (corrected) by the correct neighbors.  @type X: np.array @param X: The embeddings to use.  @type y: np.array @param y: The targets to use.  @type n_neighbors: int @param n_neighbors: The number of neighbors to use for KNN. Default is 5.  @rtype: Tuple[np.array, np.array] @return: The indices of the mismatches and the new labels.
function
isolation_forest_OOD(X: np.array, contamination: Union [ float , str ] = 'auto', n_jobs: int = -1, verbose: int = 1, random_state: Optional [ int ] = None) -> np.array: np.array
Out-of-distribution detection using Isolation Forests.  @type X: np.array @param X: The embeddings to use. @type contamination: Union[float, str] @param contamination: The contamination parameter for Isolation     Forests. Default is 'auto'. @type n_jobs: int @param n_jobs: The number of jobs to use. Default is -1, which means     all available CPUs. @type verbose: int @param verbose: The verbosity level. Default is 1. @type random_state: Optional[int] @param random_state: The random state to use. Default is None. @rtype: np.array @return: The indices of the embeddings that are in-distribution.
function
leverage_OOD(X: np.array, std_threshold: int = 3) -> np.array: np.array
Out-of-distribution detection using leverage and linear regression.  @type X: np.array @param X: The embeddings to use. @type std_threshold: int @param std_threshold: The number of standard deviations to use for     the leverage threshold. Default is 3. @rtype: np.array @return: The indices of the embeddings that are out-of-distribution.
function
function
find_representative_greedy(distance_matrix: np.ndarray, desired_size: int = 1000, seed: int = 0) -> List[int]: List[int]
Find the most representative images using a greedy algorithm. Gready search of maximally unique embeddings.  @type distance_matrix: np.array @param distance_matrix: The distance matrix to use. @type desired_size: int @param desired_size: The desired size of the representative set.     Default is 1000. @type seed: int @param seed: The index of the seed image. Default is 0. Must be in     the range [0, num_images-1]. @rtype: List[int] @return: The indices of the representative images.
function
find_representative_kmedoids(similarity_matrix: np.ndarray, desired_size: int = 1000, max_iter: int = 100, seed: int = None) -> List[int]: List[int]
Find the most representative images using k-medoids. K-medoids clustering of embeddings.  @type similarity_matrix: np.array @param similarity_matrix: The similarity matrix to use. @type desired_size: int @param desired_size: The desired size of the representative set.     Default is 1000. @type max_iter: int @param max_iter: The maximum number of iterations to use. Default is     100. @type seed: int @param seed: The random seed to use. Default is None. @rtype: list @return: The indices of the representative images.
module

luxonis_ml.embeddings.methods.duplicate

function
kde_peaks(data: np.ndarray, bandwidth: Union [ str , float ] = 'scott', plot: bool = False) -> Tuple[np.ndarray, np.ndarray, int, float]: Tuple[np.ndarray, np.ndarray, int, float]
Find peaks in a KDE distribution using scipy's argrelextrema function.  @type data: np.ndarray @param data: The data to fit the KDE. @type bandwidth: Union[str, float] @param bandwidth: The bandwidth to use for the KDE. Default is     'scott'. @type plot: bool @param plot: Whether to plot the KDE. @rtype: Tuple[np.ndarray, np.ndarray, int, float] @return: The indices of the KDE maxima, the indices of the KDE     minima, the index of the global maxima, and the standard     deviation of the data.
module

luxonis_ml.embeddings.methods.representative

function
find_representative_greedy_vectordb(vectordb_api: VectorDBAPI, desired_size: int = 1000, seed: int = None) -> List[int]: List[int]
Find the most representative embeddings using a greedy algorithm with VectorDB.  @note: Due to many requests, this function is very slow. Use     vectordb_api.retrieve_all_embeddings() and     find_representative_greedy() instead. @type vectordb_api: VectorDBAPI @param vectordb_api: The Vector database client instance to use for     searches. @type desired_size: int @param desired_size: The desired size of the representative set.     Default is 1000. @type seed: int @param seed: The ID of the seed embedding. Default is None, which     means a random seed is chosen. @rtype: List[int] @return: The IDs of the representative embeddings.
package

luxonis_ml.embeddings.utils

module
embedding
Embeddings Extractor and Storage for ONNX Models.  This module provides utility functions specifically for extracting embeddings from ONNX models, reading images from Luxonis Filesystem (LFS), and storing/retrieving embeddings to/from disk.  Key Functions:  extract_embeddings(image_paths, ort_session, lfs, ...) Extracts embeddings from an ONNX model, reading images from LFS and handling potential errors.  get_image_tensors_from_LFS(image_paths, preprocess_function, lfs) Reads images from LFS, applies preprocessing, and returns a list of tensors.  preprocess_image_cv2(img) Example preprocessing function for resizing, normalization, and channel arrangement.  Workflow:  Load the ONNX model into an ONNX Runtime session.  Provide a list of image paths (stored on LFS) to the extract_embeddings function.  Optionally, specify a custom preprocessing function for image preparation.  The function extracts embeddings in batches, handles errors, and returns the extracted embeddings.  Additional Features:  Saving and loading embeddings to/from disk (functions not yet implemented in this version).  Dependencies:  onnxruntime  cv2  numpy  Note:  Ensure the output_layer_name in extract_embeddings matches the appropriate output layer in the ONNX model.  This module specifically focuses on ONNX models and reading images from Luxonis Filesystem.
module
ldf
Utilities for generating image embeddings and inserting them into a VectorDB database.  This script provides functions for:  Extracting payloads from LuxonisDatasets, specifically for classification datasets.  Filtering new samples based on their instance IDs to avoid duplicates in the database.  Generating embeddings for new images using an ONNX runtime session.  Performing batch upserts of embeddings into a VectorDB database.  Key modules used:  luxonis_ml.data: For loading and working with LuxonisDatasets.  luxonis_ml.embeddings.utils.embedding: For extracting embeddings from images.  luxonis_ml.embeddings.utils.vectordb: For interacting with VectorDB databases.  Main functions:  _get_sample_payloads_LDF: Extracts payloads from a LuxonisDataset for classification datasets.  _filter_new_samples_by_id: Filters out samples already in the database based on instance IDs.  _batch_upsert: Performs batch upserts of embeddings into the database.  generate_embeddings: Main function to generate embeddings for a dataset and insert them into the database.  Important note:  Ensure that a VectorDB server is running and accessible before using these utilities.
module
model
Model Utility Functions.  This script provides utility functions for handling ONNX models. It allows manipulating ONNX models in order to extract intermediate outputs aka embeddings.  Functions:  load_model_onnx(model_path: str = "resnet50.onnx") -> onnx.ModelProto: Loads an ONNX model from the specified file path.  save_model_onnx(model: onnx.ModelProto, model_path_out: str = "resnet50.onnx"): Saves an ONNX model to the specified file path.  extend_output_onnx(onnx_model: onnx.ModelProto, intermediate_tensor_name: str = "/Flatten_output_0") -> onnx.ModelProto: Sets an intermediate output layer as output of the provided ONNX model. If overwrite is set to True, the second to last layer output will be set as output layer and renamed.  Dependencies:  onnx
module
qdrant
Qdrant Docker Management and Embedding Operations.  This script provides utility functions for managing Qdrant via Docker and performing operations related to embeddings.  Features include:  Docker management: Checks for Docker installation, running status, and existence of specific images or containers.  Qdrant management: Facilitates starting a Qdrant container, connecting to Qdrant, and creating collections.  Embedding operations: Supports inserting, batch inserting, searching, and retrieving embeddings within a Qdrant collection.  Dependencies:  os: Standard library module for interacting with the operating system.  docker: Library to manage Docker containers.  qdrant_client: Client library for interacting with the Qdrant service.  Usage steps:  Ensure Docker is installed and running on your system.  Utilize the QdrantManager class and its start_docker_qdrant() method to initiate a Qdrant container.  Employ the QdrantAPI class to either connect to an existing Qdrant service or create a new collection.  Use the QdrantAPI class to perform various embedding-related operations within the specified Qdrant collection.  Note:  The default collection name is 'mnist'.  It is essential that the user has appropriate permissions to execute Docker commands without sudo.  For guidance on setting this up, refer to: https://docs.docker.com/engine/install/linux-postinstall/
module
module
function
function
generate_embeddings(luxonis_dataset: LuxonisDataset, ort_session: onnxruntime.InferenceSession, vectordb_api: VectorDBAPI, output_layer_name: str = '/Flatten_output_0', transform: Callable [ [ np.ndarray ] , np.ndarray ] = None, emb_batch_size: int = 64, vectordb_batch_size: int = 64) -> Dict[str, List[float]]: Dict[str, List[float]]
Generate embeddings for a given dataset and insert them into a VectorDB.  @type luxonis_dataset: L{LuxonisDataset} @param luxonis_dataset: The dataset object. @type ort_session: L{InferenceSession} @param ort_session: ONNX runtime session. @type vectordb_api: L{VectorDBAPI} @param vectordb_api: VectorDBAPI instance. @type output_layer_name: str @param output_layer_name: Name of the output layer in the ONNX     model. @type transform: Callable[[np.ndarray], np.ndarray] @param transform: Preprocessing function for images. If None,     default preprocessing is used. @type emb_batch_size: int @param emb_batch_size: Batch size for generating embeddings. @type vectordb_batch_size: int @param vectordb_batch_size: Batch size for inserting into a vector     DB. @type: Dict[str, List[float]] @return: Dictionary of instance ID to embedding.
function
extend_output_onnx(onnx_model: onnx.ModelProto, intermediate_tensor_name: str = '/Flatten_output_0', overwrite: bool = False) -> onnx.ModelProto: onnx.ModelProto
Set an intermediate output layer as output of the provided ONNX model.  If C{overwrite} is set to True, the second to last layer output will be set as output layer and renamed.  (You need to know the name of the intermediate layer, which you can find by inspecting the ONNX model with Netron.app)
function
function
class
QdrantAPI
Class to perform various Qdrant operations related to embeddings.
class
QdrantManager
Class to manage Qdrant Docker container and perform various operations related to embeddings.
class
VectorDBAPI
Abstract class for Vector Database APIs.  This class defines a common interface for vector database operations for different implementations like Qdrant and Weaviate.
class
WeaviateAPI
Provides a Python interface for interacting with Weaviate, facilitating operations such as creating collections, managing embeddings, and querying for similar embeddings.  It only supports cosine similarity for now.
module

luxonis_ml.embeddings.utils.embedding

function
preprocess_image_cv2(img: np.ndarray) -> np.ndarray: np.ndarray
Custom preprocessing function provided by the user.
function
module

luxonis_ml.embeddings.utils.ldf

class

luxonis_ml.embeddings.utils.QdrantAPI(luxonis_ml.embeddings.utils.VectorDBAPI)

method
__init__(self, host: str = 'localhost', port: int = 6333)
Initialize the QdrantAPI without setting a specific collection.  @type host: str @param host: The host address of the Qdrant server. Default is     "localhost". @type port: int @param port: The port number of the Qdrant server. Default is     6333.
variable
method
create_collection(self, collection_name: str, properties: List [ str ], vector_size: int = 512)
Create a collection in Qdrant with specified properties.  @type collection_name: str @param collection_name: The name of the collection. @type properties: List[str] @param properties: The list of properties for the collection. @type vector_size: int @param vector_size: The size of the embedding vectors. Default     is 512.
variable
variable
method
delete_collection(self)
Delete a collection in Qdrant.
method
insert_embeddings(self, ids: List [ str ], embeddings: List [ List [ float ] ], payloads: List [ Dict [ str , Any ] ], batch_size: int = 50)
Batch insert embeddings with IDs and additional metadata into a collection.  @type ids: List[str] @param ids: The list of instance_ids for the embeddings. @type embeddings: List[List[float]] @param embeddings: The list of embedding vectors. @type payloads: List[Dict[str, Any]] @param payloads: The list of additional metadata for the     embeddings. @type batch_size: int @param batch_size: The batch size for inserting embeddings.     Default is 50.
method
search_similar_embeddings(self, embedding: List [ float ], top_k: int = 5) -> Tuple[List[str], List[float]]: Tuple[List[str], List[float]]
Search for the top similar embeddings in a Qdrant collection.  @type embedding: List[float] @param embedding: The query embedding vector. @type top_k: int @param top_k: The number of similar embeddings to retrieve.     Default is 5. @rtype: Tuple[List[str], List[float]] @return: The list of instance_ids of the similar embeddings and     the list of similarity scores.
method
get_similarity_scores(self, reference_id: str, other_ids: List [ str ], sort_distances: bool = True) -> Tuple[List[str], List[float]]: Tuple[List[str], List[float]]
Get a list of similarity scores between the reference embedding and other embeddings.  @type reference_id: int @param reference_id: The instance_id of the reference embedding. @type other_ids: List[int] @param other_ids: The list of instance_ids of other embeddings     to compare with the reference. @type sort_distances: bool @param sort_distances: Whether to sort the results by distance     or keep the original order. @rtype: Tuple[List[int], List[float] @return: The list of instance_ids of the other embeddings and     the list of similarity scores.
method
compute_similarity_matrix(self) -> List[List[float]]: List[List[float]]
Compute a full similarity matrix for all embeddings in a Qdrant collection.  @rtype: Tuple[List[str], List[List[float]]] @return: The list of instance_ids of the embeddings and the     similarity matrix. @note: This method is not recommended for large collections. It     is better to use the L{get_all_embeddings} method and     compute the similarity matrix yourself.
method
retrieve_embeddings_by_ids(self, ids: List [ str ]) -> List[List[float]]: List[List[float]]
Retrieve embeddings associated with a list of IDs from a Qdrant collection. The order of the embeddings IS preserved.  @type ids: List[str] @param ids: The list of instance_ids of the embeddings to     retrieve. @rtype: List[List[float]] @return: The list of embedding vectors.
method
retrieve_all_ids(self) -> List[str]: List[str]
Retrieve all IDs from a Qdrant collection.  @rtype: List[str] @return: The list of instance_ids of the embeddings.
method
retrieve_all_embeddings(self) -> Tuple[List[str], List[List[float]]]: Tuple[List[str], List[List[float]]]
Retrieve all embeddings and their IDs from a Qdrant collection.  @rtype: Tuple[List[str], List[List[float]]] @return: The list of instance_ids of the embeddings and the list     of embedding vectors.
method
retrieve_payloads_by_ids(self, ids: List [ str ], properties: Optional [ List [ str ] ] = None) -> List[Dict[str, Any]]: List[Dict[str, Any]]
Retrieve specified payload properties for a list of IDs from a collection. The order of the labels IS preserved.  @type ids: List[str] @param ids: The list of instance_ids of the embeddings to     retrieve. @type properties: Optional[List[str]] @param properties: The list of payload properties to retrieve.     Default is None. @rtype: List[Dict[str, Any]] @return: The list of payload dictionaries.
class

luxonis_ml.embeddings.utils.QdrantManager

method
__init__(self, image_name = 'qdrant/qdrant', container_name = 'qdrant_container')
Initialize the QdrantManager.
variable
variable
variable
method
is_docker_installed(self)
Check if Docker is installed.
method
is_docker_running(self)
Check if Docker daemon is running.
method
does_image_exist(self)
Check if a Docker image exists.
method
does_container_exist(self)
Check if a Docker container exists.
method
is_container_running(self)
Check if a Docker container is running.
method
start_docker_qdrant(self)
Start the Qdrant Docker container.  @note: Make sure the user has the appropriate permissions to run Docker commands     without sudo. Otherwise, the client_docker.images.pull() command will fail.     See U{https://docs.docker.com/engine/install/linux-postinstall/} for more details.
method
stop_docker_qdrant(self)
Stop the Qdrant Docker container.
class

luxonis_ml.embeddings.utils.VectorDBAPI(abc.ABC)

method
method
delete_collection(self)
Delete the collection from the vector database.
method
method
method
method
compute_similarity_matrix(self) -> List[List[float]]: List[List[float]]
Compute a similarity matrix for all the embeddings in the collection.
method
method
method
retrieve_all_ids(self) -> List[str]: List[str]
Retrieve all IDs from the collection.
method
class

luxonis_ml.embeddings.utils.WeaviateAPI(luxonis_ml.embeddings.utils.VectorDBAPI)

method
__init__(self, url: str = 'http://localhost:8080', grpc_url: str = 'http://localhost:50051', auth_api_key: str = None)
Initializes the Weaviate API client with connection details.  @type url: str @param url: URL of the Weaviate instance, defaults to     U{localhost:8080}. @type grpc_url: str @param grpc_url: URL of the gRPC Weaviate instance, defaults to     U{localhost:50051}. @type auth_api_key: str @param auth_api_key: API key for authentication. Defaults to     C{None}.
variable
method
create_collection(self, collection_name: str, properties: List [ str ] = None)
Creates a new collection in the Weaviate database.  @type collection_name: str @param collection_name: Name of the collection to create. @type properties: List[str] @param properties: List of properties for the collection.     Defaults to None.
variable
variable
variable
method
delete_collection(self)
Deletes a collection from the Weaviate database.
method
insert_embeddings(self, uuids: List [ str ], embeddings: List [ List [ float ] ], payloads: List [ Dict [ str , Any ] ], batch_size: int = 100)
Inserts embeddings with associated payloads into a collection.  @type uuids: List[str] @param uuids: List of UUIDs for the embeddings. @type embeddings: List[List[float]] @param embeddings: List of embeddings. @type payloads: List[Dict[str, Any]] @param payloads: List of payloads. @type batch_size: int @param batch_size: Batch size for inserting the embeddings.
method
search_similar_embeddings(self, embedding: List [ float ], top_k: int = 10) -> Tuple[List[str], List[float]]: Tuple[List[str], List[float]]
Searches for embeddings similar to a given vector.  @type embedding: List[float] @param embedding: Embedding to find similar embeddings for. @type top_k: int @param top_k: Number of similar embeddings to find. @rtype uuids: List[str] @return uuids: List of UUIDs of the similar embeddings. @rtype scores: List[float] @return scores: List of similarity scores.
method
get_similarity_scores(self, reference_id: str, other_ids: List [ str ], sort_distances: bool = False) -> Tuple[List[str], List[float]]: Tuple[List[str], List[float]]
Calculates the similarity score between the reference embedding and the specified embeddings.  @type reference_id: str @param reference_id: UUID of the reference embedding. @type other_ids: List[str] @param other_ids: List of UUIDs of the embeddings to compare to. @type sort_distances: bool @param sort_distances: Whether to sort the results by distance     or keep order of the UUIDs. Defaults to False. @rtype ids: List[str] @return ids: List of UUIDs of the embeddings. @rtype scores: List[float] @return scores: List of similarity scores.
method
compute_similarity_matrix(self) -> List[List[float]]: List[List[float]]
Calculates the similarity matrix for all the embeddings in the collection. @note: This is a very inefficient implementation. For large numbers of embeddings, calculate the similarity matrix by hand (sklearn.metrics.pairwise.cosine_similarity).  @rtype sim_matrix: List[List[float]] @return sim_matrix: Similarity matrix for all the embeddings in     the collection.
method
retrieve_embeddings_by_ids(self, uuids: List [ str ]) -> List[List[float]]: List[List[float]]
Gets the embeddings for the specified UUIDs, up to a maximum of 10000.  @type uuids: List[str] @param uuids: List of UUIDs of the embeddings to get. @rtype embeddings: List[List[float]] @return embeddings: List of embeddings.
method
retrieve_payloads_by_ids(self, uuids: List [ str ], properties: Optional [ List [ str ] ] = None) -> List[Dict[str, Any]]: List[Dict[str, Any]]
Gets the payloads for the specified UUIDs, up to a maximum of 10000.  @type uuids: List[str] @param uuids: List of UUIDs of the embeddings to get. @type properties: List[str] @param properties: List of properties to retrieve. @rtype payloads: List[Dict[str, Any]] @return payloads: List of payloads.
method
retrieve_all_ids(self) -> List[str]: List[str]
Gets all the UUIDs in the Weaviate collection.  @rtype uuids: List[str] @return uuids: List of UUIDs.
method
retrieve_all_embeddings(self) -> Tuple[List[str], List[List[float]]]: Tuple[List[str], List[List[float]]]
Gets all the embeddings and UUIDs in the Weaviate collection.  @rtype uuids: List[str] @return uuids: List of UUIDs. @rtype embeddings: List[List[float]] @return embeddings: List of embeddings.
package

luxonis_ml.enums

module
class
DatasetType
class

luxonis_ml.enums.DatasetType(str, enum.Enum)

constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
module

luxonis_ml.guard_extras

function
package

luxonis_ml.nn_archive

module
module
config
package
config_building_blocks
module
module
class
ArchiveGenerator
Generator of abstracted NN archive (.tar) files containing config and model files (executables).
class
Config
The main class of the multi/single-stage model config scheme (multi- stage models consists of interconnected single-stage models).
class
Model
Class defining a single-stage model config scheme.
function
infer_layout(shape: List [ int ]) -> str: str
Infers a layout for the given shape.  Tries to guess most common layouts for the given shape pattern. Otherwise, uses the first free letter of the alphabet for each dimension.  Example::     >>> make_default_layout([1, 3, 256, 256])     >>> "NCHW"     >>> make_default_layout([1, 19, 7, 8])     >>> "NABC"
function
is_nn_archive(path: PathType) -> bool: bool
Check if the given path is a valid NN archive file.  @type path: PathType @param path: Path to the file to check. @rtype: bool @return: True if the file is a valid NN archive file, False     otherwise.
module

luxonis_ml.nn_archive.config

constant
package

luxonis_ml.nn_archive.config_building_blocks

package
base_models
package
enums
package

luxonis_ml.nn_archive.config_building_blocks.base_models

module
module
head_metadata
module
module
module
class
Head
Represents head of a model.
class
HeadMetadata
Metadata for the basic head. It allows you to specify additional fields.
class
Input
Represents input stream of a model.
class
PreprocessingBlock
Represents preprocessing operations applied to the input data.
class
Metadata
Represents metadata of a model.
class
Output
Represents output stream of a model.
module

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata

class
HeadObjectDetectionMetadata
Metadata for the object detection head.
class
HeadObjectDetectionSSDMetadata
Metadata for the SSD object detection head.
class
HeadClassificationMetadata
Metadata for the classification head.
class
HeadSegmentationMetadata
Metadata for the segmentation head.
class
HeadYOLOMetadata
Metadata for the YOLO head.
class

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadObjectDetectionMetadata(luxonis_ml.nn_archive.config_building_blocks.base_models.HeadMetadata)

variable
classes
Names of object classes detected by the model.
variable
n_classes
Number of object classes detected by the model.
variable
iou_threshold
Non-max supression threshold limiting boxes intersection.
variable
conf_threshold
Confidence score threshold above which a detected object is considered valid.
variable
max_det
Maximum detections per image.
variable
anchors
Predefined bounding boxes of different sizes and aspect ratios. The innermost lists are length 2 tuples of box sizes. The middle lists are anchors for each output. The outmost lists go from smallest to largest output.
class

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadObjectDetectionSSDMetadata(luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadObjectDetectionMetadata)

variable
boxes_outputs
Output name corresponding to predicted bounding box coordinates.
variable
scores_outputs
Output name corresponding to predicted bounding box confidence scores.
class

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadClassificationMetadata(luxonis_ml.nn_archive.config_building_blocks.base_models.HeadMetadata)

variable
classes
Names of object classes classified by the model.
variable
n_classes
Number of object classes classified by the model.
variable
is_softmax
True, if output is already softmaxed
class

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadSegmentationMetadata(luxonis_ml.nn_archive.config_building_blocks.base_models.HeadMetadata)

variable
classes
Names of object classes segmented by the model.
variable
n_classes
Number of object classes segmented by the model.
variable
is_softmax
True, if output is already softmaxed
class

luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadYOLOMetadata(luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadObjectDetectionMetadata, luxonis_ml.nn_archive.config_building_blocks.base_models.head_metadata.HeadSegmentationMetadata)

variable
yolo_outputs
A list of output names for each of the different YOLO grid sizes.
variable
mask_outputs
A list of output names for each mask output.
variable
protos_outputs
Output name for the protos.
variable
keypoints_outputs
A list of output names for the keypoints.
variable
angles_outputs
A list of output names for the angles.
variable
subtype
YOLO family decoding subtype (e.g. yolov5, yolov6, yolov7 etc.)
variable
n_prototypes
Number of prototypes per bbox in YOLO instance segmnetation.
variable
n_keypoints
Number of keypoints per bbox in YOLO keypoint detection.
variable
is_softmax
True, if output is already softmaxed in YOLO instance segmentation
CLASS_METHOD
class

luxonis_ml.nn_archive.config_building_blocks.base_models.Head(pydantic.BaseModel, abc.ABC)

variable
name
Optional name of the head.
variable
parser
Name of the parser responsible for processing the models output.
variable
outputs
Specify which outputs are fed into the parser. If None, all outputs are fed.
variable
metadata
Metadata of the parser.
class

luxonis_ml.nn_archive.config_building_blocks.base_models.HeadMetadata(pydantic.BaseModel)

variable
postprocessor_path
Path to the postprocessor.
variable
class

luxonis_ml.nn_archive.config_building_blocks.base_models.Input(luxonis_ml.utils.BaseModelExtraForbid)

variable
name
Name of the input layer.
variable
dtype
Data type of the input data (e.g., 'float32').
variable
input_type
Type of input data (e.g., 'image').
variable
shape
Shape of the input data as a list of integers (e.g. [H,W], [H,W,C], [N,H,W,C], ...).
variable
layout
Lettercode interpretation of the input data dimensions (e.g., 'NCHW').
variable
preprocessing
Preprocessing steps applied to the input data.
method
static method
class

luxonis_ml.nn_archive.config_building_blocks.base_models.PreprocessingBlock(luxonis_ml.utils.BaseModelExtraForbid)

variable
mean
Mean values in channel order. Order depends on the order in which the model was trained on.
variable
scale
Standardization values in channel order. Order depends on the order in which the model was trained on.
variable
reverse_channels
If True input to the model is RGB else BGR.
variable
interleaved_to_planar
If True input to the model is interleaved (NHWC) else planar (NCHW).
variable
dai_type
DepthAI input type which is read by DepthAI to automatically setup the pipeline.
class

luxonis_ml.nn_archive.config_building_blocks.base_models.Metadata(pydantic.BaseModel)

variable
name
Name of the model.
variable
path
Relative path to the model executable.
variable
class

luxonis_ml.nn_archive.config_building_blocks.base_models.Output(luxonis_ml.utils.BaseModelExtraForbid)

variable
name
Name of the output layer.
variable
dtype
Data type of the output data (e.g., 'float32').
variable
variable
method
method
package

luxonis_ml.nn_archive.config_building_blocks.enums

module
module
class
DataType
Represents all existing data types used in i/o streams of the model.
class
InputType
Represents a type of input the model is expecting.
class

luxonis_ml.nn_archive.config_building_blocks.enums.DataType(enum.Enum)

constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
class

luxonis_ml.nn_archive.config_building_blocks.enums.InputType(enum.Enum)

constant
constant
class

luxonis_ml.nn_archive.ArchiveGenerator

variable
archive_name
Desired archive file name.
variable
save_path
Path to where we want to save the archive file.
variable
cfg_dict
Archive configuration dict.
variable
executables_paths
Paths to relevant model executables.
variable
compression
Type of archive file compression ("xz" for LZMA, "gz" for gzip, or "bz2" for bzip2 compression).
method
__init__(self, archive_name: str, save_path: PathType, cfg_dict: dict, executables_paths: List [ PathType ], compression: Literal [ ' xz ' , ' gz ' , ' bz2 ' ] = 'xz')
variable
method
make_archive(self) -> Path: Path
Run NN archive (.tar) file generation.
class

luxonis_ml.nn_archive.Config(luxonis_ml.utils.BaseModelExtraForbid)

variable
config_version
String representing config schema version in format 'x.y' where x is major version and y is minor version
variable
model
A Model object representing the neural network used in the archive.
CLASS_METHOD
class

luxonis_ml.nn_archive.Model(luxonis_ml.utils.BaseModelExtraForbid)

variable
metadata
Metadata object defining the model metadata.
variable
inputs
List of Input objects defining the model inputs.
variable
outputs
List of Output objects defining the model outputs.
variable
heads
List of Head objects defining the model heads. If not defined, we assume a raw output.
package

luxonis_ml.tracker

module
mlflow_plugins
module
class
LuxonisRequestHeaderProvider
class
LuxonisTracker
module

luxonis_ml.tracker.mlflow_plugins

class

luxonis_ml.tracker.LuxonisRequestHeaderProvider(mlflow.tracking.request_header.abstract_request_header_provider.RequestHeaderProvider)

method
method
class

luxonis_ml.tracker.LuxonisTracker

method
__init__(self, project_name: Optional [ str ] = None, project_id: Optional [ str ] = None, run_name: Optional [ str ] = None, run_id: Optional [ str ] = None, save_directory: PathType = 'output', is_tensorboard: bool = False, is_wandb: bool = False, is_mlflow: bool = False, is_sweep: bool = False, wandb_entity: Optional [ str ] = None, mlflow_tracking_uri: Optional [ str ] = None, rank: int = 0)
Implementation of PytorchLightning Logger that wraps various logging software. Supported loggers: TensorBoard, WandB and MLFlow.  @type project_name: Optional[str] @param project_name: Name of the project used for WandB and MLFlow.     Defaults to None.  @type project_id: Optional[str] @param project_id: Project id used for WandB and MLFlow.     Defaults to None.  @type run_name: Optional[str] @param run_name: Name of the run, if None then auto-generate random name.     Defaults to None.  @type run_id: Optional[str] @param run_id: Run id used for continuing MLFlow run.     Defaults to None.  @type save_directory: str @param save_directory: Path to save directory.     Defaults to "output".  @type is_tensorboard: bool @param is_tensorboard: Wheter use TensorBoard logging.     Defaults to False.  @type is_wandb: bool @param is_wandb: Wheter use WandB logging.     Defaults to False.  @type is_mlflow: bool @param is_mlflow: Wheter use MLFlow logging.     Defaults to False.  @type is_sweep: bool @param is_sweep: Wheter current run is part of a sweep.     Defaults to False.  @type wandb_entity: Optional[str] @param wandb_entity: WandB entity to use.     Defaults to None.  @type mlflow_tracking_uri: Optional[str] @param mlflow_tracking_uri: MLFlow tracking uri to use.     Defaults to None.  @type rank: int @param rank: Rank of the process, used when running on multiple threads.     Defaults to 0.
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
static method
LuxonisTracker.rank_zero_only(fn: Callable) -> Callable: Callable
Function wrapper that lets only processes with rank=0 execute it.
method
log_to_mlflow(self, log_fn: Callable, args, kwargs)
Attempts to log to MLflow, with retries.  Logs locally if failures persist.
method
store_log_locally(self, log_fn: Callable, args, kwargs)
Stores log data locally if logging to MLflow fails.
method
log_stored_logs_to_mlflow(self)
Attempts to log any data stored in local_logs to MLflow.
method
save_logs_locally(self)
Saves metrics, parameters, images, artifacts, and matrices locally.
property
name
Returns run name.
property
version
Returns tracker's version.
property
experiment
Creates new experiments or returns active ones if already created.
method
log_hyperparams(self, params: Dict [ str , Union [ str , bool , int , float , None ] ])
Logs hyperparameter dictionary.  @type params: Dict[str, Union[str, bool, int, float, None]] @param params: Dict of hyperparameters key-value pairs.
method
log_metric(self, name: str, value: float, step: int)
Logs metric value with name and step.  @note: step is ommited when logging with wandb to avoid problems     with inconsistent incrementation. @type name: str @param name: Metric name @type value: float @param value: Metric value @type step: int @param step: Current step
method
log_metrics(self, metrics: Dict [ str , float ], step: int)
Logs metric dictionary.  @type metrics: Dict[str, float] @param metrics: Dict of metric key-value pairs @type step: int @param step: Current step
method
log_image(self, name: str, img: np.ndarray, step: int)
Logs image with name and step. Note: step is omitted when logging with wandb is used to avoid problems with inconsistent incrementation.  @type name: str @param name: Caption of the image @type img: np.ndarray @param img: Image data @type step: int @param step: Current step
method
upload_artifact(self, path: PathType, name: Optional [ str ] = None, typ: str = 'artifact')
Uploads artifact to the logging service.  @type path: PathType @param path: Path to the artifact @type name: Optional[str] @param name: Name of the artifact, if None then use the name of     the file @type typ: str @param typ: Type of the artifact, defaults to "artifact". Only     used for WandB.
method
upload_artifact_to_mlflow(self, path: PathType, name: Optional [ str ] = None)
Uploads artifact specifically to MLflow.  @type path: PathType @param path: Path to the artifact @type name: Optional[str] @param name: Name of the artifact, if None then use the name of     the file
method
log_matrix(self, matrix: np.ndarray, name: str, step: int)
Logs matrix to the logging service.  @type matrix: np.ndarray @param matrix: The matrix to log. @type name: str @param name: The name used to log the matrix. @type step: int @param step: The current step.
method
log_images(self, imgs: Dict [ str , np.ndarray ], step: int)
Logs multiple images.  @type imgs: Dict[str, np.ndarray] @param imgs: Dict of image key-value pairs where key is image     caption and value is image data @type step: int @param step: Current step
method
close(self)
Finalizes logging and saves unsent logs locally.
module

luxonis_ml.typing

type alias
PathType: TypeAlias
A string or a `pathlib.Path` object.
type alias
PosixPathType: TypeAlias
A string or a `pathlib.PurePosixPath` object.
type alias
type alias
Labels: TypeAlias
Dictionary mappping task names to the annotations as C{np.ndarray}
type alias
LoaderOutput: TypeAlias
C{LoaderOutput} is a tuple of an image as a C{np.ndarray} and a dictionary of task group names and their annotations as L{Annotations}.
type alias
type alias
type alias
Color: TypeAlias
Color type alias.  Can be either a string (e.g. "red", "#FF5512"),  a tuple of RGB values, or a single value (in which case it is interpreted as a grayscale value).
type alias
PrimitiveType: TypeAlias
Primitive types in Python.
type alias
type alias
Params: TypeAlias
A keyword dictionary of additional parameters.  Usually loaded from a YAML file.
type alias
Kwargs: TypeAlias
A keyword dictionary of arbitrary parameters.
class
ConfigItem
Configuration schema for dynamic object instantiation. Typically used to instantiate objects stored in registries.  A dictionary with a name and a dictionary of parameters.
type variable
function
check_type(value: Any, type_: Type [ T ]) -> TypeGuard[T]: TypeGuard[T]
Checks if the value has the correct type.  @type value: Any @param value: The value to check. @type type_: Type[K] @param type_: The type to check against. @rtype: bool @return: C{True} if the value has the correct type, C{False}     otherwise.
function
all_not_none(values: Iterable [ Any ]) -> bool: bool
Checks if none of the values in the iterable is C{None}  @type values: Iterable[Any] @param values: An iterable of values @rtype: bool @return: C{True} if all values are not C{None}, C{False} otherwise
function
any_not_none(values: Iterable [ Any ]) -> bool: bool
Checks if at least one value in the iterable is not C{None}  @type values: Iterable[Any] @param values: An iterable of values @rtype: bool @return: C{True} if at least one value is not C{None}, C{False}     otherwise
class

luxonis_ml.typing.ConfigItem(pydantic.BaseModel)

variable
name
The name of the object this configuration applies to. Required.
variable
params
Additional parameters for instantiating the object. Not required.
package

luxonis_ml.utils

module
config
module
environ
module
filesystem
module
graph
module
logging
module
module
registry
module
class
LuxonisConfig
Class for storing configuration.
constant
class
LuxonisFileSystem
function
is_acyclic(graph: Dict [ str , List [ str ] ]) -> bool: bool
Tests if graph is acyclic.  @type graph: dict[str, list[str]] @param graph: Graph in a format of a dictionary of predecessors. @rtype: bool @return: True if graph is acyclic, False otherwise.
function
traverse_graph(graph: Dict [ str , List [ str ] ], nodes: Mapping [ str , T ]) -> Iterator[Tuple[str, T, List[str], List[str]]]: Iterator[Tuple[str, T, List[str], List[str]]]
Traverses the graph in topological order, starting from the nodes with no predecessors.  Example:      >>> graph = {"a": ["b"], "b": []}     >>> nodes = {"a": 1, "b": 2}     >>> for name, value, preds, rem in traverse_graph(graph, nodes):     ...     print(name, value, preds, rem)     b 2 [] ['a']     a 1 ['b'] []  @type graph: dict[str, list[str]] @param graph: Graph in a format of a dictionary of predecessors.     Keys are node names, values are node predecessors (list of node     names). @type nodes: dict[str, T] @param nodes: Dictionary mapping node names to values. @rtype: Iterator[tuple[str, T, list[str], list[str]]] @return: Iterator of tuples containing node name, node value, node     predecessors, and remaining unprocessed nodes. @raises RuntimeError: If the graph is malformed.
function
deprecated(args: str, suggest: Optional [ Dict [ str , str ] ] = None, additional_message: Optional [ str ] = None, altogether: bool = False) -> Callable[[Callable], Callable]: Callable[[Callable], Callable]
Decorator to mark a function or its parameters as deprecated.  Example:      >>> @deprecated("old_arg",     ...             "another_old_arg",     ...             suggest={"old_arg": "new_arg"},     ...             additional_message="Usage of 'old_arg' is discouraged.")     ...def my_func(old_arg, another_old_arg, new_arg=None):     ...   pass     >>> my_func("foo")     >>> # DeprecationWarning: Argument 'old_arg'     ... # in function `my_func` is deprecated and     ... # will be removed in future versions.     ... # Use 'new_arg' instead.     ... # Usage of 'old_arg' is discouraged.  @type args: str @param args: The names of the deprecated parameters. @type suggest: Dict[str, str] @param suggest: Suggested replacement parameters. @type additional_message: str @param additional_message: Additional message to display.     If provided, it will be appended to the warning message. @type altogether: bool @param altogether: If True, the whole function is     marked as deprecated. Defaults to False.
function
log_once(logger: Callable [ [ str ] , None ], message: str)
Logs a message only once.  @type logger: Logger @param logger: The logger to use. @type message: str @param message: The message to log.
function
setup_logging(level: Optional [ Literal [ ' DEBUG ' , ' INFO ' , ' WARNING ' , ' ERROR ' , ' CRITICAL ' ] ] = None, file: Optional [ PathType ] = None, kwargs)
Sets up global logging using loguru and rich.  @type level: Optional[str] @param level: Logging level. If not set, reads from the environment     variable C{LOG_LEVEL}. Defaults to "INFO". @type file: Optional[str] @param file: Path to the log file. If provided, logs will be saved     to this file. @type kwargs: Any @param kwargs: Additional keyword arguments to pass to     C{RichHandler}.
class
BaseModelExtraForbid
class
AutoRegisterMeta
Metaclass for automatically registering modules.  Can be set as a metaclass for abstract base classes. Then, all subclasses will be automatically registered under the name of the subclass.  Example:  >>> REGISTRY = Registry(name="modules") >>> class BaseClass(metaclass=AutoRegisterMeta, registry=REGISTRY): ...     pass >>> class SubClass(BaseClass): ...     pass >>> REGISTRY.get("SubClass") <class '__main__.SubClass'> >>> BaseClass.REGISTRY.get("SubClass") <class '__main__.SubClass'>
class
Registry
function
module

luxonis_ml.utils.config

type variable
module

luxonis_ml.utils.environ

class
Environ
A BaseSettings subclass for storing environment variables.
variable
class

luxonis_ml.utils.environ.Environ(pydantic_settings.BaseSettings)

variable
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
constant
module

luxonis_ml.utils.filesystem

class
PutFile
class
FSType
class

luxonis_ml.utils.filesystem.PutFile(typing.Protocol)

method
__call__(self, local_path: PathType, remote_path: PosixPathType, mlflow_instance: Optional [ ModuleType ] = None) -> str: str
class

luxonis_ml.utils.filesystem.FSType(enum.Enum)

constant
constant
module

luxonis_ml.utils.graph

type variable
module

luxonis_ml.utils.logging

module

luxonis_ml.utils.registry

type variable
class

luxonis_ml.utils.LuxonisConfig(luxonis_ml.utils.BaseModelExtraForbid)

CLASS_METHOD
get_config
Loads config from a yaml file or a dictionary.  @type cfg: Optional[Union[str, dict]] @param cfg: Path to config file or a dictionary. @type overrides: Optional[Union[dict, list[str], tuple[str, ...]]] @param overrides: List of CLI overrides in a form of a dictionary mapping     "dotted" keys to unparsed string or python values. @rtype: LuxonisConfig @return: Instance of the config class. @raise ValueError: If neither C{cfg} nor C{overrides} are provided.
method
__str__(self) -> str: str
method
__repr__(self) -> str: str
method
get_json_schema(self) -> Params: Params
Retuns dict representation of the config json schema.  @rtype: dict @return: Dictionary with config json schema.
method
save_data(self, path: PathType)
Saves config to a yaml file.  @type path: str @param path: Path to output yaml file.
method
get(self, key_merged: str, default: Any = None) -> Any: Any
Returns a value from L{Config} based on the given key.  If the key doesn't exist, the default value is returned.  @type key_merged: str @param key_merged: Key in a form of a string with levels     separated by dots. @type default: Any @param default: Default value to return if the key doesn't     exist. @rtype: Any @return: Value of the key or default value.
class

luxonis_ml.utils.LuxonisFileSystem

method
__init__(self, path: str, allow_active_mlflow_run: Optional [ bool ] = False, allow_local: Optional [ bool ] = True, cache_storage: Optional [ str ] = None, put_file_plugin: Optional [ str ] = None)
Abstraction over remote and local sources.  Helper class which abstracts uploading and downloading files from remote and local sources. Supports S3, MLflow, GCS, and local file systems.  @type path: str @param path: Input path consisting of protocol and actual path     or just path for local files @type allow_active_mlflow_run: Optional[bool] @param allow_active_mlflow_run: Flag if operations are allowed     on active MLFlow run. Defaults to False. @type allow_local: Optional[bool] @param allow_local: Flag if operations are allowed on local file     system. Defaults to True. @type cache_storage: Optional[str] @param cache_storage: Path to cache storage. No cache is used if     set to None. Defaults to None. @type put_file_plugin: Optional[str] @param put_file_plugin: The name of a registered function under     the PUT_FILE_REGISTRY to override C{self.put_file}.
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
variable
method
put_file(self, local_path: PathType, remote_path: PosixPathType, mlflow_instance: Optional [ ModuleType ] = None) -> str: str
Copy a single file to remote storage.  @type local_path: PathType @param local_path: Path to local file @type remote_path: PosixPathType @param remote_path: Relative path to remote file @type mlflow_instance: Optional[L{ModuleType}] @param mlflow_instance: MLFlow instance if uploading to active     run. Defaults to C{None}. @rtype: str @return: The full remote path of the uploded file.
property
is_mlflow
Returns True if the filesystem is MLFlow.
property
is_fsspec
Returns True if the filesystem is fsspec.
property
full_path
Returns full remote path.
method
init_fsspec_filesystem(self) -> fsspec.AbstractFileSystem: fsspec.AbstractFileSystem
Initializes L{fsspec} filesystem based on the used protocol.  @rtype: L{fsspec.AbstractFileSystem} @return: Initialized fsspec filesystem.
method
put_dir(self, local_paths: Union [ PathType , Iterable [ PathType ] ], remote_dir: PosixPathType, uuid_dict: Optional [ Dict [ str , str ] ] = None, mlflow_instance: Optional [ ModuleType ] = None, copy_contents: bool = False) -> Optional[Dict[str, str]]: Optional[Dict[str, str]]
Uploads files to remote storage.  @type local_paths: Union[PathType, Sequence[PathType]] @param local_paths: Either a string specifying a directory to     walk the files or a list of files which can be in different     directories. @type remote_dir: PosixPathType @param remote_dir: Relative path to remote directory @type uuid_dict: Optional[Dict[str, str]] @param uuid_dict: Stores paths as keys and corresponding UUIDs     as values to replace the file basename. @type mlflow_instance: Optional[L{ModuleType}] @param mlflow_instance: MLFlow instance if uploading to active     run. Defaults to None. @type copy_contents: bool @param copy_contents: If True, only copy the content of the     folder specified in local_paths. Defaults to False. @rtype: Optional[Dict[str, str]] @return: When local_paths is a list, this maps local_paths to     remote_paths
method
put_bytes(self, file_bytes: bytes, remote_path: PosixPathType, mlflow_instance: Optional [ ModuleType ] = None)
Uploads a file to remote storage directly from file bytes.  @type file_bytes: bytes @param file_bytes: the bytes for the file contents @type remote_path: PosixPathType @param remote_path: Relative path to remote file @type mlflow_instance: Optional[L{ModuleType}] @param mlflow_instance: MLFlow instance if uploading to active     run. Defaults to None.
method
get_file(self, remote_path: PosixPathType, local_path: PathType, mlflow_instance: Optional [ ModuleType ] = None) -> Path: Path
Copy a single file from remote storage.  @type remote_path: PosixPathType @param remote_path: Relative path to remote file @type local_path: PathType @param local_path: Path to local file @type mlflow_instance: Optional[L{ModuleType}] @param mlflow_instance: MLFlow instance if uploading to active     run. Defaults to C{None}. @rtype: Path @return: Path to the downloaded file.
method
delete_file(self, remote_path: PosixPathType)
Deletes a single file from remote storage.  @type remote_path: PosixPathType @param remote_path: Relative path to remote file
method
delete_files(self, remote_paths: List [ PosixPathType ])
Deletes multiple files from remote storage.  @type remote_paths: List[PosixPathType] @param remote_paths: Relative paths to remote files
method
get_dir(self, remote_paths: Union [ PosixPathType , Iterable [ PosixPathType ] ], local_dir: PathType, mlflow_instance: Optional [ ModuleType ] = None) -> Path: Path
Copies many files from remote storage to local storage.  @type remote_paths: Union[PosixPathType,     Sequence[PosixPathType]] @param remote_paths: Either a string specifying a directory to     walk the files or a list of files which can be in different     directories. @type local_dir: PathType @param local_dir: Path to local directory @type mlflow_instance: Optional[L{ModuleType}] @param mlflow_instance: MLFlow instance if uploading to active     run. Defaults to C{None}. @rtype: Path @return: Path to the downloaded directory.
method
delete_dir(self, remote_dir: PosixPathType = '', allow_delete_parent: bool = False)
Deletes a directory and all its contents from remote storage.  @type remote_dir: PosixPathType @param remote_dir: Relative path to remote directory. @type allow_delete_parent: bool @param allow_delete_parent: If True, allows deletion of the     parent directory.
method
walk_dir(self, remote_dir: PosixPathType, recursive: bool = True, typ: Literal [ ' file ' , ' directory ' , ' all ' ] = 'file') -> Iterator[str]: Iterator[str]
Walks through the individual files in a remote directory.  @type remote_dir: PosixPathType @param remote_dir: Relative path to remote directory @type recursive: bool @param recursive: If True, walks through the directory     recursively. @type typ: Literal["file", "directory", "all"] @param typ: Specifies the type of files to walk through.     Defaults to "file". @rtype: Iterator[str] @return: Iterator over the paths.
method
read_text(self, remote_path: PosixPathType) -> Union[str, bytes]: Union[str, bytes]
Reads a file into a string.  @type remote_path: PosixPathType @param remote_path: Relative path to remote file. @rtype: Union[str, bytes] @return: The string containing the file contents.
method
read_to_byte_buffer(self, remote_path: Optional [ PosixPathType ] = None) -> BytesIO: BytesIO
Reads a file into a byte buffer.  @type remote_path: Optional[PosixPathType] @param remote_path: Relative path to remote file. @rtype: BytesIO @return: The byte buffer containing the file contents.
method
get_file_uuid(self, path: PathType, local: bool = False) -> str: str
Reads a file and returns the (unique) UUID generated from file bytes.  @type path: PathType @param path: Relative path to remote file. @type local: bool @param local: Specifies a local path as opposed to a remote     path. @rtype: str @return: The generated UUID.
method
get_file_uuids(self, paths: Iterable [ PathType ], local: bool = False) -> Dict[str, str]: Dict[str, str]
Computes the UUIDs for all files stored in the filesystem.  @type paths: List[PathType] @param paths: A list of relative remote paths if remote else     local paths. @type local: bool @param local: Specifies local paths as opposed to remote paths. @rtype: Dict[str, str] @return: A dictionary mapping the paths to their UUIDs
method
is_directory(self, remote_path: PosixPathType) -> bool: bool
Checks whether the given remote path is a directory.  @type remote_path: PosixPathType @param remote_path: Relative path to remote file. @rtype: bool @return: True if the path is a directory.
method
exists(self, remote_path: PosixPathType = '') -> bool: bool
Checks whether the given remote path exists.  @type remote_path: PosixPathType @param remote_path: Relative path to remote file. Defaults to ""     (root). @rtype: bool @return: True if the path exists.
static method
LuxonisFileSystem.split_full_path(path: PathType) -> Tuple[str, str]: Tuple[str, str]
Splits the full path into protocol and absolute path.  @type path: PathType @param path: Full path @rtype: Tuple[str, str] @return: Tuple of protocol and absolute path.
static method
LuxonisFileSystem.get_protocol(path: str) -> str: str
Extracts the detected protocol from a path.  @type path: str @param path: Full path @rtype: str @return: Protocol of the path.
static method
LuxonisFileSystem.download(url: str, dest: Optional [ PathType ]) -> Path: Path
Downloads file or directory from remote storage.  Intended for downloading a single remote object, elevating the need to create an instance of L{LuxonisFileSystem}.  @type url: str @param url: URL to the file or directory @type dest: Optional[PathType] @param dest: Destination directory. If unspecified, the current     directory is used. @rtype: Path @return: Path to the downloaded file or directory.
static method
LuxonisFileSystem.upload(local_path: PathType, url: str)
Uploads file or directory to remote storage.  Intended for uploading a single local object, elevating the need to create an instance of L{LuxonisFileSystem}.  @type local_path: PathType @param local_path: Path to the local file or directory @type url: str @param url: URL to the remote file or directory @rtype: str @return: URL to the uploaded file or directory.
class

luxonis_ml.utils.BaseModelExtraForbid(pydantic.BaseModel)

variable
class

luxonis_ml.utils.AutoRegisterMeta(abc.ABCMeta)

variable
method
__new__(cls, name: str, bases: Tuple [ type , ... ], attrs: Dict [ str , type ], register: bool = True, register_name: Optional [ str ] = None, registry: Optional [ Registry ] = None)
Automatically register the class.  @type name: str @param name: Class name  @type bases: Tuple[type, ...] @param bases: Base classes  @type attrs: Dict[str, type] @param attrs: Class attributes  @type register: bool @param register: Weather to register the class. Defaults to True.     Should be set to False for abstract base classes.  @type register_name: Optional[str] @param register_name: Name used for registration.     If unset, the class name is used. Defaults to None.  @type registry: Optional[Registry] @param registry: Registry to use for registration.     Defaults to None. Has to be set in the base class.
class

luxonis_ml.utils.Registry(typing.Generic)

method
__init__(self, name: str)
A Registry class to store and retrieve modules.  @type name: str @ivar name: Name of the registry
method
__str__(self) -> str: str
method
__repr__(self) -> str: str
method
__len__(self) -> int: int
method
__getitem__(self, key: str) -> T: T
method
__setitem__(self, key: str, value: T)
method
__contains__(self, key: str) -> bool: bool
property
method
get(self, key: str) -> T: T
Retrieves the registry record for the key.  @type key: str @param key: Name of the registered item, I{e.g.} the class name     in string format. @rtype: type @return: Corresponding class if L{key} exists @raise KeyError: If L{key} is not in the registry
method
method
register(self, module: Optional [ T ] = None, name: Optional [ str ] = None, force: bool = False) -> Optional[Callable[[T], T]]: Optional[Callable[[T], T]]
Registers a module.  Can be used as a decorator or as a normal method:      >>> registry = Registry(name="modules")     >>> @registry.register()     ... class Foo:     ...     pass     >>> registry.get("Foo")     <class '__main__.Foo'>     >>> class Bar:     ...     pass     >>> registry.register(module=Bar)     >>> registry.get("Bar")     <class '__main__.Bar'>  @type name: Optional[str] @param name: Name of the module. If C{None}, then use class name.     Defaults to None.  @type module: Optional[type] @param module: Module class to be registered. Defaults to None.  @type force: bool @param force: Whether to override an existing class with the same name.     Defaults to False.  @rtype: Union[type, Callable[[type], type]] @return: Module class or register function if used as a decorator  @raise KeyError: Raised if class name already exists and C{force==False}