core

Abstract base classes and shared infrastructure for the fault-detection pipeline.

Concrete implementations of these contracts live in their own subpackages (autoencoders, anomaly_scores, threshold_selectors, data_preprocessing).

class AnomalyScore(**kwargs)

Bases: BaseEstimator, TransformerMixin, SaveLoadMixin, ABC

Anomaly score template

abstractmethod fit(x, y=None)

Fit the scorer.

Parameters:

• x (Union[DataFrame, ndarray]) – numpy 2d array with differences between prediction and actual sensor values

• y (Optional[Series]) – labels indicating whether sample is normal (True) or anomalous (False).

Return type:

AnomalyScore

abstractmethod transform(x)

Implement transform method.

Parameters:

x (Union[DataFrame, ndarray]) – input data

Return type:

Series

Returns:

Scores

class DataTransformer(**kwargs)

Bases: BaseEstimator, TransformerMixin, SaveLoadMixin, ABC

DataTransformer template.

abstractmethod fit(x, y=None)

Fit the preprocessor with training data. Should set attributes feature_names_in_ and n_features_in_. If columns are selected/dropped, should also set feature_names_out_ and columns_dropped_.

Parameters:

• x (DataFrame) – pandas dataframe with input data

• y (Series) – (optional) labels indicating whether sample is normal (True) or anomalous (False).

Return type:

DataTransformer

abstractmethod get_feature_names_out(input_features=None)

Get output feature names for transformation.

Return type:

List[str]

Returns

feature_names_out : list/np array of str objects.

abstractmethod inverse_transform(x)

Implement inverse transform method

Parameters:

x (DataFrame) – input data.

Return type:

DataFrame

abstractmethod transform(x)

Implement transform method.

Parameters:

x (DataFrame) – input data.

Return type:

DataFrame

class FaultDetectionResult(predicted_anomalies, reconstruction, recon_error, anomaly_score, bias_data=None, arcana_losses=None, tracked_bias=None)

Bases: object

Class to encapsulate results from the fault detection process.

anomaly_score: Series

Series with predicted anomaly scores.

arcana_losses: DataFrame | None = None

DataFrame containing recorded values for all losses in ARCANA. None if ARCANA was not run. Empty if losses were not tracked.

bias_data: DataFrame | None = None

DataFrame with ARCANA results (ARCANA bias). None if ARCANA was not run.

criticality(normal_idx=None, init_criticality=0, max_criticality=1000)

Criticality based on the predicted anomalies.

Parameters:

• normal_idx (Series | None) – A pandas Series with boolean values indicating normal operation, indexed by timestamp. Ignored if None.

• init_criticality (int) – The initial criticality value. Defaults to 0.

• max_criticality (int) – The maximum criticality value. Defaults to 1000.

Return type:

Series

classmethod load(directory, **kwargs)

Loads the results from CSV files in the specified directory.

Restores MultiIndex if the results were saved with one.

Parameters:

• directory (str | Path) – The directory where the CSV files are stored.

• kwargs – other keywords args for pd.read_csv (e.g., sep=’,’)

Returns:

The loaded result object.

Return type:

FaultDetectionResult

predicted_anomalies: Series

Series with a predicted anomalies (bool).

recon_error: DataFrame

DataFrame with reconstruction errors.

reconstruction: DataFrame

DataFrame with reconstruction of the input data.

save(directory, **kwargs)

Saves the results to CSV files in the specified directory.

Parameters:

• directory (str | Path) – The directory where the CSV files will be saved.

• kwargs – other keywords args for pd.DataFrame.to_csv (i.e. sep=’,’)

Return type:

None

tracked_bias: List[DataFrame] | None = None

List of DataFrames containing the ARCANA bias every 50th iteration. None if ARCANA was not run. Empty if bias was not tracked.

class ModelMetadata(model_date, model_path, train_recon_error, val_recon_error=None)

Bases: object

Class to encapsulate metadata about the FaultDetector model.

model_date: str

model_path: str | Path

train_recon_error: ndarray | DataFrame

val_recon_error: ndarray | DataFrame | None = None

class ThresholdSelector

Bases: BaseEstimator, ClassifierMixin, SaveLoadMixin

Template for threshold selectors.

Finds the threshold (fit method) of the given reconstruction errors (x), to be considered anomalous. The predict method returns an array of boolean, indicating which samples of the input are anomalous.

threshold

scores above the threshold is classified as anomaly, below is classified as normal.

fit(x, y=None)

Fit ThresholdSelector object on anomaly score values x and labels y to determine the score threshold. For prediction, every sample with a scores above this threshold is considered an anomaly.

Parameters:

• x (Union[ndarray, Series]) – array or pandas Series with calculated anomaly scores

• y (Union[ndarray, Series]) – labels indicating whether sample is normal (True) or anomalous (False) Optional, if not given, we assume all data provided represents normal behaviour

Return type:

ThresholdSelector

Returns:

The threshold selector

predict(x, **kwargs)

Return a boolean array indicating whether sample is anomalous.

Parameters:

• x (Union[ndarray, Series]) – array with calculated anomaly scores

• kwargs – other arguments for specific threshold selectors.

Return type:

Union[ndarray, Series, Tuple[Union[ndarray, Series], Union[ndarray, Series]]]

Returns:

Boolean array where true = anomaly, false = normal

• core.autoencoder
  • Autoencoder
    • Autoencoder.is_conditional
    • Autoencoder.model
    • Autoencoder.encoder
    • Autoencoder.history
    • Autoencoder.callbacks
    • Autoencoder.__call__()
    • Autoencoder.compile_model()
    • Autoencoder.create_model()
    • Autoencoder.encode()
    • Autoencoder.epochs_completed
    • Autoencoder.fit()
    • Autoencoder.get_reconstruction_error()
    • Autoencoder.load()
    • Autoencoder.predict()
    • Autoencoder.save()
    • Autoencoder.summary()
    • Autoencoder.tune()
    • Autoencoder.tune_decoder()
  • split_inputs()