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 onceall_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:from dataset file (
set_data_source()
), ora waveform generator function (
set_generator_function()
).
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 whennext_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 eitherfloat
ordatetime64
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 eitherfloat
ordatetime64
to indicate the time when the_current_sample_quantities
were measured.- Type
Union[List, DataFrame, np.ndarray]
- _data_source_type
Explicitly account for the data source type: either “function” or “dataset”
- Type
- _generator_function
A generator function which takes in at least one argument
time
which will be used innext_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
- _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: Optional[int] = 1) Dict[str, Union[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: Optional[int] = 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, Union[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]]
- next_sample(batch_size: Optional[int] = 1) Dict[str, Union[List, DataFrame, ndarray]] [source]
Fetch the latest samples from the
quantities
,time
andtarget
- 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]]
- set_data_source(quantities: Optional[Union[List, DataFrame, ndarray]] = None, target: Optional[Union[List, DataFrame, ndarray]] = None, time: Optional[Union[List, DataFrame, ndarray]] = None)[source]
This sets the data source by providing up to three iterables:
quantities
,time
andtarget
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 eitherfloat
ordatetime64
to indicate the time when thequantities
were measured.
- set_generator_function(generator_function: Optional[Callable] = None, sfreq: Optional[int] = 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 innext_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: Union[str, Tuple[str, ...]], quantity_units: Union[str, Tuple[str, ...]], misc: Optional[Any] = None)[source]
Set the quantities metadata as a
MetaData
objectDetails 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: Optional[int] = 100, cosine_freq: Optional[float] = 6.283185307179586, amplitude: Optional[float] = 1.0, initial_phase: Optional[float] = 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 500cosine_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
- class agentMET4FOF.streams.signal_streams.SineGenerator(sfreq: Optional[int] = 100, sine_freq: Optional[float] = 6.283185307179586, amplitude: Optional[float] = 1.0, initial_phase: Optional[float] = 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 500sine_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
- class agentMET4FOF.streams.signal_streams.SineWithJitterGenerator(sfreq: Optional[int] = 10, sine_freq: Optional[float] = 0.15915494309189535, amplitude: Optional[float] = 1.0, initial_phase: Optional[float] = 0.0, jitter_std: Optional[float] = 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 10sine_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: Optional[int] = 1) Dict[str, ndarray] [source]
Generate the next batch of samples from the sine function with jitter
- property jitter_std
The standard deviation of the distribution to randomly draw jitter from
Metrologically enabled base streams
- class agentMET4FOF.streams.metrological_base_streams.MetrologicalDataStreamMET4FOF(value_unc: Optional[float] = 0.0, time_unc: Optional[float] = 0.0, exp_unc: Optional[float] = None, cov_factor: Optional[float] = 1.0)[source]
Abstract class for creating datastreams with metrological information. Inherits from the
DataStreamMET4FOF
classTo create a new
MetrologicalDataStreamMET4FOF
class, inherit this class and callset_metadata()
in the constructor. Choose one of two types of datastreams to be created:from dataset file (
set_data_source()
), ora waveform generator function (
set_generator_function()
).
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 whennext_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 innext_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: Union[List, DataFrame, ndarray], values: Union[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 uncertaintyut
and measurement uncertaintyuv
to sample
- set_generator_function(generator_function: Optional[Callable] = None, uncertainty_generator: Optional[Callable] = None, sfreq: Optional[int] = None, **kwargs: Optional[Any]) 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 defaultDataStreamMET4FOF.set_generator_function()
method.- Parameters
generator_function (callable) – A generator function which takes in at least one argument
time
which will be used innext_sample()
.uncertainty_generator (callable) – An uncertainty generator function which takes in at least one argument
time
which will be used innext_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
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: Union[str, Tuple[str, ...]] = ('Length', 'Mass'), quantity_units: Union[str, Tuple[str, ...]] = ('m', 'kg'), misc: Optional[Any] = ' Generator for a linear sum of cosines', value_unc: Union[float, Iterable[float]] = 0.1, time_unc: Union[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: Union[str, Tuple[str, ...]] = 'Voltage', quantity_units: Union[str, Tuple[str, ...]] = 'V', misc: Optional[Any] = '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.