export package

Subpackages

export.drise package

Submodules

export.base module

class export.base.Exporter(name: str, predictions_spec: DataSpecification | None = None, targets_spec: DataSpecification | None = None, criterion: Callable[[int, Batch], bool | Iterable[int]] | None = None)

Bases: Trackable, ABC

Base class for an Armory sample exporter.

Criterion: alias of Callable[[int, Batch], Union[bool, Iterable[int]]]

static artifact_path(batch_idx: int, sample_idx: int, filename: str) → str

Creates the full artifact path for a particular sample export.

Parameters:

batch_idx (int) – The index/number of the sample’s batch.
sample_idx (int) – The index/number of the sample within the batch.
filename (str) – The name of the exported file.

Returns:

Full artifact path as a string.

Return type:

str

export(batch_idx: int, batch: Batch) → None

Exports the given batch.

Parameters:

batch_idx (int) – The index/number of this batch.
batch (Batch) – The batch to be exported.

abstract export_samples(batch_idx: int, batch: Batch, samples: Iterable[int]) → None

Exports samples from the given batch.

Parameters:

batch_idx (int) – The index/number of this batch.
batch (Batch) – The batch to be exported.
samples (Iterable[int]) – The indices of samples in the batch to be exported.

use_sink(sink: Sink) → None

Sets the export sink to be used by the exporter.

Parameters:: sink (Sink) – Sink

export.captum module

export.criteria module

Export criterion utilities

export.criteria.all_satisfied(*criteria: Callable[[int, Batch], bool | Iterable[int]]) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches samples that satisfy all of the given nested criteria.

Example:

from armory.export import Exporter
from armory.export.criteria import all_satisfied, every_n_samples, first_n_batches

# Only exports every other sample from the first 2 batches
exporter = Exporter(
    criterion=all_satisfied(
        every_n_samples(2),
        first_n_batches(2),
    )
)

Parameters:

*criteria –

Nested criteria that must all be satisfied for a sample to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.always() → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches all samples of all batches.

Returns:: Export criterion function

export.criteria.any_satisfied(*criteria: Callable[[int, Batch], bool | Iterable[int]]) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches samples that satisfy any of the given nested criteria.

Example:

from armory.export import Exporter
from armory.export.criteria import any_satisfied, every_n_samples, first_n_batches

# Exports every sample from the first 2 batches, then every other sample
# for all other batches
exporter = Exporter(
    criterion=any_satisfied(
        every_n_samples(2),
        first_n_batches(2),
    )
)

Parameters:

*criteria –

Nested criteria that must all be satisfied for a sample to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.batch_targets(spec: DataSpecification | None = None) → Callable[[Batch], Tensor]

Creates a batch metric callable that returns the targets from the batch.

Example:

from armory.export import Exporter
from armory.export.criteria import batch_targets, when_metric_lt

# Exports samples that have a target value less than 10
exporter = Exporter(criterion=when_metric_lt(batch_targets(), 10))

Parameters:: spec (DataSpecification, optional) – Data specification for obtaining targets in a batch, defaults to None
Returns:: Batch metric function
Return type:: Callable[[Batch], torch.Tensor]

export.criteria.every_n_batches(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches all samples from every nth batch.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import every_n_batches

# Exports every sample in every 5th batch (when batch_idx is 4, 9, 14, etc.)
exporter = Exporter(criterion=every_n_batches(5))

Parameters:: n (int) – Interval at which to export batches based on the index of the batch
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.every_n_samples(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches every nth sample in the dataset.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import every_n_samples

# Exports every 3rd sample, regardless of batch
# For example with a batch size of 4:
#  - batch 0, sample 2
#  - batch 1, sample 1
#  - batch 2, samples 0 and 3
exporter = Exporter(criterion=every_n_samples(3))

Parameters:: n (int) – Interval at which to export samples based on the index of the sample within the dataset
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.every_n_samples_of_batch(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches every nth sample in every batch.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import every_n_samples_of_batch

# Exports every other sample in every batch (when sample_idx is 1, 3, 5, etc.)
exporter = Exporter(criterion=every_n_samples_of_batch(5))

Parameters:: n (int) – Interval at which to export samples based on the index of the sample within each batch
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.first_n_batches(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches all samples from the first n batches.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import first_n_batches

# Exports every sample in the first 3 batches (when batch_idx is 0, 1, and 2)
exporter = Exporter(criterion=every_n_batches(3))

Parameters:: n (int) – Number of batches to be exported
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.first_n_samples(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches the first n samples in the dataset.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import first_n_samples

# Exports the first 5 samples, regardless of batch
# For example with a batch size of 2:
#  - batch 0, samples 0 and 1
#  - batch 1, samples 0 and 1
#  - batch 2, sample 0
exporter = Exporter(criterion=first_n_samples(5))

Parameters:: n (int) – Number of samples to be exported
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.first_n_samples_of_batch(n: int) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches the first n samples in every batch.

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import first_n_samples_of_batch

# Exports the first 2 samples in every batches (when sample_idx is 0 and 1)
exporter = Exporter(criterion=first_n_samples_of_batch(2))

:param n:Number of samples to be exported from each batch :type n: int :return: Export criterion function :rtype: Exporter.Criterion

export.criteria.not_satisfied(criterion: Callable[[int, Batch], bool | Iterable[int]]) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches samples that do not satisfy the nested criterion.

Example:

from armory.export import Exporter
from armory.export.criteria import every_n_batches, not_satisfied

# Exports every sample in every batch except every 3rd batch (when
# batch_idx is 0, 1, 3, etc.)
exporter = Exporter(criterion=not_satisfied(every_n_batches(3)))

Parameters:: criterion (Exporter.Criterion) – Nested criterion for which unsatisfied samples are to be exported
Returns:: Export criterion function
Return type:: Exporter.Criterion

export.criteria.samples(indices: Sequence[int]) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches specific samples in the dataset, by their global index (regardless of batch).

Note, if the data loader is shuffling the dataset, the samples that get exported will vary between runs.

Example:

from armory.export import Exporter
from armory.export.criteria import samples

# For a batch size of 4:
#  - batch 0, sample 2
#  - batch 1, samples 0 and 1
#  - batch 2, samples 2 and 3
exporter = Exporter(criterion=samples([2, 4, 5, 10, 11]))

Parameters:: indices (Sequence[int]) – Indices of samples within the dataset to be exported
Returns:: Export criterion function
Return type:: Exporter.Criterion

Creates an export criterion that matches when a computed metric for a batch or the samples within the batch is equal to a particular value.

Example:

import torch
from armory.data import TorchSpec
from armory.export import Exporter
from armory.export.criteria import when_metric_eq

# Exports samples that have max score of exactly 5
def max_pred(batch):
    return torch.tensor([
        torch.max(p) for p in batch.predictions.get(TorchSpec())
    ])
exporter = Exporter(criterion=when_metric_eq(max_pred, 5))

Parameters:

metric (Callable[[Batch], Union[bool, float, torch.Tensor]]) – Callable that computes a metric for a batch. The return value may be a single boolean or number, or it can be a tensor array of the computed metric values for each sample in the batch.
threshold (Union[bool, float, torch.Tensor]) – Value the computed metric (either batchwise or samplewise) must be equal to in order for the batch or samples to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.when_metric_gt(metric, threshold) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches when a computed metric for a batch or the samples within the batch is greater than a particular threshold value.

Example:

import torch
from armory.data import TorchSpec
from armory.export import Exporter
from armory.export.criteria import when_metric_gt

# Exports samples that have max score greater than 5
def max_pred(batch):
    return torch.tensor([
        torch.max(p) for p in batch.predictions.get(TorchSpec())
    ])
exporter = Exporter(criterion=when_metric_gt(max_pred, 5))

Parameters:

metric (Callable[[Batch], Union[float, torch.Tensor]]) – Callable that computes a metric for a batch. The return value may be a single number, or it can be a tensor array of the computed metric values for each sample in the batch.
threshold (Union[float, torch.Tensor]) – Value the computed metric (either batchwise or samplewise) must be less than in order for the batch or samples to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.when_metric_in(metric: Callable[[Batch], float | Tensor], threshold: Sequence[float] | Sequence[Tensor]) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches when a computed metric for a batch or the samples within the batch is one of a particular set of values.

Example:

import torch
from armory.data import TorchSpec
from armory.export import Exporter
from armory.export.criteria import when_metric_in

# Exports samples that have max score of 5 or 8
def max_pred(batch):
    return torch.tensor([
        torch.max(p) for p in batch.predictions.get(TorchSpec())
    ])
exporter = Exporter(criterion=when_metric_in(max_pred, [5, 8]))

Parameters:

metric (Callable[[Batch], Union[float, torch.Tensor]]) – Callable that computes a metric for a batch. The return value may be a single number, or it can be a tensor array of the computed metric values for each sample in the batch.
threshold (Union[Sequence[float], Sequence[torch.Tensor]]) – Value the computed metric (either batchwise or samplewise) must be less than in order for the batch or samples to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.when_metric_isclose(metric: Callable[[Batch], float | Tensor], threshold: float | Tensor, rtol: float = 1e-05, atol: float = 1e-08) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches when a computed metric for a batch or the samples within the batch is close to a particular value.

Example:

import torch
from armory.data import TorchSpec
from armory.export import Exporter
from armory.export.criteria import when_metric_isclose

# Exports samples that have max score of 5.0
def max_pred(batch):
    return torch.tensor([
        torch.max(p) for p in batch.predictions.get(TorchSpec())
    ])
exporter = Exporter(criterion=when_metric_isclose(max_pred, 5))

Parameters:

metric (Callable[[Batch], Union[float, torch.Tensor]]) – Callable that computes a metric for a batch. The return value may be a single boolean or number, or it can be a tensor array of the computed metric values for each sample in the batch.
threshold (Union[float, torch.Tensor]) – Value the computed metric (either batchwise or samplewise) must be equal to in order for the batch or samples to be exported
rtol (float, optional) – Optional, relative tolerance, defaults to 1e-05
atol (float, optional) – Optional, absolute tolerance, defaults to 1e-08

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.criteria.when_metric_lt(metric: Callable[[Batch], float | Tensor], threshold: float | Tensor) → Callable[[int, Batch], bool | Iterable[int]]

Creates an export criterion that matches when a computed metric for a batch or the samples within the batch is less than a particular threshold value.

Example:

import torch
from armory.data import TorchSpec
from armory.export import Exporter
from armory.export.criteria import when_metric_lt

# Exports samples that have max score less than 5
def max_pred(batch):
    return torch.tensor([
        torch.max(p) for p in batch.predictions.get(TorchSpec())
    ])
exporter = Exporter(criterion=when_metric_lt(max_pred, 5))

Parameters:

metric (Callable[[Batch], Union[float, torch.Tensor]]) – Callable that computes a metric for a batch. The return value may be a single number, or it can be a tensor array of the computed metric values for each sample in the batch.
threshold (Union[float, torch.Tensor]) – Value the computed metric (either batchwise or samplewise) must be less than in order for the batch or samples to be exported

Returns:

Export criterion function

Return type:

Exporter.Criterion

export.image_classification module

class export.image_classification.ImageClassificationExporter(name: str | None = None, inputs_spec: NumpyImageSpec | None = None, predictions_spec: DataSpecification | None = None, targets_spec: DataSpecification | None = None, criterion: Callable[[int, Batch], bool | Iterable[int]] | None = None)

Bases: Exporter

An exporter for image classification samples.

export_samples(batch_idx: int, batch: ImageClassificationBatch, samples: Iterable[int]) → None

Export samples

Parameters:

batch_idx (int) – Batch index
batch (ImageClassificationBatch) – Batch
samples (Iterable[int]) – Samples

export.object_detection module

class export.object_detection.ObjectDetectionExporter(name: str | None = None, score_threshold: float = 0.5, inputs_spec: NumpyImageSpec | None = None, predictions_spec: NumpyBoundingBoxSpec | None = None, targets_spec: NumpyBoundingBoxSpec | None = None, criterion: Callable[[int, Batch], bool | Iterable[int]] | None = None)

Bases: Exporter

An exporter for object detection samples.

export_samples(batch_idx: int, batch: ObjectDetectionBatch, samples: Iterable[int]) → None

Export samples

Parameters:

batch_idx (int) – Batch index
batch (ObjectDetectionBatch) – Batch
samples (Iterable[int]) – Samples

export.object_detection.draw_boxes_on_image(image: ndarray, ground_truth_boxes: ndarray | None = None, ground_truth_color: str = 'red', ground_truth_width: int = 2, pred_boxes: ndarray | None = None, pred_color: str = 'white', pred_width: int = 2) → ndarray

Draw bounding boxes for ground truth objects and predicted objects on top of an image sample.

Ground truth bounding boxes will be drawn first, then the predicted bounding boxes.

Parameters:

image (np.ndarray) – 3-dimensional array of image data in shape (C, H, W) and type uint8
ground_truth_boxes (np.ndarray, optional) – Optional array of shape (N, 4) containing ground truth bounding boxes in (xmin, ymin, xmax, ymax) format.
ground_truth_color (str, optional) – Color to use for ground truth bounding boxes. Color can be represented as PIL strings (e.g., “red”). Defaults to “red”
ground_truth_width (int, optional) – Width of ground truth bounding boxes, defaults to 2
pred_boxes (np.ndarray, optional) – Optional array of shape (N, 4) containing predicted bounding boxes in (xmin, ymin, xmax, ymax) format.
pred_color (str, optional) – Color to use for predicted bounding boxes. Color can be represented as PIL strings (e.g., “red”). Defaults to “white”
pred_width (int, optional) – Width of ground truth bounding boxes, defaults to 2

Returns:

uint8 tensor of (C, H, W) image with bounding boxes data

Return type:

np.ndarray

export.sink module

Sample export sinks/destinations

class export.sink.MlflowSink(client: MlflowClient, run_id: str)

Bases: Sink

Convenience wrapper around an MLFlow client to automatically invoke the client APIs with the run ID

log_artifact(local_path: str, artifact_path: str | None = None)

log_dict(dictionary: Dict[str, Any], artifact_file: str)

log_figure(figure: matplotlib.figure.Figure | plotly.graph_objects.Figure, artifact_file: str)

log_image(image: ndarray | PIL.Image.Image, artifact_path: str)

log_table(data: Dict[str, Any] | pandas.DataFrame, artifact_file: str)

log_text(text: str, artifact_file: str)

class export.sink.Sink

Bases: object

No-op export sink

log_artifact(local_path: str, artifact_path: str | None = None)

log_dict(dictionary: Dict[str, Any], artifact_file: str)

log_figure(figure: matplotlib.figure.Figure | plotly.graph_objects.Figure, artifact_file: str)

log_image(image: ndarray | PIL.Image.Image, artifact_path: str)

log_table(data: Dict[str, Any] | pandas.DataFrame, artifact_file: str)

log_text(text: str, artifact_file: str)

export package

Subpackages

Submodules

export.base module

export.captum module

export.criteria module

export.image_classification module

export.object_detection module

export.sink module

export.xaitksaliency module

Module contents