Documentation

The full documentation is available here. Use the side bar to select the sub-module that is most applicable.

All of these functions can be loaded from the aml package directly, for example:

     >>> import aml
>>> aml.TensorboardLoad('./')

Array

Here is the documentation for the array functionality. This includes building arrays and performing transformation functions on arrays.

aml.array.flatten(data: ndarray, start_dim: int = 0, end_dim: int = -1, copy=False) → ndarray

This class allows you to flatten an array inside a pipeline. This class was implemented to mirror the behaviour in https://pytorch.org/docs/stable/generated/torch.flatten.html.

Examples

>>> flat = flatten(
...     [[[1, 2],
...     [3, 4]],
...     [[5, 6],
...     [7, 8]]],
...     start_dim=1,
...     end_dim=-1,
...     )
[[1,2,3,4],
[5,6,7,8]]

Parameters:

• start_dim (-) – The first dim to flatten. Defaults to 0.

• end_dim (-) – The last dim to flatten. Defaults to -1.

• copy (-) – Whether to return a copied version of the array during the transform method. Defaults to False.

• data (ndarray) –

Return type:

ndarray

aml.array.make_input_roll(data: ndarray, sequence_length: int) → ndarray

This function will produce an array that is a rolled version of the original data sequence. The original sequence must be 2D.

Examples

>>> make_input_roll(np.array([[1],[2],[3],[4],[5]]), sequence_length=2)
array([[[1],
        [2]],

        [[2],
        [3]],

        [[3],
        [4]],

        [[4],
        [5]]]

similarly:

>>> make_input_roll(np.array([[1, 2],[3, 4],[5, 6]]), sequence_length=2)
array([[[1,2],
        [3,4]],

        [[3,4],
        [5,6]]]

Parameters:

• data (-) – This is the data that you want transformed. Please use the shape (n_datapoints, n_features).

• sequence_length (-) – This is an integer that contains the length of each of the returned sequences.

Returns:

- output – This is an array with the rolled data.

Return type:

ndarray:

aml.array.stratification(array: ndarray, **leq: float) → ndarray

This function allows you to replace values in a numpy array less than or equal to the given values with the keywords given in the function arguments. All values that are more than the largest argument will have a returned value of None.

Examples

Here, we stratify an array into three groups:

>>> stratification(
...     np.array(
...         [[0.1, 0.2],
...         [0.7, 0.9]]),
...     Orange=0.8,
...     Green=0.15,
...     Red=1.0,
...     )
array(
    [['Green', 'Orange'],
    ['Orange', 'Red']],
    dtype=object)

Similarly, if the largest leq argument is smaller than the largest value in the array, then None will be returned in its place:

>>> stratification(
...     np.array(
...         [[0.1, 0.2],
...         [0.7, 1.1]]),
...     Orange=0.8,
...     Green=0.15,
...     Red=1.0,
...     )
array(
    [['Green', 'Orange'],
    ['Orange', None]],
    dtype=object)

Parameters:

• array (-) – Array to perform transformation over.

• **leq (-) –

  The keyword arguments used to stratify the object. The keywords are the labels in the output array.

Returns:

- out – Array, stratified into the groups given by the arguments.

Return type:

np.ndarray:

aml.array.time_interval_range(max_hour: int, minutes_per_hour: int, minute_interval: int) → ndarray

Create an array of a time range.

Edited from https://stackoverflow.com/a/58052181/19451559.

Examples

The following will create a list of values

>>> time_interval_range(24, 60, 30)
[
    '00:00',
    '00:30',
    ...,
    '22:30',
    '23:00',
    '23:30'
    ]

Parameters:

• max_hour (-) – The number of hours to run to. This is in hours.

• minutes_per_hour (-) – The number of minutes in each hour. This is in minutes.

• minute_interval (-) – The interval for a new element. This is in minutes.

Returns:

- out – Array of time range, returned as strings.

Return type:

np.ndarray:

Data

Here is the documentation for the data functionality.

class aml.data.ECGCorruptor(dataset: Dataset, corrupt_sources: list | int | None = None, noise_level: list | float | None = None, seed: int | None = None, axis: str = 'both', x_noise_std: float = 0.1)

Bases: HelperDataset

Parameters:

• dataset (Dataset) –

• corrupt_sources (list | int | None) –

• noise_level (list | float | None) –

• seed (int | None) –

• axis (str) –

• x_noise_std (float) –

__init__(dataset: Dataset, corrupt_sources: list | int | None = None, noise_level: list | float | None = None, seed: int | None = None, axis: str = 'both', x_noise_std: float = 0.1)

ECG Data corruptor. You may pass a noise level, sources to corrupt, and the seed for determining the random events. This class allows you to corrupt either the 'x', 'y', or 'both'. This class is built specifically for use with PTB_XL (found in aml.data.datasets).

This function will work as expected on all devices.

Examples

>>> dataset = ECGCorruptor(
...     dataset=dataset_train
...     corrupt_sources=[0,1,2,3],
...     noise_level=0.5,
...     )

Parameters:

• dataset (-) – The dataset to corrupt. When iterated over, the dataset should return x, y, and source.

• corrupt_sources (-) – The sources to corrupt in the dataset. This can be a list of sources, an integer of the source, or None for no sources to be corrupted. Defaults to None.

• noise_level (-) – This is the level of noise to apply to the dataset. It can be a list of noise levels, a single noise level to use for all sources, or None for no noise. Defaults to None.

• seed (-) – This is the seed that is used to determine random events. Defaults to None.

• axis (-) –

  This is the axis to apply the corruption to. This should be either 'x', 'y', or 'both'.

  • 'x': Adds a Gaussian distribution to the 'x' values with mean=0 and std=0.1.

  • 'y': Swaps the binary label using the function 1-y_true.

  • 'both': Adds a Gaussian distribution to the 'x' values with mean=0 and std=0.1 and swaps the binary label using the function 1-y_true.

  Defaults to 'both'.

• x_noise_std (-) – This is the standard deviation of the noise that is added to x when it is corrupted. Defaults to 0.1.

property corrupt_sources

class aml.data.GroupAllBatchSampler(group: ndarray | List[Any], seed: int | None = None, shuffle_groups: bool = True, shuffle_points: bool = True)

Bases: Sampler

Parameters:

• group (ndarray | List[Any]) –

• seed (None | int) –

• shuffle_groups (bool) –

• shuffle_points (bool) –

__init__(group: ndarray | List[Any], seed: int | None = None, shuffle_groups: bool = True, shuffle_points: bool = True)

A pytorch batch sampler that returns batches containing all of the data for an entire group, in a shuffled order. There will only be as many batches as number of unique groups, since each batch contains all data from that group.

Examples

The following will batch the training dataset into batches that contains all data from a single group, given by the group argument.

>>> dl = torch.utils.data.DataLoader(
...     train_dataset,
...     batch_sampler=GroupAllBatchSampler(
...         group=train_group,
...         seed=seed,
...         )
...     )

Parameters:

• group (-) – The group of the data points. This should be the same length as the data set that is to be sampled.

• seed (-) – Random seed for group order shuffling and shuffling of points in each batch. Defaults to None.

• shuffle_groups (-) – Shuffles the order of the groups that are returned. Defaults to True.

• shuffle_points (-) – Shuffles the order of the points that are returned in the groups. Defaults to True.

class aml.data.GroupBatchSampler(group: ndarray | List[Any], seed: int | None = None, batch_size: int = 20, upsample: bool | Dict[Any, int] = False)

Bases: Sampler

Parameters:

• group (ndarray | List[Any]) –

• seed (None | int) –

• batch_size (int) –

• upsample (bool | Dict[Any, int]) –

__init__(group: ndarray | List[Any], seed: int | None = None, batch_size: int = 20, upsample: bool | Dict[Any, int] = False)

A pytorch batch sampler that returns a batch of samples with that same group. This means each batch will be drawn from only a single group.

Examples

The following will batch the training dataset into batches that contains single group, given by the group argument

>>> dl = torch.utils.data.DataLoader(
...     train_dataset,
...     batch_sampler=GroupBatchSampler(
...         group=train_group,
...         seed=seed,
...         batch_size=64,
...         )
...     )

Parameters:

• group (-) – The group of the data points. This should be the same length as the data set that is to be sampled.

• seed (-) – Random seed for group order shuffling and shuffling of points in each batch. Defaults to None.

• batch_size (-) – The size of each batch. Each batch will be smaller than or equal in size to this value. Defaults to 20.

• upsample (-) – Whether to upsample the smaller groups, so that all groups have the same size. Defaults to False.

class aml.data.HelperDataset(dataset: Dataset)

Bases: Dataset

Parameters:

dataset (Dataset) –

__init__(dataset: Dataset)

This dataset helps you to build wrappers for other datasets by ensuring that any method or attribute of the original dataset is available as a method or attribute of the new dataset.

The original dataset is available as the attribute ._dataset.

Parameters:

dataset (-) – The dataset that will be wrapped.

class aml.data.MemoryDataset(dataset: Dataset, now: bool = True, verbose: bool = True, n_jobs: int = 1)

Bases: HelperDataset

Parameters:

• dataset (Dataset) –

• now (bool) –

• verbose (bool) –

• n_jobs (int) –

__init__(dataset: Dataset, now: bool = True, verbose: bool = True, n_jobs: int = 1)

This dataset allows the user to wrap another dataset and load all of the outputs into memory, so that they are accessed from RAM instead of storage. All attributes of the original dataset will still be available, except for ._dataset and ._data_dict if they were defined. It also allows the data to be saved in memory right away or after the data is accessed for the first time.

Examples

>>> dataset = MemoryDataset(dataset, now=True)

Parameters:

• dataset (-) – The dataset to wrap and add to memory.

• now (-) – Whether to save the data to memory right away, or the first time the data is accessed. If True, then this initialisation might take some time as it will need to load all of the data. Defaults to True.

• verbose (-) – Whether to print progress as the data is being loaded into memory. This is ignored if now=False. Defaults to True.

• n_jobs (-) – The number of parallel operations when loading the data to memory. Defaults to 1.

class aml.data.MyData(*inputs: tensor)

Bases: Dataset

Parameters:

inputs (tensor) –

__init__(*inputs: tensor)

Allows the user to turn any set of tensors into a dataset.

Examples

>>> data = MyData(X, y, other)
>>> len(data) == len(X)
True

Parameters:

inputs (-) – Any tensors.

class aml.data.NumpyLoader(*args, **kwargs)

Bases: DataLoader

__init__(*args, **kwargs)

A wrapper for the torch dataloader that supplies numpy arrays instead of tensors.

It takes all of the same arguments as torch.utils.data.DataLoader and has the same default arguments: https://pytorch.org/docs/stable/data.html#torch.utils.data.DataLoader

The collate_fn function is changed to allow for numpy arrays.

The code was inspired by: https://jax.readthedocs.io/en/latest/notebooks/Neural_Network_and_Data_Loading.html

batch_size: int | None

dataset: Dataset[T_co]

drop_last: bool

num_workers: int

pin_memory: bool

pin_memory_device: str

prefetch_factor: int | None

sampler: Sampler | Iterable

timeout: float

class aml.data.PTB_XL(data_path: str = './', train: bool = True, sampling_rate: Literal[100, 500] = 100, source_name: Literal['nurse', 'site', 'device'] = 'nurse', return_sources: bool = True, binary: bool = False, subset=False)

Bases: Dataset

Parameters:

• data_path (str) –

• train (bool) –

• sampling_rate (Literal[100, 500]) –

• source_name (Literal['nurse', 'site', 'device']) –

• return_sources (bool) –

• binary (bool) –

__init__(data_path: str = './', train: bool = True, sampling_rate: Literal[100, 500] = 100, source_name: Literal['nurse', 'site', 'device'] = 'nurse', return_sources: bool = True, binary: bool = False, subset=False)

ECG Data, as described here: https://physionet.org/content/ptb-xl/1.0.2/. A positive class when binary=True, indicates that the ECG Data is abnormal.

Examples

>>> dataset = PTB_XL(
...     data_path='../../data/',
...     train=True,
...     source_name='nurse',
...     sampling_rate=500,
...     return_sources=False,
...     )

Parameters:

• data_path (-) – The path that the data is saved or will be saved. Defaults to './'.

• train (-) – Whether to load the training or testing set. Defaults to True.

• sampling_rate (-) – The sampling rate. This should be in [100, 500]. Defaults to 100.

• source_name (-) – Which of the three attributes should be interpretted as the data sources. This should be in ['nurse', 'site', 'device']. This is ignored if return_sources=False. Defaults to 'nurse'.

• return_sources (-) – Whether to return the sources alongside the data and targets. For example, with return_sources=True, for every index this dataset will return data, target, source. Defaults to True.

• binary (-) – Whether to return classes based on whether the ecg is normal or not, and so a binary classification problem. Defaults to False.

• subset (-) – If True, only the first 1000 items of the training and test set will be returned. Defaults to False.

aggregate_diagnostic()

download()

static single_diagnostic(y_dict, agg_df)

class aml.data.WrapperDataset(dataset: ~torch.utils.data.dataset.Dataset, functions_index: ~typing.List[int] | int | None = None, functions: ~typing.Callable | ~typing.List[~typing.Callable] = <function WrapperDataset.<lambda>>)

Bases: HelperDataset

Parameters:

• dataset (Dataset) –

• functions_index (List[int] | int | None) –

• functions (Callable | List[Callable]) –

__init__(dataset: ~torch.utils.data.dataset.Dataset, functions_index: ~typing.List[int] | int | None = None, functions: ~typing.Callable | ~typing.List[~typing.Callable] = <function WrapperDataset.<lambda>>)

This allows you to wrap a dataset with a set of functions that will be applied to each returned data point. You can apply a single function to all outputs of a data point, or a different function to each of the different outputs.

Examples

The following would multiply all of the first returned values in the dataset by 2.

>>> WrapperDataset(
...     dataset
...     functions_index=0,
...     functions=lambda x: x*2
...     )

The following would multiply all of the returned values in the dataset by 2.

>>> WrapperDataset(
...     dataset
...     functions_index=None,
...     functions=lambda x: x*2
...     )

The following would multiply all of the first returned values in the dataset by 2, and the second by 2.

>>> WrapperDataset(
...     dataset
...     functions_index=[0, 1],
...     functions=lambda x: x*2
...     )

The following would multiply all of the first returned values in the dataset by 2, and the second by 3.

>>> WrapperDataset(
...     dataset
...     functions_index=[0, 1],
...     functions=[lambda x: x*2, lambda x: x*3]
...     )

Parameters:

• dataset (-) – The dataset to be wrapped.

• functions_index (-) –

  The index of the functions to be applied to.

  • If None, then if the functions is callable, it will be applied to all outputs of the data points, or if the functions is a list, it will be applied to the corresponding output of the data point.

  • If list then the corresponding index will have the functions applied to them. If functions is a list, then it will be applied to the corresponding indicies given in functions_index of the data point. If functions is callable, it will be applied to all of the indicies in functions_index

  • If int, then the functions must be callable, and will be applied to the output of this index.

  • If 'all', then the functions must be callable, and will be applied to the output of the dataset. This allows you to build a function that can act over all of the outputs of dataset. The returned value will be the data that is returned by the dataset.

  Defaults to None.

• functions (-) – This is the function, or list of functions to apply to the corresponding indices in functions_index. Please see the documentation for the functions_index argument to understand the behaviour of different input types. Defaults to lambda x:x.

aml.data.sequential_samplers_sampler(*samplers) → Sampler

This sampler allows you to combine several samplers sequentially, so that each one can be run after the other.

Parameters:

samplers (-) – The samplers to combine.

Returns:

- sampler – A single sampler that will use each of the given samplers in sequence.

Return type:

torch.utils.data.Sampler:

aml.data.split_sampler(sampler: Sampler, splits: None | List[float] = None) → List[Sampler]

This allows you to split a sampler into multiple samplers based on proportions. The splits will be taken in order given and each sampler will contain data between each pair of proportions.

Examples

The following is an example of how to split a sampler into two:

>>> sampler = [[0,1], [2,3], [4,5]]
>>> s1, s2 = split_sampler(sampler, splits=[0.5, 0.5])
>>> for i in zip(s1, s2):
...     print(i)
([0], [1])
([2], [3])
([4], [5])

The following is what would happen if batches were of odd length:

>>> sampler = [[0,1,2], [3,4,5], [6,7,8]]
>>> s1, s2 = split_sampler(sampler, splits=[0.5, 0.5])
>>> for i in zip(s1, s2):
...     print(i)
([0], [1, 2])
([3], [4, 5])
([6], [7, 8])

The following is how you would split a sampler into three:

>>> sampler = [[0,1,2], [3,4,5], [6,7,8]]
>>> s1, s2, s3 = split_sampler(sampler, splits=[0.34, 0.34, 0.34])
>>> for i in zip(s1, s2, s3):
...     print(i)
([0], [1], [2])
([3], [4], [5])
([6], [7], [8])

And an example with this in use with GroupAllBatchSampler:

>>> sampler = GroupAllBatchSampler(
...     group=np.array([1,1,1,2,2,2,3,3,3,4,4,4,5,5,5]),
...     seed=42,
...     shuffle_groups=True,
...     shuffle_points=True,
...     )
>>> sampler_splits = split_sampler(sampler, splits=[0.34, 0.34, 0.34])
>>> for i in zip(*sampler_splits):
...     print(i)
([11], [9], [10])
([0], [1], [2])
([14], [13], [12])
([5], [4], [3])
([8], [6], [7])

Parameters:

• sampler (-) – This is the sampler to split.

• splits (-) – This is the proportions of each batch to be contained in each new sampler. If None, then the original sampler is returned in a list. Please see the examples. Defaults to None.

Returns:

- samplers – A list of the new samplers.

Return type:

list:

Dataframe

Here is the documentation for the dataframe functionality.

aml.dataframe.interpolate_nans(x: Series, **kwargs)

Interpolate a pandas series.

Parameters:

• x (-) – The series to interpolate. The index will be taken as the x values and the values will be taken as the y values.

• kwargs (-) – Keyword arguments that are passed to the np.interp function.

Returns:

- res – The interpolated pandas data series.

Return type:

pd.Series:

Metrics

Here is the documentation for the metric functionality.

aml.metric.auc_precision_recall_curve(y_true: ndarray, y_proba: ndarray, pos_label=None, sample_weight=None) → float

A function that calculates the area under the precision-recall curve between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> auc_precision_recall_curve(
...     y_true=np.array([0,1,0,1,0]),
...     y_proba=np.array([0,0,0,1,0]),
...     )
0.85

Parameters:

• y_true (-) – The array of true values.

• y_proba (-) – The array of predicted score values. If y_pred has shape (N,2), and y_true has two unique values, then the probability of a positive class will be assumed to be y_proba[:,1].

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• sample_weight (-) – Sample weights. Defualts to None.

Returns:

- auc – The area under the precision-recall curve.

Return type:

float:

aml.metric.countna(array: ndarray, normalise: bool = True) → float

Function to calculate the number of NAs in an array.

Parameters:

• array (-) – The array to calculate the number of missing values on.

• normalise (-) – Whether to return the values as a percentage. Defaults to True.

Returns:

- countna – A float equal to the number or proportion of missing values in an array.

Return type:

float:

aml.metric.format_mean_ci(values: ndarray, confidence: float = 0.95, string: str = '{mean:.2f} ({confidence_l:.2f} - {confidence_u:.2f})', distribution: Literal['norm', 't'] = 't') → str

A function useful for formatting a table with information on the mean and confidence intervals of a set of values. You may use the normal distribution or the t-distribution to calculate the confidence intervals.

Examples

>>> import numpy as np
>>> format_mean_confidence(
...     values=np.array([1,2,3,4,5]),
...     confidence=0.95,
...     string="{mean:.2f} ({confidence_l:.2f} - {confidence_u:.2f})",
...     distribution='t',
... )
'3.00 (1.04 - 4.96)'

Parameters:

• values (-) – The array to calculate the values on.

• confidence (-) – The confidence interval to calculate. Defaults to 0.95.

• string (-) – The string that dictates the output. This should include somewhere {mean}, {confidence_l}, and {confidence_u}. Defaults to "{mean:.2f} ({confidence_l:.2f} - {confidence_u:.2f})".

• distribution (-) – The distribution to use for calculating the confidence intervals. Can be either 'norm' or 't'. Defaults to 't'.

Returns:

- stats – A string of the desired format with the statistics included.

Return type:

str:

aml.metric.format_mean_iqr_missing(values: ndarray, string: str = '{mean:.2f} ({iqr:.2f}) (({count_na:.0f}%))') → str

A function useful for formatting a table with information on the mean, IQR and missing rate of an attribute.

Examples

>>> import numpy as np
>>> format_mean_iqr_missing(
...     values=np.array([1,2,3,4,5]),
...     string="{mean:.2f} ({iqr:.2f}) (({count_na:.0f}%))",
...     )
'3.00 (2) ((0%))'

Parameters:

• values (-) – The array to calculate the values on.

• string (-) – The string that dictates the output. This should include somewhere {mean}, {iqr}, and {count_na}. Defaults to "{mean:.2f} ({iqr:.2f}) (({count_na:.0f}%))".

Returns:

- stats – A string of the desired format with the statistics included.

Return type:

str:

aml.metric.format_mean_std(values: ndarray, string: str = '{mean:.2f} ({std:.2f})') → str

A function useful for formatting a table with information on the mean and standard deviation an attribute.

Examples

>>> import numpy as np
>>> format_mean_std(
...     values=np.array([1,2,3,4,5]),
...     string="{mean:.2f} ({std:.2f})",
...     )
'3.00 (1.41)'

Parameters:

• values (-) – The array to calculate the values on.

• string (-) – The string that dictates the output. This should include somewhere {mean} and {std}. Defaults to "{mean:.2f} ({std:.2f})".

Returns:

- stats – A string of the desired format with the statistics included.

Return type:

str:

aml.metric.interquartile_range(values: ndarray, lower: float = 25, upper: float = 75) → float

Function to calculate the interquartile range of an array.

Parameters:

• array (-) – The array to calculate the IQR of.

• lower (-) – The percentile of the lower quartile. Defaults to 25.

• upper (-) – The percentile of the upper quartile. Defaults to 75.

• values (ndarray) –

Returns:

- iqr – A float equal to the interquartile range.

Return type:

float:

aml.metric.make_confusion_matrix(cfm: ndarray, group_names: None | List[str] = None, categories: None | List[str] = 'auto', count: bool = True, percent: bool = True, cbar: bool = True, xyticks: bool = True, xyplotlabels: bool = True, sum_stats: bool = True, figsize: tuple | list | None = None, cmap: str = 'Blues', title: str | None = None)

This function was edited from https://github.com/DTrimarchi10/confusion_matrix/blob/master/cfm_matrix.py.

This function will plot a confusion matrix, based on the array given.

Examples

>>> from sklearn.metrics import confusion_matrix
>>> make_confusion_matrix(
...     confusion_matrix(
...         y_test,
...         predictions_test
...         )
...     )

Parameters:

• cfm (-) – Confusion matrix to be plotted.

• group_names (-) – List of strings that represent the labels row by row to be shown in each square. Defaults to None.

• categories (-) – List of strings containing the categories to be displayed on the x,y axis. Defaults to 'auto'.

• count (-) – If True, show the raw number in the confusion matrix. Defaults to True.

• percent (-) – If True, show the proportions for each category. Defaults to True.

• cbar (-) – If True, show the color bar. The cbar values are based off the values in the confusion matrix. Defaults to True.

• xyticks (-) – If True, show x and y ticks. Defaults to True.

• xyplotlabels (-) – If True, show ‘True Label’ and ‘Predicted Label’ on the figure. Defaults to True.

• sum_stats (-) – If True, display summary statistics below the figure. Defaults to True.

• figsize (-) – Tuple representing the figure size. If None, the matplotlib rcParams value will be used. Defaults to None.

• cmap (-) – Colormap of the values displayed from matplotlib.pyplot.cm. See http://matplotlib.org/examples/color/colormaps_reference.html. Defaults to 'Blues'.

• title (-) – Title for the heatmap. Defaults to None.

Returns:

• - figure (matplotlib.pyplot.figure:) – The figure containing the axes.

• - ax (matplotlib.pyplot.axes:) – The axes containing the plot.

aml.metric.mean_confusion_matrix(y_pred: List[ndarray], y_true: List[ndarray], **kwargs) → ndarray

This function allows you to calculate the mean of confusion matrices based on multiple y_pred and y_true arrays.

Parameters:

• y_pred (-) – A list arrays of the predicted values.

• y_true (-) – A list arrays of the true values.

• **kwargs (-) –

  Other keyword arguments are passed to the sklearn.metrics.confusion_matrix function.

Returns:

- cfm_mean – A numpy array, the mean confusion matrix.

Return type:

np.ndarray:

aml.metric.npv_score(y_true: ndarray, y_pred: ndarray, labels: ndarray | None = None, pos_label: str | int = 1, average: str | None = 'binary', sample_weight: ndarray | None = None, zero_division: int | str = 'warn')

A function that calculates the negative predictive value between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> npv_score(
...     y_true=np.array([0,1,0,1,0]),
...     y_pred=np.array([0,0,0,1,0]),
...     )
0.75

Parameters:

• y_true (-) – The array of true values.

• y_pred (-) – The array of predicted values.

• labels (-) – The set of labels to include when average != 'binary', and their order if average is None. Labels present in the data can be excluded, for example to calculate a multiclass average ignoring a majority negative class, while labels not present in the data will result in 0 components in a macro average. For multilabel targets, labels are column indices. By default, all labels in y_true and y_pred are used in sorted order. Defaults to None.

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• average (-) –

  If None, the scores for each class are returned. Otherwise, this determines the type of averaging performed on the data:

  • 'binary': Only report results for the class specified by pos_label. This is applicable only if targets (y_{true,pred}) are binary.

  • 'micro': Calculate metrics globally by counting the total true positives, false negatives and false positives.

  • 'macro': Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.

  • 'weighted': Calculate metrics for each label, and find their average weighted by support (the number of true instances for each label). This alters ‘macro’ to account for label imbalance; it can result in an F-score that is not between precision and recall.

  • 'samples': Calculate metrics for each instance, and find their average (only meaningful for multilabel classification where this differs from accuracy_score).

  Defaults to 'binary'.

• sample_weight (-) – Sample weights. Defualts to None.

• zero_division (-) –

  Sets the value to return when there is a zero division:

  • npv: when there are no negative predictions

  If set to "warn", this acts as 0, but warnings are also raised. Defaults to "warn".

Returns:

- npv – The positive predictive value score.

Return type:

float:

aml.metric.ppv_score(y_true: ndarray, y_pred: ndarray, labels: ndarray | None = None, pos_label: str | int = 1, average: str | None = 'binary', sample_weight: ndarray | None = None, zero_division: int | str = 'warn')

A function that calculates the positive predictive value between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> ppv_score(
...     y_true=np.array([0,1,0,1,0]),
...     y_pred=np.array([0,0,0,1,0]),
...     )
1.0

Parameters:

• y_true (-) – The array of true values.

• y_pred (-) – The array of predicted values.

• labels (-) – The set of labels to include when average != 'binary', and their order if average is None. Labels present in the data can be excluded, for example to calculate a multiclass average ignoring a majority negative class, while labels not present in the data will result in 0 components in a macro average. For multilabel targets, labels are column indices. By default, all labels in y_true and y_pred are used in sorted order. Defaults to None.

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• average (-) –

  If None, the scores for each class are returned. Otherwise, this determines the type of averaging performed on the data:

  • 'binary': Only report results for the class specified by pos_label. This is applicable only if targets (y_{true,pred}) are binary.

  • 'micro': Calculate metrics globally by counting the total true positives, false negatives and false positives.

  • 'macro': Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.

  • 'weighted': Calculate metrics for each label, and find their average weighted by support (the number of true instances for each label). This alters ‘macro’ to account for label imbalance; it can result in an F-score that is not between precision and recall.

  • 'samples': Calculate metrics for each instance, and find their average (only meaningful for multilabel classification where this differs from accuracy_score).

  Defaults to 'binary'.

• sample_weight (-) – Sample weights. Defualts to None.

• zero_division (-) –

  Sets the value to return when there is a zero division:

  • ppv: when there are no positive predictions

  If set to "warn", this acts as 0, but warnings are also raised. Defaults to "warn".

Returns:

- ppv – The positive predictive value score.

Return type:

float:

aml.metric.sensitivity_score(y_true: ndarray, y_pred: ndarray, labels: ndarray | None = None, pos_label: str | int = 1, average: str | None = 'binary', sample_weight: ndarray | None = None, zero_division: int | str = 'warn') → float

A function that calculates the sensitivity between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> sensitivity_score(
...     y_true=np.array([0,1,0,1,0]),
...     y_pred=np.array([0,0,0,1,0]),
...     )
0.5

Parameters:

• y_true (-) – The array of true values.

• y_pred (-) – The array of predicted values.

• labels (-) – The set of labels to include when average != 'binary', and their order if average is None. Labels present in the data can be excluded, for example to calculate a multiclass average ignoring a majority negative class, while labels not present in the data will result in 0 components in a macro average. For multilabel targets, labels are column indices. By default, all labels in y_true and y_pred are used in sorted order. Defaults to None.

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• average (-) –

  If None, the scores for each class are returned. Otherwise, this determines the type of averaging performed on the data:

  • 'binary': Only report results for the class specified by pos_label. This is applicable only if targets (y_{true,pred}) are binary.

  • 'micro': Calculate metrics globally by counting the total true positives, false negatives and false positives.

  • 'macro': Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.

  • 'weighted': Calculate metrics for each label, and find their average weighted by support (the number of true instances for each label). This alters ‘macro’ to account for label imbalance; it can result in an F-score that is not between precision and recall.

  • 'samples': Calculate metrics for each instance, and find their average (only meaningful for multilabel classification where this differs from accuracy_score).

  Defaults to 'binary'.

• sample_weight (-) – Sample weights. Defualts to None.

• zero_division (-) –

  Sets the value to return when there is a zero division:

  • specificity: when there are no negative labels

  If set to "warn", this acts as 0, but warnings are also raised. Defaults to "warn".

Returns:

- sensitivity – The sensitivity score.

Return type:

float:

aml.metric.sensitivity_specificity_ppv_npv(y_true: ndarray, y_pred: ndarray, labels: ndarray | None = None, pos_label: str | int = 1, average: str | None = None, warn_for: str | Tuple[str] = ('sensitivity', 'specificity'), sample_weight: ndarray | None = None, zero_division: int | str = 'warn') → Tuple[float]

A function that calculates the sensitivity, specificity, ppv, and npv between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> sensitivity_specificity_ppv_npv(
...     y_true=np.array([0,1,0,1,0]),
...     y_pred=np.array([0,0,0,1,0]),
...     )
(0.5, 1.0, 1.0, 0.75)

Parameters:

• y_true (-) – The array of true values.

• y_pred (-) – The array of predicted values.

• labels (-) – The set of labels to include when average != 'binary', and their order if average is None. Labels present in the data can be excluded, for example to calculate a multiclass average ignoring a majority negative class, while labels not present in the data will result in 0 components in a macro average. For multilabel targets, labels are column indices. By default, all labels in y_true and y_pred are used in sorted order. Defaults to None.

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• average (-) –

  If None, the scores for each class are returned. Otherwise, this determines the type of averaging performed on the data:

  • 'binary': Only report results for the class specified by pos_label. This is applicable only if targets (y_{true,pred}) are binary.

  • 'micro': Calculate metrics globally by counting the total true positives, false negatives and false positives.

  • 'macro': Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.

  • 'weighted': Calculate metrics for each label, and find their average weighted by support (the number of true instances for each label). This alters ‘macro’ to account for label imbalance; it can result in an F-score that is not between precision and recall.

  • 'samples': Calculate metrics for each instance, and find their average (only meaningful for multilabel classification where this differs from accuracy_score).

  Defaults to None.

• warn_for (-) – This determines which warnings will be made in the case that this function is being used to return only one of its metrics. Defaults to ("sensitivity", "specificity").

• sample_weight (-) – Sample weights. Defualts to None.

• zero_division (-) –

  Sets the value to return when there is a zero division:

  • sensitivity: when there are no positive labels

  • specificity: when there are no negative labels

  • ppv: when there are no positive predictions

  • npv: when there are no negative predictions

Returns:

• - sensitivity (float:) – The sensitivity score.

• - specificity (float:) – The specificity score.

• - ppv (float:) – The positive predictive value score.

• - npv (float:) – The negative predictive value score.

Return type:

Tuple[float]

aml.metric.specificity_score(y_true: ndarray, y_pred: ndarray, labels: ndarray | None = None, pos_label: str | int = 1, average: str | None = 'binary', sample_weight: ndarray | None = None, zero_division: int | str = 'warn')

A function that calculates the sensitivity between two arrays. This is modelled on the Scikit-Learn recall_score, precision_score, accuracy_score, and f1_score.

Examples

>>> import numpy as np
>>> specificity_score(
...     y_true=np.array([0,1,0,1,0]),
...     y_pred=np.array([0,0,0,1,0]),
...     )
1.0

Parameters:

• y_true (-) – The array of true values.

• y_pred (-) – The array of predicted values.

• labels (-) – The set of labels to include when average != 'binary', and their order if average is None. Labels present in the data can be excluded, for example to calculate a multiclass average ignoring a majority negative class, while labels not present in the data will result in 0 components in a macro average. For multilabel targets, labels are column indices. By default, all labels in y_true and y_pred are used in sorted order. Defaults to None.

• pos_label (-) – The class to report if average='binary' and the data is binary. If the data are multiclass or multilabel, this will be ignored; setting labels=[pos_label] and average != 'binary' will report scores for that label only. Defaults to 1.

• average (-) –

  If None, the scores for each class are returned. Otherwise, this determines the type of averaging performed on the data:

  • 'binary': Only report results for the class specified by pos_label. This is applicable only if targets (y_{true,pred}) are binary.

  • 'micro': Calculate metrics globally by counting the total true positives, false negatives and false positives.

  • 'macro': Calculate metrics for each label, and find their unweighted mean. This does not take label imbalance into account.

  • 'weighted': Calculate metrics for each label, and find their average weighted by support (the number of true instances for each label). This alters ‘macro’ to account for label imbalance; it can result in an F-score that is not between precision and recall.

  • 'samples': Calculate metrics for each instance, and find their average (only meaningful for multilabel classification where this differs from accuracy_score).

  Defaults to 'binary'.

• sample_weight (-) – Sample weights. Defualts to None.

• zero_division (-) –

  Sets the value to return when there is a zero division:

  • specificity: when there are no negative labels

  If set to "warn", this acts as 0, but warnings are also raised. Defaults to "warn".

Returns:

- sensitivity – The sensitivity score.

Return type:

float:

aml.metric.std_confusion_matrix(y_pred: List[ndarray], y_true: List[ndarray], **kwargs) → ndarray

This function allows you to calculate the std of confusion matrices based on multiple y_pred and y_true arrays.

Parameters:

• y_pred (-) – A list arrays of the predicted values.

• y_true (-) – A list arrays of the true values.

• **kwargs (-) –

  Other keyword arguments are passed to the sklearn.metrics.confusion_matrix function.

Returns:

- cfm_std – A numpy array, the std confusion matrix.

Return type:

np.ndarray:

Models

Here is the documentation for the model functionality and classes.

class aml.model.AEModel(n_input: int, n_embedding: int, n_layers: int = 2, dropout: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

Bases: BaseLightningModule

Parameters:

• n_input (int) –

• n_embedding (int) –

• n_layers (int) –

• dropout (float) –

• optimizer (dict) –

• criterion (str) –

• n_epochs (int) –

__init__(n_input: int, n_embedding: int, n_layers: int = 2, dropout: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

An auto-encoder model, built to be run similar to sklearn models.

Examples

>>> ae_model = AEModel(n_input=100,
...     n_embedding=5,
...     n_layers=2,
...     n_epochs = 2,
...     verbose=True,
...     batch_size=10,
...     optimizer={'adam':{'lr':0.01}},
...     criterion='mseloss',
...     )
>>> X = torch.tensor(np.random.random((10000,100))).float()
>>> X_val = torch.tensor(np.random.random((10000,100))).float()
>>> training_metrics = ae_model.fit(X=X, X_val=X_val)
>>> output = ae_model.transform(X_test=X)

Parameters:

• n_input (-) – The size of the input feature dimension.

• n_embedding (-) – The number of features that the embedding will have.

• n_layers (-) – The number of layers in the encoder model. The decoder model will have the same number of layers. Defaults to 2.

• dropout (-) – The dropout value in each of the layers. Defaults to 0.2

• criterion (-) – The criterion that is used to calculate the loss. If using a string, please use one of ['mseloss', 'celoss'] Defaults to mseloss.

• optimizer (-) – A dictionary containing the optimizer name as keys and a dictionary as values containing the arguments as keys. For example: {'adam':{'lr':0.01}}. The key can also be a torch.optim class, but not initiated. For example: {torch.optim.Adam:{'lr':0.01}}. Defaults to {'adam':{'lr':0.01}}.

• n_epochs (-) – The number of epochs to run the training for. Defaults to 10.

• accelerator (-) – The device to use for training. Please use any of (“cpu”, “gpu”, “tpu”, “ipu”, “hpu”, “auto”). Defaults to 'auto'

• kwargs (-) – These keyword arguments will be passed to dcarte_transform.model.base_model.BaseModel.

fit(X: array | None = None, y=None, train_loader: DataLoader | None = None, X_val: array | None = None, y_val=None, val_loader: DataLoader | None = None, **kwargs)

This is used to fit the model. Please either use the train_loader or X and y. This corresponds to using either a torch DataLoader or a numpy array as the training data. If using the train_loader, ensure each iteration returns [X, X].

Parameters:

• X (-) – The input array to fit the model on. Defaults to None.

• train_loader (-) – The training data, which contains the input and the targets. Defaults to None.

• X_val (-) – The validation input to calculate validation loss on when training the model. Defaults to None

• val_loader (-) – The validation data, which contains the input and the targets. Defaults to None.

forward(X)

Same as torch.nn.Module.forward().

Parameters:

• *args – Whatever you decide to pass into the forward method.

• **kwargs – Keyword arguments are also possible.

Returns:

Your model’s output

predict(X: array | None = None, test_loader: DataLoader | None = None, reconstruct: bool = False, **kwargs)

Method for predicting data based on the fit AE.

Parameters:

• X_test (-) – The input array to test the model on. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

• reconstruct (-) – Whether to reconstruct the input data.

• X (array | None) –

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_step(batch, batch_idx: int)

Step function called during predict(). By default, it calls forward(). Override to add any processing logic.

The predict_step() is used to scale inference on multi-devices.

To prevent an OOM error, it is possible to use BasePredictionWriter callback to write the predictions to disk or database after each batch or on epoch end.

The BasePredictionWriter should be used while using a spawn based accelerator. This happens for Trainer(strategy="ddp_spawn") or training on 8 TPU cores with Trainer(accelerator="tpu", devices=8) as predictions won’t be returned.

Example

class MyModel(LightningModule):

    def predict_step(self, batch, batch_idx, dataloader_idx=0):
        return self(batch)

dm = ...
model = MyModel()
trainer = Trainer(accelerator="gpu", devices=2)
predictions = trainer.predict(model, dm)

Parameters:

• batch – Current batch.

• batch_idx (int) – Index of current batch.

• dataloader_idx – Index of the current dataloader.

Returns:

Predicted output

training_step(batch, batch_idx)

Here you compute and return the training loss and some additional metrics for e.g. the progress bar or logger.

Parameters:

• batch (Tensor | (Tensor, …) | [Tensor, …]) – The output of your DataLoader. A tensor, tuple or list.

• batch_idx (int) – Integer displaying index of this batch

Returns:

Any of.

• Tensor - The loss tensor

• dict - A dictionary. Can include any keys, but must include the key 'loss'

• None - Training will skip to the next batch. This is only for automatic optimization.

  This is not supported for multi-GPU, TPU, IPU, or DeepSpeed.

In this step you’d normally do the forward pass and calculate the loss for a batch. You can also do fancier things like multiple forward passes or something model specific.

Example:

def training_step(self, batch, batch_idx):
    x, y, z = batch
    out = self.encoder(x)
    loss = self.loss(out, x)
    return loss

To use multiple optimizers, you can switch to ‘manual optimization’ and control their stepping:

def __init__(self):
    super().__init__()
    self.automatic_optimization = False


# Multiple optimizers (e.g.: GANs)
def training_step(self, batch, batch_idx):
    opt1, opt2 = self.optimizers()

    # do training_step with encoder
    ...
    opt1.step()
    # do training_step with decoder
    ...
    opt2.step()

Note

When accumulate_grad_batches > 1, the loss returned here will be automatically normalized by accumulate_grad_batches internally.

transform(X: array | None = None, test_loader: DataLoader | None = None, **kwargs)

Method for transforming data based on the fit AE.

Parameters:

• X_test (-) – The input array to test the model on. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

• X (array | None) –

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

validation_step(val_batch, batch_idx)

Operates on a single batch of data from the validation set. In this step you’d might generate examples or calculate anything of interest like accuracy.

Parameters:

• batch – The output of your DataLoader.

• batch_idx – The index of this batch.

• dataloader_idx – The index of the dataloader that produced this batch. (only if multiple val dataloaders used)

Returns:

• Any object or value

• None - Validation will skip to the next batch

# if you have one val dataloader:
def validation_step(self, batch, batch_idx):
    ...


# if you have multiple val dataloaders:
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    ...

Examples:

# CASE 1: A single validation dataset
def validation_step(self, batch, batch_idx):
    x, y = batch

    # implement your own
    out = self(x)
    loss = self.loss(out, y)

    # log 6 example images
    # or generated text... or whatever
    sample_imgs = x[:6]
    grid = torchvision.utils.make_grid(sample_imgs)
    self.logger.experiment.add_image('example_images', grid, 0)

    # calculate acc
    labels_hat = torch.argmax(out, dim=1)
    val_acc = torch.sum(y == labels_hat).item() / (len(y) * 1.0)

    # log the outputs!
    self.log_dict({'val_loss': loss, 'val_acc': val_acc})

If you pass in multiple val dataloaders, validation_step() will have an additional argument. We recommend setting the default value of 0 so that you can quickly switch between single and multiple dataloaders.

# CASE 2: multiple validation dataloaders
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    # dataloader_idx tells you which dataset this is.
    ...

Note

If you don’t need to validate you don’t need to implement this method.

Note

When the validation_step() is called, the model has been put in eval mode and PyTorch gradients have been disabled. At the end of validation, the model goes back to training mode and gradients are enabled.

class aml.model.BaseLightningModule(optimizer: Dict[str | Optimizer, Dict[str, Any]] | Optimizer = {'adam': {'lr': 0.01}}, criterion: str | Module = 'mseloss', n_epochs: int = 10, batch_size: int = 10, shuffle: bool = True, verbose: bool = True, dl_kwargs: dict = {}, accelerator: str = 'auto', enable_model_summary: bool = False, enable_checkpointing: bool = False, pl_trainer_kwargs: dict = {}, callbacks: list = [], early_stopping: int | None = None, validation_fraction: float = 0.1, logging: bool = False, log_every_n_steps: int = 20)

Bases: TrainingHelper, LightningModule

Parameters:

• optimizer (Dict[str | Optimizer, Dict[str, Any]] | Optimizer) –

• criterion (str | Module) –

• n_epochs (int) –

• batch_size (int) –

• shuffle (bool) –

• verbose (bool) –

• dl_kwargs (dict) –

• accelerator (str) –

• enable_model_summary (bool) –

• enable_checkpointing (bool) –

• pl_trainer_kwargs (dict) –

• callbacks (list) –

• early_stopping (int) –

• validation_fraction (float) –

• logging (bool) –

• log_every_n_steps (int) –

__init__(optimizer: Dict[str | Optimizer, Dict[str, Any]] | Optimizer = {'adam': {'lr': 0.01}}, criterion: str | Module = 'mseloss', n_epochs: int = 10, batch_size: int = 10, shuffle: bool = True, verbose: bool = True, dl_kwargs: dict = {}, accelerator: str = 'auto', enable_model_summary: bool = False, enable_checkpointing: bool = False, pl_trainer_kwargs: dict = {}, callbacks: list = [], early_stopping: int | None = None, validation_fraction: float = 0.1, logging: bool = False, log_every_n_steps: int = 20)

A base model class, built to be run similar to sklearn models. This is built on top of pytorch_lightning.LightningModule and includes some helper functions.

Parameters:

• optimizer (-) – A dictionary containing the optimizer name as keys and a dictionary as values containing the arguments as keys. For example: {'adam':{'lr':0.01}}. The key can also be a torch.optim class, but not initiated. For example: {torch.optim.Adam:{'lr':0.01}}. Defaults to {'adam':{'lr':0.01}}.

• criterion (-) – The criterion that is used to calculate the loss. If using a string, please use one of ['mseloss', 'celoss'] Defaults to mseloss.

• n_epochs (-) – The number of epochs to run the training for. Defaults to 10.

• batch_size (-) – The batch size to use in training and transforming. Only used if the input data is not a torch DataLoader. Defaults to 10.

• shuffle (-) – Whether to shuffle the training data when training. Only used if the input data is not a torch DataLoader. Defaults to True.

• verbose (-) – Whether to print information whilst training. Defaults to True.

• dl_kwargs (-) – A dictionary of keyword arguments that will be passed to the training, validation and testing data loader. These will be passed to torch.utils.data.DataLoader. Only used if the input data is not a torch DataLoader. Defaults to {}.

• accelerator (-) – The device to use for training. Please use any of (“cpu”, “gpu”, “tpu”, “ipu”, “hpu”, “auto”). Defaults to 'auto'`.

• enable_model_summary (-) – Whether to pint a model summary when training. Defaults to False.

• enable_checkpointing (-) – Whether to save the model periodically. Defaults to False.

• pl_trainer_kwargs (-) – These are keyword arguments that will be passed to pytorch_lightning.Trainer. Defaults to {}.

• callbacks (-) – These are there callbacks passed to the pytorch_lightning.Trainer class. Please don’t pass a progress bar to this list, as the TQDM progress bar is passed to this list within this class. Defaults to [].

• early_stopping (-) – Whether to use early stopping. If an X_val and y_val are passed during the fitting, these will be used. Otherwise a sample from the training data will be used. If None, early stopping will not be used, and if an integer, this will be used as the patience. Defaults to None.

• validation_fraction (-) – If early_stopping is not None, and no validation data is passed, then a fraction of the training data will be used. This parameter defines that fraction of the training data. Defaults to 0.1.

• logging (-) – Whether to log the run data. Defaults to False.

• log_every_n_steps (-) – How many steps to train before logging metrics. Defaults to 20.

configure_optimizers()

This is required for pytorch lightning. If overwriting with your own function, please return a dictionary with keys 'optimizer' and 'lr_scheduler', if one is being used. The optimizer is also saved in this class as an attribute .optimizer, which is built from the input, saved in .passed_optimizer, when the fit method is called.

fit(X: array | None = None, y: array | None = None, train_loader: DataLoader | None = None, X_val: array | None = None, y_val: array | None = None, val_loader: DataLoader | None = None, ckpt_path=None, **kwargs)

This is used to fit the model. Please either use the train_loader, or X and y. This corresponds to using either a torch DataLoader or numpy arrays as the training data.

Parameters:

• X (-) – The input array to fit the model on. Defaults to None.

• y (-) – The target array to fit the model on. Defaults to None.

• train_loader (-) – The training data, which contains the input and the targets. Defaults to None.

• X_val (-) – The validation input to calculate validation loss on when training the model. Defaults to None

• X_val – The validation target to calculate validation loss on when training the model. Defaults to None

• val_loader (-) – The validation data, which contains the input and the targets. Defaults to None.

• y_val (array | None) –

predict(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making predictions on a test loader or numpy arrays. Please either use the test_loader, or X and y. This corresponds to using either a torch DataLoader or numpy arrays as the training data.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then the batches given to the pytorch lightning model will just contain X. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The results from the predictions

Return type:

torch.tensor:

class aml.model.GaussianMixPrior(L: int, num_components: int)

Bases: Module

Parameters:

• L (int) –

• num_components (int) –

__init__(L: int, num_components: int)

Based on code here: https://github.com/jmtomczak/intro_dgm/blob/main/vaes/vae_priors_example.ipynb

Parameters:

• L (-) – The number of latent dimensions.

• num_components (-) – The number of Gaussian components. Defaults to 1.

forward(z: Tensor, type: Literal['log_prob', 'sample'] = 'log_prob') → Tensor

The forward function.

Parameters:

• z (-) – The sample.

• type (-) – Whether to sample from the distribution or calculate the log probability. Defaults to 'log_prob'.

Returns:

- out – A sample or the log probability.

Return type:

torch.Tensor:

log_prob(z: Tensor) → Tensor

The log probability of a sample.

Parameters:

z (-) – The sample

Returns:

- out – The log probability.

Return type:

torch.Tensor:

sample(batch_size: int) → Tensor

Sample the prior.

Parameters:

batch_size (-) – The number of samples to draw.

Returns:

- out – The samples.

Return type:

torch.Tensor:

training: bool

class aml.model.KDAttributeTree(data: ndarray, *args, attributes: List[ndarray] | None | ndarray = None, n_jobs: int = 1, **kwargs)

Bases: KDTree

Parameters:

• data (ndarray) –

• attributes (List[ndarray] | None | ndarray) –

• n_jobs (int) –

__init__(data: ndarray, *args, attributes: List[ndarray] | None | ndarray = None, n_jobs: int = 1, **kwargs)

A wrapper for scipy.spatial.KDTree that allows you to pass attributes that will be returned along with the results from the query and query_ball_point methods. All arguments to the initialisation and methods will be passed to scipy.spatial.KDTree.

Examples

>>> tree = KDAttributeTree(
...     data=np.arange(10).reshape(-1,1),
...     attributes=[2*np.arange(10).reshape(-1,1), 3*np.arange(20).reshape(-1,2)],
...     )

Parameters:

• data (-) – The n data points of dimension m to be indexed. This array is not copied unless this is necessary to produce a contiguous array of doubles, and so modifying this data will result in bogus results. The data are also copied if the kd-tree is built with copy_data=True.

• attributes (-) – The list of attributes to be returned with the data. If None, then all of the outputted attributes will be None. If this is a 1D array, it will be transformed to a 2D array. You may also pass a list of attributes. Each attribute should be the same length as data. Defaults to None.

• n_jobs (-) – The number of parallel processes to use when querying the tree. Defaults to 1.

query(*args, **kwargs) → Tuple[float | ndarray, int | ndarray, ndarray | List[ndarray]]

Query the tree to return the distace, indices, and attributes of the points that are the k closest to the given data point.

Examples

When k=1 is given with a single point, the returned distance and index are floats and integers:

>>> tree.query(5, k=1)
(0.0, 5, [array([10]), array([30, 33])])

when k=1 is given with multiple points, the returned distance and indices are 1d arrays:

>>> tree.query(np.array([[1],[2]]), k=1)
(array([0., 0.]),
array([1, 2], dtype=int64),
[array([[2],
        [4]]),

array([[ 6,  9],
        [12, 15]])])

However, when an array of data points is queried with k more than 1, the distance and index are returned as 2D arrays:

>>> tree.query(np.array([[1],[2]]), k=2)
(array([[0., 1.],
        [0., 1.]]),
array([[1, 0],
        [2, 1]], dtype=int64),
[array([[[2],
        [0]],

        [[4],
        [2]]]),

array([[[ 6,  9],
        [ 0,  3]],

        [[12, 15],
        [ 6,  9]]])])

Note that if attributes was an array in the initialisation, the returned attributes will be an array instead of a list.

Parameters:

• *args (-) –

  All arguments from scipy.spatial.KDTree are accepted.

• **kwargs (-) –

  All keyword arguments from scipy.spatial.KDTree are accepted.

Returns:

• - d (typing.Union[float, np.ndarray]:) – The distance of the k closest points to the queried point.

• - i (typing.Union[int, np.ndarray]:) – The index of the k closest points to the queried point.

• - a (typing.Union[np.ndarray, typing.List[np.ndarray]]:) – A list of numpy arrays that are the attributes of the data points returned in i. If attributes was an array in the initialisation, the returned attributes will be an array instead of a list.

Return type:

Tuple[float | ndarray, int | ndarray, ndarray | List[ndarray]]

query_ball_point(*args, **kwargs) → Tuple[List[int], List[ndarray] | List[List[ndarray]]]

Find all pairs of points between the given point and the tree whose distance is at most r. It will return the attributes of these points along side the index.

Examples

The following example is what the query looks like on a single point:

>>> tree.query_ball_point(2, r=1)
([1, 2, 3],
[[array([2]), array([6, 9])],
[array([4]), array([12, 15])],
[array([6]), array([18, 21])]])

And on multiple points:

>>> tree.query_ball_point(np.array([[1],[2]]), r=1)
(array([list([0, 1, 2]), list([1, 2, 3])], dtype=object),
[[array([[0],
        [2],
        [4]]),
array([[ 0,  3],
        [ 6,  9],
        [12, 15]])],

[array([[2],
        [4],
        [6]]),
array([[ 6,  9],
        [12, 15],
        [18, 21]])]])

Parameters:

• *args (-) –

  All arguments from scipy.spatial.KDTree are accepted.

• **kwargs (-) –

  All keyword arguments from scipy.spatial.KDTree are accepted.

Returns:

• - d (typing.List[int]:) – The index of all of the points within r of the queried point.

• - a (typing.Union[typing.List[np.ndarray], typing.List[typing.List[np.ndarray]]]:) – The list of lists of numpy arrays that are the attributes of the points within r of the queried point. If attributes was an array in the initialisation, the returned attributes will be a list of arrays instead of a list of lists.

Return type:

Tuple[List[int], List[ndarray] | List[List[ndarray]]]

class aml.model.LambdaModule(f: Callable[[Tensor], Tensor])

Bases: Module

Parameters:

f (Callable[[Tensor], Tensor]) –

__init__(f: Callable[[Tensor], Tensor])

This is a wrapper class for a function to be used as a module in a nn.Sequential.

Examples

>>> from aml.model.module_utils import LambdaModule
>>> import torch.nn as nn
>>> import torch
>>> model = nn.Sequential(
...      LambdaModule(lambda x: x**2),
...      nn.Linear(10, 1)
... )
>>> x = torch.randn(10)
>>> model(x)

Parameters:

f (-) – The function to be wrapped as a module. All arguments and keyword arguments passed using the forward method will be passed to the function.

forward(*args, **kwargs)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

training: bool

class aml.model.MLPModel(n_input: int, n_output: int, hidden_layer_sizes: List[int] = (100,), activation: str | Module = 'relu', use_softmax: bool = True, dropout: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

Bases: BaseLightningModule

Parameters:

• n_input (int) –

• n_output (int) –

• hidden_layer_sizes (List[int]) –

• activation (str | Module) –

• use_softmax (bool) –

• dropout (float) –

• optimizer (dict) –

• criterion (str) –

• n_epochs (int) –

__init__(n_input: int, n_output: int, hidden_layer_sizes: List[int] = (100,), activation: str | Module = 'relu', use_softmax: bool = True, dropout: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

A simple MLP model that can be used for classification and built to be run similar to sklearn models.

Examples

>>> mlp_model = MLPModel(n_input=100,
...     n_output=2,
...     hidden_layer_sizes=(100,100,50),
...     n_epochs = 2,
...     verbose=True,
...     batch_size=10,
...     optimizer={'adam':{'lr':0.01}},
...     criterion='mseloss',
...     )
>>> X = torch.tensor(np.random.random((10000,100))).float()
>>> X_val = torch.tensor(np.random.random((10000,100))).float()
>>> training_metrics = mlp_model.fit(X=X, X_val=X_val)
>>> output = mlp_model.transform(X_test=X)

Parameters:

• n_input (-) – The size of the input feature dimension.

• n_output (-) – The output dimension sizes.

• hidden_layer_sizes (-) – The hidden layer sizes. Defaults to (100,).

• activation (-) – The activation function to be used in the hidden layers to add non-linearity. You may pass a str of the form: - ‘identity’: The identity function. - ‘logistic’: The logistic sigmoid function. - ‘tanh’, the hyperbolic tan function. - ‘relu’, the rectified linear unit function. You may also pass a torch module itself, which should be callable, taking a tensor as input and outputting a tensor. Defaults to relu.

• use_softmax (-) – Whether to use a softmax at the end of the fully connected layers. Defaults to True

• dropout (-) – The dropout value in each of the layers. Defaults to 0.2

• criterion (-) – The criterion that is used to calculate the loss. If using a string, please use one of ['mseloss', 'celoss'] Defaults to mseloss.

• optimizer (-) – A dictionary containing the optimizer name as keys and a dictionary as values containing the arguments as keys. For example: {'adam':{'lr':0.01}}. The key can also be a torch.optim class, but not initiated. For example: {torch.optim.Adam:{'lr':0.01}}. Defaults to {'adam':{'lr':0.01}}.

• n_epochs (-) – The number of epochs to run the training for. Defaults to 10.

• accelerator (-) – The device to use for training. Please use any of (“cpu”, “gpu”, “tpu”, “ipu”, “hpu”, “auto”). Defaults to 'auto'

• kwargs (-) – These keyword arguments will be passed to dcarte_transform.model.base_model.BaseModel.

fit(X: array | None = None, y=None, train_loader: DataLoader | None = None, X_val: array | None = None, y_val=None, val_loader: DataLoader | None = None, **kwargs)

This is used to fit the model. Please either use the train_loader or X and y. This corresponds to using either a torch DataLoader or a numpy array as the training data. If using the train_loader, ensure each iteration returns [X, X].

Parameters:

• X (-) – The input array to fit the model on. Defaults to None.

• train_loader (-) – The training data, which contains the input and the targets. Defaults to None.

• X_val (-) – The validation input to calculate validation loss on when training the model. Defaults to None

• val_loader (-) – The validation data, which contains the input and the targets. Defaults to None.

forward(X)

Same as torch.nn.Module.forward().

Parameters:

• *args – Whatever you decide to pass into the forward method.

• **kwargs – Keyword arguments are also possible.

Returns:

Your model’s output

predict(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_proba(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making probability predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_step(batch, batch_idx: int)

Step function called during predict(). By default, it calls forward(). Override to add any processing logic.

The predict_step() is used to scale inference on multi-devices.

To prevent an OOM error, it is possible to use BasePredictionWriter callback to write the predictions to disk or database after each batch or on epoch end.

The BasePredictionWriter should be used while using a spawn based accelerator. This happens for Trainer(strategy="ddp_spawn") or training on 8 TPU cores with Trainer(accelerator="tpu", devices=8) as predictions won’t be returned.

Example

class MyModel(LightningModule):

    def predict_step(self, batch, batch_idx, dataloader_idx=0):
        return self(batch)

dm = ...
model = MyModel()
trainer = Trainer(accelerator="gpu", devices=2)
predictions = trainer.predict(model, dm)

Parameters:

• batch – Current batch.

• batch_idx (int) – Index of current batch.

• dataloader_idx – Index of the current dataloader.

Returns:

Predicted output

training_step(batch, batch_idx)

Here you compute and return the training loss and some additional metrics for e.g. the progress bar or logger.

Parameters:

• batch (Tensor | (Tensor, …) | [Tensor, …]) – The output of your DataLoader. A tensor, tuple or list.

• batch_idx (int) – Integer displaying index of this batch

Returns:

Any of.

• Tensor - The loss tensor

• dict - A dictionary. Can include any keys, but must include the key 'loss'

• None - Training will skip to the next batch. This is only for automatic optimization.

  This is not supported for multi-GPU, TPU, IPU, or DeepSpeed.

In this step you’d normally do the forward pass and calculate the loss for a batch. You can also do fancier things like multiple forward passes or something model specific.

Example:

def training_step(self, batch, batch_idx):
    x, y, z = batch
    out = self.encoder(x)
    loss = self.loss(out, x)
    return loss

To use multiple optimizers, you can switch to ‘manual optimization’ and control their stepping:

def __init__(self):
    super().__init__()
    self.automatic_optimization = False


# Multiple optimizers (e.g.: GANs)
def training_step(self, batch, batch_idx):
    opt1, opt2 = self.optimizers()

    # do training_step with encoder
    ...
    opt1.step()
    # do training_step with decoder
    ...
    opt2.step()

Note

When accumulate_grad_batches > 1, the loss returned here will be automatically normalized by accumulate_grad_batches internally.

validation_step(val_batch, batch_idx)

Operates on a single batch of data from the validation set. In this step you’d might generate examples or calculate anything of interest like accuracy.

Parameters:

• batch – The output of your DataLoader.

• batch_idx – The index of this batch.

• dataloader_idx – The index of the dataloader that produced this batch. (only if multiple val dataloaders used)

Returns:

• Any object or value

• None - Validation will skip to the next batch

# if you have one val dataloader:
def validation_step(self, batch, batch_idx):
    ...


# if you have multiple val dataloaders:
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    ...

Examples:

# CASE 1: A single validation dataset
def validation_step(self, batch, batch_idx):
    x, y = batch

    # implement your own
    out = self(x)
    loss = self.loss(out, y)

    # log 6 example images
    # or generated text... or whatever
    sample_imgs = x[:6]
    grid = torchvision.utils.make_grid(sample_imgs)
    self.logger.experiment.add_image('example_images', grid, 0)

    # calculate acc
    labels_hat = torch.argmax(out, dim=1)
    val_acc = torch.sum(y == labels_hat).item() / (len(y) * 1.0)

    # log the outputs!
    self.log_dict({'val_loss': loss, 'val_acc': val_acc})

If you pass in multiple val dataloaders, validation_step() will have an additional argument. We recommend setting the default value of 0 so that you can quickly switch between single and multiple dataloaders.

# CASE 2: multiple validation dataloaders
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    # dataloader_idx tells you which dataset this is.
    ...

Note

If you don’t need to validate you don’t need to implement this method.

Note

When the validation_step() is called, the model has been put in eval mode and PyTorch gradients have been disabled. At the end of validation, the model goes back to training mode and gradients are enabled.

class aml.model.ResNet1DModel(input_dim: int = 4096, input_channels: int = 64, n_output: int = 10, kernel_size: int = 16, dropout_rate: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

Bases: BaseLightningModule

Parameters:

• input_dim (int) –

• input_channels (int) –

• n_output (int) –

• kernel_size (int) –

• dropout_rate (float) –

• optimizer (dict) –

• criterion (str) –

• n_epochs (int) –

__init__(input_dim: int = 4096, input_channels: int = 64, n_output: int = 10, kernel_size: int = 16, dropout_rate: float = 0.2, optimizer: dict = {'adam': {'lr': 0.01}}, criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

Model with 4 :code:`ResBlock`s, in which the number of channels increases linearly and the output dimensions decreases exponentially. This model will require the input dimension to be of at least 256 in size. This model is designed for sequences, and not images. The expected input is of the type:

[n_batches, n_filters, sequence_length]

Examples

>>> model = ResNet(
        input_dim=4096,
        input_channels=64,
        kernel_size=16,
        n_output=5,
        dropout_rate=0.2,
        )
>>> model.fit(torch.rand(1,64,4096))

Parameters:

• input_dim (-) – The input dimension of the input. This is the size of the final dimension, and the sequence length. Defaults to 4096.

• input_channels (-) – The number of channels in the input. This is the second dimension. It is the number of features for each sequence element. Defaults to 64.

• n_output (-) – The number of output classes in the prediction. Defaults to 10.

• kernel_size (-) – The size of the kernel filters that will act over the sequence. Defaults to 16.

• dropout_rate (-) – The dropout rate of the ResNet blocks. This should be a value between 0 and 1. Defaults to 0.2.

• criterion (-) – The criterion that is used to calculate the loss. If using a string, please use one of ['mseloss', 'celoss'] Defaults to mseloss.

• optimizer (-) – A dictionary containing the optimizer name as keys and a dictionary as values containing the arguments as keys. For example: {'adam':{'lr':0.01}}. The key can also be a torch.optim class, but not initiated. For example: {torch.optim.Adam:{'lr':0.01}}. Defaults to {'adam':{'lr':0.01}}.

• n_epochs (-) – The number of epochs to run the training for. Defaults to 10.

• accelerator (-) – The device to use for training. Please use any of (“cpu”, “gpu”, “tpu”, “ipu”, “hpu”, “auto”). Defaults to 'auto'

• kwargs (-) – These keyword arguments will be passed to dcarte_transform.model.base_model.BaseModel.

fit(X: array | None = None, y=None, train_loader: DataLoader | None = None, X_val: array | None = None, y_val=None, val_loader: DataLoader | None = None, **kwargs)

This is used to fit the model. Please either use the train_loader or X and y. This corresponds to using either a torch DataLoader or a numpy array as the training data. If using the train_loader, ensure each iteration returns [X, X].

Parameters:

• X (-) – The input array to fit the model on. Defaults to None.

• train_loader (-) – The training data, which contains the input and the targets. Defaults to None.

• X_val (-) – The validation input to calculate validation loss on when training the model. Defaults to None

• val_loader (-) – The validation data, which contains the input and the targets. Defaults to None.

forward(X)

Same as torch.nn.Module.forward().

Parameters:

• *args – Whatever you decide to pass into the forward method.

• **kwargs – Keyword arguments are also possible.

Returns:

Your model’s output

predict(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_proba(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making probability predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_step(batch, batch_idx: int)

Step function called during predict(). By default, it calls forward(). Override to add any processing logic.

The predict_step() is used to scale inference on multi-devices.

To prevent an OOM error, it is possible to use BasePredictionWriter callback to write the predictions to disk or database after each batch or on epoch end.

The BasePredictionWriter should be used while using a spawn based accelerator. This happens for Trainer(strategy="ddp_spawn") or training on 8 TPU cores with Trainer(accelerator="tpu", devices=8) as predictions won’t be returned.

Example

class MyModel(LightningModule):

    def predict_step(self, batch, batch_idx, dataloader_idx=0):
        return self(batch)

dm = ...
model = MyModel()
trainer = Trainer(accelerator="gpu", devices=2)
predictions = trainer.predict(model, dm)

Parameters:

• batch – Current batch.

• batch_idx (int) – Index of current batch.

• dataloader_idx – Index of the current dataloader.

Returns:

Predicted output

training_step(batch, batch_idx)

Here you compute and return the training loss and some additional metrics for e.g. the progress bar or logger.

Parameters:

• batch (Tensor | (Tensor, …) | [Tensor, …]) – The output of your DataLoader. A tensor, tuple or list.

• batch_idx (int) – Integer displaying index of this batch

Returns:

Any of.

• Tensor - The loss tensor

• dict - A dictionary. Can include any keys, but must include the key 'loss'

• None - Training will skip to the next batch. This is only for automatic optimization.

  This is not supported for multi-GPU, TPU, IPU, or DeepSpeed.

In this step you’d normally do the forward pass and calculate the loss for a batch. You can also do fancier things like multiple forward passes or something model specific.

Example:

def training_step(self, batch, batch_idx):
    x, y, z = batch
    out = self.encoder(x)
    loss = self.loss(out, x)
    return loss

To use multiple optimizers, you can switch to ‘manual optimization’ and control their stepping:

def __init__(self):
    super().__init__()
    self.automatic_optimization = False


# Multiple optimizers (e.g.: GANs)
def training_step(self, batch, batch_idx):
    opt1, opt2 = self.optimizers()

    # do training_step with encoder
    ...
    opt1.step()
    # do training_step with decoder
    ...
    opt2.step()

Note

When accumulate_grad_batches > 1, the loss returned here will be automatically normalized by accumulate_grad_batches internally.

validation_step(val_batch, batch_idx)

Operates on a single batch of data from the validation set. In this step you’d might generate examples or calculate anything of interest like accuracy.

Parameters:

• batch – The output of your DataLoader.

• batch_idx – The index of this batch.

• dataloader_idx – The index of the dataloader that produced this batch. (only if multiple val dataloaders used)

Returns:

• Any object or value

• None - Validation will skip to the next batch

# if you have one val dataloader:
def validation_step(self, batch, batch_idx):
    ...


# if you have multiple val dataloaders:
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    ...

Examples:

# CASE 1: A single validation dataset
def validation_step(self, batch, batch_idx):
    x, y = batch

    # implement your own
    out = self(x)
    loss = self.loss(out, y)

    # log 6 example images
    # or generated text... or whatever
    sample_imgs = x[:6]
    grid = torchvision.utils.make_grid(sample_imgs)
    self.logger.experiment.add_image('example_images', grid, 0)

    # calculate acc
    labels_hat = torch.argmax(out, dim=1)
    val_acc = torch.sum(y == labels_hat).item() / (len(y) * 1.0)

    # log the outputs!
    self.log_dict({'val_loss': loss, 'val_acc': val_acc})

If you pass in multiple val dataloaders, validation_step() will have an additional argument. We recommend setting the default value of 0 so that you can quickly switch between single and multiple dataloaders.

# CASE 2: multiple validation dataloaders
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    # dataloader_idx tells you which dataset this is.
    ...

Note

If you don’t need to validate you don’t need to implement this method.

Note

When the validation_step() is called, the model has been put in eval mode and PyTorch gradients have been disabled. At the end of validation, the model goes back to training mode and gradients are enabled.

class aml.model.StandardNormalPrior(L: int)

Bases: Module

Parameters:

L (int) –

__init__(L: int)

A prior that can be used in the VAE. This prior is a standard normal distribution.

Parameters:

L (-) – The dimensionality of the latent space.

forward(z: Tensor, type: Literal['log_prob', 'sample'] = 'log_prob') → Tensor

The forward function.

Parameters:

• z (-) – The sample.

• type (-) – Whether to sample from the distribution or calculate the log probability. Defaults to 'log_prob'.

Returns:

- out – A sample or the log probability.

Return type:

torch.Tensor:

log_prob(z: Tensor) → Tensor

The log probability of a sample.

Parameters:

z (-) – The sample

Returns:

- out – The log probability.

Return type:

torch.Tensor:

sample(batch_size: int) → Tensor

Sample the prior.

Parameters:

batch_size (-) – The number of samples to draw.

Returns:

- out – The samples.

Return type:

torch.Tensor:

training: bool

class aml.model.TransformerEncoderModel(n_output: int, n_input: int | None = None, seq_len: int | None = None, n_heads: int = 5, lr: float = 0.001, weight_decay: float = 0.0, dim_feedforward: int = 2048, transformer_encoder_dim_feedforward: int = 2048, dropout: float = 0.1, activation: str = 'relu', max_len: int = 5000, optimizer: str = 'adam', criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

Bases: BaseLightningModule

Parameters:

• n_output (int) –

• n_input (int) –

• seq_len (int) –

• n_heads (int) –

• lr (float) –

• weight_decay (float) –

• dim_feedforward (int) –

• transformer_encoder_dim_feedforward (int) –

• dropout (float) –

• activation (str) –

• max_len (int) –

• optimizer (str) –

• criterion (str) –

• n_epochs (int) –

__init__(n_output: int, n_input: int | None = None, seq_len: int | None = None, n_heads: int = 5, lr: float = 0.001, weight_decay: float = 0.0, dim_feedforward: int = 2048, transformer_encoder_dim_feedforward: int = 2048, dropout: float = 0.1, activation: str = 'relu', max_len: int = 5000, optimizer: str = 'adam', criterion: str = 'mseloss', n_epochs: int = 10, accelerator='auto', **kwargs)

A simple Transformer Encoder model and built to be run similar to sklearn models.

Examples

>>> transformer_model = TransformerEncoderModel(
...     n_input=100,
...     n_output=2,
...     n_heads=5,
...     n_epochs = 2,
...     verbose=True,
...     batch_size=10,
...     optimizer={'adam':{'lr':0.01}},
...     criterion='mseloss',
...     )
>>> X = torch.tensor(np.random.random((5, 10, 100))).float()
>>> X_val = torch.tensor(np.random.random((5, 10, 100))).float()
>>> training_metrics = transformer_model.fit(X=X, X_val=X_val)
>>> output = transformer_model.predict(X_test=X)

Parameters:

• n_output (-) – The size of the output dimension.

• n_input (int | None) –

• seq_len (int | None) –

• n_heads (int) –

• lr (float) –

• weight_decay (float) –

• dim_feedforward (int) –

• transformer_encoder_dim_feedforward (int) –

• dropout (float) –

• activation (str) –

• max_len (int) –

• optimizer (str) –

• criterion (str) –

• n_epochs (int) –

• n_input: int:

  The size of the input feature dimension. Defaults to None.

• seq_len: int:

  The sequence length of the input. Defaults to None.

• n_heads: int, optional:

  The number of heads in the multi-head attention. Defaults to 5.

• dim_feedforward: int, optional:

  The dimension of the feedforward network model. Defaults to 2048.

• transformer_encoder_dim_feedforward: int, optional:

  The dimension of the feedforward network model in the transformer encoder. Defaults to 2048.

• lr: float, optional:

  The learning rate to be used in the optimizer. Defaults to 0.001.

• weight_decay: float, optional:

  The weight decay to be used in the optimizer. Defaults to 0.0.

• dropout: float, optional:

  The dropout value in each of the layers in the transformer encoder. Defaults to 0.1.

• activation: str, optional:

  The activation function to be used in the hidden layers within the transformer encoder. Defaults to relu.

• max_len: int, optional:

  The maximum length of the input sequence. Defaults to 5000.

• criterion: str or torch.nn.Module:

  The criterion that is used to calculate the loss. If using a string, please use one of ['mseloss', 'celoss'] Defaults to mseloss.

• optimizer: dict, optional:

  A dictionary containing the optimizer name as keys and a dictionary as values containing the arguments as keys. For example: {'adam':{'lr':0.01}}. The key can also be a torch.optim class, but not initiated. For example: {torch.optim.Adam:{'lr':0.01}}. If using a string, please use one of ['adam', 'sgd', 'adagrad'] and provide lr and weight_decay. Defaults to {'adam':{'lr':0.01}}.

• n_epochs: int, optional:

  The number of epochs to run the training for. Defaults to 10.

• accelerator: str, optional:

  The device to use for training. Please use any of (“cpu”, “gpu”, “tpu”, “ipu”, “hpu”, “auto”). Defaults to 'auto'

• kwargs: optional:

  These keyword arguments will be passed to dcarte_transform.model.base_model.BaseModel.

fit(X: array | None = None, y=None, train_loader: DataLoader | None = None, X_val: array | None = None, y_val=None, val_loader: DataLoader | None = None, **kwargs)

This is used to fit the model. Please either use the train_loader or X and y. This corresponds to using either a torch DataLoader or a numpy array as the training data. If using the train_loader, ensure each iteration returns [X, X].

Parameters:

• X (-) – The input array to fit the model on. Defaults to None.

• train_loader (-) – The training data, which contains the input and the targets. Defaults to None.

• X_val (-) – The validation input to calculate validation loss on when training the model. Defaults to None

• val_loader (-) – The validation data, which contains the input and the targets. Defaults to None.

forward(X)

Same as torch.nn.Module.forward().

Parameters:

• *args – Whatever you decide to pass into the forward method.

• **kwargs – Keyword arguments are also possible.

Returns:

Your model’s output

predict(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_proba(X: array | None = None, y: array | None = None, test_loader: DataLoader | None = None)

Method for making probability predictions on a test loader.

Parameters:

• X (-) – The input array to test the model on. Defaults to None.

• y (-) – The target array to test the model on. If set to None, then targets_too will automatically be set to False. Defaults to None.

• test_loader (-) – A data loader containing the test data. Defaults to None.

Returns:

- output – The resutls from the predictions

Return type:

torch.tensor:

predict_step(batch, batch_idx: int)

Step function called during predict(). By default, it calls forward(). Override to add any processing logic.

The predict_step() is used to scale inference on multi-devices.

To prevent an OOM error, it is possible to use BasePredictionWriter callback to write the predictions to disk or database after each batch or on epoch end.

The BasePredictionWriter should be used while using a spawn based accelerator. This happens for Trainer(strategy="ddp_spawn") or training on 8 TPU cores with Trainer(accelerator="tpu", devices=8) as predictions won’t be returned.

Example

class MyModel(LightningModule):

    def predict_step(self, batch, batch_idx, dataloader_idx=0):
        return self(batch)

dm = ...
model = MyModel()
trainer = Trainer(accelerator="gpu", devices=2)
predictions = trainer.predict(model, dm)

Parameters:

• batch – Current batch.

• batch_idx (int) – Index of current batch.

• dataloader_idx – Index of the current dataloader.

Returns:

Predicted output

training_step(batch, batch_idx)

Here you compute and return the training loss and some additional metrics for e.g. the progress bar or logger.

Parameters:

• batch (Tensor | (Tensor, …) | [Tensor, …]) – The output of your DataLoader. A tensor, tuple or list.

• batch_idx (int) – Integer displaying index of this batch

Returns:

Any of.

• Tensor - The loss tensor

• dict - A dictionary. Can include any keys, but must include the key 'loss'

• None - Training will skip to the next batch. This is only for automatic optimization.

  This is not supported for multi-GPU, TPU, IPU, or DeepSpeed.

In this step you’d normally do the forward pass and calculate the loss for a batch. You can also do fancier things like multiple forward passes or something model specific.

Example:

def training_step(self, batch, batch_idx):
    x, y, z = batch
    out = self.encoder(x)
    loss = self.loss(out, x)
    return loss

To use multiple optimizers, you can switch to ‘manual optimization’ and control their stepping:

def __init__(self):
    super().__init__()
    self.automatic_optimization = False


# Multiple optimizers (e.g.: GANs)
def training_step(self, batch, batch_idx):
    opt1, opt2 = self.optimizers()

    # do training_step with encoder
    ...
    opt1.step()
    # do training_step with decoder
    ...
    opt2.step()

Note

When accumulate_grad_batches > 1, the loss returned here will be automatically normalized by accumulate_grad_batches internally.

validation_step(val_batch, batch_idx)

Operates on a single batch of data from the validation set. In this step you’d might generate examples or calculate anything of interest like accuracy.

Parameters:

• batch – The output of your DataLoader.

• batch_idx – The index of this batch.

• dataloader_idx – The index of the dataloader that produced this batch. (only if multiple val dataloaders used)

Returns:

• Any object or value

• None - Validation will skip to the next batch

# if you have one val dataloader:
def validation_step(self, batch, batch_idx):
    ...


# if you have multiple val dataloaders:
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    ...

Examples:

# CASE 1: A single validation dataset
def validation_step(self, batch, batch_idx):
    x, y = batch

    # implement your own
    out = self(x)
    loss = self.loss(out, y)

    # log 6 example images
    # or generated text... or whatever
    sample_imgs = x[:6]
    grid = torchvision.utils.make_grid(sample_imgs)
    self.logger.experiment.add_image('example_images', grid, 0)

    # calculate acc
    labels_hat = torch.argmax(out, dim=1)
    val_acc = torch.sum(y == labels_hat).item() / (len(y) * 1.0)

    # log the outputs!
    self.log_dict({'val_loss': loss, 'val_acc': val_acc})

If you pass in multiple val dataloaders, validation_step() will have an additional argument. We recommend setting the default value of 0 so that you can quickly switch between single and multiple dataloaders.

# CASE 2: multiple validation dataloaders
def validation_step(self, batch, batch_idx, dataloader_idx=0):
    # dataloader_idx tells you which dataset this is.
    ...

Note

If you don’t need to validate you don’t need to implement this method.

Note

When the validation_step() is called, the model has been put in eval mode and PyTorch gradients have been disabled. At the end of validation, the model goes back to training mode and gradients are enabled.

class aml.model.VAE(encoder: Module, decoder: Module, prior: Module, KL_beta: float = 1.0)

Bases: Module

Parameters:

• encoder (Module) –

• decoder (Module) –

• prior (Module) –

• KL_beta (float) –

__init__(encoder: Module, decoder: Module, prior: Module, KL_beta: float = 1.0)

A Variational Autoencoder (VAE). This class is a wrapper around the encoder, decoder, and prior.

Examples

>>> encoder = Encoder(...)
>>> decoder = Decoder(...)
>>> prior = Prior(...)
>>> vae = VAE(encoder, decoder, prior)

Parameters:

• encoder (-) – The encoder. Please see the documentation for VAEEncoder for more details.

• decoder (-) – The decoder. Please see the documentation for VAEDecoder for more details.

• prior (-) – The prior. This should have methods for sampling and calculating the log probability. Specifically, it should implement the following methods: - sample(batch_size: int) -> torch.Tensor[batch_size, latent_dim] - log_prob(z: torch.Tensor) -> torch.Tensor[batch_size, latent_dim]

• KL_beta (-) – The KL weight used in the loss. This will likely need to be tuned for each dataset. Defaults to 1.0.

forward(x: Tensor, reduction: Literal['mean', 'sum', None] = 'mean') → Tensor

The forward pass through the VAE. This will return the negative ELBO.

Examples

The following provides the loss for a batch of data:

>>> vae = VAE(...)
>>> x = torch.randn(64, 784)
>>> loss = vae(x)

Parameters:

• x (-) – The input data.

• reduction (-) – Whether to sum or mean the loss before returning. Defaults to 'mean'.

Returns:

- out – The loss.

Return type:

torch.Tensor:

reconstruct(x: Tensor) → Tensor

Reconstruct the input data.

Examples

>>> vae = VAE(...)
>>> x = torch.randn(64, 784)
>>> reconstruction = vae.reconstruct(x)

Parameters:

x (-) – The input data.

Returns:

- out – The reconstructions.

Return type:

torch.Tensor:

sample(batch_size: int = 64) → Tensor

Sample from the VAE.

Examples

>>> vae = VAE(...)
>>> samples = vae.sample(64)

Parameters:

batch_size (-) – The number of samples to generate. Defaults to 64.

Returns:

- out – The samples.

Return type:

torch.Tensor:

training: bool

class aml.model.VAEDecoder(decoder_net: Module, distribution: Literal['categorical', 'bernoulli', 'standard_normal'] = 'standard_normal', num_values: int | None = None, decoder_std: bool = False, out_shape: Tuple[int, ...] | None = None, std_method: Literal['learned', 'fixed'] = 'fixed')

Bases: Module

Parameters:

• decoder_net (Module) –

• distribution (Literal['categorical', 'bernoulli', 'standard_normal']) –

• num_values (int | None) –

• decoder_std (bool) –

• out_shape (Tuple[int, ...] | None) –

• std_method (Literal['learned', 'fixed']) –

__init__(decoder_net: Module, distribution: Literal['categorical', 'bernoulli', 'standard_normal'] = 'standard_normal', num_values: int | None = None, decoder_std: bool = False, out_shape: Tuple[int, ...] | None = None, std_method: Literal['learned', 'fixed'] = 'fixed')

The decoder network that maps the latent sample to the parameters of the distribution over the data. This is used as part of a VAE model.

Examples

>>> decoder_net = nn.Sequential(
...     nn.Linear(32, 256),
...     nn.ReLU(),
...     nn.Linear(256, 256),
...     nn.ReLU(),
...     nn.Linear(256, 32),
... )
>>> decoder = VAEDecoder(decoder_net, distribution='bernoulli')

Parameters:

• decoder_net (-) –

  The decoder network that maps the latent sample to the parameters of the distribution over the data.

  • categorical distribution: the output of this network should be of shape (batch_size, data_shape, num_values). The softmax function will be applied to this output to ensure that the values in the last dimension sum to 1.

  • bernoulli distribution: the output of this network should be of shape (batch_size, data_shape). The sigmoid function will be applied to this output to ensure that the output is between 0 and 1.

  • standard normal distribution: the output of this network should be of shape (batch_size, data_shape) if the decoder_decoder_std argument is False, else the output of this network should be of shape (2, batch_size, data_shape). No functions will be applied to this output.

• distribution (-) – The distribution of the output. Defaults to 'categorical'.

• num_values (-) – The number of values in the categorical distribution if used. Defaults to None.

• decoder_std (-) – Whether the decoder network outputs the standard deviation of the distribution. This is only used in the case of the standard normal distribution. If it does not, then the network should output shape (batch_size, data_shape). Defaults to False.

• out_shape (-) – The shape of the output data. This is only used in the case of the decoder_std=False. Defaults to None.

• std_method (-) – Whether the standard deviation of the normal distribution is learned or fixed. If "learned", then the standard deviation will be learned. If "fixed", then the standard deviation will be fixed to a value of 1. This will only be used if decoder_std=False. Defaults to "fixed".

decode(z: Tensor) → Tensor

Decode the latent sample to the parameters of the distribution over the data.

Parameters:

z (-) – The latent sample.

Returns:

- out – The parameters of the distribution over the data.

Return type:

torch.Tensor:

forward(z: Tensor, x: None | Tensor = None, type: Literal['log_prob', 'sample'] = 'log_prob') → Tensor

The forward pass through the decoder network.

Parameters:

• z (-) – The latent sample.

• x (-) – The data. No data is required if using type="sample". Defaults to None.

• type (-) – Whether to return a sample or the log probability. Defaults to 'log_prob'.

Returns:

- out – A sample or the log probability.

Return type:

torch.Tensor:

log_prob(x: Tensor, z: Tensor) → Tensor

The log probability of x under p(x|z).

Parameters:

• x (-) – The data.

• z (-) – The latent sample.

Returns:

- out – The log probability of x under p(x|z).

Return type:

torch.Tensor:

sample(z: Tensor) → Tensor

Samples p(x|z).

Parameters:

z (-) – The latent sample to decode and use to generate a sample of x.

Returns:

- out – The generated sample

Return type:

torch.Tensor:

training: bool

class aml.model.VAEEncoder(encode_net: Module)

Bases: Module

Parameters:

encode_net (Module) –

__init__(encode_net: Module)

The encoder network for the VAE model. This can be used to wrap any module for use in a VAE.

Examples

The following example shows how to wrap a simple MLP for use in a VAE. The MLP takes in a tensor of shape (batch_size, 32) and outputs a tensor of shape (2, batch_size, latent_dim). The first half of the output is used as the mean of the latent distribution, and the second half is used as the standard deviation of the latent distribution.

>>> class MLPEncoder(nn.Module):
...     def __init__(self):
...         super().__init__()
...         self.fc = nn.Sequential(
...             nn.Linear(32, 256),
...             nn.ReLU(),
...             nn.Linear(256, 256),
...             nn.ReLU(),
...             nn.Linear(256, 32*2),
...         )
...
...     def forward(self, x):
...         h = self.fc(x)
...         mu, std = torch.chunk(h, 2, dim=1)
...         return mu, F.softplus(std) + 1e-7
>>> encoder_net = MLPEncoder()
>>> encoder = Encoder(encoder_net)

Parameters:

encode_net (-) – The network that will be used to encode the data. This network should output a tensor of shape (2, batch_size, latent_dim). This is so that the first half of the output is used as the mean of the latent distribution, and the second half is used as the standard deviation of the latent distribution.

encode(x: Tensor) → Tuple[Tensor, Tensor]

Encode the input data into the mean and standard deviation of the latent distribution.

Parameters:

x (-) – The input data.

Returns:

• - mu (torch.Tensor:) – The mean of the latent distribution.

• - std (torch.Tensor:) – The standard deviation of the latent distribution.

Return type:

Tuple[Tensor, Tensor]

forward(x: Tensor) → Tuple[Tensor, Tensor, Tensor]

The forward pass of the encoder network.

Parameters:

x (-) – The input data.

Returns:

• - z (torch.Tensor:) – The sample from the latent distribution.

• - mu (torch.Tensor:) – The mean of the latent distribution.

• - std (torch.Tensor:) – The standard deviation of the latent distribution.

Return type:

Tuple[Tensor, Tensor, Tensor]

log_prob(x: None | Tensor = None, mu: None | Tensor = None, std: None | Tensor = None, z: None | Tensor = None) → Tensor

Calculate the probability of the sample under the latent distribution.

Parameters:

• x (-) – The input data. This is only used if mu, std, and x are not provided. Defaults to None.

• mu (-) – The mean of the latent distribution. If this is not provided, then the input data will be encoded to obtain this value. Defaults to None.

• std (-) – The standard deviation of the latent distribution. If this is not provided, then the input data will be encoded to obtain this value. Defaults to None.

• z (-) – The sample from the latent distribution. If this is not provided, then the input data will be encoded to obtain this value. Defaults to None.

Returns:

- prob – The probability of the sample under the latent distribution.

Return type:

torch.Tensor:

sample(x: None | Tensor = None, mu: None | Tensor = None, std: None | Tensor = None) → Tensor

Sample from the latent distribution.

Parameters:

• x (-) – The input data. This is only used if mu and std are not provided. Defaults to None.

• mu (-) – The mean of the latent distribution. If this is not provided, then the input data will be encoded to obtain this value. Defaults to None.

• std (-) – The standard deviation of the latent distribution. If this is not provided, then the input data will be encoded to obtain this value. Defaults to None.

Returns:

- z – The sample from the latent distribution.

Return type:

torch.Tensor:

training: bool

Model Selection

Here is the documentation for the model selection functionality.

class aml.model_selection.TensorboardLoad(path: str, level: int | None = None, verbose: bool = True, n_jobs=1)

Bases: object

Parameters:

• path (str) –

• level (int | None) –

• verbose (bool) –

__init__(path: str, level: int | None = None, verbose: bool = True, n_jobs=1)

This class allows you to load tensorboard files from a directory.

Parameters:

• path (-) – The path of the directory containing the files.

• level (-) – The maximum number of levels to dive into when loading the files. If None then all levels are loaded. Note that level=-1 will behave as if level=None, and will return all of the files, and not just the deepest level. Defaults to None.

• verbose (-) – Whether to print progress when loading the files. Defaults to True.

• n_jobs (-) – The number of parallel operations when loading the data. Defaults to 1.

audio(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the audio written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy audio from all files.

>>> tbload = TensorboardLoad('./')
>>> tbload.audio(
...     'Accuracy',
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

distributions(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the distributions written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy distributions from all files.

>>> tbload = TensorboardLoad('./')
>>> tbload.distributions(
...     'Accuracy',
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

histograms(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the histograms written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy histograms from all files.

>>> tbload = TensorboardLoad('./')
>>> tbload.histograms(
...     'Accuracy',
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

images(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the images written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy images from all files.

>>> tbload = TensorboardLoad('./')
>>> tbload.images(
...     'Accuracy',
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

scalars(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the scalars written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy values from all files, in which the accuracy is more than or equal to 0.5.

>>> tbload = TensorboardLoad('./')
>>> tbload.scalars(
...     'Accuracy',
...     query_expression="value >= 0.5",
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

property tags

This returns the available tags that can be used to filter the results when loading files.

tensors(tags: List[str] | str | None = None, query_expression: str | None = None) → Dict[int, DataFrame]

This function collects all of the tensors written to the tensorboard, with the given tag and querying expression.

Examples

The following would load the accuracy tensors from all files.

>>> tbload = TensorboardLoad('./')
>>> tbload.tensors(
...     'Accuracy',
...     )

Parameters:

• tags (-) – The tag of the results that are required. If None then all tags are returned. Defaults to None.

• query_expression (-) – An expression that will be passed to the pandas dataframe that allows the user to filter out un-wanted rows before that dataframe is concatenated with the results. Defaults to None.

Raises:

TypeError – If the tag is not a string, a list of strings or None.:

Returns:

- out – Pandas dataframe containing the results.

Return type:

Dict[int, pd.DataFrame]:

type_func_dict = {'audio': 'Audio', 'distributions': 'CompressedHistograms', 'graph': 'Graphs', 'histograms': 'Histograms', 'images': 'Images', 'meta_graph': 'MetaGraph', 'run_metadata': 'RunMetadata', 'scalars': 'Scalars', 'tensors': 'Tensors'}

Parallel

Here is the documentation for the parallel compute functionality.

class aml.parallel.ParallelModelling(models: Any, n_jobs: int = 1, backend: str = 'threading', **kwargs)

Bases: object

Parameters:

• models (Any) –

• n_jobs (int) –

• backend (str) –

__init__(models: Any, n_jobs: int = 1, backend: str = 'threading', **kwargs)

This class allows you to train, predict, or anything else on models in parallel.

In fact, this class is general enough to accept any object to perform parallel methods on.

Examples

>>> n_runs = 20
>>> pf = ParallelModelling(
...     models=[
...         ResNetLearning(
...             seed=seed,
...             model_name=f'RN-seed_{seed}',
...             **model_args
...             )
...         for seed in range(n_runs)
...         ],
...     n_jobs=2
...     )
>>> pf.fit(train_loader=train_dl, val_loader=val_dl)
>>> predictions = pf.predict(train_loader=train_dl, val_loader=val_dl)

Similarly, if different models need to be trained with different train loaders, you may also use the list__ prefix. If doing this, please ensure that the argument lists are the same size as the number of models being trained. An example would be:

>>> pf.fit(list__train_loader=train_dl_list, list__val_loader=val_dl_list)

The parallel computations are done using the backend='threading' by default, which means that memory is shared across the processes. This will ensure that datasets can be shared across the computations. If this is not desired, please use backend='loky'.

The returned values for the methods will be a list of the returned values for the method on each of the models. You may also call attributes of the models, which will be returned in the same way.

Note, that if an attribute is callable, then it will be returned as a parallel function, which means that if an attribute is callable and has its own attributes, it will not be accessible in this way. An example of this special case is when trying to access the weights of pytorch models in parallel. pf.layer1.weight will raise an error, as pf.layer1 is callable. To access the weights, please instead use: [model.layer1.weight for model in pf.models].

Parameters:

• models (-) – The models to be wrapped and accessed in parallel. Please ensure that they all have the attribute that you are hoping to call.

• n_jobs (-) – The number of jobs to run in parallel. Be mindful that the models and related computations might be expensive for the CPU, GPU, or RAM. Defaults to 1.

• backend (-) – The backend used for the parallel compute. This should be an acceptable value for joblib.Parallel. Defaults to threading.

• kwargs (-) – All other keyword arguments are passed to joblib.Parallel.

- models

The models supplied in the __init__ function.

- parallel_args

The arguments passed to the joblib.Parallel __init__ function. This will include the number of jobs and the backend being used. Arguments not shown here, but used in joblib.Parallel, will be set to the default values.

Type:

dict:

class aml.parallel.ProgressParallel(tqdm_bar: tqdm | None = None, verbose: bool = True, desc: str = 'In Parallel', total: int | None = None, tqdm_style: Dict[str, Any] = {'ascii': '▏▎▍▋▊▉', 'dynamic_ncols': True}, *args, **kwargs)

Bases: Parallel

Parameters:

• tqdm_bar (tqdm | None) –

• verbose (bool) –

• desc (str) –

• total (int) –

• tqdm_style (Dict[str, Any]) –

__init__(tqdm_bar: tqdm | None = None, verbose: bool = True, desc: str = 'In Parallel', total: int | None = None, tqdm_style: Dict[str, Any] = {'ascii': '▏▎▍▋▊▉', 'dynamic_ncols': True}, *args, **kwargs)

This is a wrapper for the joblib Parallel class that allows for a progress bar to be passed into the __init__ function so that the progress can be viewed.

Recall that using backend='threading' allows for shared access to variables!

Examples

>>> pbar = tqdm.tqdm(total=5)
>>> result = ProgressParallel(
...     tqdm_bar=pbar,
...     n_jobs=10,
...     )(
...         joblib.delayed(f_parallel)(i)
...         for i in range(5)
...     )

Alternatively, you do not need to pass a tqdm bar:

>>> result = ProgressParallel(
...     n_jobs=10,
...     total=20,
...     desc='In Parallel',
...     )(
...         joblib.delayed(f_parallel)(i)
...         for i in range(20)
...     )

Parameters:

• tqdm_bar (-) – The tqdm bar that will be used in the progress updates. Every time progress is displayed, tqdm_bar.update(n) will be called, where n is the number of updates made. If None, then a bar is created inside this class. Defaults to None.

• verbose (-) – If tqdm_bar=None, then this argument allows the user to stop the progress bar from printing at all. Defaults to True.

• desc (-) – If tqdm_bar=None, then this argument allows the user to add a description to the progress bar. Defaults to 'In Parallel'.

• total (-) – If tqdm_bar=None, then this argument allows the user to add a total to the progress bar, rather than let the bar automatically update it as it finds new tasks. If None, then the total might update multiple times as the parallel process queues jobs. Defaults to None.

• tqdm_style (-) – A dictionary passed to the tqdm object which can be used to pass kwargs. desc, total, and disable (verbose) cannot be passed here. Please use the arguments above. Defaults to aml_tqdm_style (see aml.tqdm_style).

print_progress()

Display the process of the parallel execution only a fraction of time, controlled by self.verbose.

aml.parallel.p_apply(func: Callable, n_jobs: int = 1, backend: str = 'threading', verbose: bool = True, **kwargs) → List[Any]

This class allows you to parallelise any function over some inputs.

You may use the prefix list__ to any argument for each element to be parallelised, and not prefix an argument for it to be consistent between parallel computations.

This is more easily seen through example.

Examples

In the following example, the function is parallelised over the y argument, since this is prefixed with list__. This means that x is added to each of the :code:`y`s.

>>> p_apply(
...     lambda x,y: x+y,
...     x=np.array([0,1,2]),
...     list__y=np.array([0,1,2]),
...     )
Parallel function: 3it [00:00, 2000.78it/s]
[array([0, 1, 2]), array([1, 2, 3]), array([2, 3, 4])]

In the next example, the function is parallelised over none of the arguments, since none are prefixed with list__. This means that x is added to y.

>>> p_apply(
...     lambda x,y: x+y,
...     x=np.array([0,1,2]),
...     y=np.array([0,1,2]),
...     )
Parallel function: 1it [00:00, ?it/s]
[array([0, 2, 4])]

Parameters:

• func (-) – The function to be used.

• n_jobs (-) – The number of jobs to run in parallel. Be mindful that the functions and related computations might be expensive for the CPU, GPU, or RAM. Defaults to 1.

• backend (-) – The backend used for the parallel compute. This should be an acceptable value for joblib.Parallel. Defaults to threading.

• verbose (-) – Whether to print progress. Defaults to True.

Returns:

- out – A list of the outputs. The list iterates over the parallel computes.

Return type:

List[Any]:

aml.parallel.p_map(func: Callable, n_jobs: int = 1, backend: str = 'threading', verbose: bool = True) → List[Any]

This class allows you to parallelise any function over some inputs.

You may use the prefix list__ to any argument for each element to be parallelised, and not prefix an argument for it to be consistent between parallel computations.

This is more easily seen through example.

NOTE: This is a more refined version of p_apply, and is recommended to be used instead. The difference being that this function only takes a function as an input, and wraps it. This means you may use some of the same arguments in this function as is in the function you are wrapping.

Examples

In the following example, the function is parallelised over the y argument, since this is prefixed with list__. This means that x is added to each of the :code:`y`s.

>>> p_map(
...     lambda x,y: x+y,
... )(
...     x=np.array([0,1,2]),
...     list__y=np.array([0,1,2]),
... )
Parallel function: 3it [00:00, 2000.78it/s]
[array([0, 1, 2]), array([1, 2, 3]), array([2, 3, 4])]

In the next example, the function is parallelised over none of the arguments, since none are prefixed with list__. This means that x is added to y.

>>> p_map(
...     lambda x,y: x+y,
... )(
...     x=np.array([0,1,2]),
...     y=np.array([0,1,2]),
... )
Parallel function: 1it [00:00, ?it/s]
[array([0, 2, 4])]

Parameters:

• func (-) – The function to be used.

• n_jobs (-) – The number of jobs to run in parallel. Be mindful that the functions and related computations might be expensive for the CPU, GPU, or RAM. Defaults to 1.

• backend (-) – The backend used for the parallel compute. This should be an acceptable value for joblib.Parallel. Defaults to threading.

• verbose (-) – Whether to print progress. Defaults to True.

Returns:

- out – A parallel version of the function given.

Return type:

Callable:

Progress

Here is the documentation for the progress functionality.

class aml.progress.PLTQDMProgressBar(refresh_rate: int = 1, process_position: int = 0)

Bases: TQDMProgressBar

Parameters:

• refresh_rate (int) –

• process_position (int) –

__init__(refresh_rate: int = 1, process_position: int = 0)

Parameters:

• refresh_rate (int) –

• process_position (int) –

init_predict_tqdm() → Tqdm

Override this to customize the tqdm bar for predicting.

Return type:

Tqdm

init_sanity_tqdm() → Tqdm

Override this to customize the tqdm bar for the validation sanity run.

Return type:

Tqdm

init_test_tqdm() → Tqdm

Override this to customize the tqdm bar for testing.

Return type:

Tqdm

init_train_tqdm() → Tqdm

Override this to customize the tqdm bar for training.

Return type:

Tqdm

init_validation_tqdm() → Tqdm

Override this to customize the tqdm bar for validation.

Return type:

Tqdm

on_train_epoch_start(trainer: Trainer, *_: Any) → None

Called when the train epoch begins.

Parameters:

• trainer (Trainer) –

• _ (Any) –

Return type:

None

Random

Here is the documentation for the random functionality.

class aml.random.RandomState(random_state=None)

Bases: object

__init__(random_state=None)

next(n=1)

Utils

Here is the documentation for the util functionality.

class aml.utils.ArgFake(**arguments: Any)

Bases: object

Parameters:

arguments (Any) –

__init__(**arguments: Any)

Fake argparse arguments

Parameters:

arguments (-) – Keyword arguments that will be accessable as attributes of this class.

class aml.utils.NPEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

aml.utils.dirtree(path: str, level: int | None = None, files_only: bool = False, file_path: bool = False) → dict

This function will produce a dictionary of the file structure. All keys with value of None are files, and if files_only=True all values that are part of a list are files.

Examples

>>> dirtree('./')

Parameters:

• path (-) – The path to search over.

• level (-) – The number of levels to recursively search. level=0 is the files in the directory of the path, and level=1 would be all of the files in the directories of the directory of the path, etc. None searches recursively until there are no more directories in the leaves. Note that level=-1 will not return the tree from the last level, but instead act as if level=None. Defaults to None.

• files_only (-) – Whether to only show the files, or the folders too. True will only return the files. Defaults to False.

• file_path (-) – If True, then the returned names will contain the full paths. Defaults to False.

Returns:

- directory_dict – The dictionary containing the file structure.

Return type:

dict:

aml.utils.module_from_file(module_name: str, file_path: str)

Will open a module from a file path.

Edited from https://stackoverflow.com/a/51585877/19451559.

Examples

>>> model_trainer = module_from_file(
...     'model_trainer',
...     './some_code/model_trainer.py'
...     )
>>> train_function = model_trainer.train
<function model_trainer.train(...)>

Parameters:

• module_name (-) – The name of the module to load.

• file_path (-) – File path to that module.

Returns:

- out – A python module that can be used to access objects from within it.

Return type:

module: