dispel.processing.epochs module#

Epoch-specific processing steps.

class dispel.processing.epochs.CreateLevelEpochStep[source]#

Bases: LevelEpochDefinitionMixIn, TransformStepChainMixIn, MutateDataSetProcessingStepBase

A processing step to create epochs.

This class provides a convenient way to create epochs from one or more data sets by specifying their id, their level_ids or level filter, a transformation function and an epoch definition.

Examples

Assuming you have a data set and a method that derives specific epochs from this data set that are leveraged down the line for further processing. The following example illustrates creating level epochs from raw data sets.

First, we specify a definition for our epochs to be extracted:

>>> from dispel.data.epochs import EpochDefinition
>>> definition = EpochDefinition(
...     id_='epoch-id',
...     name='Epoch name',
...     description='A detailed description of the epoch'
... )

Next, we create a processing step that leverages a data set and returns the start and end of our bouts.

>>> import pandas as pd
>>> from scipy import signal
>>> from dispel.processing.data_set import transformation
>>> from dispel.processing.epochs import CreateLevelEpochStep
>>> class MyCreateLevelEpochStep(CreateLevelEpochStep):
...     data_set_ids = 'data-set-id'
...     definition = definition
...     @transformation
...     def detect_bouts(self, data: pd.DataFrame) -> pd.DataFrame:
...         offset = pd.Timedelta(seconds=5)
...         peaks = signal.find_peaks(data['column'])
...         ts = data.index.iloc[peaks].to_series()
...         return pd.DataFrame(dict(start=ts - offset, end=ts + offset))

The example above inspects the data set for peaks and returns epochs that start five seconds before the peak and end five seconds after.

epoch_data_set_id: str | None = None#: If provided, the epochs will be additionally stored as a data set

get_epoch_data_set(epochs, **kwargs)[source]#

Get raw data set representation of a sequence of epochs.

Parameters:: epochs (Sequence[LevelEpoch]) –
Return type:: RawDataSet

get_epoch_data_set_id(**_kwargs)[source]#

Get the data set id for the newly created epoch data set.

Return type:: str | None

storage_error = 'concatenate'#: The behavior to handle multiple epochs being processed.

transform_data_frame_to_epochs(data, definition, **kwargs)[source]#

Get a sequence of epochs for a data frame.

This function is called if the result from a transformation function was a pandas.DataFrame. It converts each row into an epoch.

Parameters:

data (DataFrame) – The data frame containing the epoch information. Each row is passed to transform_row_to_epoch() to convert it to an epoch.
definition (EpochDefinition) – The epoch definition.
kwargs – Optional keyword arguments pushed down from the processing function.

Returns:

A sequence of epochs representing the provided epochs from data.

Return type:

List[LevelEpoch]

transform_row_to_epoch(row, definition, **_kwargs)[source]#

Get epoch representation for a row in a returned data set.

This function is called by transform_data_frame_to_epochs() to transform a row into an epoch. It is applied when the transformation function returned a pandas.DataFrame.

Parameters:

row (Series) – The row in the data frame to be transformed into an epoch.
definition (EpochDefinition) – The epoch definition.
_kwargs – Optional keyword arguments pushed down from the processing function. This can be used to implement custom logic in transforming rows to epochs based on processing arguments.

Returns:

The epoch representing the provided row. It uses the start and end column/names of the provided series as start and end of the epoch, respectively.

Return type:

LevelEpoch

wrap_result(res, level, reading, **kwargs)[source]#

Wrap the result from the processing function into a class.

Parameters:

res (Epoch | LevelEpoch | DataFrame) –
The result passed from the transformation function. Supported values are Epoch, LevelEpoch, and pandas.DataFrame.

If an Epoch was provided, the start and end times are copied to a new LevelEpoch with the definition obtained from get_definition(). If a LevelEpoch was returned, both values and flag will be copied over. If a pandas.DataFrame was handed back, the data frame will be transformed using transform_data_frame_to_epochs().
level (Level) – The current level
reading (Reading) – The current reading
kwargs (Any) – Additional kwargs

Yields:

LevelProcessingResult – The processing result

Raises:

ValueError – Will be risen if the value returned from the transformation function is of any other type than Epoch or pandas.DataFrame.

Return type:

Generator[LevelProcessingResult | RawDataSetProcessingResult, None, None]

class dispel.processing.epochs.DefaultLevelEpochFilter[source]#

Bases: LevelEpochFilter

A default level epoch filter that passes all epochs for processing.

filter(epochs)[source]#

Filter level epochs.

Parameters:: epochs (Iterable[LevelEpoch]) –
Return type:: Sequence[LevelEpoch]

class dispel.processing.epochs.LevelEpochDefinitionMixIn[source]#

Bases: object

A mixin-class for processing steps producing epoch measure sets.

Parameters:: definition (dispel.data.epochs.EpochDefinition | None) – An optional epoch definition. If no epoch definition is provided, the definition class variable will be used. Alternatively, one can overwrite get_definition() to provide the definition.

definition#

The epoch definition. This will be used in get_definition() by default. You can overwrite the function to implement custom logic.

Type:: dispel.data.epochs.EpochDefinition | None

__init__(*args, **kwargs)[source]#

definition: EpochDefinition | None = None#

get_definition(**_kwargs)[source]#

Get the measure definition.

Parameters:: _kwargs – Optional parameters that will be passed along to the creation of epoch definitions. This can be used to implement custom logic in the epoch definition that depends on processing arguments.
Returns:: The definition of the epoch
Return type:: EpochDefinition

class dispel.processing.epochs.LevelEpochExtractStep[source]#

Bases: LevelEpochProcessingStepMixIn, ExtractStep

A measure extraction processing step for epochs.

Examples

Assuming you have a set of epochs that for which you would like to extract a specific measure from a data set leveraging the data between the start and the end of each epoch, you can do this as follows:

>>> from dispel.data.values import AbbreviatedValue as AV
>>> from dispel.data.measures import MeasureValueDefinition
>>> from dispel.processing.data_set import transformation
>>> from dispel.processing.epochs import LevelEpochExtractStep, LevelEpochIdFilter
>>> class MyLevelEpochExtractStep(LevelEpochExtractStep):
...     data_set_ids = 'data_set_id'
...     epoch_filter = LevelEpochIdFilter('a')
...     definition = MeasureValueDefinition(
...         task_name=AV('My Task', 'MT'),
...         measure_name=AV('Maximum', 'max'),
...         data_type='int16'
...     )
...
...     @transformation
...     def max(self, data: pd.DataFrame) -> float:
...        return data['column'].max()

The example above will create a measure value for each epoch containing the maximum of the ‘column’ column in the data set with the ‘data_set_id’.

get_value(value, **kwargs)[source]#

Get a measure value based on the definition.

Parameters:

value (Any) – The value
kwargs – Optional arguments passed to get_definition().

Returns:

The value wrapped with the definition from get_definition().

Return type:

MeasureValue

wrap_result(res, level, reading, **kwargs)[source]#

Wrap the result from the processing function into a class.

Parameters:

res (Any) – Any result returned by the extraction step. If res is a WrappedResult, the flag contained in the object will be automatically added to the MeasureValue hence the flagged wrapped results will always translate into flagged MeasureValue .
level (Level) – The current level
reading (Reading) – The current reading
kwargs (Any) – Additional kwargs

Yields:

LevelProcessingResult – The processing result

Return type:

Generator[LevelProcessingResult | RawDataSetProcessingResult, None, None]

class dispel.processing.epochs.LevelEpochFilter[source]#

Bases: ABC

A base class to filter level epochs during processing.

LevelEpochFilter provides a basic mechanism for processing steps using LevelEpochProcessingStepMixIn to filter epochs to be processed. Each filter has to implement the filter() method that consumes an iterable of level epochs and returns a list of epochs to be considered during processing.

filter(epochs)[source]#

Filter level epochs.

Parameters:: epochs (Iterable[LevelEpoch]) – The epochs to be filtered.
Raises:: NotImplementedError – This method is not implemented since there is no unambiguous definition of filters.
Return type:: Sequence[LevelEpoch]

class dispel.processing.epochs.LevelEpochIdFilter[source]#

Bases: LevelEpochFilter

A level epoch filter that returns epochs with a specific id.

Parameters:: id – The definition id of the epoch to be matched.

__init__(id_)[source]#

filter(epochs)[source]#

Filter all epochs matching the provided id.

Parameters:: epochs (Iterable[LevelEpoch]) –
Return type:: Sequence[LevelEpoch]

class dispel.processing.epochs.LevelEpochProcessingResult[source]#

Bases: LevelProcessingResult

A processing result originating from processing epochs.

__init__(level, step, sources, result, epoch)#

Parameters:

level (Level) –
step (ProcessingStep) –
sources (Iterable[Reading | Level | RawDataSet | MeasureValue | LevelEpoch] | Reading | Level | RawDataSet | MeasureValue | LevelEpoch) –
result (Level | MeasureValue | MeasureSet | LevelEpoch | RawDataSet) –
epoch (LevelEpoch) –

Return type:

None

epoch: LevelEpoch#

class dispel.processing.epochs.LevelEpochProcessingStepMixIn[source]#

Bases: DataSetProcessingStepProtocol

A mixin class for all processing steps using epochs to create measures.

Parameters:: epoch_filter (dispel.processing.epochs.LevelEpochFilter) – The filter to be used when processing epochs.

Examples

The most common use case will be extracting measures for epochs using LevelEpochExtractStep. The mixin class can also be combined with CreateLevelEpochStep to create new epochs from existing epochs.

>>> import pandas as pd
>>> from dispel.processing.data_set import transformation
>>> from dispel.processing.epochs import (LevelEpochIdFilter,
...     CreateLevelEpochStep, LevelEpochProcessingStepMixIn)
>>> class MyStep(LevelEpochProcessingStepMixIn, CreateLevelEpochStep):
...     data_set_ids = 'data-set-id'
...     epoch_filter = LevelEpochIdFilter('existing-epoch-id')
...     definition = EpochDefinition(
...         id_='epoch-id',
...         name='Epoch name',
...         description='The new epochs derived from existing-epoch-id'
...     )
...
...     @transformation
...     def detect_epochs(self, data: pd.DataFrame) -> pd.DataFrame:
...         new_epoch_data_set = ...
...         return new_epoch_data_set

The above example passes for each epoch with existing-epoch-id the view of data-set-id to the detect_epochs function. The returned data frame in turn will be converted to new epochs defined in MyStep.definition.

__init__(*args, epoch_filter=None, **kwargs)[source]#

Parameters:: epoch_filter (LevelEpochFilter | None) –

epoch_filter: LevelEpochFilter = <dispel.processing.epochs.DefaultLevelEpochFilter object>#: The filter to be used when processing epochs.

get_epoch_data_set_view(epoch, data_set)[source]#

Get the view of a data set specific to an epoch.

This method can be overwritten to implement custom logic to retrieve relevant parts of the passed data_set.

Parameters:

epoch (LevelEpoch) – The epoch for which to return the data set view
data_set (DataFrame) – The data set for which to return a view

Returns:

The data_set view specific to the passed epoch.

Return type:

pandas.DataFrame

get_epoch_data_set_views(data_sets, level, reading, **kwargs)[source]#

Get epoch based data set views for processing.

Parameters:

data_sets (Sequence[DataFrame]) – An iterable of pandas.DataFrame to be processed.
level (Level) – The current level
reading (Reading) – The reading
kwargs – Additional arguments passed from processing.

Returns:

A sequence of tuples that contain the epoch and the respective views of data sets to be processed.

Return type:

Sequence[Tuple[LevelEpoch, Sequence[pandas.DataFrame]]]

get_epoch_filter()[source]#

Get the epoch filter.

This function is called by LevelEpochProcessingStepMixIn.get_epochs() to filter down relevant epochs for processing.

Returns:: The filter to be used for processing.
Return type:: LevelEpochFilter

get_epochs(level, _reading, **_kwargs)[source]#

Get epochs to be processed.

Parameters:

level (Level) – The current level
_reading (Reading) – The reading being processed. This parameter is not used in the default implementation, but can be used in any inheriting class to implement custom logic.
_kwargs – Additional arguments passed from processing.

Returns:

The epochs to be processed. Those are the epochs of the level that passed the epoch filter returned by LevelEpochProcessingStepMixIn.get_epoch_filter().

Return type:

Iterable[LevelEpoch]

process_data_sets(data_sets, level, reading, **kwargs)[source]#

Process the provided data sets.

Parameters:

data_sets (Sequence[DataFrame]) –
level (Level) –
reading (Reading) –

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]