agentMET4FOF streams

Base streams

class agentMET4FOF.streams.base_streams.DataStreamMET4FOF[source]

Abstract class for creating datastreams

Data can be fetched sequentially using next_sample() or all at once all_samples(). This increments the internal sample index _sample_idx.

For sensors data, we assume:

  • The format shape for 2D data stream (timesteps, n_sensors)

  • The format shape for 3D data stream (num_cycles, timesteps , n_sensors)

To create a new DataStreamMET4FOF class, inherit this class and call set_metadata() in the constructor. Choose one of two types of datastreams to be created:

Alternatively, override the next_sample() function if neither option suits the application. For generator functions, sfreq is a required variable to be set on init which sets the sampling frequency and the time-step which occurs when next_sample() is called.

For an example implementation of using generator function, see the built-in SineGenerator class. See tutorials for more implementations.

_quantities

Measured quantities such as sensors readings

Type:

Union[List, DataFrame, np.ndarray]

_target

Target label in the context of machine learning. This can be Remaining Useful Life in predictive maintenance application. Note this can be an unobservable variable in real-time and applies only for validation during offline analysis.

Type:

Union[List, DataFrame, np.ndarray]

_time

dtype can be either float or datetime64 to indicate the time when the _quantities were measured.

Type:

Union[List, DataFrame, np.ndarray]

_current_sample_quantities

Last returned measured quantities from a call to next_sample()

Type:

Union[List, DataFrame, np.ndarray]

_current_sample_target

Last returned target labels from a call to next_sample()

Type:

Union[List, DataFrame, np.ndarray]

_current_sample_time

dtype can be either float or datetime64 to indicate the time when the _current_sample_quantities were measured.

Type:

Union[List, DataFrame, np.ndarray]

_sample_idx

Current sample index

Type:

int

_n_samples

Total number of samples

Type:

int

_data_source_type

Explicitly account for the data source type: either “function” or “dataset”

Type:

str

_generator_function

A generator function which takes in at least one argument time which will be used in next_sample()

Type:

Callable

_generator_parameters

Any additional keyword arguments to be supplied to the generator function. The generator function call for every sample will be supplied with the **generator_parameters.

Type:

Dict

sfreq

Sampling frequency

Type:

int

_metadata

The quantities metadata as time_series_metadata.scheme.MetaData

Type:

MetaData

_default_generator_function(time)[source]

This is the default generator function used, if non was specified

Parameters:

time (Union[List, DataFrame, np.ndarray]) – the time stamps at which to evaluate the function

Returns:

\(f(x) = \sin (2 \pi \cdot ext{self.sfreq} \cdot x)\) evaluated at time

Return type:

np.ndarray

_next_sample_data_source(batch_size: int | None = 1) → Dict[str, List | DataFrame | ndarray][source]

Internal method for fetching latest samples from a dataset

Parameters:

batch_size (int, optional) – number of batches to get from data stream, defaults to 1

Returns:

latest samples in the form:

dict like {
    "quantities": <time series data as a list, np.ndarray or
        pd.Dataframe>,
    "target": <target labels as a list, np.ndarray or pd.Dataframe>,
    "time": <time stamps as a list, np.ndarray or pd.Dataframe of
        float or np.datetime64>
}

Return type:

Dict[str, Union[List, DataFrame, np.ndarray]]

_next_sample_generator(batch_size: int | None = 1) → Dict[str, ndarray][source]

Internal method to generate a batch of samples from the generator function

Parameters:

batch_size (int, optional) – number of batches to get from data stream, defaults to 1

Returns:

latest samples in the form:

dict like {
    "quantities": <time series data as a list, np.ndarray or
        pd.Dataframe>,
    "time": <time stamps as a list, np.ndarray or pd.Dataframe of
        float or np.datetime64>
}

Return type:

Dict[str, Union[List, DataFrame, np.ndarray]]

_set_data_source_type(dt_type: str = 'function')[source]

To explicitly account for the type of data source: either from dataset, or a generator function.

Parameters:

dt_type (str) – Either “function” or “dataset”

all_samples() → Dict[str, List | DataFrame | ndarray][source]

Return all the samples in the data stream

Returns:

all samples in the form:

dict like {
    "quantities": <time series data as a list, np.ndarray or
        pd.Dataframe>,
    "target": <target labels as a list, np.ndarray or pd.Dataframe>,
    "time": <time stamps as a list, np.ndarray or pd.Dataframe of
        float or np.datetime64>
}

Return type:

Dict[str, Union[List, DataFrame, np.ndarray]]

has_more_samples() → bool[source]

Tell if there are more samples to extract

next_sample(batch_size: int | None = 1) → Dict[str, List | DataFrame | ndarray][source]

Fetch the latest samples from the quantities, time and target

Parameters:

batch_size (int, optional) – number of batches to get from data stream, defaults to 1

Returns:

latest samples in the form:

dict like {
    "quantities": <time series data as a list, np.ndarray or
        pd.Dataframe>,
    "target": <target labels as a list, np.ndarray or pd.Dataframe>,
    "time": <time stamps as a list, np.ndarray or pd.Dataframe of
        float or np.datetime64>
}

Return type:

Dict[str, Union[List, DataFrame, np.ndarray]]

randomize_data()[source]

Randomizes the provided quantities, useful in machine learning contexts

reset()[source]

Set the sample count to zero to prepare for new extractions

set_data_source(quantities: List | DataFrame | ndarray = None, target: List | DataFrame | ndarray | None = None, time: List | DataFrame | ndarray | None = None)[source]

This sets the data source by providing up to three iterables: quantities , time and target which are assumed to be aligned.

For sensors data, we assume: The format shape for 2D data stream (timesteps, n_sensors) The format shape for 3D data stream (num_cycles, timesteps , n_sensors)

Parameters:

  • quantities (Union[List, DataFrame, np.ndarray]) – Measured quantities such as sensors readings.

  • target (Optional[Union[List, DataFrame, np.ndarray]]) – Target label in the context of machine learning. This can be Remaining Useful Life in predictive maintenance application. Note this can be an unobservable variable in real-time and applies only for validation during offline analysis.

  • time (Optional[Union[List, DataFrame, np.ndarray]]) – dtype can be either float or datetime64 to indicate the time when the quantities were measured.

set_generator_function(generator_function: Callable | None = None, sfreq: int | None = 50, **kwargs: Any)[source]

Sets the data source to a generator function. By default, this function resorts to a sine wave generator function. Initialisation of the generator’s parameters should be done here such as setting the sampling frequency and wave frequency. For setting it with a dataset instead, see set_data_source().

Parameters:

  • generator_function (Callable, optional) – A generator function which takes in at least one argument time which will be used in next_sample(). Parameters of the function can be fixed by providing additional arguments such as the wave frequency.

  • sfreq (int, optional) – Sampling frequency.

  • **kwargs (Any) – Any additional keyword arguments to be supplied to the generator function. The **kwargs will be saved as _generator_parameters. The generator function call for every sample will be supplied with the **generator_parameters.

set_metadata(device_id: str, time_name: str, time_unit: str, quantity_names: str | Tuple[str, ...], quantity_units: str | Tuple[str, ...], misc: Any | None = None)[source]

Set the quantities metadata as a MetaData object

Details you find in the time_series_metadata.scheme.MetaData documentation.

Parameters:

  • device_id (str) – Name of the represented generator

  • time_name (str) – Name for the time dimension

  • time_unit (str) – Unit for the time

  • quantity_names (iterable of str or str) – A string or an iterable of names of the represented quantities’ values

  • quantity_units (iterable of str or str) – An iterable of units for the quantities’ values

  • misc (Any, optional) – This parameter can take any additional metadata which will be handed over to the corresponding attribute of the created Metadata object

Signal streams

class agentMET4FOF.streams.signal_streams.CosineGenerator(sfreq: int | None = 100, cosine_freq: float | None = 6.283185307179586, amplitude: float | None = 1.0, initial_phase: float | None = 0.0)[source]

Streaming cosine wave generator

Parameters:

  • sfreq (int, optional) – sampling frequency which determines the time step when next_sample() is called, defaults to 500

  • cosine_freq (float, optional) – frequency of wave function, defaults to 50.0

  • amplitude (float, optional) – amplitude of the wave function, defaults to 1.0

  • initial_phase (float, optional) – initial phase of the wave function, defaults to 0.0

static cosine_wave_function(time, cosine_freq, amplitude, initial_phase)[source]

A simple cosine wave generator

Parameters:

  • time (Union[List, DataFrame, np.ndarray]) – the time stamps at which to evaluate the function

  • cosine_freq (float) – frequency of wave function

  • amplitude (float) – amplitude of the wave function

  • initial_phase (float) – initial phase of the wave function

Returns:

the cosine values of the specified curve at time

Return type:

np.ndarray

class agentMET4FOF.streams.signal_streams.SineGenerator(sfreq: int | None = 100, sine_freq: float | None = 6.283185307179586, amplitude: float | None = 1.0, initial_phase: float | None = 0.0)[source]

Streaming sine wave generator

Parameters:

  • sfreq (int, optional) – sampling frequency which determines the time step when next_sample() is called, defaults to 500

  • sine_freq (float, optional) – frequency of wave function, defaults to 50.0

  • amplitude (float, optional) – amplitude of the wave function, defaults to 1.0

  • initial_phase (float, optional) – initial phase of the wave function, defaults to 0.0

static sine_wave_function(time, sine_freq, amplitude, initial_phase)[source]

A simple sine wave generator

Parameters:

  • time (Union[List, DataFrame, np.ndarray]) – the time stamps at which to evaluate the function

  • sine_freq (float) – frequency of wave function

  • amplitude (float) – amplitude of the wave function

  • initial_phase (float) – initial phase of the wave function

Returns:

the sine values of the specified curve at time

Return type:

np.ndarray

class agentMET4FOF.streams.signal_streams.SineWithJitterGenerator(sfreq: int | None = 10, sine_freq: float | None = 0.15915494309189535, amplitude: float | None = 1.0, initial_phase: float | None = 0.0, jitter_std: float | None = 0.02)[source]

Represents a streamed sine signal with jitter

Parameters:

  • sfreq (int, optional) – sampling frequency which determines the time step when next_sample() is called, defaults to 10

  • sine_freq (float, optional) – frequency of wave function, defaults to \(\frac{1}{2 \pi}\)

  • amplitude (float, optional) – amplitude of the wave function, defaults to 1.0

  • initial_phase (float, optional) – initial phase of the wave function, defaults to 0.0

  • jitter_std (float, optional) – the standard deviation of the distribution to randomly draw jitter from, defaults to 0.02

_next_sample_generator(batch_size: int | None = 1) → Dict[str, ndarray][source]

Generate the next batch of samples from the sine function with jitter

Parameters:

batch_size (int, optional) – number of batches to get from data stream, defaults to 1

Returns:

latest samples of the sine signal with jitter in the form:

dict like {
    "quantities": <time series data as np.ndarray>,
    "time": <time stamps as np.ndarray>
}

Return type:

Dict[str, np.ndarray]

property jitter_std

The standard deviation of the distribution to randomly draw jitter from

class agentMET4FOF.streams.signal_streams.StaticSineWithJitterGenerator(num_cycles=1000, jitter_std=0.02)[source]

Represents a fixed length sine signal with jitter

Parameters:

  • num_cycles (int, optional) – numbers of cycles, determines the signal length by \(\pi \cdot num_cycles\), defaults to 1000

  • jitter_std (float, optional) – the standard deviation of the distribution to randomly draw jitter from, defaults to 0.02

Metrologically enabled base streams

class agentMET4FOF.streams.metrological_base_streams.MetrologicalDataStreamMET4FOF(value_unc: float | None = 0.0, time_unc: float | None = 0.0, exp_unc: float | None = None, cov_factor: float | None = 1.0)[source]

Abstract class for creating datastreams with metrological information. Inherits from the DataStreamMET4FOF class

To create a new MetrologicalDataStreamMET4FOF class, inherit this class and call set_metadata() in the constructor. Choose one of two types of datastreams to be created:

Alternatively, override the next_sample() function if neither option suits the application. For generator functions, sfreq is a required variable to be set on init which sets the sampling frequency and the time-step which occurs when next_sample() is called.

For an example implementation of using generator function, see the built-in MetrologicalSineGenerator class. See tutorials for more implementations.

_generator_function_unc

A generator function for the time and quantity uncertainties which takes in at least one argument time which will be used in next_sample(). The return value must be a 2-tuple of time and value uncertainties each of one of the three types:

  • np.ndarray

  • pandas DataFrame

  • list

Type:

Callable

_uncertainty_parameters

Any additional keyword arguments to be supplied to the generator function. Both the calls of the value generator function and of the uncertainty generator function will be supplied with the **_uncertainty_parameters.

Type:

Dict

_default_uncertainty_generator(time: List | DataFrame | ndarray, values: List | DataFrame | ndarray) → Tuple[ndarray, ndarray][source]

Default (standard) uncertainty generator function

Parameters:

  • time (Union[List, DataFrame, np.ndarray]) – timestamps

  • values (Union[List, DataFrame, np.ndarray]) – values corresponding to timestamps

Returns:

constant time and value uncertainties each of the same shape as time

Return type:

Tuple[np.ndarray, np.ndarray]

_next_sample_generator(batch_size: int = 1) → ndarray[source]

Internal method for generating a batch of samples from the generator function. Overrides DataStreamMET4FOF._next_sample_generator(). Adds time uncertainty ut and measurement uncertainty uv to sample

set_generator_function(generator_function: Callable = None, uncertainty_generator: Callable = None, sfreq: int = None, **kwargs: Any | None) → Callable[source]

Set value and uncertainty generators based on user-defined functions. By default, this function resorts to a sine wave generator function and a constant (zero) uncertainty. Initialisation of the generator’s parameters should be done here such as setting the sampling frequency and wave frequency. For setting it with a dataset instead, see set_data_source(). Overwrites the default DataStreamMET4FOF.set_generator_function() method.

Parameters:

  • generator_function (callable) – A generator function which takes in at least one argument time which will be used in next_sample().

  • uncertainty_generator (callable) – An uncertainty generator function which takes in at least one argument time which will be used in next_sample().

  • sfreq (int) – Sampling frequency.

  • **kwargs (Optional[Dict[str, Any]]) – Any additional keyword arguments to be supplied to the generator function. The **kwargs will be saved as _uncertainty_parameters. Both the calls of the value generator function and of the uncertainty generator function will be supplied with the **uncertainty_parameters.

Returns:

The uncertainty generator function

Return type:

Callable

property time_unc: float | Iterable[float]

uncertainties associated with timestamps

Type:

Union[float, Iterable[float]]

property value_unc: float | Iterable[float]

uncertainties associated with the values

Type:

Union[float, Iterable[float]]

Metrologically enabled signal streams

class agentMET4FOF.streams.metrological_signal_streams.MetrologicalMultiWaveGenerator(sfreq: int = 500, freq_arr: array = array([50]), amplitude_arr: array = array([1.]), initial_phase_arr: array = array([0.]), intercept: float = 0, device_id: str = 'MultiWaveDataGenerator', time_name: str = 'time', time_unit: str = 's', quantity_names: str | Tuple[str, ...] = ('Length', 'Mass'), quantity_units: str | Tuple[str, ...] = ('m', 'kg'), misc: Any | None = ' Generator for a linear sum of cosines', value_unc: float | Iterable[float] = 0.1, time_unc: float | Iterable[float] = 0, noisy: bool = True)[source]

Class to generate data as a sum of cosine wave and additional Gaussian noise.

Values with associated uncertainty are returned.

Parameters:

  • sfreq (float) – sampling frequency which determines the time step when next_sample is called.

  • intercept (float) – constant intercept of the signal

  • freq_arr (np.ndarray of float) – array with frequencies of components included in the signal

  • amplitude_arr (np.ndarray of float) – array with amplitudes of components included in the signal

  • initial_phase_arr (np.ndarray of float) – array with initial phases of components included in the signal

  • noisy (bool) – boolean to determine whether the generated signal should be noisy or “clean” defaults to True

class agentMET4FOF.streams.metrological_signal_streams.MetrologicalSineGenerator(sfreq: int = 500, sine_freq: float = 50, amplitude: float = 1.0, initial_phase: float = 0.0, device_id: str = 'SineGenerator', time_name: str = 'time', time_unit: str = 's', quantity_names: str | Tuple[str, ...] = 'Voltage', quantity_units: str | Tuple[str, ...] = 'V', misc: Any | None = 'Simple sine wave generator', value_unc: float = 0.1, time_unc: float = 0)[source]

Built-in class of sine wave generator

Parameters:

  • sfreq (int, optional) – Sampling frequency which determines the time step when next_sample() is called. Defaults to 500.

  • sine_freq (float, optional) – Frequency of the wave function. Defaults to 50.0.

  • amplitude (float, optional) – Amplitude of the wave function. Defaults to 1.0.

  • initial_phase (float, optional) – Initial phase of the wave function. Defaults to 0.0.

  • device_id (str, optional) – Name of the represented generator. Defaults to ‘SineGenerator’.

  • time_name (str, optional) – Name for the time dimension. Defaults to ‘time’.

  • time_unit (str, optional) – Unit for the time. Defaults to ‘s’.

  • quantity_names (iterable of str or str, optional) – An iterable of names of the represented quantities’ values. Defaults to (‘Voltage’)

  • quantity_units (iterable of str or str, optional) – An iterable of units for the quantities’ values. Defaults to (‘V’)

  • misc (Any, optional) – This parameter can take any additional metadata which will be handed over to the corresponding attribute of the created Metadata object. Defaults to ‘Simple sine wave generator’.

  • value_unc (iterable of floats or float, optional) – standard uncertainty(ies) of the quantity values. Defaults to 0.1.

  • time_unc (iterable of floats or float, optional) – standard uncertainty of the time stamps. Defaults to 0.0.

_sine_wave_function(time, sine_freq, amplitude, initial_phase)[source]

A simple sine wave generator