import numpy as np
import pandas as pd
from ..events.events_find import _events_find_label
from ..misc import listify
[docs]
def epochs_create(
data,
events=None,
sampling_rate=1000,
epochs_start=0,
epochs_end="from_events",
event_labels=None,
event_conditions=None,
baseline_correction=False,
):
"""**Create Epochs**
Create epochs of a signal or a dataframe.
Parameters
----------
data : DataFrame
A DataFrame containing the different signal(s) as different columns.
If a vector of values is passed, it will be transformed in a DataFrame
with a single 'Signal' column.
events : list or ndarray or dict
Events onset location. If a dict is passed (e.g., from ``events_find()``),
will select only the 'onset' list. If an integer is passed,
will use this number to create an evenly spaced list of events. If None,
will chunk the signal into successive complete blocks of the set duration,
starting at the first sample and including a final full-length block.
sampling_rate : int
The sampling frequency of the signal (in Hz, i.e., samples/second).
epochs_start : int, list
Epochs start relative to events_onsets (in seconds). The start can be negative to start
epochs before a given event (to have a baseline for instance). An integer can be specified
to have the same start for all epochs. A list of equal length to the events can be
specified to have a different start for each epoch.
epochs_end : int, list
Epochs end relative to events_onsets (in seconds). An integer can be specified to have the
same end for all epochs. A list of equal length to the events can be specified to have a
different end for each epoch. If ``"from_events"``, events must be a dict (from
:func:`.events_find`). Duration from events will be used as ``epochs_end``.
event_labels : list
A list containing unique event identifiers. If ``None``, will use the event index number.
event_conditions : list
An optional list containing, for each event, for example the trial category, group or
experimental conditions.
baseline_correction : bool
Whether to subtract a baseline computed separately for each epoch. The baseline
extends from the epoch start to time zero. If the epoch starts after zero,
its first sample is used instead. Defaults to False.
Returns
----------
dict
A dict containing DataFrames for all epochs.
See Also
----------
events_find, events_plot, epochs_to_df, epochs_plot
Examples
----------
* **Example 1**: Find events
.. ipython:: python
import neurokit2 as nk
# Get data
data = nk.data("bio_eventrelated_100hz")
# Find events
events = nk.events_find(data["Photosensor"],
threshold_keep='below',
event_conditions=["Negative", "Neutral", "Neutral", "Negative"])
@savefig p_epochs_create1.png scale=100%
nk.events_plot(events, data)
@suppress
plt.close()
* **Example 2**: Create epochs
.. ipython:: python
epochs = nk.epochs_create(data, events, sampling_rate=100, epochs_end=3)
@savefig p_epochs_create2.png scale=100%
nk.epochs_plot(epochs)
@suppress
plt.close()
* **Example 3**: Baseline correction
.. ipython:: python
epochs = nk.epochs_create(data, events, sampling_rate=100,
epochs_end=3, baseline_correction=True)
@savefig p_epochs_create3.png scale=100%
nk.epochs_plot(epochs)
@suppress
plt.close()
* **Example 4**: Arbitrary epoching
.. ipython:: python
# Chunk into n blocks of 1 second
epochs = nk.epochs_create(data, sampling_rate=100, epochs_end=1)
* **Example 5**: Epochs with list of starting points
.. ipython:: python
epochs = nk.epochs_create(data, events, sampling_rate=100,
epochs_start=[0, -1, -1, 0],
epochs_end=[1, 0, 0, 1], baseline_correction=True)
[len(epochs[i]) for i in epochs.keys()]
"""
# Santize data input
if isinstance(data, tuple): # If a tuple of data and info is passed
data = data[0]
if isinstance(data, (list, np.ndarray, pd.Series)):
data = pd.DataFrame({"Signal": list(data)})
# Sanitize events input
if events is None:
max_duration = (np.max(epochs_end) - np.min(epochs_start)) * sampling_rate
events = np.arange(0, len(data) - max_duration + 1, max_duration)
# Align the beginning of each block, rather than its event, with the data.
events -= np.min(epochs_start) * sampling_rate
if isinstance(events, int):
events = np.linspace(0, len(data), events + 2)[1:-1]
if isinstance(events, dict) is False:
events = _events_find_label(
{"onset": events},
event_labels=event_labels,
event_conditions=event_conditions,
)
event_onsets = list(events["onset"])
event_labels = list(events["label"])
if "condition" in events.keys():
event_conditions = list(events["condition"])
# Create epochs
if epochs_end == "from_events":
if "duration" not in events.keys():
events["duration"] = list(np.diff(events["onset"])) + [len(data) - 1]
epochs_end = [i / sampling_rate for i in events["duration"]]
parameters = listify(
onset=event_onsets,
label=event_labels,
condition=event_conditions,
start=epochs_start,
end=epochs_end,
)
# Find the maximum numbers of samples in an epoch
parameters["duration"] = list(np.array(parameters["end"]) - np.array(parameters["start"]))
epoch_max_duration = int(max(i * sampling_rate for i in parameters["duration"]))
# Extend data by the max samples in epochs * NaN (to prevent non-complete data)
length_buffer = epoch_max_duration
# First createa buffer of the same dtype as data and fill with it 0s
buffer = pd.DataFrame(0, index=range(length_buffer), columns=data.columns).astype(dtype=data.dtypes)
# Only then, we convert the non-integers to nans (because regular numpy's ints cannot be nan)
cols = buffer.select_dtypes(exclude=["int", "int64"]).columns
buffer[cols] = buffer[cols].replace({0.0: np.nan})
# Now we can combine the buffer with the data
data = pd.concat([buffer, data, buffer], ignore_index=True, sort=False)
# Adjust the Onset of the events for the buffer
parameters["onset"] = [i + length_buffer for i in parameters["onset"]]
epochs = {}
for i, label in enumerate(parameters["label"]):
# Find indices
start = parameters["onset"][i] + (parameters["start"][i] * sampling_rate)
end = parameters["onset"][i] + (parameters["end"][i] * sampling_rate)
# Slice dataframe
epoch = data.iloc[int(start) : int(end)].copy()
# Correct index
epoch["Index"] = epoch.index.values - length_buffer
epoch.index = np.linspace(
start=parameters["start"][i],
stop=parameters["end"][i],
num=len(epoch),
endpoint=True,
)
if baseline_correction is True:
baseline_end = 0 if parameters["start"][i] <= 0 else parameters["start"][i]
epoch = epoch - epoch.loc[:baseline_end].mean()
# Add additional
epoch["Label"] = parameters["label"][i]
if parameters["condition"][i] is not None:
epoch["Condition"] = parameters["condition"][i]
# Store
epochs[label] = epoch
# Sanitize dtype of individual columns
for i, epoch_df in epochs.items():
for colname, column in epoch_df.select_dtypes(include=["object", "string"]).items():
# Check whether columns are indices or label/condition
values = column.unique().tolist()
zero_or_one = all(x in [0, 1] for x in values)
if zero_or_one:
# Force to int64
epoch_df[colname] = epoch_df[colname].astype("int64")
else:
epoch_df[colname] = epoch_df[colname].astype("string")
return epochs
