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()
), or - a 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: str
-
_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[T], pandas.core.frame.DataFrame, numpy.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, numpy.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[T], pandas.core.frame.DataFrame, numpy.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[T], pandas.core.frame.DataFrame, numpy.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: Union[List[T], pandas.core.frame.DataFrame, numpy.ndarray] = None, target: Union[List[T], pandas.core.frame.DataFrame, numpy.ndarray, None] = None, time: Union[List[T], pandas.core.frame.DataFrame, numpy.ndarray, None] = 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)[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
.
- generator_function (Callable, optional) – A generator function which takes in at least one argument
-
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.
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 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
- sfreq (int, optional) – sampling frequency which determines the time step when
-
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 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
- sfreq (int, optional) – sampling frequency which determines the time step when
-
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 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: Optional[int] = 1) → Dict[str, numpy.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]
-
jitter_std
¶ The standard deviation of the distribution to randomly draw jitter from
- sfreq (int, optional) – sampling frequency which determines the time step when
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()
), or - a 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[T], pandas.core.frame.DataFrame, numpy.ndarray], values: Union[List[T], pandas.core.frame.DataFrame, numpy.ndarray]) → Tuple[numpy.ndarray, numpy.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) → numpy.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: Callable = None, uncertainty_generator: Callable = None, sfreq: int = None, **kwargs) → 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
- generator_function (callable) – A generator function which takes in at least one argument
- from dataset file (
Metrologically enabled signal streams¶
-
class
agentMET4FOF.streams.metrological_signal_streams.
MetrologicalMultiWaveGenerator
(sfreq: int = 500, freq_arr: numpy.array = array([50]), amplitude_arr: numpy.array = array([1.]), initial_phase_arr: numpy.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.
- sfreq (int, optional) – Sampling frequency which determines the time step when