Cognitive load weakens neural speech tracking without altering response timing.
The 8 matches
- [1] § Methods › Source estimation of TRF weights difference ↔ plot_scripts/plot_trf_sourcelocation.py, lines 80–153 · score 0.87 · ad hoc, Norm, SNR, depth, loose, MNE
- [2] § Methods › EEG recording and preprocessing ↔ calculate_scripts/calculate_power.py, lines 20–27 · score 0.70 · 0.5–4 Hz, frequency bands, 4–8 Hz, power, beta, gamma
- [3] § Methods › Region of interest ↔ plot_scripts/plot_connectivity.py, lines 51–137 · score 0.69 · selected electrodes, occipital, map, parietal, frontal, brain
- [4] § Methods › Network topology analysis ↔ plot_scripts/plot_connectivity.py, lines 1238–1289 · score 0.67 · global efficiency, clustering coefficient, shortest, density, thresholded, metrics
- [5] § Methods › Key difference subnetwork extraction ↔ plot_scripts/plot_connectivity.py, lines 906–946 · score 0.62 · largest connected component, surface, subnetwork, nodes, Edges, weights
- [6] § Methods › Source estimation of TRF weights difference ↔ plot_scripts/plot_trf_sourcelocation.py, lines 1–19 · score 0.57 · source localization, MNE, template, Python, fsaverage, TRF
- [7] § Results › Graph metrics indicate a more clustered and more strongly small-world delta-band topology under HCL ↔ plot_scripts/plot_connectivity.py, lines 1238–1289 · score 0.57 · clustering coefficient, Global efficiency, metrics, density, thresholded, edge
- [8] § Results › Cognitive load reduces the speech-envelope information recoverable from the EEG ↔ plot_scripts/plot_pcc.py, lines 41–43 · score 0.55 · Pearson correlation coefficient, ADT Network, decoding
Paper
Loaded from Europe PMC by your browser, not stored by OSCR: doi.org · Europe PMC
The paper is loaded when this pane is shown.
The authors' code
Python · 1,445 lines · 65 KB · no license · 4 matches
- import numpy as np
- import pandas as pd
- import matplotlib.pyplot as plt
- import seaborn as sns
- from scipy import stats
- from scipy.signal import welch
- from scipy.stats import permutation_test
- import mne
- from mne_connectivity.spectral import spectral_connectivity_epochs
- import networkx as nx
- from nilearn import datasets, plotting, surface
- from nilearn.surface import vol_to_surf
- import warnings
- import math
- from statsmodels.stats.multitest import multipletests
- import json
- warnings.filterwarnings('ignore')
- # Set English font
- plt.rcParams['font.family'] = 'Times New Roman'
- class BrainConnectivityAnalyzer:
- def __init__(self, data_path='results/delta/eeg_delta.npy'):
- """Initialize Analyzer"""
- self.data_path = data_path
- self.data = None
- self.low_load_data = None
- self.high_load_data = None
- self.channel_names = None
- self.sfreq = 128 # Sampling frequency
- self.connectivity_matrices = {}
- self.statistical_results = {}
- self.network_metrics = {}
- # Define regions of interest and channel mapping
- self.regions = {
- 'Frontal': ['Fp1', 'Fp2', 'F7', 'F8', 'F3', 'F4'],
- 'Central': ['C3', 'C4'],
- 'Parietal': ['P3', 'P4'],
- 'Temporal': ['T7', 'T8', 'P7', 'P8'],
- 'Occipital': ['O1', 'O2']
- }
- # All selected channels
- self.selected_channels = [ch for region in self.regions.values() for ch in region]
- print(f"Selected channels ({len(self.selected_channels)}): {self.selected_channels}")
- # Load channel info
- self.load_channel_info()
- def load_channel_info(self):
- """Load channel info, keep only selected electrodes"""
- ch_pos = {'Fp1': [-0.026133014404070235, 0.08078401376909139, -0.00400108454195971],
- 'Fpz': [0.0, 0.08498123361344626, -0.0017860385037488254],
- 'Fp2': [0.026133014404070235, 0.08078401376909139, -0.00400108454195971],
- 'AF7': [-0.049709431314887954, 0.0686910763510323, -0.005958898227616103],
- 'AF3': [-0.03148279679848069, 0.07615276676848455, 0.020846813167733094],
- 'AF4': [0.03148279679848069, 0.07615276676848455, 0.020846813167733094],
- 'AF8': [0.04966890402811598, 0.06872089942163155, -0.00595297869371352],
- 'F7': [-0.06842333502695395, 0.04987137794892023, -0.0074895183600262386],
- 'F5': [-0.06305822186454818, 0.05403788811325115, 0.018846813167733094],
- 'F3': [-0.048200427317538846, 0.05755106339309903, 0.039869711671018535],
- 'F1': [-0.026042093389975367, 0.0599127302448179, 0.054380824988956196],
- 'Fz': [0.0, 0.06073848094846255, 0.059462903831491894],
- 'F2': [0.02602543804214761, 0.059874412766011756, 0.05443097712368926],
- 'F4': [0.04814259646845235, 0.05758402610681053, 0.0398919834378528],
- 'F6': [0.06304473912257513, 0.05402633404653858, 0.018846813167733094],
- 'F8': [0.06838359029760958, 0.049926526811881665, -0.007485085070400889],
- 'FT7': [-0.08041001431187064, 0.026207503532593573, -0.008508604877055598],
- 'FC5': [-0.07624736450995313, 0.028762823435357604, 0.024166906986885676],
- 'FC3': [-0.059274978176089226, 0.030955284953191532, 0.05247139502329684],
- 'FC1': [-0.03235137713122826, 0.03243618389873952, 0.07159806122933911],
- 'FCz': [0.0, 0.03292788363525597, 0.07836296624875198],
- 'FC2': [0.03235137713122826, 0.03243618389873952, 0.07159806122933911],
- 'FC4': [0.059274978176089226, 0.030955284953191532, 0.05247139502329684],
- 'FC6': [0.07624736450995313, 0.028762823435357604, 0.024166906986885676],
- 'FT8': [0.08041001431187064, 0.026207503532593573, -0.008508604877055598],
- 'T7': [-0.08453853863965728, 5.176482896376251e-18, -0.008845082513531118],
- 'C5': [-0.08083154804902482, 4.9495048294181866e-18, 0.02629183979865601],
- 'C3': [-0.06317128071259069, 3.868125336135656e-18, 0.056871691491734856],
- 'C1': [-0.03453737403184571, 2.114804227952738e-18, 0.07766704445892345],
- 'Cz': [0.0, 5.204748896376251e-18, 0.085],
- 'C2': [0.03460920316454118, 2.11920249382479e-18, 0.07763506331752111],
- 'C4': [0.06316731016557851, 3.867882210251189e-18, 0.05687610154050295],
- 'C6': [0.08083154804902482, 4.9495048294181866e-18, 0.02629183979865601],
- 'T8': [0.08453853863965728, 5.176482896376251e-18, -0.008845082513531118],
- 'TP7': [-0.08041001431187064, -0.02620750353259356, -0.008508604877055598],
- 'CP5': [-0.07624736450995313, -0.028762823435357594, 0.024166906986885676],
- 'CP3': [-0.05927497817608923, -0.03095528495319151, 0.05247139502329684],
- 'CP1': [-0.03235137713122826, -0.032436183898739514, 0.07159806122933911],
- 'CP2': [0.03235137713122826, -0.032436183898739514, 0.07159806122933911],
- 'CP4': [0.05927497817608923, -0.03095528495319151, 0.05247139502329684],
- 'CP6': [0.07624736450995313, -0.028762823435357594, 0.024166906986885676],
- 'TP8': [0.08038510202811598, -0.026284771809974194, -0.008505652714920988],
- 'P7': [-0.06842333502695394, -0.049871377948920236, -0.0074895183600262386],
- 'P5': [-0.06305822186454818, -0.05403788811325114, 0.018846813167733094],
- 'P3': [-0.04820042731753886, -0.057551063393099025, 0.039869711671018535],
- 'P1': [-0.026042093389975374, -0.05991273024481789, 0.054380824988956196],
- 'Pz': [7.438318627860717e-18, -0.06073848094846255, 0.059462903831491894],
- 'P2': [0.026025438042147617, -0.05987441276601174, 0.05443097712368926],
- 'P4': [0.04814259646845235, -0.05758402610681054, 0.0398919834378528],
- 'P6': [0.06304473912257513, -0.054026334046538574, 0.018846813167733094],
- 'P8': [0.0683835902976096, -0.04992652681188165, -0.007485085070400889],
- 'PO7': [-0.04970943131488796, -0.0686910763510323, -0.005958898227616103],
- 'PO5': [-0.042951551117820216, -0.07280389857511523, 0.008930565565948855],
- 'PO3': [-0.03148279679848071, -0.07615276676848454, 0.020846813167733094],
- 'POz': [9.677837321860717e-18, -0.07902553885914161, 0.059462903831491894],
- 'PO4': [0.03148279679848071, -0.07615276676848454, 0.020846813167733094],
- 'PO6': [0.042951551117820216, -0.07280389857511523, 0.008930565565948855],
- 'PO8': [0.04966890402811598, -0.06872089942163155, -0.00595297869371352],
- 'O1': [-0.026133014404070225, -0.0807840137690914, -0.00400108454195971],
- 'Oz': [1.0407199573230042e-17, -0.08498123361344626, -0.0017860385037488254],
- 'O2': [0.026133014404070225, -0.0807840137690914, -0.00400108454195971]}
- # Keep only selected electrodes
- ch_pos = {k: v for k, v in ch_pos.items() if k in self.selected_channels}
- self.channel_names = list(ch_pos.keys())
- self.channel_positions = ch_pos
- # Create brain region color map
- self.region_colors = {
- 'Frontal': 'red',
- 'Central': 'blue',
- 'Parietal': 'green',
- 'Temporal': 'purple',
- 'Occipital': 'orange'
- }
- # Assign region color to each channel
- self.channel_colors = []
- for ch in self.channel_names:
- for region, channels in self.regions.items():
- if ch in channels:
- self.channel_colors.append(self.region_colors[region])
- break
- print(f"Set {len(self.channel_names)} channels")
- def load_data(self):
- """Load data, keep only selected electrodes"""
- print("Loading data...")
- self.data = np.load(self.data_path)
- print(f"Original data shape: {self.data.shape}")
- # Data format is (19, 12, 30720, 61) - (subjects, trials, time_points, channels)
- # Keep only selected electrodes
- # Get all original channel names
- original_channels = list(self.channel_positions.keys()) # Original 61 channels order
- # Find indices of selected electrodes in original data
- selected_indices = [original_channels.index(ch) for ch in self.selected_channels]
- # Extract selected electrodes from data
- self.data = self.data[:, :, :, selected_indices]
- print(f"Filtered data shape: {self.data.shape}")
- # Separate low and high cognitive load data
- self.low_load_data = self.data[:, :6, :, :] # First 6 trials
- self.high_load_data = self.data[:, 6:, :, :] # Last 6 trials
- print(f"Low cognitive load data shape: {self.low_load_data.shape}")
- print(f"High cognitive load data shape: {self.high_load_data.shape}")
- def preprocess_data(self, window_size=1024, overlap=0.5):
- """Preprocess data"""
- print("Preprocessing data...")
- # Calculate window parameters
- step_size = int(window_size * (1 - overlap))
- # Check if data length is sufficient
- data_length = self.low_load_data.shape[2] # Number of time points (3rd dimension)
- print(f"Data length: {data_length} time points")
- print(f"Window size: {window_size} time points")
- print(f"Step size: {step_size} time points")
- if data_length < window_size:
- print(f"Warning: Data length ({data_length}) is smaller than window size ({window_size})")
- print("Adjusting window size...")
- window_size = min(window_size, data_length // 2)
- step_size = int(window_size * (1 - overlap))
- print(f"Adjusted window size: {window_size}")
- print(f"Adjusted step size: {step_size}")
- # Segment for each subject and trial
- self.low_load_segments = []
- self.high_load_segments = []
- for subj in range(self.low_load_data.shape[0]):
- low_load_subj_segments = []
- high_load_subj_segments = []
- for trial in range(self.low_load_data.shape[1]):
- # Low cognitive load - Data format is (time_points, channels)
- data = self.low_load_data[subj, trial, :, :] # (3720, 16)
- segments = []
- for start in range(0, data.shape[0] - window_size + 1, step_size):
- segment = data[start:start + window_size, :] # (1024, 16)
- segments.append(segment)
- low_load_subj_segments.extend(segments)
- # High cognitive load - Data format is (time_points, channels)
- data = self.high_load_data[subj, trial, :, :] # (3720, 16)
- segments = []
- for start in range(0, data.shape[0] - window_size + 1, step_size):
- segment = data[start:start + window_size, :] # (1024, 16)
- segments.append(segment)
- high_load_subj_segments.extend(segments)
- # Add all segments of current subject to total list
- self.low_load_segments.append(low_load_subj_segments)
- self.high_load_segments.append(high_load_subj_segments)
- if len(self.low_load_segments) == 0 or len(self.high_load_segments) == 0:
- raise ValueError("No data segments generated! Please check window size and data length.")
- self.low_load_segments = np.array(self.low_load_segments)
- self.high_load_segments = np.array(self.high_load_segments)
- print(f"Segment data shape: {self.low_load_segments.shape}")
- def compute_connectivity(self, method='coh', fmin=4, fmax=8, batch_size=50):
- """Compute connectivity matrix"""
- print("Computing connectivity matrix...")
- # Check channel count match
- print(f"Number of channel names: {len(self.channel_names)}")
- print(f"Number of channels in segment data: {self.low_load_segments.shape[-1]}") # 3rd dimension is channels
- # Create MNE Info object
- print("Creating MNE Info object...")
- # Ensure channel names are strings
- channel_names = [str(ch) for ch in self.channel_names]
- print(f"Channel name type check: {type(channel_names[0])}")
- print(f"Channel name example: {channel_names[:5]}")
- # Create Info object, explicitly specify channel type
- ch_types = ['eeg'] * len(channel_names)
- info = mne.create_info(ch_names=channel_names, sfreq=self.sfreq, ch_types=ch_types)
- print(f"MNE Info object channel count: {len(info.ch_names)}")
- print(f"MNE Info object channel names: {info.ch_names[:5]}...")
- # Compute low cognitive load connectivity
- low_load_conn = []
- n_segments = len(self.low_load_segments)
- for sub_seg in self.low_load_segments:
- # Convert data from μV to V (1μV = 1e-6 V)
- data_v = sub_seg * 1e-6 # Convert to Volts
- low_load_epochs = mne.EpochsArray(np.transpose(data_v,(0,2,1)), info)
- try:
- con = spectral_connectivity_epochs(
- low_load_epochs, method=method, mode='multitaper',
- fmin=fmin, fmax=fmax, faverage=True,
- mt_adaptive=True, mt_bandwidth=None, mt_low_bias=True,
- cwt_freqs=None, cwt_n_cycles=7, sfreq=self.sfreq,
- tmin=None, tmax=None,
- names=None, n_jobs=1, verbose=False)
- low_load_conn.append(np.reshape(con.get_data(),(len(channel_names), len(channel_names))))
- except Exception as e:
- print(f"Failed to compute connectivity: {e}")
- low_load_conn=np.array(low_load_conn)
- high_load_conn = []
- for sub_seg in self.high_load_segments:
- # Convert data from μV to V (1μV = 1e-6 V)
- data_v = sub_seg * 1e-6 # Convert to Volts
- high_load_epochs = mne.EpochsArray(np.transpose(data_v,(0,2,1)), info)
- try:
- con = spectral_connectivity_epochs(
- high_load_epochs, method=method, mode='multitaper',
- fmin=fmin, fmax=fmax, faverage=True,
- mt_adaptive=True, mt_bandwidth=None, mt_low_bias=True,
- cwt_freqs=None, cwt_n_cycles=7, sfreq=self.sfreq,
- tmin=None, tmax=None,
- names=None, n_jobs=1, verbose=False)
- high_load_conn.append(np.reshape(con.get_data(),(len(channel_names), len(channel_names))))
- except Exception as e:
- print(f"Failed to compute connectivity: {e}")
- high_load_conn=np.array(high_load_conn)
- print(low_load_conn.shape)
- print(high_load_conn.shape)
- # Check range of connectivity values
- print(f"Low load connectivity range: {np.min(low_load_conn):.6f} - {np.max(low_load_conn):.6f}")
- print(f"High load connectivity range: {np.min(high_load_conn):.6f} - {np.max(high_load_conn):.6f}")
- print(f"Low load connectivity mean: {np.mean(low_load_conn):.6f}")
- print(f"High load connectivity mean: {np.mean(high_load_conn):.6f}")
- # Compute average connectivity matrix
- self.connectivity_matrices['low_load'] = np.mean(low_load_conn, axis=0)
- self.connectivity_matrices['high_load'] = np.mean(high_load_conn, axis=0)
- # Save original connectivity data for statistical test
- self.low_load_conn = low_load_conn
- print(f"low_load_conn: {low_load_conn}")
- print(f"low_load_conn.shape: {self.low_load_conn}")
- self.high_load_conn = high_load_conn
- print("Connectivity matrix computation completed")
- def statistical_test(self, alpha=0.05):
- """Statistical test - using permutation test (lower triangle)"""
- print("Performing paired permutation test (lower triangle)...")
- # Get matrix shape
- matrix_shape = self.connectivity_matrices['low_load'].shape
- n_channels = matrix_shape[0]
- # Only consider lower triangle (exclude diagonal) to avoid duplicate calculation of symmetric connections
- lower_tri_indices = np.tril_indices(n_channels, k=-1) # k=-1 excludes diagonal
- # Prepare data - extract only lower triangle part
- low_data_lower = []
- high_data_lower = []
- for subj_low, subj_high in zip(self.low_load_conn, self.high_load_conn):
- low_data_lower.append(subj_low[lower_tri_indices])
- high_data_lower.append(subj_high[lower_tri_indices])
- low_data_lower = np.array(low_data_lower) # (subjects, lower_tri_connections)
- high_data_lower = np.array(high_data_lower)
- n_subjects, n_connections = low_data_lower.shape
- # Compute paired difference matrix
- diff_matrix = low_data_lower - high_data_lower # (subjects, connections)
- # Compute original statistic (mean of paired differences)
- original_stat_lower = np.mean(diff_matrix, axis=0)
- # Perform permutation test
- print("Performing paired permutation test...")
- n_permutations = 5000
- p_values_lower = np.zeros(n_connections)
- for i in range(n_connections):
- # Perform paired permutation test for each connection
- low_conn = low_data_lower[:, i] # Low load data
- high_conn = high_data_lower[:, i] # High load data
- # Compute original statistic (mean of paired differences)
- original_stat_conn = np.mean(low_conn - high_conn)
- # Perform paired permutation
- perm_stats = []
- for _ in range(n_permutations):
- # Randomly swap low and high load labels for each subject
- # This maintains the integrity of the pairing
- perm_low = np.zeros_like(low_conn)
- perm_high = np.zeros_like(high_conn)
- for subj in range(n_subjects):
- if np.random.random() < 0.5:
- # Keep original order
- perm_low[subj] = low_conn[subj]
- perm_high[subj] = high_conn[subj]
- else:
- # Swap order
- perm_low[subj] = high_conn[subj]
- perm_high[subj] = low_conn[subj]
- # Compute statistic after permutation
- perm_stat = np.mean(perm_low - perm_high)
- perm_stats.append(perm_stat)
- # Compute p-value
- perm_stats = np.array(perm_stats)
- # Two-tailed test: calculate proportion of extreme values
- extreme_count = np.sum(np.abs(perm_stats) >= np.abs(original_stat_conn))
- p_values_lower[i] = extreme_count / n_permutations
- # FDR correction
- # _, p_corrected_lower, _, _ = multipletests(p_values_lower, alpha=alpha, method='fdr_bh')
- # print(f"p_corrected_lower: {p_corrected_lower}")
- p_corrected_lower = p_values_lower
- print(f"p_corrected_lower: {p_corrected_lower}")
- print(len(p_corrected_lower))
- # Create full matrix
- p_values = np.zeros(matrix_shape)
- p_corrected = np.zeros(matrix_shape)
- effect_sizes = np.zeros(matrix_shape)
- significant = np.zeros(matrix_shape, dtype=bool)
- # Fill lower triangle results into full matrix
- p_values[lower_tri_indices] = p_values_lower
- p_corrected[lower_tri_indices] = p_corrected_lower
- effect_sizes[lower_tri_indices] = original_stat_lower
- significant[lower_tri_indices] = p_corrected_lower < alpha
- # Copy to upper triangle (symmetric)
- p_values = p_values + p_values.T - np.diag(np.diag(p_values))
- p_corrected = p_corrected + p_corrected.T - np.diag(np.diag(p_corrected))
- effect_sizes = effect_sizes + effect_sizes.T - np.diag(np.diag(effect_sizes))
- significant = significant | significant.T
- print(significant)
- self.statistical_results = {
- 'p_values': p_values,
- 'p_corrected': p_corrected,
- 'effect_sizes': effect_sizes,
- 'significant': significant
- }
- print(f"Number of significant connections: {np.sum(self.statistical_results['significant'])}")
- def statistical_test_t_test(self, alpha=0.1):
- """Statistical test using paired T-test (lower triangle matrix)"""
- # Get matrix shape
- matrix_shape = self.connectivity_matrices['low_load'].shape
- n_channels = matrix_shape[0]
- # Only consider lower triangle (exclude diagonal)
- lower_tri_indices = np.tril_indices(n_channels, k=-1) # k=-1 excludes diagonal
- # Prepare data - extract only lower triangle part
- low_data_lower = []
- high_data_lower = []
- for subj_low, subj_high in zip(self.low_load_conn, self.high_load_conn):
- low_data_lower.append(subj_low[lower_tri_indices])
- high_data_lower.append(subj_high[lower_tri_indices])
- low_data_lower = np.array(low_data_lower) # (subjects, lower_tri_connections)
- high_data_lower = np.array(high_data_lower)
- n_subjects, n_connections = low_data_lower.shape
- # Use paired t-test for statistical testing
- p_values_lower = np.zeros(n_connections)
- effect_sizes_lower = np.zeros(n_connections)
- t_stats_lower = np.zeros(n_connections)
- for i in range(n_connections):
- # Perform paired t-test for each connection
- t_stat, p_val = stats.ttest_rel(low_data_lower[:, i], high_data_lower[:, i])
- t_stats_lower[i] = t_stat
- p_values_lower[i] = p_val
- effect_sizes_lower[i] = np.mean(low_data_lower[:, i] - high_data_lower[:, i])
- # FDR correction
- # _, p_corrected_lower, _, _ = multipletests(p_values_lower, alpha=alpha, method='fdr_bh')
- p_corrected_lower = p_values_lower
- # Create full matrix
- p_values = np.zeros(matrix_shape)
- p_corrected = np.zeros(matrix_shape)
- effect_sizes = np.zeros(matrix_shape)
- t_stats = np.zeros(matrix_shape)
- significant = np.zeros(matrix_shape, dtype=bool)
- # Fill lower triangle results into full matrix
- p_values[lower_tri_indices] = p_values_lower
- p_corrected[lower_tri_indices] = p_corrected_lower
- effect_sizes[lower_tri_indices] = effect_sizes_lower
- t_stats[lower_tri_indices] = t_stats_lower
- significant[lower_tri_indices] = p_corrected_lower < alpha
- # Copy to upper triangle (symmetric)
- p_values = p_values + p_values.T - np.diag(np.diag(p_values))
- print(p_values)
- p_corrected = p_corrected + p_corrected.T - np.diag(np.diag(p_corrected))
- effect_sizes = effect_sizes + effect_sizes.T - np.diag(np.diag(effect_sizes))
- t_stats = t_stats + t_stats.T - np.diag(np.diag(t_stats))
- significant = significant | significant.T
- self.t_test_results = {
- 'p_values': p_values,
- 'p_corrected': p_corrected,
- 'effect_sizes': effect_sizes,
- 't_stats': t_stats,
- 'significant': significant
- }
- print(f"Number of T-test significant connections: {np.sum(self.t_test_results['significant'])}")
- def compare_test_results(self):
- """Compare results of Permutation Test and T-Test"""
- perm_sig = self.statistical_results['significant']
- ttest_sig = self.t_test_results['significant']
- # Calculate overlap
- overlap = np.sum(perm_sig & ttest_sig)
- perm_only = np.sum(perm_sig & ~ttest_sig)
- ttest_only = np.sum(~perm_sig & ttest_sig)
- neither = np.sum(~perm_sig & ~ttest_sig)
- print(f"Permutation test significant connections: {np.sum(perm_sig)}")
- print(f"T-test significant connections: {np.sum(ttest_sig)}")
- print(f"Connections significant in both methods: {overlap}")
- print(f"Significant only in Permutation test: {perm_only}")
- print(f"Significant only in T-test: {ttest_only}")
- print(f"Not significant in either: {neither}")
- # Create matrix showing only lower triangle
- def create_lower_tri_matrix(matrix):
- n = matrix.shape[0]
- lower_tri_matrix = np.zeros_like(matrix)
- lower_tri_indices = np.tril_indices(n, k=-1)
- lower_tri_matrix[lower_tri_indices] = matrix[lower_tri_indices]
- return lower_tri_matrix
- # Plot comparison
- fig, axes = plt.subplots(1, 3, figsize=(18, 6))
- # Permutation test results (lower triangle only)
- perm_effect_lower = create_lower_tri_matrix(
- self.statistical_results['effect_sizes'] * self.statistical_results['significant']
- )
- im1 = axes[0].imshow(perm_effect_lower, cmap='RdBu_r', aspect='auto')
- axes[0].set_title('Channel-wise Permutation Test')
- axes[0].set_xlabel('Channel')
- axes[0].set_ylabel('Channel')
- plt.colorbar(im1, ax=axes[0])
- # T-test results (lower triangle only)
- ttest_effect_lower = create_lower_tri_matrix(
- self.t_test_results['effect_sizes'] * self.t_test_results['significant']
- )
- im2 = axes[1].imshow(ttest_effect_lower, cmap='RdBu_r', aspect='auto')
- axes[1].set_title('Channel-wise T-Test')
- axes[1].set_xlabel('Channel')
- axes[1].set_ylabel('Channel')
- plt.colorbar(im2, ax=axes[1])
- # Difference plot (lower triangle only)
- diff_sig = perm_sig.astype(int) - ttest_sig.astype(int)
- diff_sig_lower = create_lower_tri_matrix(diff_sig)
- im3 = axes[2].imshow(diff_sig_lower, cmap='RdBu_r', aspect='auto')
- axes[2].set_title('Method Difference\n(1=Perm Test Only, -1=T-Test Only, 0=Both)')
- axes[2].set_xlabel('Channel')
- axes[2].set_ylabel('Channel')
- plt.colorbar(im3, ax=axes[2])
- plt.tight_layout()
- plt.savefig('results/test_comparison.png', dpi=300, bbox_inches='tight')
- plt.show()
- def plot_connectivity_matrices(self):
- """Plot connectivity matrices"""
- fig, axes = plt.subplots(2, 2, figsize=(10, 8))
- # Create matrix showing only lower triangle
- def create_lower_tri_matrix(matrix):
- n = matrix.shape[0]
- lower_tri_matrix = np.zeros_like(matrix)
- lower_tri_indices = np.tril_indices(n, k=-1) # k=-1 excludes diagonal
- lower_tri_matrix[lower_tri_indices] = matrix[lower_tri_indices]
- return lower_tri_matrix
- # Low cognitive load connectivity matrix (lower triangle only)
- low_load_lower = create_lower_tri_matrix(self.connectivity_matrices['low_load'])
- im1 = axes[0,0].imshow(low_load_lower,
- cmap='RdBu_r', aspect='auto')
- axes[0,0].set_title('LCW Connectivity Matrix')
- axes[0,0].set_xlabel('Channel')
- axes[0,0].set_ylabel('Channel')
- plt.colorbar(im1, ax=axes[0,0])
- # High cognitive load connectivity matrix (lower triangle only)
- high_load_lower = create_lower_tri_matrix(self.connectivity_matrices['high_load'])
- im2 = axes[0,1].imshow(high_load_lower,
- cmap='RdBu_r', aspect='auto')
- axes[0,1].set_title('HCW Connectivity Matrix')
- axes[0,1].set_xlabel('Channel')
- axes[0,1].set_ylabel('Channel')
- plt.colorbar(im2, ax=axes[0,1])
- # Difference matrix (lower triangle only)
- diff_matrix = self.connectivity_matrices['high_load'] - self.connectivity_matrices['low_load']
- diff_lower = create_lower_tri_matrix(diff_matrix)
- im3 = axes[1,0].imshow(diff_lower, cmap='RdBu_r', aspect='auto')
- axes[1,0].set_title('Difference Matrix (HCW - LCW)')
- axes[1,0].set_xlabel('Channel')
- axes[1,0].set_ylabel('Channel')
- plt.colorbar(im3, ax=axes[1,0])
- # Significant matrix (lower triangle only)
- significant_matrix = diff_matrix * self.statistical_results['significant']
- significant_lower = create_lower_tri_matrix(significant_matrix)
- im4 = axes[1,1].imshow(significant_lower, cmap='RdBu_r', aspect='auto')
- axes[1,1].set_title('Significant Difference Matrix')
- axes[1,1].set_xlabel('Channel')
- axes[1,1].set_ylabel('Channel')
- plt.colorbar(im4, ax=axes[1,1])
- plt.tight_layout()
- plt.savefig('results/delta/connectivity_matrices_wpli.svg', format='svg', bbox_inches='tight')
- plt.show()
- def plot_network_graph(self):
- """Plot network graph, colored by brain region, using only lower triangle matrix"""
- fig, axes = plt.subplots(1, 3, figsize=(18, 6))
- # Create brain region legend
- legend_elements = [plt.Line2D([0], [0], marker='o', color='w',
- markerfacecolor=color, markersize=10, label=region)
- for region, color in self.region_colors.items()]
- # Create network graph
- for i, (title, matrix) in enumerate([
- ('LCW Network', self.connectivity_matrices['low_load']),
- ('HCW Network', self.connectivity_matrices['high_load']),
- ('Significant Difference Network', self.statistical_results['significant'] *
- self.statistical_results['effect_sizes']) # Use effect size instead of difference
- ]):
- G = nx.Graph()
- # Add nodes, project 3D coordinates to 2D
- for j, ch in enumerate(self.channel_names):
- pos_3d = self.channel_positions[ch]
- pos_2d = (pos_3d[0], pos_3d[1])
- G.add_node(ch, pos=pos_2d)
- # Use only lower triangle (excluding diagonal) for stats and threshold
- n_ch = len(self.channel_names)
- lower_tri_indices = np.tril_indices(n_ch, k=-1)
- matrix_lower = matrix[lower_tri_indices]
- matrix_abs_lower = np.abs(matrix_lower)
- max_val = np.max(matrix_abs_lower)
- mean_val = np.mean(matrix_abs_lower)
- std_val = np.std(matrix_abs_lower)
- # Increase threshold: use mean + 2 std devs, and not exceed 90% of max value
- threshold = min(mean_val, max_val * 0.9)
- # Further ensure threshold is not too low
- if threshold < max_val * 0.3:
- threshold = max_val * 0.3
- print(f"{title} - Max: {max_val:.4f}, Mean: {mean_val:.4f}, Threshold: {threshold:.4f}")
- # Iterate only lower triangle, add edges
- edge_count = 0
- for j in range(n_ch):
- for k in range(0, j): # Only iterate lower triangle
- weight = matrix[j, k]
- if 'Significant' in title:
- # Significant network: show all non-zero connections (i.e. significant connections)
- if weight != 0:
- G.add_edge(self.channel_names[j], self.channel_names[k], weight=weight)
- edge_count += 1
- else:
- # Other networks: use threshold filtering
- if abs(weight) > threshold:
- G.add_edge(self.channel_names[j], self.channel_names[k], weight=weight)
- edge_count += 1
- print(f"{title} - Added {edge_count} edges")
- # Add extra info for significant network
- if 'Significant' in title:
- significant_count = np.sum(self.statistical_results['significant'])
- print(f"Significant Network - Total statistically significant connections: {significant_count}")
- print(f"Significant Network - Actual plotted connections: {edge_count}")
- print(f"Significant Network - Non-zero elements in matrix: {np.sum(matrix != 0)}")
- if edge_count != significant_count:
- print(f"Warning: Plotted connection count ({edge_count}) does not match significant connection count ({significant_count})!")
- print(f"Possible reason: Some connections became 0 after multiplying significant matrix with difference matrix")
- # Check specific case for O2 and F3
- try:
- o2_idx = self.channel_names.index('O2')
- f3_idx = self.channel_names.index('F3')
- # Print index for all electrodes
- for i, ch in enumerate(self.channel_names):
- print(f"{ch}: {i}")
- # Check significance
- print(o2_idx, f3_idx)
- is_significant = self.statistical_results['significant'][o2_idx, f3_idx]
- # Check effect size
- effect_size = self.statistical_results['effect_sizes'][o2_idx, f3_idx]
- # Check p-value
- p_value = self.statistical_results['p_corrected'][o2_idx, f3_idx]
- # Check value after multiplication
- multiplied_value = matrix[o2_idx, f3_idx]
- print(f"O2-F3 Connection Details:")
- print(f" Significant: {is_significant}")
- print(f" Effect Size: {effect_size:.6f}")
- print(f" P-value: {p_value:.6f}")
- print(f" Value after multiplication: {multiplied_value:.6f}")
- print(f" Is plotted: {multiplied_value != 0}")
- except ValueError:
- print("O2 or F3 not in electrode list")
- # Plot network
- pos = nx.get_node_attributes(G, 'pos')
- weights = [G[u][v]['weight'] for u, v in G.edges()]
- # Plot nodes (colored by region)
- nx.draw_networkx_nodes(G, pos, ax=axes[i], node_size=300,
- node_color=self.channel_colors)
- # Plot edges
- if len(weights) > 0:
- nx.draw_networkx_edges(G, pos, ax=axes[i],
- edge_color=weights, edge_cmap=plt.cm.RdBu_r,
- width=2)
- # Add labels
- nx.draw_networkx_labels(G, pos, ax=axes[i], font_size=8)
- axes[i].set_title(f'{title}\n(Threshold: {threshold:.3f}, Edge Count: {edge_count})')
- # Add legend
- fig.legend(handles=legend_elements, loc='lower center', ncol=5)
- plt.tight_layout()
- plt.subplots_adjust(bottom=0.15) # Leave space for legend
- plt.savefig('results/delta/network_graphs.svg', format='svg', bbox_inches='tight')
- plt.show()
- # def plot_brain_surface(self):
- # """在脑表面绘制关键连接,仅使用下三角矩阵"""
- # fig, axes = plt.subplots(1, 2, figsize=(15, 6))
- # significant_connections = self.statistical_results['significant']
- # effect_sizes = self.statistical_results['effect_sizes']
- # # 只考虑下三角
- # n_ch = effect_sizes.shape[0]
- # lower_tri_indices = np.tril_indices(n_ch, k=-1)
- # effect_sizes_lower = effect_sizes[lower_tri_indices]
- # significant_lower = significant_connections[lower_tri_indices]
- # # 找到下三角中最强的连接
- # max_connections = 20
- # abs_effect = np.abs(effect_sizes_lower)
- # top_indices_flat = np.argsort(abs_effect)[-max_connections:]
- # # 创建连接强度矩阵(只填下三角)
- # connection_strength = np.zeros_like(effect_sizes)
- # for idx in top_indices_flat:
- # i = lower_tri_indices[0][idx]
- # j = lower_tri_indices[1][idx]
- # if significant_connections[i, j]:
- # connection_strength[i, j] = effect_sizes[i, j]
- # # 不复制到上三角,保持只显示下三角
- # # 绘制连接强度热图(只显示下三角)
- # def create_lower_tri_matrix(matrix):
- # n = matrix.shape[0]
- # lower_tri_matrix = np.zeros_like(matrix)
- # lower_tri_indices = np.tril_indices(n, k=-1)
- # lower_tri_matrix[lower_tri_indices] = matrix[lower_tri_indices]
- # return lower_tri_matrix
- # connection_strength_lower = create_lower_tri_matrix(connection_strength)
- # im1 = axes[0].imshow(connection_strength_lower, cmap='RdBu_r', aspect='auto')
- # axes[0].set_title('Key Connection Strength Distribution')
- # axes[0].set_xlabel('Channel')
- # axes[0].set_ylabel('Channel')
- # plt.colorbar(im1, ax=axes[0])
- # # 绘制连接数量统计(只统计下三角)
- # connection_counts = np.sum(significant_connections, axis=1)
- # axes[1].bar(range(len(connection_counts)), connection_counts)
- # axes[1].set_title('Number of Significant Connections per Channel')
- # axes[1].set_xlabel('Channel Index')
- # axes[1].set_ylabel('Connection Count')
- # plt.tight_layout()
- # plt.savefig('results/brain_connectivity_summary.png', dpi=300, bbox_inches='tight')
- # plt.show()
- def project_electrodes_to_surface(self):
- """Project electrode positions onto cortical surface"""
- print("Projecting electrode positions onto cortical surface...")
- # Load cortical surface
- fsaverage = datasets.fetch_surf_fsaverage()
- left_pial = surface.load_surf_mesh(fsaverage['pial_left'])
- right_pial = surface.load_surf_mesh(fsaverage['pial_right'])
- # Merge left and right hemispheres
- all_vertices = np.vstack([left_pial[0], right_pial[0]])
- all_faces = np.vstack([left_pial[1], right_pial[1] + len(left_pial[0])])
- # Create KDTree for nearest neighbor search
- from scipy.spatial import cKDTree
- kdtree = cKDTree(all_vertices)
- # Project each electrode to the nearest cortical point
- projected_positions = {}
- for ch, pos in self.channel_positions.items():
- # Convert position to mm (assuming original is meters)
- pos_mm = np.array(pos) * 1000
- # Find nearest cortical point
- distance, idx = kdtree.query(pos_mm)
- projected_pos = all_vertices[idx]
- # Move slightly outward to avoid embedding
- normal = projected_pos - np.mean(all_vertices, axis=0)
- normal /= np.linalg.norm(normal)
- projected_pos += normal * 2 # Move outward by 2mm
- projected_positions[ch] = projected_pos
- return projected_positions
- def plot_brain_surface(self, top_n=20):
- """Plot key connections on brain surface (Top view only) - Plot only statistically significant connections"""
- print("Plotting key brain surface connections (Top view only)...")
- # Project electrode positions to cortical surface
- projected_positions = self.project_electrodes_to_surface()
- # Get significant connections
- significant = self.statistical_results['significant']
- effect_sizes = self.statistical_results['effect_sizes']
- # Consider only significant connections
- significant_indices = np.where(significant)
- if len(significant_indices[0]) == 0:
- print("No significant connections to plot")
- return
- # Extract effect sizes of significant connections
- significant_effects = effect_sizes[significant_indices]
- # Find top_n connections with largest effect sizes
- top_indices = np.argsort(np.abs(significant_effects))[-top_n:]
- # Create connection strength matrix
- connection_strength = np.zeros_like(effect_sizes)
- for idx in top_indices:
- i = significant_indices[0][idx]
- j = significant_indices[1][idx]
- connection_strength[i, j] = effect_sizes[i, j]
- connection_strength[j, i] = effect_sizes[i, j] # Symmetric position
- # Create figure (Top view only)
- fig = plt.figure(figsize=(6, 6))
- ax = fig.add_subplot(1, 1, 1, projection='3d')
- # Load cortical surface
- fsaverage = datasets.fetch_surf_fsaverage()
- # Plot left hemisphere
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_left'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.2)
- # Plot right hemisphere
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_right'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.2)
- # Set top view
- ax.view_init(elev=90, azim=0)
- ax.set_title('Top View')
- ax.set_axis_off()
- # Plot electrode positions (using projected positions)
- for ch, pos in projected_positions.items():
- ax.scatter(pos[0], pos[1], pos[2], s=50, color='blue', alpha=0.8)
- # Plot key connections (using projected positions)
- for i in range(len(self.channel_names)):
- for j in range(i+1, len(self.channel_names)):
- if abs(connection_strength[i, j]) > 0:
- ch_i = self.channel_names[i]
- ch_j = self.channel_names[j]
- pos_i = projected_positions[ch_i]
- pos_j = projected_positions[ch_j]
- # Set color and width based on effect size
- effect = connection_strength[i, j]
- color = 'red' if effect < 0 else 'blue' # Negative value means enhanced in high load, positive means reduced
- width = 1 + 3 * abs(effect) / np.max(np.abs(connection_strength))
- # Plot connection line
- ax.plot([pos_i[0], pos_j[0]],
- [pos_i[1], pos_j[1]],
- [pos_i[2], pos_j[2]],
- color=color, linewidth=width, alpha=0.7)
- # # Add legend
- # plt.figtext(0.5, 0.05,
- # f"Red: High Load Connection Enhanced | Blue: High Load Connection Reduced",
- # ha='center', fontsize=12)
- plt.tight_layout()
- plt.subplots_adjust(bottom=0.1) # Leave space for legend
- plt.savefig('results/delta/brain_key_connections_topview.svg', format='svg', bbox_inches='tight')
- plt.show()
- print("Brain surface key connections plot (Top view) saved")
- def plot_key_subnetwork(self, min_connections=5, max_nodes=20):
- """Plot key subnetwork - using electrode positions projected to cortex"""
- print("Plotting key subnetwork...")
- # Project electrode positions to cortex
- projected_positions = self.project_electrodes_to_surface()
- # Get significant connections and effect sizes
- significant = self.statistical_results['significant']
- effect_sizes = self.statistical_results['effect_sizes']
- # Create network graph
- G = nx.Graph()
- # Add nodes (using projected positions)
- for i, ch in enumerate(self.channel_names):
- region = self.get_region(ch)
- G.add_node(ch, pos=projected_positions[ch], region=region)
- # Add significant connections as edges
- n_channels = len(self.channel_names)
- for i in range(n_channels):
- for j in range(i+1, n_channels):
- if significant[i, j]:
- weight = effect_sizes[i, j]
- G.add_edge(self.channel_names[i], self.channel_names[j], weight=weight)
- # Check if there are significant connections
- if G.number_of_edges() == 0:
- print("No significant connections to plot")
- return
- # Extract largest connected component
- largest_cc = max(nx.connected_components(G), key=len)
- subgraph = G.subgraph(largest_cc).copy()
- print(f"Largest connected component contains {len(subgraph.nodes())} nodes and {len(subgraph.edges())} edges")
- # If subnetwork is too large, extract core subnetwork
- if len(subgraph.nodes()) > max_nodes:
- print(f"Subnetwork too large ({len(subgraph.nodes())} nodes), extracting core subnetwork...")
- # Compute node centrality
- centrality = nx.betweenness_centrality(subgraph, weight='weight')
- # Select nodes with highest centrality
- top_nodes = sorted(centrality, key=centrality.get, reverse=True)[:max_nodes]
- subgraph = subgraph.subgraph(top_nodes).copy()
- # Keep only connections between these nodes
- for u, v in list(subgraph.edges()):
- if u not in top_nodes or v not in top_nodes:
- subgraph.remove_edge(u, v)
- # Check if subnetwork meets minimum connection requirement
- if len(subgraph.edges()) < min_connections:
- print(f"Subnetwork connections insufficient ({len(subgraph.edges())}), adding more connections...")
- # Add extra connections with largest effect sizes
- all_edges = sorted(G.edges(data=True), key=lambda x: abs(x[2]['weight']), reverse=True)
- for u, v, data in all_edges:
- if u in subgraph and v in subgraph and not subgraph.has_edge(u, v):
- subgraph.add_edge(u, v, weight=data['weight'])
- if len(subgraph.edges()) >= min_connections:
- break
- print(f"Final subnetwork: {len(subgraph.nodes())} nodes, {len(subgraph.edges())} edges")
- # Create figure
- fig = plt.figure(figsize=(10, 6))
- # Load cortical surface
- fsaverage = datasets.fetch_surf_fsaverage()
- # Create three views: Left, Right, Top
- views = [
- {'title': 'Left Hemisphere', 'hemi': 'left', 'view': 'lateral'},
- {'title': 'Right Hemisphere', 'hemi': 'right', 'view': 'lateral'},
- {'title': 'Top View', 'hemi': 'both', 'view': 'dorsal'}
- ]
- axes = []
- for i in range(3):
- axes.append(fig.add_subplot(1, 3, i+1, projection='3d'))
- # Plot for each view
- for i, view in enumerate(views):
- ax = axes[i]
- # Plot cortical surface
- if view['hemi'] == 'left':
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_left'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.1)
- elif view['hemi'] == 'right':
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_right'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.1)
- else: # Both hemispheres
- # Left hemisphere
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_left'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.1)
- # Right hemisphere
- vertices, faces = surface.load_surf_mesh(fsaverage['pial_right'])
- ax.plot_trisurf(vertices[:, 0], vertices[:, 1], vertices[:, 2],
- triangles=faces, color='lightgray', alpha=0.1)
- # Set view
- if view['view'] == 'lateral':
- ax.view_init(elev=10, azim=180 if view['hemi'] == 'left' else 0)
- elif view['view'] == 'dorsal':
- ax.view_init(elev=90, azim=0)
- ax.set_title(view['title'])
- ax.set_axis_off()
- # Plot subnetwork nodes
- node_colors = []
- for node in subgraph.nodes():
- region = subgraph.nodes[node]['region']
- node_colors.append(self.region_colors[region])
- pos = subgraph.nodes[node]['pos']
- ax.scatter(pos[0], pos[1], pos[2], s=150,
- color=self.region_colors[region], alpha=0.9,
- edgecolors='black', zorder=10)
- # Add node labels
- ax.text(pos[0], pos[1], pos[2], node,
- fontsize=9, ha='center', va='center', zorder=11)
- # Plot subnetwork connections
- max_weight = max(abs(data['weight']) for _, _, data in subgraph.edges(data=True))
- for u, v, data in subgraph.edges(data=True):
- pos_u = subgraph.nodes[u]['pos']
- pos_v = subgraph.nodes[v]['pos']
- # Set color and width based on weight
- weight = data['weight']
- color = 'red' if weight < 0 else 'blue' # Negative means enhanced in high load, positive means reduced
- width = 1 + 4 * abs(weight) / max_weight
- # Plot connection line
- ax.plot([pos_u[0], pos_v[0]],
- [pos_u[1], pos_v[1]],
- [pos_u[2], pos_v[2]],
- color=color, linewidth=width, alpha=0.8, zorder=5)
- # Add legend
- legend_elements = [plt.Line2D([0], [0], marker='o', color='w',
- markerfacecolor=color, markersize=10, label=region)
- for region, color in self.region_colors.items()]
- plt.figlegend(handles=legend_elements, loc='lower center', ncol=5, fontsize=10)
- # Add color explanation
- # plt.figtext(0.5, 0.02,
- # "Red: High Load Connection Enhanced | Blue: High Load Connection Reduced",
- # ha='center', fontsize=12)
- plt.tight_layout()
- plt.subplots_adjust(bottom=0.15) # Leave space for legend
- plt.savefig('results/delta/key_subnetwork.svg', format='svg', bbox_inches='tight')
- plt.show()
- print("Key subnetwork plot saved")
- # Return subnetwork info
- return subgraph
- def analyze_subnetwork(self, subgraph):
- """Analyze key subnetwork characteristics"""
- print("\nAnalyzing key subnetwork...")
- # Basic stats
- n_nodes = subgraph.number_of_nodes()
- n_edges = subgraph.number_of_edges()
- density = nx.density(subgraph)
- print(f"Nodes: {n_nodes}, Edges: {n_edges}, Density: {density:.4f}")
- # Node distribution
- region_counts = {}
- for node in subgraph.nodes():
- region = subgraph.nodes[node]['region']
- region_counts[region] = region_counts.get(region, 0) + 1
- print("\nNode region distribution:")
- for region, count in region_counts.items():
- print(f"{region}: {count} nodes ({count/n_nodes*100:.1f}%)")
- # Connection characteristics
- positive_edges = 0
- negative_edges = 0
- total_weight = 0
- for u, v, data in subgraph.edges(data=True):
- weight = data['weight']
- total_weight += abs(weight)
- if weight > 0:
- positive_edges += 1
- else:
- negative_edges += 1
- print(f"\nConnection characteristics:")
- print(f"Enhanced connections: {positive_edges} ({positive_edges/n_edges*100:.1f}%)")
- print(f"Reduced connections: {negative_edges} ({negative_edges/n_edges*100:.1f}%)")
- print(f"Total connection strength: {total_weight:.4f}")
- # Central node analysis
- centrality = nx.betweenness_centrality(subgraph, weight='weight')
- top_nodes = sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:5]
- print("\nNodes with highest centrality:")
- for node, cent in top_nodes:
- region = subgraph.nodes[node]['region']
- print(f"{node} ({region}): {cent:.4f}")
- # Plot subnetwork graph
- plt.figure(figsize=(10, 6))
- # Node positions
- pos = nx.get_node_attributes(subgraph, 'pos')
- pos_2d = {node: (p[0], p[1]) for node, p in pos.items()} # Use x,y coordinates
- # Node colors
- node_colors = [self.region_colors[subgraph.nodes[node]['region']] for node in subgraph.nodes()]
- # Edge colors and widths
- edge_colors = []
- edge_widths = []
- max_weight = max(abs(data['weight']) for _, _, data in subgraph.edges(data=True))
- for u, v, data in subgraph.edges(data=True):
- if data['weight'] < 0: # Negative means enhanced in high load
- edge_colors.append('red')
- else: # Positive means reduced in high load
- edge_colors.append('blue')
- edge_widths.append(1 + 3 * abs(data['weight']) / max_weight)
- # Plot network
- nx.draw_networkx_nodes(subgraph, pos_2d, node_size=800,
- node_color=node_colors, alpha=0.9,
- edgecolors='black')
- nx.draw_networkx_edges(subgraph, pos_2d, edge_color=edge_colors,
- width=edge_widths, alpha=0.7)
- nx.draw_networkx_labels(subgraph, pos_2d, font_size=10)
- # Add legend
- legend_elements = [plt.Line2D([0], [0], marker='o', color='w',
- markerfacecolor=color, markersize=10, label=region)
- for region, color in self.region_colors.items()]
- plt.legend(handles=legend_elements, loc='best')
- plt.title("Key Subnetwork")
- plt.axis('off')
- plt.tight_layout()
- plt.savefig('results/delta/subnetwork_graph.svg', format='svg', bbox_inches='tight')
- plt.show()
- return {
- 'n_nodes': n_nodes,
- 'n_edges': n_edges,
- 'density': density,
- 'region_distribution': region_counts,
- 'positive_edges': positive_edges,
- 'negative_edges': negative_edges,
- 'top_nodes': top_nodes
- }
- def compute_network_metrics(self):
- """Calculate network metrics for each subject, then statistics (mean & std), using only lower triangle matrix"""
- print("Computing network metrics (per subject)...")
- metrics = {}
- metrics_detail = {}
- for condition in ['low_load', 'high_load']:
- # Get connectivity matrices for all subjects
- if condition == 'low_load':
- subj_conn = self.low_load_conn
- else:
- subj_conn = self.high_load_conn
- n_subjects = subj_conn.shape[0]
- n_ch = subj_conn.shape[1]
- lower_tri_indices = np.tril_indices(n_ch, k=-1)
- # Use connectivity values from all subjects to calculate global threshold (ensure consistency)
- all_matrix_lower = []
- for subj in range(n_subjects):
- matrix = subj_conn[subj]
- matrix_lower = matrix[lower_tri_indices]
- all_matrix_lower.append(matrix_lower)
- all_matrix_lower = np.concatenate(all_matrix_lower)
- matrix_abs_lower = np.abs(all_matrix_lower)
- max_val = np.max(matrix_abs_lower)
- mean_val = np.mean(matrix_abs_lower)
- std_val = np.std(matrix_abs_lower)
- threshold = min(mean_val, max_val * 0.5)
- if threshold > max_val * 0.3:
- threshold = max_val * 0.1
- print(f"{condition} Network Metrics Calculation - Threshold: {threshold:.4f}")
- # Calculate metrics for each subject
- density_list = []
- clustering_list = []
- path_length_list = []
- efficiency_list = []
- edge_count_list = []
- for subj in range(n_subjects):
- matrix = subj_conn[subj]
- G = nx.Graph()
- edge_count = 0
- # Iterate only lower triangle (excluding diagonal)
- for i in range(n_ch):
- for j in range(0, i):
- if matrix[i, j] > threshold:
- G.add_edge(i, j, weight=matrix[i, j])
- edge_count += 1
- edge_count_list.append(edge_count)
- # Calculate metrics
- if len(G.edges()) > 0:
- # Density
- density = nx.density(G)
- # Clustering coefficient
- clustering = nx.average_clustering(G)
- # Average path length
- try:
- path_length = nx.average_shortest_path_length(G)
- except:
- path_length = np.nan
- # Global efficiency
- try:
- efficiency = nx.global_efficiency(G)
- except:
- efficiency = np.nan
- else:
- density = 0
- clustering = 0
- path_length = np.nan
- efficiency = np.nan
- density_list.append(density)
- clustering_list.append(clustering)
- path_length_list.append(path_length)
- efficiency_list.append(efficiency)
- # Calculate mean and std
- metrics[condition] = {
- 'Density': np.nanmean(density_list),
- 'Clustering': np.nanmean(clustering_list),
- 'Path_Length': np.nanmean(path_length_list),
- 'Efficiency': np.nanmean(efficiency_list),
- 'Edge_Count': np.nanmean(edge_count_list),
- 'Threshold': threshold,
- 'Density_std': np.nanstd(density_list),
- 'Clustering_std': np.nanstd(clustering_list),
- 'Path_Length_std': np.nanstd(path_length_list),
- 'Efficiency_std': np.nanstd(efficiency_list),
- 'Edge_Count_std': np.nanstd(edge_count_list)
- }
- # Also save detailed metrics for each subject
- metrics_detail[condition] = {
- 'Density': density_list,
- 'Clustering': clustering_list,
- 'Path_Length': path_length_list,
- 'Efficiency': efficiency_list,
- 'Edge_Count': edge_count_list
- }
- self.network_metrics = metrics
- self.network_metrics_detail = metrics_detail
- # Plot metrics comparison (mean + std)
- fig, axes = plt.subplots(2, 3, figsize=(6, 4))
- metric_names = ['Density', 'Clustering', 'Path_Length', 'Efficiency', 'Edge_Count', 'Threshold']
- for i, metric in enumerate(metric_names):
- ax = axes[i//3, i%3]
- low_val = metrics['low_load'][metric]
- high_val = metrics['high_load'][metric]
- if metric != 'Threshold':
- low_std = metrics['low_load'][metric + '_std']
- high_std = metrics['high_load'][metric + '_std']
- t, p = stats.ttest_rel(np.array(self.network_metrics_detail['low_load'][metric]), np.array(self.network_metrics_detail['high_load'][metric]))
- print(f"{metric} t-test: {t:.3f}, p={p:.3f}")
- # Plot bars for LCW and HCW, blue for LCW, red for HCW, alpha 0.5, width 0.3, no fill
- bars = []
- bar_lcw = ax.bar(
- ['LCW'], [low_val], yerr=[low_std], capsize=4,
- color='none', edgecolor='blue', alpha=0.5, width=0.3, label='LCW'
- )
- bar_hcw = ax.bar(
- ['HCW'], [high_val], yerr=[high_std], capsize=4,
- color='none', edgecolor='red', alpha=0.5, width=0.3, label='HCW'
- )
- bars = [bar_lcw[0], bar_hcw[0]]
- else:
- # No error bar for threshold
- bar_lcw = ax.bar(
- ['LCW'], [low_val], color='none', edgecolor='blue', alpha=0.5, width=0.3, label='LCW'
- )
- bar_hcw = ax.bar(
- ['HCW'], [high_val], color='none', edgecolor='red', alpha=0.5, width=0.3, label='HCW'
- )
- bars = [bar_lcw[0], bar_hcw[0]]
- ax.set_title(metric)
- #ax.set_ylabel('Value')
- # # Add value labels
- # for bar, val in zip(bars, [low_val, high_val]):
- # if not np.isnan(val):
- # ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.01,
- # f'{val:.3f}', ha='center', va='bottom', fontsize=12)
- # # Add legend only once
- # if i == 0:
- # ax.legend(['LCW', 'HCW'])
- plt.tight_layout()
- plt.savefig('results/delta/network_metrics.svg', format='svg', bbox_inches='tight')
- plt.show()
- def save_results_table(self):
- """Save results table"""
- print("Saving results table...")
- # Create results DataFrame
- results_data = []
- # Save only lower triangle (excluding diagonal) connectivity results
- for i in range(1, len(self.channel_names)):
- for j in range(i):
- results_data.append({
- 'Channel_From': self.channel_names[i],
- 'Channel_To': self.channel_names[j],
- 'Region_From': self.get_region(self.channel_names[i]),
- 'Region_To': self.get_region(self.channel_names[j]),
- 'Low_Load_Connectivity': self.connectivity_matrices['low_load'][i, j],
- 'High_Load_Connectivity': self.connectivity_matrices['high_load'][i, j],
- 'Difference': (self.connectivity_matrices['high_load'][i, j] -
- self.connectivity_matrices['low_load'][i, j]),
- 'P_Value': self.statistical_results['p_values'][i, j],
- 'P_Corrected': self.statistical_results['p_corrected'][i, j],
- 'Significant': self.statistical_results['significant'][i, j],
- 'Effect_Size': self.statistical_results['effect_sizes'][i, j]
- })
- # Create DataFrame
- df = pd.DataFrame(results_data)
- # Save to CSV
- df.to_csv('results/delta/connectivity_results.csv', index=False)
- # Save network metrics
- metrics_df = pd.DataFrame(self.network_metrics).T
- metrics_df.to_csv('results/delta/network_metrics.csv')
- print("Results table saved to results/ directory")
- def get_region(self, channel_name):
- """Get brain region for a channel"""
- for region, channels in self.regions.items():
- if channel_name in channels:
- return region
- return 'Unknown'
- def run_analysis(self, window_size=1024, overlap=0.5, method='wpli',
- fmin=4, fmax=8, alpha=0.2, batch_size=50):
- """Run full analysis"""
- print("Starting brain network connectivity analysis...")
- # Load data
- self.load_data()
- # Preprocess
- self.preprocess_data(window_size, overlap)
- # Compute connectivity
- self.compute_connectivity(method, fmin, fmax, batch_size)
- # Statistical test
- self.statistical_test(alpha)
- # Paired T-test comparison
- self.statistical_test_t_test(alpha)
- # Compare two test methods
- self.compare_test_results()
- # Plot results
- self.plot_connectivity_matrices()
- self.plot_network_graph()
- self.plot_brain_surface()
- # Compute network metrics
- self.compute_network_metrics()
- # Save results
- self.save_results_table()
- # Plot key subnetwork
- subgraph = self.plot_key_subnetwork(min_connections=10, max_nodes=15)
- # Analyze subnetwork characteristics
- if subgraph:
- subnetwork_stats = self.analyze_subnetwork(subgraph)
- # Save subnetwork analysis results
- with open('results/delta/subnetwork_analysis.json', 'w') as f:
- json.dump(subnetwork_stats, f, indent=4)
- print("Analysis completed!")
- # Main program
- if __name__ == "__main__":
- # Create analyzer
- analyzer = BrainConnectivityAnalyzer()
- # Run analysis
- analyzer.run_analysis(
- window_size=512, # 8 second window (128Hz sampling rate)
- overlap=0, # 50% overlap
- method='pli', # Coherence
- fmin=1, # alpha band lower limit
- fmax=4, # alpha band upper limit
- alpha=0.05, # Significance level
- )
plot_connectivity.py at commit 64e7e01, no license · at the source
Overview
- Department of Biomedical Engineering, China Medical University,Shenyang, Liaoning China
- Shanghai Institute of Microsystem and Information Technology, Chinese Academy of Sciences,Shanghai, China
- School of Life Science, China Medical University,Shenyang, Liaoning China
Abstract
The abstract is not reproduced here: the paper's license (CC BY-NC-ND) does not allow it. Read it in the paper, at the publisher or on Europe PMC.
Repositories
Its files are read in the Code ↔ Paper reader above, with 8 matches between paragraphs and lines of code.
ruix6/cognitiveload_audio_analysis
64e7e019d149d2fd24d558df25a47e2ffee5e273, 17 December 2025Availability: 1 check, the latest on 28 September 2026: the link answers
- 28 September 2026: the link answers
12 files
- calculate_scripts/
calculate_power.py , Python, 127 lines, 1 match - calculate_scripts/
calculate_se.py , Python, 92 lines - convert_data_to_ndvars.p
y , Python, 145 lines - eelbrain_trf.py, Python, 166 lines
- generate_data_all.py, Python, 145 lines
- main.py, Python, 256 lines
- plot_scripts/
plot_connectivity.py , Python, 1,445 lines, 4 matches - plot_scripts/
plot_pcc.py , Python, 49 lines, 1 match - plot_scripts/
plot_peak.py , Python, 83 lines - plot_scripts/
plot_trf.py , Python, 187 lines - plot_scripts/
plot_trf_sourcelocation. , Python, 270 lines, 2 matchespy - README.md, Text, 41 lines
Zenodo 20161055
Availability: 1 check, the latest on 28 September 2026: the link answers (HTTP 200)
- 28 September 2026: the link answers (HTTP 200)
Code availability statement
The paper has a code availability statement. Its license (CC BY-NC-ND) does not allow reproducing it here; in short, from what the harvester recognized in it:
- it points to the authors' code: ruix6/
cognitiveload_audio_anal , Zenodo 20161055ysis
Read it in the paper: doi.org/10.1038/s42003-026-10394-7.
Tracing map
Proposed by the machine: these links were found in the paper and verified at the source, without human review. The map will receive a Zenodo DOI once one of the paper's authors has validated it with their ORCID.
What the map holds:
- 2 repositories of the authors' code, each at its verified commit, with its license and how the link was found in the paper;
- 11 scripts, each with its path and the digest of its content;
- 8 matches between paragraphs of the paper and lines of the code (method lexical-v1);
- neither the text of the paper nor the code itself.
Its JSON (tracing-map.json) is deposited on Zenodo with its DOI once the map is validated.
Data
No dataset and no data link were found in the paper.
Data availability statement
The paper has a data availability statement. Its license (CC BY-NC-ND) does not allow reproducing it here; in short, from what the harvester recognized in it:
- it says that the data are available on request
Read it in the paper: doi.org/10.1038/s42003-026-10394-7.
Versions
The history of this record: each version stored by the harvester or made by a correction of its authors or of the maintainers of its code, and what changed in its facts. The texts of the paper (its abstract, its availability statements) are not part of it; versions that changed only those are not listed.
Version 1, 28 September 2026: the first record
Recorded: type, language, journal, volume, issue, pages, dates, 12 authors, 2 keywords, 11 MeSH terms, 1 funder, 45 references.
Cite
This paper
Liu, R., Zhang, H., Liu, C., Xu, X., Zhao, X., Wang, X., Zhang, F., Sha, X., Sun, L., Wang, Z., Li, S., & Chang, S. (2026). Cognitive load weakens neural speech tracking without altering response timing. Communications biology, 9(1), 1139. https://
BibTeX
@article{liu2026cognitiv
author = {Liu, Ruixiang and Zhang, Huan and Liu, Chang and Xu, Xinmeng and Zhao, Xu and Wang, Xuewen and Zhang, Fulong and Sha, Xianzheng and Sun, Limin and Wang, Zhi and Li, Shuo and Chang, Shijie},
title = {{Cognitive load weakens neural speech tracking without altering response timing}},
journal = {Communications biology},
year = {2026},
month = may,
volume = {9},
number = {1},
pages = {1139},
publisher = {Nature Publishing Group},
issn = {2399-3642},
doi = {10.1038/
url = {https://
pmid = {42215620},
pmcid = {PMC13507231}
}
RIS
TY - JOUR
AU - Liu, Ruixiang
AU - Zhang, Huan
AU - Liu, Chang
AU - Xu, Xinmeng
AU - Zhao, Xu
AU - Wang, Xuewen
AU - Zhang, Fulong
AU - Sha, Xianzheng
AU - Sun, Limin
AU - Wang, Zhi
AU - Li, Shuo
AU - Chang, Shijie
TI - Cognitive load weakens neural speech tracking without altering response timing
T2 - Communications biology
J2 - Commun Biol
PY - 2026
DA - 2026/
VL - 9
IS - 1
SP - 1139
SN - 2399-3642
PB - Nature Publishing Group
DO - 10.1038/
UR - https://
LA - en
ER -
CSL-JSON
{
"id": "10.1038/
"type": "article-journal",
"title": "Cognitive load weakens neural speech tracking without altering response timing",
"container-title": "Communications biology",
"author": [
{
"family": "Liu",
"given": "Ruixiang"
},
{
"family": "Zhang",
"given": "Huan"
},
{
"family": "Liu",
"given": "Chang"
},
{
"family": "Xu",
"given": "Xinmeng"
},
{
"family": "Zhao",
"given": "Xu"
},
{
"family": "Wang",
"given": "Xuewen"
},
{
"family": "Zhang",
"given": "Fulong"
},
{
"family": "Sha",
"given": "Xianzheng"
},
{
"family": "Sun",
"given": "Limin"
},
{
"family": "Wang",
"given": "Zhi"
},
{
"family": "Li",
"given": "Shuo"
},
{
"family": "Chang",
"given": "Shijie"
}
],
"container-title-short":
"volume": "9",
"issue": "1",
"page": "1139",
"DOI": "10.1038/
"PMID": "42215620",
"PMCID": "PMC13507231",
"ISSN": "2399-3642",
"publisher": "Nature Publishing Group",
"URL": "https://
"language": "en",
"issued": {
"date-parts": [
[
2026,
5,
29
]
]
}
}
The tracing map gets a citation of its own once an author has validated it and it has a DOI.
Similar papers
The papers with a page that share the most with this one: the tools found in their code, their categories, datasets, cited references and authors, the rarest counting most.
- [1] doi:10.1093/nc/niag029 [code]
- A data-driven approach to identifying and evaluating connectivity-based neural correlates of conscious visual perception.Journal: Neuroscience of consciousnessIn common: MNE-Connectivity, MNE-Python, Nilearn, 7 other tools, cognitive, 1 reference
- [2] doi:10.1097/j.pain.0000000000004044 [code]
- No effect of rhythmic visual stimulation on experimental pain perception.Journal: PainIn common: MNE-Connectivity, MNE-Python, Nilearn, 6 other tools, EEG, cognitive, 1 reference
- [3] doi:10.1093/braincomms/fcag261 [code]
- Cortical speech envelope tracking reflects lesion-symptom profiles in post-stroke aphasia.Journal: Brain communicationsIn common: cognitive, 8 references
- [4] doi:10.1162/imag.a.1245 [code]
- Towards precision EEG connectomics: Evaluating the benefits of dense sampling.Journal: Imaging neuroscience (Cambridge, Mass.)In common: MNE-Connectivity, MNE-Python, Nilearn, 6 other tools, EEG, 1 reference
- [5] doi:10.1523/eneuro.0041-26.2026 [code]
- Ocular Speech Tracking Persists in Blindness, but Its Dynamics and Oculo-Cerebral Connectivity Depend on Visual Status.Journal: eNeuroIn common: MNE-Connectivity, MNE-Python, statsmodels, 5 other tools, cognitive, 2 references
- [6] doi:10.3389/fncom.2026.1786996 [code]
- Schumann-anchored golden ratio organization of human neural oscillations.Journal: Frontiers in computational neuroscienceIn common: MNE-Connectivity, MNE-Python, NetworkX, 6 other tools, EEG, 1 reference
- [7] doi:10.3390/s26134019 [code]
- NeuroStat: An Open-Source EEG Connectivity Platform for Randomised Controlled Trials.Journal: Sensors (Basel, Switzerland)In common: MNE-Connectivity, MNE-Python, NetworkX, 4 other tools, EEG, 3 references
- [8] doi:10.1016/j.ibneur.2026.05.013 [code]
- Anticipatory slow potentials before auditory feedback show posterior predominance but limited condition effects in speech-in-noise.Journal: IBRO neuroscience reportsIn common: MNE-Python, statsmodels, pandas, 3 other tools, EEG, cognitive, 4 references
- [9] doi:10.1038/s41597-026-07350-9 [code]
- An open multi-center MEG-EEG dataset for studying conscious visual perception.Journal: Scientific dataIn common: MNE-Connectivity, MNE-Python, Nilearn, 6 other tools, EEG
- [10] doi:10.1038/s41597-026-07377-y [code]
- An open-access multi-site fMRI dataset for investigating conscious visual perception.Journal: Scientific dataIn common: MNE-Connectivity, MNE-Python, Nilearn, 6 other tools
Contribute
The authors of this paper can claim it, correct its record and validate its tracing map, and the maintainers of its code (its owner, or a public member of its organization) correct what it says of their repository; anyone signed in can ask for its removal. Every request goes to OSCR's own machine, which answers it; your account page follows them.
Sign in with ORCID to claim this paper as one of its authors, correct its record or validate its tracing map: when the paper's metadata lists your ORCID iD, you are recognized at once. Maintainers of its code: sign in with GitHub, then claim the repository on your account page.
Claim this paper
Correct its record
Say what each link of this record is, remove the ones that are not the paper's, add the ones that are missing. The correction becomes a new version of the record, in its Versions section.
Validate its tracing map
You validate the map as this page shows it: 2 repositories of the authors' code, each at its verified commit and with its license, 11 scripts, and 8 matches between paragraphs and code (see the Code and Map sections). It then receives a DOI on Zenodo, with you (your ORCID iD) and OSCR as its creators; the code itself is not deposited.
The map's fingerprint: sha256:5bf6c78b72a841c2…
Add the badge to its README
The badge links the code to this page. Copy one of these into the README of the paper's code: only you decide where it goes, and nothing is changed for you.
Markdown
[, paste the snippet at the top, then “Commit changes…” and, to review it first, “Create a new branch and start a pull request”. You open the pull request; OSCR asks for no permission.
Request its removal
To ask OSCR to remove this record, the copies of its authors' scripts or its tracing map, use the removal request page: signed in, you say who you are, what to remove and why, then review and confirm the request. Published rules decide every request (how).
Discussion, reproductions, activity
Discussion: questions and error reports about this paper and its code, from signed-in readers and its authors. It opens with sign-in.
Reproductions: reports from readers who ran the authors' code: what they reproduced, with which environment, commit and data. It opens with sign-in.
Activity: what happens around this paper: new versions of its record, its map's validation, discussions and reproductions. It opens with sign-in.
