saftig.evaluation

Tooling to automate evaluation of filtering techniques on datasets.

Submodules

Classes

EvaluationDataset

A representation of a dataset for the evaluation of noise mitigation methods.

EvaluationMetric

Parent class for evaluation metrics

EvaluationMetricScalar

Parent class for evaluation metrics that yield a scalar value

EvaluationMetricPlottable

Parent class for evaluation metrics that provide a plotting feature

MSEMetric

The MSE of the residual signal

RMSMetric

The RMS of the residual signal

BandwidthPowerMetric

The signal power on a given frequency range

PSDMetric

Plots the PSD of the given signal

EvaluationRun

Representation of an evaluation run

TestDataGenerator

Generate simple test data for correlated noise mitigation techniques

ReportElement

Parent class for elements placed in a report object that generate latex code

Functions

rms(a)

Calculate the root mean square value of an array

total_power(a)

calculate the total power of a signal (square or RMS)

residual_power_ratio(target, prediction[, start, ...])

Calculate the ratio between residual power of the residual and the target signal

residual_amplitude_ratio(*args, **kwargs)

Calculate the ratio between residual amplitude of the residual and the target signal

measure_runtime(filter_classes[, n_samples, n_filter, ...])

Measure the runtime of filers for a specific scenario

Package Contents

saftig.evaluation.rms(a)

Calculate the root mean square value of an array

Parameters:

a (collections.abc.Sequence[float] | numpy.typing.NDArray[numpy.floating]) –

Return type:

float

saftig.evaluation.total_power(a)

calculate the total power of a signal (square or RMS)

>>> import saftig, numpy
>>> signal = numpy.ones(10) * 2
>>> saftig.evaluation.total_power(signal)
4.0
Parameters:

a (collections.abc.Sequence[float] | numpy.typing.NDArray[numpy.floating]) –

Return type:

float

class saftig.evaluation.EvaluationDataset(sample_rate, witness_conditioning, target_conditioning, witness_evaluation, target_evaluation, signal_conditioning=None, signal_evaluation=None, name='Unnamed', target_unit='1')

A representation of a dataset for the evaluation of noise mitigation methods.

Provided sequences will be stored as immutable float64 numpy arrays.

Parameters:

  • sample_rate (float) – Sample rate in Hz

  • witness_conditioning (collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]) – witness channel data for the conditioning format: witness_conditioning[sequence_idx][channel_idx][sample_idx]

  • target_conditioning (collections.abc.Sequence[NDArrayF]) – target channel data for the conditioning format: witness_conditioning[sequence_idx][sample_idx]

  • witness_evaluation (collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]) – witness channel data for the evaluation

  • target_evaluation (collections.abc.Sequence[NDArrayF]) – target channel data for the evaluation

  • signal_conditioning (collections.abc.Sequence[NDArrayF] | None) – (Optional) A signal that can be subtracted from the target for performance metrics

  • signal_evaluation (collections.abc.Sequence[NDArrayF] | None) – (Optional) A signal that can be subtracted from the target for performance metrics

  • name (str) – (Optional) a string describing the dataset

  • target_unit (str) –

sample_rate: float
witness_conditioning: collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]
target_conditioning: collections.abc.Sequence[NDArrayF]
witness_evaluation: collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]
target_evaluation: collections.abc.Sequence[NDArrayF]
signal_conditioning: collections.abc.Sequence[NDArrayF] | None
signal_evaluation: collections.abc.Sequence[NDArrayF] | None
name: str
target_unit: str
static _prepare_dataset(witness_inp, target_inp, signal_inp=None)

Convert input to immutable np.float64 arrays and check shape

Parameters:

  • witness_inp (collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]) –

  • target_inp (collections.abc.Sequence[NDArrayF]) –

  • signal_inp (collections.abc.Sequence[NDArrayF] | None) –

Return type:

tuple[collections.abc.Sequence[collections.abc.Sequence[NDArrayF]], collections.abc.Sequence[NDArrayF], collections.abc.Sequence[NDArrayF] | None]

get_min_sequence_len(separate=False)

Get the length of the shortest sequence in the dataset

Parameters:

separate (bool) – If True, returns the minimum separately for conditioning and evaluation data.

Return type:

int | tuple[int, int]

static _hash_wts_data(witness, target, signal=None)

Calculate a hash value for a set of witness, target, signal data

Parameters:

  • witness (collections.abc.Sequence[collections.abc.Sequence[NDArrayF]]) –

  • target (collections.abc.Sequence[NDArrayF]) –

  • signal (collections.abc.Sequence[NDArrayF] | None) –

hash_bytes()

return a hash over the dataset data as a bytes object

Return type:

bytes

__hash__()
Return type:

int

class saftig.evaluation.EvaluationMetric(**kwargs)

Bases: abc.ABC

Parent class for evaluation metrics

applied = False
prediction: collections.abc.Sequence[numpy.typing.NDArray]
dataset: saftig.evaluation.dataset.EvaluationDataset
residual: collections.abc.Sequence[numpy.typing.NDArray]
parameters: dict
name: str
method_hash_value: bytes
static init_wrapper(func)

A decorator for the __init__function

Saves a hash value for the configuration

apply(prediction, dataset)

Apply this filter

Parameters:
Return type:

Self

property result_full: tuple
Abstractmethod:

Return type:

tuple

The raw data of the result

property result: Any

The result of the metric evaluation

Return type:

Any

classmethod result_to_text(result_full)

String indicating the evaluation result

Parameters:

result_full (tuple[float | numpy.floating, Ellipsis]) –

Return type:

str

property text

The text representation of the evaluation result

classmethod _file_hash()

Calculates a hash value based on the file in which this method was defined.

Return type:

bytes

property method_hash: bytes

A hash representing the configured metric as a bytes object

Return type:

bytes

property method_hash_str: str

A hash representing the configured metric as a base64 like string

Return type:

str

static result_full_wrapper(func)

A decorator for the result_full member function.

Raises an exception if result is accessed on an object that was not applied to data. Caches the result to prevent double calculation.

class saftig.evaluation.EvaluationMetricScalar(**kwargs)

Bases: EvaluationMetric

Parent class for evaluation metrics that yield a scalar value

property result: float

The raw data of the result

Return type:

float

classmethod result_to_text(result_full)

String indicating the evaluation result

Parameters:

result_full (tuple[float | numpy.floating, Ellipsis]) –

Return type:

str

class saftig.evaluation.EvaluationMetricPlottable(**kwargs)

Bases: EvaluationMetric

Parent class for evaluation metrics that provide a plotting feature

abstract plot(ax)

Generate a result plot on the given axes object

Parameters:

ax (matplotlib.axes.Axes) –

save_plot(fname, figsize=(6, 3), tight_layout=True)

Save the plot to a file

Parameters:

  • fname (str | pathlib.Path) –

  • figsize (tuple[int, int]) –

  • tight_layout (bool) –

class saftig.evaluation.MSEMetric(**kwargs)

Bases: EvaluationMetricScalar

The MSE of the residual signal

name = 'Residual MSE'
property result_full: tuple[numpy.floating | float]

The raw data of the result

Return type:

tuple[numpy.floating | float]

class saftig.evaluation.RMSMetric(**kwargs)

Bases: EvaluationMetricScalar

The RMS of the residual signal

name = 'Residual RMS'
property result_full: tuple[numpy.floating | float]

The raw data of the result

Return type:

tuple[numpy.floating | float]

static result_to_text(result_full)

String indicating the evaluation result

Parameters:

result_full (tuple[float | numpy.floating, Ellipsis]) –

Return type:

str

class saftig.evaluation.BandwidthPowerMetric(f_start, f_stop, n_fft=1024, window='hann')

Bases: EvaluationMetricScalar

The signal power on a given frequency range

The spectrum is calculated with welch on each sequence. An average weighted by the sequence length is used to combine spectra from the sequences. The closes bins to f_start and f_stop is chosen as the integration borders.

Parameters:

  • f_start (float) – The frequency at which the power integration starts

  • f_stop (float) – The frequency at which the power integration stops

  • n_fft (int) – Sample count per FFT block used by welch

  • window – The FFT window type

name = 'Residual power on frequency range'
f_start
f_stop
n_fft = 1024
window = 'hann'
property result_full

The raw data of the result

class saftig.evaluation.PSDMetric(n_fft=1024, window='hann', logx=True, logy=True, show_target=True)

Bases: EvaluationMetricPlottable

Plots the PSD of the given signal

The spectrum is calculated with Welch on each sequence. An average weighted by the sequence length is used to combine spectra from the sequences. The closes bins to f_start and f_stop is chosen as the integration borders.

Parameters:

  • n_fft (int) – Sample count per FFT block used by Welch’s method

  • window (str) – fft window type

  • logx (bool) – Logarithmic x scale

  • logy (bool) – Logarithmic y scale

  • show_target (bool) –

name = 'Power spectral density'
n_fft = 1024
window = 'hann'
logx = True
logy = True
show_target = True
_welch_multiple_sequences(signal)

apply welch_multiple_sequences() with correct settings

Parameters:

signal (collections.abc.Sequence[numpy.typing.NDArray]) –

property result_full: tuple[numpy.typing.NDArray, numpy.typing.NDArray]

The raw data of the result

Return type:

tuple[numpy.typing.NDArray, numpy.typing.NDArray]

plot(ax)

Plot to the given Axes object

Parameters:

ax (matplotlib.axes.Axes) –

class saftig.evaluation.EvaluationRun(method_configurations, dataset, optimization_metric, metrics=None, name='unnamed', directory='.')

Representation of an evaluation run

Parameters:
multi_sequence_support = True
method_configurations
dataset
optimization_metric
metrics = []
name = 'unnamed'
directory
all_configurations_list: list | None = None
get_all_configurations()

Returns a list of all unique (filter_technique, configuration) pairs.

Return type:

list

_create_folder_structure()

Create standardized folder structure for results

Return type:

None

static save_np_array_list(data, filename)

Save a list of numpy arrays to a .npz file

Parameters:

  • data (collections.abc.Sequence[collections.abc.Sequence[NDArrayF]] | collections.abc.Sequence[NDArrayF] | NDArrayF) –

  • filename (str | pathlib.Path) –

Return type:

None

static load_np_array_list(filename)

Load a list of numpy arrays from a .npz file

Parameters:

filename (str | pathlib.Path) –

Return type:

collections.abc.Sequence[NDArrayF]

get_prediction(filter_technique, conf)

Load the prediction created by applying the given filter and configuration to the dataset

Parameters:
run(select=None)

Execute the evaluation run

Parameters:

select (int | None) – select one specific run from the list yielded by get_all_configurations

Returns:

generates (Prediction, optimization_metric, other_metrics) objects

Return type:

collections.abc.Generator[tuple[collections.abc.Sequence[NDArrayF], saftig.evaluation.metrics.EvaluationMetricScalar, collections.abc.Sequence[saftig.evaluation.metrics.EvaluationMetric]], None, None]

class saftig.evaluation.TestDataGenerator(witness_noise_level=0.1, target_noise_level=0, transfer_function=1, sample_rate=1.0, rng_seed=None)

Generate simple test data for correlated noise mitigation techniques The channel count is implicitly defined by the shape of witness_noise_level

Parameters:

  • witness_noise_level (float | collections.abc.Sequence) – amplitude ratio of the sensor noise to the correlated noise in the witness sensor Scalar or 1D-vector for multiple sensors

  • target_noise_level (float) – amplitude ratio of the sensor noise to the correlated noise in the target sensor

  • transfer_functon – ratio between the amplitude in the target and witness signals

  • sample_rate (float) – The outputs are referenced to an ASD of 1/sqrt(Hz) if a sample rate is provided

  • transfer_function (float) –

  • rng_seed (int | None) – 

>>> import saftig as sg
>>> # create data with two witness sensors with relative noise amplitudes of 0.1
>>> tdg = sg.evaluation.TestDataGenerator(witness_noise_level=[0.1, 0.1])
>>> # generate a dataset with 1000 samples
>>> witness, target = tdg.generate(1000)
>>> witness.shape, target.shape
((2, 1000), (1000,))
rng: Any
witness_noise_level
target_noise_level
transfer_function
sample_rate = 1.0
scaled_whitenoise(shape)

Generate whitenoise with an ASD of one

Parameters:

shape – shape of the new array

Returns:

Array of white noise

Return type:

NDArrayF

generate(n)

Generate sequences of samples

Parameters:

  • N – number of samples

  • n (int) –

Returns:

witness signal, target signal

Return type:

tuple[NDArrayF, NDArrayF]

generate_multiple(n)

Generate sequences of samples

Parameters:

  • N – Tuple with the length of the sequences

  • n (collections.abc.Sequence[int] | NDArrayU) –

Returns:

witness signals, target signals

Return type:

tuple[collections.abc.Sequence, collections.abc.Sequence]

dataset(n_condition, n_evaluation, sample_rate=1.0, name=None)

Generate an EvaluationDataset

Parameters:

  • n_condition (collections.abc.Sequence[int] | numpy.typing.NDArray[numpy.uint]) – Sequence of integers indicating the number of conditioning samples generated per sample sequence

  • n_evaluation (collections.abc.Sequence[int] | numpy.typing.NDArray[numpy.uint]) – Number of evaluation samples

  • sample_rate (float) – (Optional) Sample rate for the generate EvaluationDataset

  • name (str | None) – (Optional) Specify the name of the EvaluationDataset

Return type:

saftig.evaluation.dataset.EvaluationDataset

Example: >>> # generate two sequences of 100 samples each of conditioning data and one 100 sample sequence of evaluation data >>> import saftig as sg >>> ds = sg.evaluation.TestDataGenerator().dataset((100, 100), (100,))

saftig.evaluation.residual_power_ratio(target, prediction, start=None, stop=None, remove_dc=True)

Calculate the ratio between residual power of the residual and the target signal

Parameters:

  • target (collections.abc.Sequence) – target signal array

  • prediction (collections.abc.Sequence) – prediction array (same length as target

  • start (int | None) – use only a section of the arrays, start at this index

  • stop (int | None) – use only a section of the arrays, stop at this index

  • component (remove DC) – remove DC component before calculation

  • remove_dc (bool) –

Return type:

float

saftig.evaluation.residual_amplitude_ratio(*args, **kwargs)

Calculate the ratio between residual amplitude of the residual and the target signal

Parameters:

  • target – target signal array

  • prediction – prediction array (same length as target

  • start – use only a section of the arrays, start at this index

  • stop – use only a section of the arrays, stop at this index

  • component (remove DC) – remove DC component before calculation

Return type:

float

saftig.evaluation.measure_runtime(filter_classes, n_samples=int(10000.0), n_filter=128, idx_target=0, n_channel=1, additional_filter_settings=None, repititions=1)

Measure the runtime of filers for a specific scenario Be aware that this gives no feedback upon how much multithreading is used!

Parameters:

  • n_samples (int) – Length of the test data

  • n_filter (int) – Length of the FIR filters / input block size

  • idx_target (int) – Position of the prediction

  • n_channel (int) – Number of witness sensor channels

  • additional_filter_settings (collections.abc.Sequence[dict] | None) – optional settings passed to the filters

  • repititions (int) – how manu repititions to perform during the timing measurement

  • filter_classes (collections.abc.Sequence[saftig.filtering.FilterBase]) –

Returns:

(time_conditioning, time_apply) each in seconds

Return type:

tuple[collections.abc.Sequence, collections.abc.Sequence]

class saftig.evaluation.ReportElement

Bases: abc.ABC

Parent class for elements placed in a report object that generate latex code

abstract latex()

Generate latex code

Return type:

str