macro_eeg_model.simulation

This package contains the simulation scripts for the model simulation.

Submodules

• macro_eeg_model.simulation.data_processor
• macro_eeg_model.simulation.delay_calculator
• macro_eeg_model.simulation.distributions
• macro_eeg_model.simulation.eeg_analyzer
• macro_eeg_model.simulation.global_simulation
• macro_eeg_model.simulation.simulation_info
• macro_eeg_model.simulation.simulator
• macro_eeg_model.simulation.stationary_model_developer

Classes

DataProcessor	A class responsible for processing EEG data by filtering and segmenting it.
DelayCalculator	A class to calculate delay distributions based on distance and various statistical parameters.
LagDistributions	An enumeration for different types of lag distributions.
DistributionFactory	A factory class responsible for creating different types of distributions based on the provided type.
InverseGEV	A class representing the inverse generalized extreme value (GEV) distribution.
InverseGEVSum	A class representing the sum of two inverse GEV distributions.
EEGAnalyzer	The EEGAnalyzer class is responsible computing the power spectrum of EEG data.
GlobalSimulation	A class responsible for orchestrating the entire simulation process.
SimulationInfo	A class responsible for storing and retrieving information about a simulation.
Simulator	The Simulator class is responsible for simulating EEG data using a vector autoregression (VAR) model.
StationaryModelDeveloper	A class to develop a stationary vector autoregression (VAR) model from given parameters.

Package Contents

class macro_eeg_model.simulation.DataProcessor

A class responsible for processing EEG data by filtering and segmenting it.

static filter_data(data, sample_rate, pass_frequency, stop_frequency)

Filters the data using a high-pass Butterworth filter based on the specified passband and stopband frequencies.

Parameters:

• data (numpy.ndarray) – The input data to be filtered (a 2D array where rows represent time points and columns represent channels/nodes).

• sample_rate (int) – The sample rate of the data in Hz.

• pass_frequency (float) – The passband edge frequency in Hz.

• stop_frequency (float) – The stopband edge frequency in Hz.

Returns:

The filtered data with the same shape as the input data.

Return type:

numpy.ndarray

Raises:

AssertionError – If the frequency values are invalid.

static segment_data(data, sample_rate, nr_nodes)

Segments the data into epochs of 1 second each, evenly dividing the data based on the sample rate.

Parameters:

• data (numpy.ndarray) – The input data to be segmented (a 2D array where rows represent time points and columns represent channels/nodes).

• sample_rate (int) – The sample rate of the data in Hz.

• nr_nodes (int) – The number of nodes (channels) in the data.

Returns:

A 3D array where each slice along the third dimension represents a 1-second epoch of the data. The shape of the array is (t_samples, nr_nodes, nr_epochs), where t_samples is the number of samples per second.

Return type:

numpy.ndarray

class macro_eeg_model.simulation.DelayCalculator(shape_param, scale_param, location_param, truncation_percentile)

A class to calculate delay distributions based on distance and various statistical parameters. The delay is modeled using inverse generalized extreme value (GEV) distributions.

_shape_param

The shape parameter (xi) for the GEV distribution.

Type:

float

_scale_param

The scale parameter (sigma) for the GEV distribution.

Type:

float

_location_param

The location parameter (mu) for the GEV distribution.

Type:

float

_truncation_percentile

The percentile at which to truncate the resulting delay distribution.

Type:

float

_velocity_factor

A constant velocity factor (= 6) used to calculate the scale coefficient from the distance. Expressed in meters per second per micron diameter.

Type:

float

__init__(shape_param, scale_param, location_param, truncation_percentile)

Initializes the DelayCalculator with specified parameters for the GEV distribution and the truncation percentile.

Parameters:

• shape_param (float) – The shape parameter (xi) for the GEV distribution.

• scale_param (float) – The scale parameter (sigma) for the GEV distribution.

• location_param (float) – The location parameter (mu) for the GEV distribution.

• truncation_percentile (float) – The percentile at which to truncate the resulting delay distribution. Must be in the range [0, 1).

get_delays_distribution(tempx, distance)

Generates a probability density function (PDF) for delays using inverse GEV distributions. The distributions are also scaled with parameter computed by _calculate_scale_coefficient().

Depending on whether the distance is a single value or a tuple (in case of a relay station), it either sums inverse GEV distributions or uses a single inverse GEV distribution (see src.simulation.distributions.LagDistributions and src.simulation.distributions.DistributionFactory). It then truncates the resulting PDF using _truncate_result().

Parameters:

• tempx (numpy.ndarray) – The array of time points (x-axis) over which to calculate the delay distribution.

• distance (float or tuple) – The distance(s) over which to calculate the delay. If a tuple, the method will use the sum of two inverse GEV distributions.

Returns:

The truncated probability density function (PDF) representing the delay distribution.

Return type:

numpy.ndarray

Raises:

AssertionError – If the distribution cannot be created.

_calculate_scale_coefficient(distance)

Calculates the scale coefficient for the GEV distribution based on the given distance.

Parameters:

distance (float) – The distance for which to calculate the scale coefficient.

Returns:

The scale coefficient used in the GEV distribution.

Return type:

float

_truncate_result(tempx, result)

Truncates the PDF by setting values beyond a certain index to zero, based on the cumulative distribution function (CDF) and the truncation percentile.

Parameters:

• tempx (numpy.ndarray) – The array of time points (x-axis) corresponding to the PDF.

• result (numpy.ndarray) – The PDF to be truncated.

Returns:

The truncated PDF.

Return type:

numpy.ndarray

Raises:

AssertionError – If the truncation percentile is outside the valid range [0, 1).

class macro_eeg_model.simulation.LagDistributions(*args, **kwds)

Bases: enum.Enum

An enumeration for different types of lag distributions.

INVERSE_GEV

Represents an inverse generalized extreme value (GEV) distribution.

Type:

str

INVERSE_GEV_SUM

Represents the sum of two inverse GEV distributions.

Type:

str

__init__(*args, **kwds)

class macro_eeg_model.simulation.DistributionFactory

A factory class responsible for creating different types of distributions based on the provided type.

static get_distribution(distribution_type, **kwargs)

Creates and returns a distribution object based on the specified type.

Parameters:

• distribution_type (LagDistributions) – The type of distribution to create (e.g., INVERSE_GEV, INVERSE_GEV_SUM).

• kwargs (dict) – The parameters required to initialize the distribution.

Returns:

An instance of a distribution class (e.g., InverseGEV, InverseGEVSum).

Return type:

rv_continuous

Raises:

ValueError – If an unknown distribution type is provided.

class macro_eeg_model.simulation.InverseGEV(lmbd, mu, sigma, xi, *args, **kwargs)

Bases: scipy.stats.rv_continuous

A class representing the inverse generalized extreme value (GEV) distribution.

This class extends scipy.stats.rv_continuous to model the inverse GEV distribution.

lmbd

A scaling parameter applied to the distribution.

Type:

float

mu

The location parameter of the GEV distribution.

Type:

float

sigma

The scale parameter of the GEV distribution.

Type:

float

xi

The shape parameter of the GEV distribution.

Type:

float

__init__(lmbd, mu, sigma, xi, *args, **kwargs)

Initializes the InverseGEV distribution with the specified parameters.

Parameters:

• lmbd (float) – A scaling parameter applied to the distribution.

• mu (float) – The location parameter of the GEV distribution.

• sigma (float) – The scale parameter of the GEV distribution.

• xi (float) – The shape parameter of the GEV distribution.

_argcheck(*args)

Validates the distribution parameters.

Returns:

True if the parameters are valid, False otherwise.

Return type:

bool

_cdf(x, *args)

Calculates the cumulative distribution function (CDF) for the inverse GEV.

Parameters:

x (array_like) – The quantiles at which to evaluate the CDF.

Returns:

The CDF evaluated at the given quantiles.

Return type:

array_like

_pdf(x, *args)

Calculates the probability density function (PDF) for the inverse GEV.

Parameters:

x (array_like) – The quantiles at which to evaluate the PDF.

Returns:

The PDF evaluated at the given quantiles.

Return type:

array_like

_ppf(q, *args)

Calculates the percent point function (PPF), also known as the quantile function, for the inverse GEV.

Parameters:

q (array_like) – The quantiles for which to evaluate the PPF.

Returns:

The PPF evaluated at the given quantiles.

Return type:

array_like

_rvs(*args, size=None, random_state=None)

Generates random variates from the inverse GEV distribution.

Parameters:

• size (int or tuple of ints, optional) – The number of random variates to generate.

• random_state (np.random.RandomState, optional) – A random state instance for reproducibility.

Returns:

The generated random variates.

Return type:

array_like

class macro_eeg_model.simulation.InverseGEVSum(lmbd1, lmbd2, mu, sigma, xi, *args, **kwargs)

Bases: scipy.stats.rv_continuous

A class representing the sum of two inverse GEV distributions.

This class extends scipy.stats.rv_continuous to model the sum of two inverse GEV distributions. The sum is approximated using a kernel density estimate (KDE) of the sum of samples from the two distributions.

lmbd1

A scaling parameter for the first inverse GEV distribution.

Type:

float

lmbd2

A scaling parameter for the second inverse GEV distribution.

Type:

float

mu

The location parameter of the GEV distributions.

Type:

float

sigma

The scale parameter of the GEV distributions.

Type:

float

xi

The shape parameter of the GEV distributions.

Type:

float

_kde

The kernel density estimate of the sum of the two inverse GEV distributions.

Type:

KernelDensity

__init__(lmbd1, lmbd2, mu, sigma, xi, *args, **kwargs)

Initializes the InverseGEVSum distribution with the specified parameters.

Parameters:

• lmbd1 (float) – A scaling parameter for the first inverse GEV distribution.

• lmbd2 (float) – A scaling parameter for the second inverse GEV distribution.

• mu (float) – The location parameter of the GEV distributions.

• sigma (float) – The scale parameter of the GEV distributions.

• xi (float) – The shape parameter of the GEV distributions.

_get_kde()

Generates a kernel density estimate (KDE) for the sum of two inverse GEV distributions.

Returns:

A KDE fitted to the sum of samples from the two inverse GEV distributions.

Return type:

KernelDensity

_argcheck(*args)

Validates the distribution parameters.

Returns:

True if the parameters are valid, False otherwise.

Return type:

bool

_pdf(x, *args)

Evaluates the kernel density estimate (KDE) at the given quantiles to approximate the probability density function (PDF) for the sum of inverse GEVs.

Parameters:

x (array_like) – The quantiles at which to evaluate the PDF.

Returns:

The PDF evaluated at the given quantiles.

Return type:

array_like

class macro_eeg_model.simulation.EEGAnalyzer

The EEGAnalyzer class is responsible computing the power spectrum of EEG data.

static calculate_power(data, sample_rate)

Applies the Fast Fourier Transform (FFT) to the EEG data to calculate the power spectrum. It returns the frequencies and the average power spectrum across epochs/samples per second.

Parameters:

• data (numpy.ndarray) – The EEG data to be analyzed (a 3D array with dimensions (time, nodes, epochs)).

• sample_rate (int) – The sample rate of the EEG data in Hz.

Returns:

A tuple containing:

• frequencies (numpy.ndarray): The array of frequencies corresponding to the power spectrum.

• power (numpy.ndarray): The calculated power spectrum for each frequency and node.

Return type:

tuple

Raises:

ValueError – If the user-defined frequencies are outside the valid range determined by the Nyquist frequency.

static plot_power(frequencies, power, nodes, plots_dir)

Visualizes the power spectrum of the EEG data (for each node/channel) as a line plot.

Parameters:

• frequencies (numpy.ndarray) – The array of frequencies corresponding to the power spectrum.

• power (numpy.ndarray) – The calculated power spectrum for each frequency and node.

• nodes (list[str]) – The list of node/channel names corresponding to the data.

• plots_dir (pathlib.Path) – The directory where the plots are saved.

Raises:

AssertionError – If the plots directory does not exist.

class macro_eeg_model.simulation.GlobalSimulation(config)

A class responsible for orchestrating the entire simulation process. It integrates the development of a stationary model, the simulation of EEG data, data processing, and data analysis.

The class uses: - src.simulation.stationary_model_developer.StationaryModelDeveloper to create a stationary model from the provided configuration. - src.simulation.simulator.Simulator to generate synthetic EEG data based on the model. - src.simulation.data_processor.DataProcessor to filter and segment the simulated data. - src.simulation.eeg_analyzer.EEGAnalyzer to calculate and plot the power spectrum of the EEG data. - src.simulation.simulation_info.SimulationInfo to save the simulation results.

__init__(config)

Initializes the GlobalSimulation class with the provided configuration.

Parameters:

config (ModelConfig) – The configuration object containing parameters for the simulation (instance of the src.config.model_config.ModelConfig class).

run(save_data=False, make_plots=False, verbose=False, simulation_name=None)

Runs the global simulation process, including model development, data simulation, processing, analysis, and optional saving/plotting.

Parameters:

• save_data (bool, optional) – If True, saves the simulation results (default is False).

• make_plots (bool, optional) – If True, generates and saves plots of the connectivity and power spectrum (default is False).

• verbose (bool, optional) – If True, displays progress bars and detailed information during the simulation (default is False).

• simulation_name (str, optional) – The name of the simulation, used for saving the results (default is None).

Returns:

A tuple containing:

• simulation_data (numpy.ndarray): The simulated EEG data.

• frequencies (numpy.ndarray): The array of frequencies corresponding to the power spectrum.

• power (numpy.ndarray): The power spectrum of the simulated EEG data.

Return type:

tuple

class macro_eeg_model.simulation.SimulationInfo(output_dir, nodes=None, distances=None, connectivity_weights=None, sample_rate=None, lag_connectivity_weights=None, simulation_data=None, frequencies=None, power=None)

A class responsible for storing and retrieving information about a simulation. It handles saving and loading the data related to a simulation, such as nodes, distances, connectivity weights, and results.

nodes

The array of nodes used in the simulation.

Type:

numpy.ndarray

distances

The distance matrix between nodes used in the simulation.

Type:

numpy.ndarray

connectivity_weights

The connectivity weights matrix between nodes.

Type:

numpy.ndarray

sample_rate

The sample rate of the simulation in Hz.

Type:

int

lag_connectivity_weights

The lagged connectivity weights matrix used in the VAR model.

Type:

numpy.ndarray

simulation_data

The simulated EEG data.

Type:

numpy.ndarray

frequencies

The array of frequencies corresponding to the power spectrum.

Type:

numpy.ndarray

power

The power spectrum calculated from the simulation data.

Type:

numpy.ndarray

_output_dir

The directory path where simulation results are saved.

Type:

pathlib.Path

__init__(output_dir, nodes=None, distances=None, connectivity_weights=None, sample_rate=None, lag_connectivity_weights=None, simulation_data=None, frequencies=None, power=None)

Initializes the SimulationInfo class with the provided simulation parameters and data.

Parameters:

• output_dir (pathlib.Path) – The path to the output directory where simulation results are saved.

• nodes (numpy.ndarray, optional) – The array of nodes used in the simulation.

• distances (numpy.ndarray, optional) – The distance matrix between nodes used in the simulation.

• connectivity_weights (numpy.ndarray, optional) – The connectivity weights matrix between nodes.

• sample_rate (int, optional) – The sample rate of the simulation in Hz.

• lag_connectivity_weights (numpy.ndarray, optional) – The lagged connectivity weights matrix used in the VAR model.

• simulation_data (numpy.ndarray, optional) – The simulated EEG data.

• frequencies (numpy.ndarray, optional) – The array of frequencies corresponding to the power spectrum.

• power (numpy.ndarray, optional) – The power spectrum calculated from the simulation data.

Raises:

AssertionError – If the output directory does not exist.

save_simulation_info()

Saves the simulation data to the output directory as .npy files. The data includes nodes, distances, connectivity weights, sample rate, lag connectivity weights, simulation data, frequencies, and power spectrum.

load_simulation_info()

Loads all the relevant data of the simulation from the output directory and assigns them to the corresponding attributes of the class.

Raises:

FileNotFoundError – If any of the required files are not found in the output directory.

class macro_eeg_model.simulation.Simulator(lag_connectivity_weights, sample_rate, nr_lags, nr_nodes, t_secs, t_burnit, noise_color, std_noise)

The Simulator class is responsible for simulating EEG data using a vector autoregression (VAR) model. It generates synthetic EEG signals based on the provided lagged connectivity weights, noise characteristics, and other simulation parameters.

_lag_connectivity_weights

The lagged connectivity weights matrix used for the VAR model.

Type:

numpy.ndarray

_sample_rate

The sample rate of the simulation in Hz.

Type:

int

_nr_lags

The number of lags (p) in the VAR(p) model.

Type:

int

_nr_nodes

The number of nodes (channels) in the simulation.

Type:

int

_t_secs

The total time of the simulation in seconds.

Type:

int

_t_burnit

The burn-in time for the simulation in seconds.

Type:

int

_noise_color

The color of the noise to be used in the simulation (‘white’ or ‘pink’).

Type:

str

_std_noise

The standard deviation of the noise to be used in the simulation.

Type:

float

__init__(lag_connectivity_weights, sample_rate, nr_lags, nr_nodes, t_secs, t_burnit, noise_color, std_noise)

Initializes the Simulator with the provided parameters.

Parameters:

• lag_connectivity_weights (numpy.ndarray) – The lagged connectivity weights matrix used for the VAR model.

• sample_rate (int) – The sample rate of the simulation in Hz.

• nr_lags (int) – The number of lags (p) in the VAR(p) model.

• nr_nodes (int) – The number of nodes (channels) in the simulation.

• t_secs (int) – The total time of the simulation in seconds.

• t_burnit (int) – The burn-in time for the simulation in seconds.

• noise_color (str) – The color of the noise to be used in the simulation (‘white’ or ‘pink’).

• std_noise (float) – The standard deviation of the noise to be used in the simulation.

simulate(verbose=False)

The simulation generates synthetic EEG signals by applying the VAR model to the provided lagged connectivity weights and adding noise.

Parameters:

verbose (bool, optional) – If True, displays a progress bar during the simulation (default is False).

Returns:

A 2D array of shape (samples, nodes) containing the simulated EEG data.

Return type:

numpy.ndarray

Raises:

• ValueError – If an invalid noise color is provided.

• AssertionError – If any of the input parameters are invalid (e.g., non-positive values for number of lags, time, or std).

class macro_eeg_model.simulation.StationaryModelDeveloper(nr_lags, nr_nodes, nodes, distances, connectivity_weights, sample_rate, delay_calculator)

A class to develop a stationary vector autoregression (VAR) model from given parameters.

_nr_lags

The number of lags (p) in the VAR(p) model.

Type:

int

_nr_nodes

The number of nodes in the model.

Type:

int

_nodes

The list of node names.

Type:

list[str]

_distances

A matrix containing the distances between nodes.

Type:

numpy.ndarray

_connectivity_weights

The initial connectivity weights between nodes.

Type:

numpy.ndarray

_sample_rate

The sample rate used for the model.

Type:

int

_delay_calculator

An instance of the src.simulation.delay_calculator.DelayCalculator class used to calculate delay distributions.

Type:

DelayCalculator

_tempx

The array of lag indices.

Type:

numpy.ndarray

_delays_x

The array of delay values based on the sample rate.

Type:

numpy.ndarray

__init__(nr_lags, nr_nodes, nodes, distances, connectivity_weights, sample_rate, delay_calculator)

Initializes the StationaryModelDeveloper with the provided parameters.

Parameters:

• nr_lags (int) – The number of lags (p) in the VAR(p) model.

• nr_nodes (int) – The number of nodes in the model.

• nodes (list[str]) – The list of node names.

• distances (numpy.ndarray) – A matrix containing the distances between nodes.

• connectivity_weights (numpy.ndarray) – The initial connectivity weights between nodes.

• sample_rate (int) – The sample rate used for the model.

• delay_calculator (DelayCalculator) – An instance of the src.simulation.delay_calculator.DelayCalculator class used to calculate delay distributions.

develop(verbose=False)

Develops a stationary VAR(p) model.

It calculates the lag connectivity weights using _calculate_lag_connectivity_weights(), and adjusts the overall connectivity weights using _adjust_connectivity_weights() until the model becomes stationary (check with _is_stationary()).

Parameters:

verbose (bool, optional) – If True, displays progress information during the model development (default is False).

Returns:

The lag connectivity weights matrix for the stationary model.

Return type:

numpy.ndarray

_adjust_connectivity_weights()

Adjusts the connectivity weights by scaling them down (preserving the relative weights).

_is_stationary(lag_connectivity_weights)

Determines whether the model is stationary.

It constructs an augmented matrix from the lag connectivity weights and checks if all eigenvalues are within the unit circle.

Parameters:

lag_connectivity_weights (numpy.ndarray) – The matrix of lag connectivity weights.

Returns:

True if the model is stationary (i.e., all eigenvalues are within the unit circle), False otherwise.

Return type:

bool

_calculate_lag_connectivity_weights()

Computes the connectivity weights for each lag between all pairs of nodes using _get_lag_distribution().

Returns:

The matrix of lag connectivity weights.

Return type:

numpy.ndarray

_get_lag_distribution(node1, node2)

Calculates the lag distribution (using :py:attr:_delay_calculator and :py:meth:src.simulation.delay_calculator.DelayCalculator.get_delays_distribution) between two nodes based on their delays and connectivity weights. If the nodes are the same, the distribution is set to zero.

Parameters:

• node1 (int) – The index of the first node.

• node2 (int) – The index of the second node.

Returns:

The lag distribution values, or 0 if the nodes are the same.

Return type:

numpy.ndarray or int

plot_connectivity(lag_connectivity_weights, plots_dir)

Visualizes the lag connectivity weights between nodes as a line plot, showing the relative strength of connections over different delays.

Parameters:

• lag_connectivity_weights (numpy.ndarray) – The matrix of lag connectivity weights to be plotted.

• plots_dir (pathlib.Path) – The directory where the plots are saved.

Raises:

AssertionError – If the plots directory does not exist.