diff --git a/flares.py b/flares.py index 520d81a..09e617e 100644 --- a/flares.py +++ b/flares.py @@ -205,6 +205,8 @@ NOISE_MODEL: str BINS: int N_JOBS: int +JSON_LOCATION: str + TIME_WINDOW_START: int TIME_WINDOW_END: int MAX_WORKERS: int @@ -1960,331 +1962,6 @@ def fold_channels(raw: BaseRaw, p_name: str, progress_queue=None) -> dict[str, l -def individual_significance(raw_haemo, glm_est): - - fig_individual_significances = [] # List to store figures - - # TODO: BAD! - cha = glm_est.to_dataframe() - - unique_annotations = set(raw_haemo.annotations.description) - - for cond in unique_annotations: - - ch_summary = cha.query(f"Condition.str.startswith('{cond}_delay_') and Chroma == 'hbo'", engine='python') - - print(ch_summary.head()) - - channel_averages = ch_summary.groupby('ch_name')['theta'].mean().reset_index() - print(channel_averages.head()) - - - activity_ch_summary = ch_summary.query( - f"Chroma == 'hbo' and Condition.str.startswith('{cond}_delay_')", engine='python' - ) - - # Function to correct p-values per channel - def fdr_correct_per_channel(df): - df = df.copy() - df['pval_fdr'] = multipletests(df['p_value'], method='fdr_bh')[1] - return df - - # Apply FDR correction grouped by channel - corrected = activity_ch_summary.groupby("ch_name", group_keys=False).apply(fdr_correct_per_channel) - - # Determine which channels are significant across any delay - sig_channels = ( - corrected.groupby('ch_name') - .apply(lambda df: (df['pval_fdr'] < 0.05).any()) - .reset_index(name='significant') - ) - - # Merge with mean theta (optional for plotting) - mean_theta = activity_ch_summary.groupby('ch_name')['theta'].mean().reset_index() - sig_channels = sig_channels.merge(mean_theta, on='ch_name') - # print(sig_channels) - - - # For example, take the minimum corrected p-value per channel - summary_pvals = corrected.groupby('ch_name')['pval_fdr'].min().reset_index() - # print(summary_pvals) - - - def parse_ch_name(ch_name): - # Extract numbers after S and D in names like 'S10_D5 hbo' - match = re.match(r'S(\d+)_D(\d+)', ch_name) - if match: - return int(match.group(1)), int(match.group(2)) - else: - return None, None - - - min_pvals = corrected.groupby('ch_name')['pval_fdr'].min().reset_index() - - # Merge the real p-values into sig_channels / avg_df - avg_df = sig_channels.merge(min_pvals, on='ch_name') - - # Rename columns for consistency - avg_df = avg_df.rename(columns={'theta': 't_or_theta', 'pval_fdr': 'p_value'}) - - # Add Source and Detector columns again - avg_df['Source'], avg_df['Detector'] = zip(*avg_df['ch_name'].map(parse_ch_name)) - - # Keep relevant columns - avg_df = avg_df[['Source', 'Detector', 't_or_theta', 'p_value']].dropna() - - ABS_SIGNIFICANCE_THETA_VALUE = 1 - ABS_SIGNIFICANCE_T_VALUE = 1 - P_THRESHOLD = 0.05 - SOURCE_DETECTOR_SEPARATOR = "_" - - t_or_theta = 'theta' - #holy log noise - # for _, row in avg_df.iterrows(): # type: ignore - # print(f"Source {row['Source']} <-> Detector {row['Detector']}: " - # f"Avg {t_or_theta}-value = {row['t_or_theta']:.3f}, Avg p-value = {row['p_value']:.3f}") - - # Extract the cource and detector positions from raw - src_pos: dict[int, tuple[float, float]] = {} - det_pos: dict[int, tuple[float, float]] = {} - for ch in getattr(raw_haemo, "info")["chs"]: - ch_name = ch['ch_name'] - if not ch_name or not ch['loc'].any(): - continue - parts = ch_name.split()[0] - src_str, det_str = parts.split(SOURCE_DETECTOR_SEPARATOR) - src_num = int(src_str[1:]) - det_num = int(det_str[1:]) - src_pos[src_num] = ch['loc'][3:5] - det_pos[det_num] = ch['loc'][6:8] - - # Set up the plot - fig, ax = plt.subplots(figsize=(8, 6)) # type: ignore - - # Plot the sources - for pos in src_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='o', edgecolors='white', linewidths=1, zorder=3) # type: ignore - - # Plot the detectors - for pos in det_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='s', edgecolors='white', linewidths=1, zorder=3) # type: ignore - - # Ensure that the colors stay within the boundaries even if they are over or under the max/min values - if t_or_theta == 't': - norm = mcolors.Normalize(vmin=-ABS_SIGNIFICANCE_T_VALUE, vmax=ABS_SIGNIFICANCE_T_VALUE) - elif t_or_theta == 'theta': - norm = mcolors.Normalize(vmin=-ABS_SIGNIFICANCE_THETA_VALUE, vmax=ABS_SIGNIFICANCE_THETA_VALUE) - - cmap: mcolors.Colormap = plt.get_cmap('seismic') - - # Plot connections with avg t-values - for row in avg_df.itertuples(): - src: int = cast(int, row.Source) # type: ignore - det: int = cast(int, row.Detector) # type: ignore - tval: float = cast(float, row.t_or_theta) # type: ignore - pval: float = cast(float, row.p_value) # type: ignore - - - if src in src_pos and det in det_pos: - x = [src_pos[src][0], det_pos[det][0]] - y = [src_pos[src][1], det_pos[det][1]] - style = '-' if pval <= P_THRESHOLD else '--' - ax.plot(x, y, linestyle=style, color=cmap(norm(tval)), linewidth=4, alpha=0.9, zorder=2) # type: ignore - - # Format the Colorbar - sm = plt.cm.ScalarMappable(cmap=cmap, norm=norm) - sm.set_array([]) - cbar = plt.colorbar(sm, ax=ax, shrink=0.85) # type: ignore - cbar.set_label(f'Average {cond} {t_or_theta} value (hbo)', fontsize=11) # type: ignore - - # Formatting the subplots - ax.set_aspect('equal') - ax.set_title(f"Average {t_or_theta} values for {cond} (HbO)", fontsize=14) # type: ignore - ax.set_xlabel('X position (m)', fontsize=11) # type: ignore - ax.set_ylabel('Y position (m)', fontsize=11) # type: ignore - ax.grid(True, alpha=0.3) # type: ignore - - # Set axis limits to be 1cm more than the optode positions - all_x = [pos[0] for pos in src_pos.values()] + [pos[0] for pos in det_pos.values()] - all_y = [pos[1] for pos in src_pos.values()] + [pos[1] for pos in det_pos.values()] - ax.set_xlim(min(all_x)-0.01, max(all_x)+0.01) - ax.set_ylim(min(all_y)-0.01, max(all_y)+0.01) - - fig.tight_layout() - - fig_individual_significances.append((f"Condition {cond}", fig)) - - return fig_individual_significances - -# TODO: Hardcoded -# def group_significance( -# raw_haemo, -# all_cha: pd.DataFrame, -# condition: str, -# correction: str = "fdr_bh" -# ) -> plt.Figure: -# """ -# Compute group-level significance using weighted Stouffer's method and plot results. - -# Args: -# raw_haemo: Raw haemoglobin MNE object (used for optode positions) -# all_cha: DataFrame with columns including 'ID', 'Condition', 'p_value', 'theta', 'df', 'ch_name', 'Chroma' -# condition: condition prefix, e.g., 'Activity' -# correction: p-value correction method ('fdr_bh' or 'bonferroni') - -# Returns: -# Matplotlib Figure with group-level theta values and significance. -# """ - -# assert "ID" in all_cha.columns, "'ID' column missing in input data" -# assert len(raw_haemo) >= 1, "At least one raw haemoglobin object is required" - -# condition_prefix = f"{condition}_delay" - -# # Filter relevant data -# ch_summary = all_cha.query( -# "Condition.str.startswith(@condition_prefix) and Chroma == 'hbo'", -# engine='python' -# ).copy() - - -# logger.info("=== ch_summary head ===") -# logger.info(ch_summary.head()) - -# logger.info("\nSummary stats:") -# logger.info(f"Total rows: {len(ch_summary)}") -# logger.info(f"Unique subjects: {ch_summary['ID'].nunique() if 'ID' in ch_summary.columns else 'ID column missing'}") -# logger.info(f"Unique conditions: {ch_summary['Condition'].unique()}") -# logger.info(f"Unique channels (Source-Detector pairs): {ch_summary.groupby(['Source', 'Detector']).ngroups}") - -# logger.info("\nSample p_values:") -# logger.info(ch_summary['p_value'].describe()) - -# if ch_summary.empty: -# raise ValueError(f"No data found for condition prefix: {condition_prefix}") - -# # --- For debugging -# logger.info(f"Total rows after filtering for condition '{condition_prefix}': {len(ch_summary)}") -# logger.info(f"Unique channels: {ch_summary['ch_name'].nunique()}") -# logger.info(f"Participants: {ch_summary['ID'].nunique()}") - -# # Step 1: Select the peak regressor (~6s after stimulus onset) -# peak_regressor = f"{condition}_delay_6" -# peak_data = ch_summary[ch_summary["Condition"] == peak_regressor].copy() - -# logger.info(f"\n=== Logging all values for {peak_regressor} ===") -# for row in peak_data.itertuples(index=False): -# logger.info( -# f"Subject: {row.ID}, " -# f"Channel: {row.ch_name}, " -# f"Source: {row.Source}, Detector: {row.Detector}, " -# f"theta: {row.theta:.4f}, " -# f"p_value: {row.p_value:.6f}, " -# f"df: {row.df}" -# ) - -# if peak_data.empty: -# raise ValueError(f"No data found for peak regressor: {peak_regressor}") - -# # Step 2: Combine per-channel stats across subjects -# group_results = [] - -# for (src, det), group in peak_data.groupby(["Source", "Detector"]): -# pvals = group["p_value"].values -# thetas = group["theta"].values -# dfs = group["df"].values - -# # Weighted Stouffer's method -# weights = np.sqrt(dfs) -# z_scores = norm.isf(pvals) -# combined_z = np.sum(weights * z_scores) / np.sqrt(np.sum(weights**2)) -# combined_p = norm.sf(combined_z) - -# theta_avg = np.average(thetas, weights=weights) - -# group_results.append({ -# "Source": src, -# "Detector": det, -# "theta_avg": theta_avg, -# "combined_p": combined_p -# }) - -# # Step 3: Create combined_df -# combined_df = pd.DataFrame(group_results) - -# # Step 4: Multiple comparisons correction -# _, pvals_corr, _, significant = multipletests( -# combined_df["combined_p"], alpha=0.05, method=correction -# ) - -# combined_df["pval_corr"] = pvals_corr -# combined_df["significant"] = significant - -# logger.info(f"Used peak regressor: {peak_regressor}") -# logger.info(f"Channels tested: {len(combined_df)}") -# logger.info(f"Significant channels after correction: {combined_df['significant'].sum()}") -# # Get optode positions from the first raw file -# raw = raw_haemo -# src_pos, det_pos = {}, {} -# for ch in raw.info["chs"]: -# ch_name = ch["ch_name"] -# if not ch_name or not ch["loc"].any(): -# continue -# parts = ch_name.split()[0] -# src_str, det_str = parts.split("_") -# src_num = int(src_str[1:]) -# det_num = int(det_str[1:]) -# src_pos[src_num] = ch["loc"][3:5] -# det_pos[det_num] = ch["loc"][6:8] - -# # Plotting parameters -# ABS_SIGNIFICANCE_THETA_VALUE = 1 -# P_THRESHOLD = 0.05 -# cmap = plt.get_cmap("seismic") -# norm = mcolors.Normalize(vmin=-ABS_SIGNIFICANCE_THETA_VALUE, vmax=ABS_SIGNIFICANCE_THETA_VALUE) - -# fig, ax = plt.subplots(figsize=(8, 6)) - -# # Plot optodes -# for pos in src_pos.values(): -# ax.scatter(*pos, s=120, c="k", marker="o", edgecolors="white", linewidths=1, zorder=3) -# for pos in det_pos.values(): -# ax.scatter(*pos, s=120, c="k", marker="s", edgecolors="white", linewidths=1, zorder=3) - -# # Plot connections colored by average theta, solid if significant -# for row in combined_df.itertuples(): -# src, det = int(row.Source), int(row.Detector) -# tval, pval = row.theta_avg, row.pval_corr -# if src in src_pos and det in det_pos: -# x = [src_pos[src][0], det_pos[det][0]] -# y = [src_pos[src][1], det_pos[det][1]] -# linestyle = "-" if pval <= P_THRESHOLD else "--" -# ax.plot(x, y, linestyle=linestyle, color=cmap(norm(tval)), linewidth=4, alpha=0.9, zorder=2) - -# # Colorbar -# sm = plt.cm.ScalarMappable(cmap=cmap, norm=norm) -# sm.set_array([]) -# cbar = plt.colorbar(sm, ax=ax, shrink=0.85) -# cbar.set_label(f"Average {condition_prefix.rstrip('_')} θ-value (HbO)", fontsize=11) - -# # Format axes -# ax.set_aspect("equal") -# ax.set_title(f"Group-level θ-values for {condition_prefix.rstrip('_')} (HbO)", fontsize=14) -# ax.set_xlabel("X position (m)", fontsize=11) -# ax.set_ylabel("Y position (m)", fontsize=11) -# ax.grid(True, alpha=0.3) - -# all_x = [p[0] for p in src_pos.values()] + [p[0] for p in det_pos.values()] -# all_y = [p[1] for p in src_pos.values()] + [p[1] for p in det_pos.values()] -# ax.set_xlim(min(all_x) - 0.01, max(all_x) + 0.01) -# ax.set_ylim(min(all_y) - 0.01, max(all_y) + 0.01) - -# fig.tight_layout() -# fig.show() - - - - def plot_glm_results(file_path, raw_haemo, glm_est, design_matrix): fig_glms = [] # List to store figures @@ -3043,6 +2720,9 @@ def load_snirf(file_path: str) -> tuple[BaseRaw, Figure]: raw = read_raw_snirf(file_path, preload=True, verbose=VERBOSITY) # type: ignore raw.load_data(verbose=VERBOSITY) # type: ignore + # TODO: Why was this commented again? + # Maybe this should be a bypass parameter? + # # Strip the specified amount of seconds from the start of the file # total_duration = getattr(raw, "times")[-1] # if total_duration > SECONDS_TO_STRIP: @@ -3082,436 +2762,570 @@ def load_snirf(file_path: str) -> tuple[BaseRaw, Figure]: # Data science requires good statistics. This section is requiring work. Am working on figuring this out. Will likely add this to a new viewer. -from scipy import stats -import matplotlib.cm as cm +import os +import json +import warnings import numpy as np import pandas as pd +import scipy.stats as stats import matplotlib.pyplot as plt -import matplotlib.colors as mcolors -from scipy import stats +import seaborn as sns +import statsmodels.formula.api as smf from statsmodels.stats.multitest import multipletests +from statsmodels.tools.sm_exceptions import ConvergenceWarning +from mne_nirs.statistics import statsmodels_to_results +from mne_nirs.visualisation import plot_glm_group_topo +import logging + + +def run_roi_second_level_analysis(df_roi_all, df_cha_all=None, raw_haemo=None, + p_threshold=0.05, min_subjects=5, + correction_method='fdr_bh', target_chroma='hbo', + graph_bounds=None, roi_config=None, + threshold_topo=False): # Added parameter + """ + Perform group-level ROI analysis, prints stats to console, plots the ROI bar chart, + and dynamically plots isolated channel-level group topography maps based on a JSON config. + """ + # 1. Validation checks + required_cols = ['ROI', 'Condition', 'Chroma', 'theta', 'ID'] + if not all(col in df_roi_all.columns for col in required_cols): + raise ValueError(f"Input ROI DataFrame must include: {required_cols}") + + # 2. Filter ROI data for the targeted chromophore + df_chroma = df_roi_all[df_roi_all['Chroma'] == target_chroma].copy() + df_chroma = df_chroma.dropna(subset=['theta']) + + # 3. Perform 1-sample t-test against zero for each ROI + rois = df_chroma['ROI'].unique() + group_results = [] + + for roi in rois: + roi_data = df_chroma[df_chroma['ROI'] == roi] + sub_data = roi_data.groupby('ID', as_index=False)['theta'].mean() + + n_subs = sub_data['ID'].nunique() + if n_subs < min_subjects: + continue + + Y = sub_data['theta'].values + t_val, p_val = stats.ttest_1samp(Y, 0) + mean_beta = np.mean(Y) + std_err = stats.sem(Y) + + group_results.append({ + 'ROI': roi, + 't_val': t_val, + 'p_val': p_val, + 'mean_beta': mean_beta, + 'std_err': std_err, + 'n_subjects': n_subs + }) + + if not group_results: + print("\n[ERROR] No ROIs met the minimum subject threshold.\n") + return pd.DataFrame() + + df_group = pd.DataFrame(group_results) + + # 4. Multiple comparisons correction + if correction_method is not None: + reject, p_corrected, _, _ = multipletests( + df_group['p_val'].values, method=correction_method + ) + df_group['p_corrected'] = p_corrected + df_group['significant'] = reject + else: + df_group['p_corrected'] = df_group['p_val'] + df_group['significant'] = df_group['p_val'] <= p_threshold + + # Print results table to terminal + print("\n" + "="*65) + print(f" GROUP-LEVEL ROI STATISTICAL RESULTS ({target_chroma.upper()})") + print("="*65) + df_print = df_group.copy() + df_print['mean_beta'] = df_print['mean_beta'].apply(lambda x: f"{x:.4f}") + df_print['t_val'] = df_print['t_val'].apply(lambda x: f"{x:.3f}") + df_print['p_val'] = df_print['p_val'].apply(lambda x: f"{x:.4f}") + df_print['p_corrected'] = df_print['p_corrected'].apply(lambda x: f"{x:.4f}") + print(df_print[['ROI', 'mean_beta', 't_val', 'p_val', 'p_corrected', 'significant']].to_string(index=False)) + print("="*65 + "\n") + + # 5. Plotting ROI Bar Chart + sns.set_theme(style="whitegrid") + fig, ax = plt.subplots(figsize=(8, 6)) + + df_sub_avg = df_chroma.groupby(['ROI', 'ID'], as_index=False)['theta'].mean() + + sns.barplot( + data=df_sub_avg, x='ROI', y='theta', + ax=ax, errorbar=('ci', 95), capsize=0.1, + color='lightgray', edgecolor='black', linewidth=1.5, zorder=1 + ) + sns.swarmplot( + data=df_sub_avg, x='ROI', y='theta', + ax=ax, color='darkblue', size=8, alpha=0.7, zorder=2 + ) + + global_max = df_sub_avg['theta'].max() + global_min = df_sub_avg['theta'].min() + y_top = global_max * 1.35 if global_max > 0 else 0.5e-6 + y_bottom = global_min * 1.1 if global_min < 0 else -0.1 * global_max + ax.set_ylim(y_bottom, y_top) + + if graph_bounds is not None and graph_bounds > 0.0: + if graph_bounds < 0.5: + ax.set_ylim(-graph_bounds, graph_bounds) + + for idx, row in df_group.iterrows(): + roi_name = row['ROI'] + p_val_corr = row['p_corrected'] + + roi_points = df_sub_avg[df_sub_avg['ROI'] == roi_name]['theta'] + max_y = roi_points.max() if len(roi_points) > 0 else 0 + text_y = max_y + (global_max * 0.03) + + if p_val_corr < 0.001: + sig_symbol = "***" + elif p_val_corr < 0.01: + sig_symbol = "**" + elif p_val_corr < p_threshold: + sig_symbol = "*" + else: + sig_symbol = "n.s." + + sig_text = f"{sig_symbol}\np_corr = {p_val_corr:.3f}" + x_pos = list(rois).index(roi_name) + ax.text( + x_pos, text_y, sig_text, + ha='center', va='bottom', fontsize=11, + fontweight='bold', color='red' if p_val_corr < p_threshold else 'gray' + ) + + ax.axhline(0, color='black', linewidth=1, linestyle='--') + ax.set_ylabel(r'Hemodynamic Response ($\Delta$ HbO / $\mu$mol/L)' if global_max > 1e-3 else r'Hemodynamic Response ($\Delta$ HbO / mol/L)', fontsize=12) + ax.set_xlabel('Region of Interest (ROI)', fontsize=12) + + correction_lbl = f"({correction_method} corrected)" if correction_method else "(uncorrected)" + ax.set_title( + f"Group-Level ROI Activation ({target_chroma.upper()})\nSignificance threshold: p < {p_threshold} {correction_lbl}", + fontsize=13, fontweight='bold', pad=15 + ) + plt.tight_layout() + plt.show() + + # === 6. Segmented Topography Plotting (No Hardcoded Regions) === + if df_cha_all is not None and raw_haemo is not None: + print(f"--> Fitting group-level channel LME for {target_chroma.upper()} topography...") + try: + val_col = 'effect' if 'effect' in df_cha_all.columns else 'theta' + ch_col = 'ch_name' if 'ch_name' in df_cha_all.columns else 'channel' + + con_summary = df_cha_all[df_cha_all['Chroma'] == target_chroma].copy() + raw_picked = raw_haemo.copy().pick(picks=target_chroma) + + # Fit channel LME (suppress ConvergenceWarning locally) + model_formula = f"{val_col} ~ -1 + {ch_col}:Chroma" + with warnings.catch_warnings(): + warnings.simplefilter("ignore", ConvergenceWarning) + con_model = smf.mixedlm(model_formula, con_summary, groups=con_summary["ID"]).fit(method="nm") + + # Map statsmodels output to MNE result format + con_model_df = statsmodels_to_results(con_model, order=raw_picked.ch_names) + + # --- DYNAMIC ROI PARSING --- + roi_mapping = {} + if roi_config is not None: + raw_json = None + if isinstance(roi_config, str) and os.path.exists(roi_config): + with open(roi_config, 'r') as f: + raw_json = json.load(f) + elif isinstance(roi_config, dict): + raw_json = roi_config + + if raw_json: + if "regions_of_interest" in raw_json: + for roi_item in raw_json["regions_of_interest"]: + roi_name = roi_item.get("name") + channels = roi_item.get("channels", []) + if roi_name and channels: + roi_mapping[roi_name] = channels + else: + roi_mapping = raw_json + + if roi_mapping: + ch_to_roi = {} + for roi_name, channels in roi_mapping.items(): + for ch in channels: + ch_to_roi[ch] = roi_name + ch_to_roi[ch.split()[0]] = roi_name + + con_summary['ROI'] = con_summary[ch_col].apply( + lambda x: ch_to_roi.get(x, ch_to_roi.get(x.split()[0], None) if isinstance(x, str) else None) + ) + + unique_rois = [] + if 'ROI' in con_summary.columns: + unique_rois = [r for r in con_summary['ROI'].dropna().unique() if r != ""] + + if not unique_rois: + print("--> Warning: No ROI mappings detected. Plotting as a unified grid.") + unique_rois = ['All_Channels'] + con_summary['ROI'] = 'All_Channels' + + # Calculate shared symmetric color limits + vlim = (None, None) + if 'Coef.' in con_model_df.columns: + clean_vals = con_model_df['Coef.'].dropna().values + if len(clean_vals) > 0: + max_abs = np.max(np.abs(clean_vals)) + if max_abs > 1e-9: + vlim = (-max_abs, max_abs) + + fig_topo, ax_topo = plt.subplots(figsize=(6, 6)) + + # Dynamic loop: Plot each region independently + for i, roi_name in enumerate(unique_rois): + roi_ch_names = con_summary[con_summary['ROI'] == roi_name][ch_col].unique().tolist() + roi_ch_names = [ch for ch in roi_ch_names if ch in raw_picked.ch_names] + + if not roi_ch_names: + continue + + raw_roi = raw_picked.copy().pick(picks=roi_ch_names) + show_colorbar = (i == len(unique_rois) - 1) + + # === FIX 2: Filter stats dataframe first to prevent "Reducing GLM results..." warnings === + roi_con_model_df = con_model_df[con_model_df['ch_name'].isin(roi_ch_names)].copy() + + plot_glm_group_topo( + raw_roi, + roi_con_model_df, + colorbar=show_colorbar, + threshold=threshold_topo, # Now uses the parameter! + axes=ax_topo, + cmap='RdBu_r', + vlim=vlim + ) + + threshold_text = "p < 0.05 Masked" if threshold_topo else "Unthresholded" + ax_topo.set_title( + f"Group-Level {target_chroma.upper()} Activation Map\n(Regions Isolated Dynamically, {threshold_text})", + fontsize=11, fontweight='bold', pad=10 + ) + plt.tight_layout() + plt.show() + + except Exception as e: + logger.error(f"Could not generate topography plot: {e}") + + return df_group + + + +import os +import json +import warnings +import numpy as np +import pandas as pd +import scipy.stats as stats +import matplotlib.pyplot as plt +import seaborn as sns +from statsmodels.stats.multitest import multipletests +from mne_nirs.visualisation import plot_glm_group_topo import logging logger = logging.getLogger(__name__) - -def run_second_level_analysis(df_contrasts, raw, p, bounds, min_subjects=5, - correction_method='fdr_bh', color_by='t_val'): +def clean_subject_id(path_or_id): """ - Perform one-sample second-level analysis using contrast data from - multiple participants (tests each channel's group mean against zero). - - Parameters - ---------- - df_contrasts : pd.DataFrame - Combined contrast results from multiple participants. - Must include: ['ch_name', 'effect', 'ID'] - raw : mne.io.Raw - Raw object containing sensor geometry info. - p : float - P-value significance threshold (e.g., 0.05) for solid vs. dashed lines. - Applied to the *corrected* p-value. - bounds : float - Symmetric colormap limits (e.g., t-value limits like 4.0, or beta - value limits like 1e-6). Must match `color_by`. - min_subjects : int, default 5 - Minimum number of subjects required per channel to run a t-test. - n=2 gives 1 degree of freedom and is not a reliable estimate. - correction_method : str or None, default 'fdr_bh' - Multiple comparisons correction passed to - statsmodels.stats.multitest.multipletests. None skips correction. - color_by : {'t_val', 'mean_beta'}, default 't_val' - Which statistic drives the colormap. Match `bounds` to whichever is - chosen (t-values ~3-5; beta/concentration values are much smaller, - e.g. ~1e-6). + Cleans file paths and ID strings to get a standardized subject identifier. + E.g., 'C:/path/Sub-01_haemo.snirf' -> 'Sub-01' """ + if not isinstance(path_or_id, str): + return str(path_or_id) + base = os.path.basename(path_or_id) + for ext in ['.snirf', '.nirs', '.fif', '.csv', '.pkl', '_haemo']: + if base.endswith(ext): + base = base[:-len(ext)] + if base.endswith('_haemo'): + base = base[:-6] + return base - if not all(col in df_contrasts.columns for col in ['ch_name', 'effect', 'ID']): - raise ValueError("Input DataFrame must include 'ch_name', 'effect', and 'ID' columns.") - if color_by not in ('t_val', 'mean_beta'): - raise ValueError("color_by must be 't_val' or 'mean_beta'.") - n_before = len(df_contrasts) - df_contrasts = df_contrasts.dropna(subset=['effect']) - n_dropped = n_before - len(df_contrasts) - if n_dropped: - logger.warning(f"Dropped {n_dropped} rows with NaN 'effect' values.") +def run_cross_group_second_level_analysis(df_roi_all, file_paths_a, file_paths_b, + group_a_name="Group A", group_b_name="Group B", + df_cha_all=None, raw_haemo=None, + p_threshold=0.05, min_subjects=3, + correction_method='fdr_bh', target_chroma='hbo', + selected_event=None, graph_bounds=None, + roi_config=None, threshold_topo=False): + """ + Perform cross-group independent statistical analyses (Group A vs Group B), + renders a grouped bar chart with significance brackets, and plots a group-contrast topography map. + """ + # 1. Align IDs and filter dataset to selected Event & Chromophore + clean_a = set(file_paths_a) + clean_b = set(file_paths_b) + + df_roi_all = df_roi_all.copy() + # df_roi_all['clean_ID'] = df_roi_all['ID'].apply(clean_subject_id) + df_roi_all['clean_ID'] = df_roi_all['ID'] - channels = df_contrasts['ch_name'].unique() + # Filter for active experimental conditions + df_filtered = df_roi_all[ + (df_roi_all['Chroma'] == target_chroma) & + (df_roi_all['Condition'] == selected_event) + ].copy() + + + df_a = df_filtered[df_filtered['clean_ID'].isin(clean_a)].copy() + df_b = df_filtered[df_filtered['clean_ID'].isin(clean_b)].copy() + + print(f"DEBUG: Filtering for Event: {selected_event}") + print(f"DEBUG: Unique IDs in df_filtered: {df_filtered['clean_ID'].unique()}") + print(f"DEBUG: Clean IDs from Group A: {clean_a}") + print(f"DEBUG: Clean IDs from Group B: {clean_b}") + print(f"DEBUG: Rows in df_a: {len(df_a)}, Rows in df_b: {len(df_b)}") + + + if df_a.empty or df_b.empty: + print("[ERROR] Missing data for one or both cohorts. Check file selection/IDs.") + return pd.DataFrame() + + # 2. ROI-Level Welch's T-Test (Independent Two-Sample) + rois = df_filtered['ROI'].dropna().unique() group_results = [] - for ch in channels: - ch_data = df_contrasts[df_contrasts['ch_name'] == ch] - - # Collapse to one value per subject first (averages multiple - # runs/sessions per subject) to preserve the t-test's independence - # assumption. - ch_data = ch_data.groupby('ID', as_index=False)['effect'].mean() - - if ch_data['ID'].nunique() < min_subjects: - logger.warning( - f"Skipping channel {ch} — only {ch_data['ID'].nunique()} subject(s), " - f"need at least {min_subjects}." - ) + for roi in rois: + vals_a = df_a[df_a['ROI'] == roi].groupby('clean_ID')['theta'].mean().values + vals_b = df_b[df_b['ROI'] == roi].groupby('clean_ID')['theta'].mean().values + + n_a, n_b = len(vals_a), len(vals_b) + if n_a < min_subjects or n_b < min_subjects: continue - Y = ch_data['effect'].values - t_val, p_val = stats.ttest_1samp(Y, 0) - mean_beta = np.mean(Y) + # Welch's t-test (assumes unequal variances) + t_val, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False) + mean_a, mean_b = np.mean(vals_a), np.mean(vals_b) + diff_val = mean_a - mean_b group_results.append({ - 'ch_name': ch, + 'ROI': roi, + 'mean_A': mean_a, + 'mean_B': mean_b, + 'mean_diff': diff_val, 't_val': t_val, 'p_val': p_val, - 'mean_beta': mean_beta, - 'n_subjects': len(Y) + 'n_A': n_a, + 'n_B': n_b }) if not group_results: - fig, ax = plt.subplots(figsize=(8, 4)) - ax.text(0.5, 0.5, - f"Second-Level Analysis Aborted\n\n" - f"Reason: All {len(channels)} channels skipped.\n" - f"Requirement: At least {min_subjects} subjects (IDs) per channel.\n" - f"Current Subject Count: {df_contrasts['ID'].nunique()}", - ha='center', va='center', fontsize=12, color='darkred', - bbox=dict(facecolor='white', alpha=0.5, edgecolor='red')) - ax.set_axis_off() - plt.show() + print("\n[ERROR] No ROIs met the subject requirements for BOTH groups.\n") return pd.DataFrame() df_group = pd.DataFrame(group_results) + # Apply FDR correction if correction_method is not None: - reject, p_corrected, _, _ = multipletests( - df_group['p_val'].values, method=correction_method - ) + reject, p_corrected, _, _ = multipletests(df_group['p_val'].values, method=correction_method) df_group['p_corrected'] = p_corrected df_group['significant'] = reject else: - logger.warning("No multiple comparisons correction applied — " - "raw p-values will be used for thresholding.") df_group['p_corrected'] = df_group['p_val'] - df_group['significant'] = df_group['p_val'] <= p + df_group['significant'] = df_group['p_val'] <= p_threshold - logger.info("Second-level results:\n%s", df_group) + # Print clean terminal report + print("\n" + "="*85) + print(f" CROSS-GROUP ROI CONTRAST: {group_a_name.upper()} vs {group_b_name.upper()} ({target_chroma.upper()})") + print(f" Event Condition: {selected_event}") + print("="*85) + df_print = df_group.copy() + df_print['mean_A'] = df_print['mean_A'].apply(lambda x: f"{x:.4f}") + df_print['mean_B'] = df_print['mean_B'].apply(lambda x: f"{x:.4f}") + df_print['mean_diff'] = df_print['mean_diff'].apply(lambda x: f"{x:.4f}") + df_print['t_val'] = df_print['t_val'].apply(lambda x: f"{x:.3f}") + df_print['p_corrected'] = df_print['p_corrected'].apply(lambda x: f"{x:.4f}") + print(df_print[['ROI', 'mean_A', 'mean_B', 'mean_diff', 't_val', 'p_corrected', 'significant']].to_string(index=False)) + print("="*85 + "\n") - # Extract source/detector positions (ignoring auxiliary/stim channels) - src_pos: dict[int, tuple[float, float]] = {} - det_pos: dict[int, tuple[float, float]] = {} - for ch in getattr(raw, "info")["chs"]: - ch_name = ch['ch_name'] - if not ch_name or not ch['loc'].any() or '_' not in ch_name: - continue - try: - parts = ch_name.split()[0] - src_str, det_str = parts.split('_') - src_num = int(src_str[1:]) - det_num = int(det_str[1:]) - src_pos[src_num] = ch['loc'][3:5] - det_pos[det_num] = ch['loc'][6:8] - except Exception as e: - logger.debug(f"Skipping non-fNIRS channel geometry for {ch_name}: {e}") + # 3. Double Grouped Bar Plot (Side-by-Side) + sns.set_theme(style="whitegrid") + fig, ax = plt.subplots(figsize=(10, 6)) - fig, ax = plt.subplots(figsize=(8, 6)) + # Construct unified dataframe for seaborn grouped layouts + df_a_tidy = df_a.groupby(['ROI', 'clean_ID'], as_index=False)['theta'].mean() + df_a_tidy['Group'] = group_a_name + df_b_tidy = df_b.groupby(['ROI', 'clean_ID'], as_index=False)['theta'].mean() + df_b_tidy['Group'] = group_b_name + combined_df = pd.concat([df_a_tidy, df_b_tidy], ignore_index=True) - for pos in src_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='o', edgecolors='white', linewidths=1, zorder=3) - for pos in det_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='s', edgecolors='white', linewidths=1, zorder=3) - - norm = mcolors.Normalize(vmin=-bounds, vmax=bounds) - cmap = plt.get_cmap('seismic') - - for _, row in df_group.iterrows(): - ch = row['ch_name'] - tval = row['t_val'] - mean_val = row['mean_beta'] - is_sig = row['significant'] - - if '_' not in ch: - continue - - src_str, det_str = ch.split('_') - det_parts = det_str.split() - detector_id = det_parts[0] - hemo_type = det_parts[1].lower() if len(det_parts) > 1 else '' - - if hemo_type != 'hbo': - continue - - try: - src = int(src_str[1:]) - det = int(detector_id[1:]) - except Exception: - continue - - if src in src_pos and det in det_pos: - x = [src_pos[src][0], det_pos[det][0]] - y = [src_pos[src][1], det_pos[det][1]] - style = '-' if is_sig else '--' - color_val = tval if color_by == 't_val' else mean_val - color = cmap(norm(color_val)) - ax.plot(x, y, linestyle=style, color=color, linewidth=4, alpha=0.9, zorder=2) - - sm = plt.cm.ScalarMappable(cmap=cmap, norm=norm) - sm.set_array([]) - cbar = plt.colorbar(sm, ax=ax, shrink=0.85) - cbar.set_label( - 'Group t-value (HbO)' if color_by == 't_val' else 'Group mean beta (HbO)', - fontsize=11 + # Plot Bars + sns.barplot( + data=combined_df, x='ROI', y='theta', hue='Group', + hue_order=[group_a_name, group_b_name], order=rois, + ax=ax, errorbar=('ci', 95), capsize=0.08, + palette=['#2b5c8f', '#d95f02'], edgecolor='black', linewidth=1.5, zorder=1 + ) + # Plot Individual Dots (Dodged over the specific bar widths) + sns.swarmplot( + data=combined_df, x='ROI', y='theta', hue='Group', + hue_order=[group_a_name, group_b_name], order=rois, + ax=ax, size=6, color='black', alpha=0.5, dodge=True, zorder=2, + legend=False ) - ax.set_aspect('equal') - correction_label = correction_method if correction_method else 'uncorrected' - ax.set_title(f"Group-Level Activation Map (HbO), p<{p} ({correction_label})", fontsize=14) - ax.set_xlabel('X position (m)', fontsize=11) - ax.set_ylabel('Y position (m)', fontsize=11) - ax.grid(True, alpha=0.3) + global_max = combined_df['theta'].max() + global_min = combined_df['theta'].min() + y_top = global_max * 1.45 if global_max > 0 else 0.5e-6 + y_bottom = global_min * 1.15 if global_min < 0 else -0.15 * global_max + ax.set_ylim(y_bottom, y_top) - all_x = [pos[0] for pos in src_pos.values()] + [pos[0] for pos in det_pos.values()] - all_y = [pos[1] for pos in src_pos.values()] + [pos[1] for pos in det_pos.values()] - ax.set_xlim(min(all_x)-0.01, max(all_x)+0.01) - ax.set_ylim(min(all_y)-0.01, max(all_y)+0.01) + if graph_bounds is not None and graph_bounds > 0.0 and graph_bounds < 0.5: + ax.set_ylim(-graph_bounds, graph_bounds) - fig.tight_layout() + # Draw professional brackets over paired bars + for idx, row in df_group.iterrows(): + roi_name = row['ROI'] + p_val_corr = row['p_corrected'] + + roi_points = combined_df[combined_df['ROI'] == roi_name]['theta'] + max_y = roi_points.max() if len(roi_points) > 0 else 0 + + x_a = idx - 0.2 # Approximate left bar X offset + x_b = idx + 0.2 # Approximate right bar X offset + y_bracket = max_y + (global_max * 0.08) + h_tick = global_max * 0.02 + + if p_val_corr < p_threshold: + sig_symbol = "***" if p_val_corr < 0.001 else "**" if p_val_corr < 0.01 else "*" + # Draw standard bracket line + ax.plot([x_a, x_a, x_b, x_b], [y_bracket - h_tick, y_bracket, y_bracket, y_bracket - h_tick], color='black', lw=1.2) + ax.text( + idx, y_bracket + (global_max * 0.02), f"{sig_symbol}\np_corr = {p_val_corr:.3f}", + ha='center', va='bottom', fontsize=9, fontweight='bold', color='red' + ) + else: + ax.text( + idx, y_bracket, "n.s.", + ha='center', va='bottom', fontsize=9, color='gray' + ) + + ax.axhline(0, color='black', linewidth=1, linestyle='--') + ax.set_ylabel(r'Hemodynamic Response ($\Delta$ HbO)', fontsize=12) + ax.set_xlabel('Region of Interest (ROI)', fontsize=12) + ax.set_title(f"Cross-Group Comparison: {group_a_name} vs {group_b_name}\n({target_chroma.upper()} - {selected_event})", fontsize=13, fontweight='bold', pad=15) + plt.tight_layout() plt.show() + # 4. Channel-by-Channel Group-Contrast Topography Map (Zero Hardcoding) + if df_cha_all is not None and raw_haemo is not None: + print(f"--> Computing group-level channel contrasts for topography...") + try: + val_col = 'effect' if 'effect' in df_cha_all.columns else 'theta' + ch_col = 'ch_name' if 'ch_name' in df_cha_all.columns else 'channel' + + # Match channel levels and clean IDs + con_summary = df_cha_all[ + (df_cha_all['Chroma'] == target_chroma) & + (df_cha_all['Condition'] == selected_event) + ].copy() + con_summary['clean_ID'] = con_summary['ID'].apply(clean_subject_id) + + raw_picked = raw_haemo.copy().pick(picks=target_chroma) + + # --- Perform manual Channel-by-Channel Two-Sample t-tests --- + contrast_data = [] + for ch in raw_picked.ch_names: + ch_a = con_summary[(con_summary['clean_ID'].isin(clean_a)) & (con_summary[ch_col] == ch)] + ch_b = con_summary[(con_summary['clean_ID'].isin(clean_b)) & (con_summary[ch_col] == ch)] + + vals_a = ch_a[val_col].dropna().values + vals_b = ch_b[val_col].dropna().values + + if len(vals_a) >= min_subjects and len(vals_b) >= min_subjects: + t_stat, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False) + mean_diff = np.mean(vals_a) - np.mean(vals_b) + else: + t_stat, p_val, mean_diff = 0.0, 1.0, 0.0 + + contrast_data.append({ + 'ch_name': ch, + 'Coef.': mean_diff, # Represents Mean A - Mean B + 't': t_stat, + 'P>|t|': p_val # For threshold masking + }) + + con_model_df = pd.DataFrame(contrast_data) + + # --- DYNAMIC ROI PARSING --- + roi_mapping = {} + if roi_config is not None and os.path.exists(roi_config): + with open(roi_config, 'r') as f: + raw_json = json.load(f) + if "regions_of_interest" in raw_json: + for roi_item in raw_json["regions_of_interest"]: + roi_mapping[roi_item.get("name")] = roi_item.get("channels", []) + + if roi_mapping: + ch_to_roi = {} + for roi_name, channels in roi_mapping.items(): + for ch in channels: + ch_to_roi[ch] = roi_name + ch_to_roi[ch.split()[0]] = roi_name + con_summary['ROI'] = con_summary[ch_col].apply(lambda x: ch_to_roi.get(x, ch_to_roi.get(x.split()[0], None) if isinstance(x, str) else None)) + + unique_rois = [r for r in con_summary['ROI'].dropna().unique() if r != ""] if 'ROI' in con_summary.columns else ['All_Channels'] + + # Shared symmetric limits for the color bar + max_abs = np.max(np.abs(con_model_df['Coef.'].dropna().values)) if len(con_model_df['Coef.']) > 0 else 1.0 + vlim = (-max_abs, max_abs) if max_abs > 1e-9 else (None, None) + + fig_topo, ax_topo = plt.subplots(figsize=(6, 6)) + + # Isolated dynamic plotting loop to prevent spatial bleeding + for i, roi_name in enumerate(unique_rois): + roi_ch_names = con_summary[con_summary['ROI'] == roi_name][ch_col].unique().tolist() if 'ROI' in con_summary.columns else raw_picked.ch_names + roi_ch_names = [ch for ch in roi_ch_names if ch in raw_picked.ch_names] + + if not roi_ch_names: + continue + + raw_roi = raw_picked.copy().pick(picks=roi_ch_names) + show_colorbar = (i == len(unique_rois) - 1) + + # Filter contrast DF to current ROI channels + roi_con_df = con_model_df[con_model_df['ch_name'].isin(roi_ch_names)].copy() + + plot_glm_group_topo( + raw_roi, + roi_con_df, + colorbar=show_colorbar, + threshold=threshold_topo, + axes=ax_topo, + cmap='RdBu_r', + vlim=vlim + ) + + threshold_text = "p < 0.05 Masked" if threshold_topo else "Unthresholded Contrast" + ax_topo.set_title(f"Group Contrast: {group_a_name} - {group_b_name}\n({target_chroma.upper()} - {threshold_text})", fontsize=11, fontweight='bold', pad=10) + plt.tight_layout() + plt.show() + + except Exception as e: + logger.error(f"Could not generate group-contrast topography plot: {e}", exc_info=True) + return df_group -def run_second_level_analysis_two_sample(df_contrasts, raw, p, bounds, - group_a_ids, group_b_ids, - min_subjects=5, - correction_method='fdr_bh', - color_by='t_val', - group_a_label='Group A', - group_b_label='Group B'): - """ - Compare the same contrast between two independent groups of participants - (e.g. two age ranges, patients vs controls) using Welch's t-test per - channel. This is NOT the same statistical test as run_second_level_analysis - — that function asks "is this contrast non-zero within one group?"; this - one asks "does this contrast differ between two separate groups of - people?" Group sizes and variances are not assumed equal (Welch's - correction), since two independently recruited groups will rarely match - on either. - Parameters - ---------- - df_contrasts : pd.DataFrame - Combined contrast results from multiple participants across BOTH - groups. Must include: ['ch_name', 'effect', 'ID'] - raw : mne.io.Raw - Raw object containing sensor geometry info. - p : float - P-value significance threshold (e.g., 0.05) for solid vs. dashed lines. - Applied to the *corrected* p-value. - bounds : float - Symmetric colormap limits. Must match `color_by`: - - 't_val': typically ~3-5 - - 'mean_diff': match your effect units (e.g. ~1e-6) - group_a_ids, group_b_ids : list - The 'ID' values (e.g. file paths) belonging to each group. IDs not - present in either list are excluded from the comparison — this lets - you pass a df_contrasts containing more groups than the two you're - currently comparing. - min_subjects : int, default 5 - Minimum number of subjects required in EACH group for a channel to - be tested. A channel is skipped if either group falls short, not - just the combined total — an imbalanced 8-vs-2 split is not the same - as a balanced 5-vs-5 split even though both total 10. - correction_method : str or None, default 'fdr_bh' - Multiple comparisons correction passed to - statsmodels.stats.multitest.multipletests. None skips correction. - color_by : {'t_val', 'mean_diff'}, default 't_val' - Which statistic drives the colormap. - group_a_label, group_b_label : str - Display labels for the two groups (used in the title only). - """ - if not all(col in df_contrasts.columns for col in ['ch_name', 'effect', 'ID']): - raise ValueError("Input DataFrame must include 'ch_name', 'effect', and 'ID' columns.") - if color_by not in ('t_val', 'mean_diff'): - raise ValueError("color_by must be 't_val' or 'mean_diff'.") - - overlap = set(group_a_ids) & set(group_b_ids) - if overlap: - raise ValueError( - f"{len(overlap)} ID(s) appear in both groups — a participant can't be " - f"in both the A and B samples of an independent two-sample test: {overlap}" - ) - - n_before = len(df_contrasts) - df_contrasts = df_contrasts.dropna(subset=['effect']) - n_dropped = n_before - len(df_contrasts) - if n_dropped: - logger.warning(f"Dropped {n_dropped} rows with NaN 'effect' values.") - - # Keep only rows belonging to one of the two groups being compared. - keep_ids = set(group_a_ids) | set(group_b_ids) - df_contrasts = df_contrasts[df_contrasts['ID'].isin(keep_ids)] - - channels = df_contrasts['ch_name'].unique() - group_results = [] - - for ch in channels: - ch_data = df_contrasts[df_contrasts['ch_name'] == ch] - - # Collapse to one value per subject first (averages multiple - # runs/sessions per subject). - ch_data = ch_data.groupby('ID', as_index=False)['effect'].mean() - - Y_a = ch_data[ch_data['ID'].isin(group_a_ids)]['effect'].values - Y_b = ch_data[ch_data['ID'].isin(group_b_ids)]['effect'].values - - if len(Y_a) < min_subjects or len(Y_b) < min_subjects: - logger.warning( - f"Skipping channel {ch} — group sizes are {len(Y_a)} (A) and " - f"{len(Y_b)} (B), need at least {min_subjects} in EACH group." - ) - continue - - # Welch's t-test: does not assume equal variance or equal N between - # the two independent groups. - t_val, p_val = stats.ttest_ind(Y_a, Y_b, equal_var=False) - mean_a = np.mean(Y_a) - mean_b = np.mean(Y_b) - mean_diff = mean_a - mean_b - - group_results.append({ - 'ch_name': ch, - 't_val': t_val, - 'p_val': p_val, - 'mean_a': mean_a, - 'mean_b': mean_b, - 'mean_diff': mean_diff, - 'n_a': len(Y_a), - 'n_b': len(Y_b), - }) - - if not group_results: - fig, ax = plt.subplots(figsize=(8, 4)) - ax.text(0.5, 0.5, - f"Two-Sample Second-Level Analysis Aborted\n\n" - f"Reason: All {len(channels)} channels skipped.\n" - f"Requirement: At least {min_subjects} subjects per group, per channel.\n" - f"Group A total IDs: {len(set(group_a_ids))} | " - f"Group B total IDs: {len(set(group_b_ids))}", - ha='center', va='center', fontsize=12, color='darkred', - bbox=dict(facecolor='white', alpha=0.5, edgecolor='red')) - ax.set_axis_off() - plt.show() - return pd.DataFrame() - - df_group = pd.DataFrame(group_results) - - if correction_method is not None: - reject, p_corrected, _, _ = multipletests( - df_group['p_val'].values, method=correction_method - ) - df_group['p_corrected'] = p_corrected - df_group['significant'] = reject - else: - logger.warning("No multiple comparisons correction applied — " - "raw p-values will be used for thresholding.") - df_group['p_corrected'] = df_group['p_val'] - df_group['significant'] = df_group['p_val'] <= p - - logger.info("Two-sample second-level results:\n%s", df_group) - - # Extract source/detector positions - src_pos: dict[int, tuple[float, float]] = {} - det_pos: dict[int, tuple[float, float]] = {} - for ch in getattr(raw, "info")["chs"]: - ch_name = ch['ch_name'] - if not ch_name or not ch['loc'].any() or '_' not in ch_name: - continue - try: - parts = ch_name.split()[0] - src_str, det_str = parts.split('_') - src_num = int(src_str[1:]) - det_num = int(det_str[1:]) - src_pos[src_num] = ch['loc'][3:5] - det_pos[det_num] = ch['loc'][6:8] - except Exception as e: - logger.debug(f"Skipping non-fNIRS channel geometry for {ch_name}: {e}") - - fig, ax = plt.subplots(figsize=(8, 6)) - - for pos in src_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='o', edgecolors='white', linewidths=1, zorder=3) - for pos in det_pos.values(): - ax.scatter(pos[0], pos[1], s=120, c='k', marker='s', edgecolors='white', linewidths=1, zorder=3) - - norm = mcolors.Normalize(vmin=-bounds, vmax=bounds) - cmap = plt.get_cmap('seismic') - - for _, row in df_group.iterrows(): - ch = row['ch_name'] - tval = row['t_val'] - mean_diff = row['mean_diff'] - is_sig = row['significant'] - - if '_' not in ch: - continue - - src_str, det_str = ch.split('_') - det_parts = det_str.split() - detector_id = det_parts[0] - hemo_type = det_parts[1].lower() if len(det_parts) > 1 else '' - - if hemo_type != 'hbo': - continue - - try: - src = int(src_str[1:]) - det = int(detector_id[1:]) - except Exception: - continue - - if src in src_pos and det in det_pos: - x = [src_pos[src][0], det_pos[det][0]] - y = [src_pos[src][1], det_pos[det][1]] - style = '-' if is_sig else '--' - color_val = tval if color_by == 't_val' else mean_diff - color = cmap(norm(color_val)) - ax.plot(x, y, linestyle=style, color=color, linewidth=4, alpha=0.9, zorder=2) - - sm = plt.cm.ScalarMappable(cmap=cmap, norm=norm) - sm.set_array([]) - cbar = plt.colorbar(sm, ax=ax, shrink=0.85) - cbar.set_label( - f'{group_a_label} vs {group_b_label}: t-value (HbO)' if color_by == 't_val' - else f'{group_a_label} vs {group_b_label}: mean difference (HbO)', - fontsize=11 - ) - - ax.set_aspect('equal') - correction_label = correction_method if correction_method else 'uncorrected' - ax.set_title( - f"Group Difference Map (HbO): {group_a_label} vs {group_b_label}, " - f"p<{p} ({correction_label})", - fontsize=14 - ) - ax.set_xlabel('X position (m)', fontsize=11) - ax.set_ylabel('Y position (m)', fontsize=11) - ax.grid(True, alpha=0.3) - - all_x = [pos[0] for pos in src_pos.values()] + [pos[0] for pos in det_pos.values()] - all_y = [pos[1] for pos in src_pos.values()] + [pos[1] for pos in det_pos.values()] - ax.set_xlim(min(all_x)-0.01, max(all_x)+0.01) - ax.set_ylim(min(all_y)-0.01, max(all_y)+0.01) - - fig.tight_layout() - plt.show() - - return df_group @@ -3528,6 +3342,7 @@ def calculate_dpf(file_path): wavelengths = sorted(wavelengths, reverse=True) age = float(AGE) logger.info(f"Their age was {AGE}") + # where the hell did I get these from again? a = 223.3 b = 0.05624 c = 0.8493 @@ -4295,6 +4110,7 @@ def process_participant(file_path, progress_callback=None): fig_individual: dict[str, Figure] = {} logger.info(f"Folding Bypass: {FOLDING_BYP}") + JSON_LOCATION = r"C:\Users\tyler\Desktop\research\flares\regions.json" # Step 1: Preprocessing raw = load_snirf(file_path) @@ -4534,14 +4350,14 @@ def process_participant(file_path, progress_callback=None): logger.info("Step 21 Completed.") # Step 18: Design Matrix - design_matrix, fig_design_matrix = make_design_matrix(raw_haemo, short_chans) + df_design_matrix, fig_design_matrix = make_design_matrix(raw_haemo, short_chans) fig_individual["Design Matrix"] = fig_design_matrix if progress_callback: progress_callback(22) logger.info("Step 22 Completed.") # Step 19: Run GLM - glm_est = run_glm(raw_haemo, design_matrix, noise_model=NOISE_MODEL, bins=BINS, n_jobs=N_JOBS, verbose=VERBOSITY) + glm_est = run_glm(raw_haemo, df_design_matrix, noise_model=NOISE_MODEL, bins=BINS, n_jobs=N_JOBS, verbose=VERBOSITY) # Not used AppData\Local\Packages\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0\LocalCache\local-packages\Python313\site-packages\nilearn\glm\contrasts.py # Yes used AppData\Local\Packages\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0\LocalCache\local-packages\Python313\site-packages\mne_nirs\utils\_io.py @@ -4558,50 +4374,93 @@ def process_participant(file_path, progress_callback=None): # Step 20: Generate GLM Results if "derivative" not in HRF_MODEL.lower(): - fig_glm_result = plot_glm_results(file_path, raw_haemo, glm_est, design_matrix) + fig_glm_result = plot_glm_results(file_path, raw_haemo, glm_est, df_design_matrix) for name, fig in fig_glm_result: fig_individual[f"GLM {name}"] = fig if progress_callback: progress_callback(24) logger.info("24") # Step 21: Generate Channel Significance - if HRF_MODEL == "fir": - fig_significance = individual_significance(raw_haemo, glm_est) - for name, fig in fig_significance: - fig_individual[f"Significance {name}"] = fig + # if HRF_MODEL == "fir": + # fig_significance = individual_significance(raw_haemo, glm_est) + # for name, fig in fig_significance: + # fig_individual[f"Significance {name}"] = fig if progress_callback: progress_callback(25) logger.info("25") # Step 22: Generate Channel, Region of Interest, and Contrast Results - cha = glm_est.to_dataframe() + df_cha = glm_est.to_dataframe() # HACK: Comment out line 588 (self._renderer.show()) in _brain.py from MNE # brain_thing = brain_3d_visualization(cha, raw_haemo) # brain_individual.append(brain_thing) # C++ objects made this get rendered on the fly - rois = dict(AllChannels=range(len(raw_haemo.ch_names))) - # Calculate ROI for all conditions - conditions = design_matrix.columns - # Compute output metrics by ROI - df_ind = glm_est.to_dataframe_region_of_interest(rois, conditions) - df_ind["ID"] = file_path + + import json + + try: + with open(JSON_LOCATION, 'r') as f: + roi_data = json.load(f) + + rois_formatted = {} + for region in roi_data.get("regions_of_interest", []): + roi_name = region["name"] + channels = region["channels"] + + # Map JSON names to MNE's expected names + mne_channels = [] + for ch in channels: + mne_channels.append(f"{ch} hbo") + mne_channels.append(f"{ch} hbr") + + # Find the INTEGER index for each valid channel in raw_haemo + valid_indices = [] + for ch in mne_channels: + if ch in raw_haemo.ch_names: + # Get the integer index of this channel name + idx = raw_haemo.ch_names.index(ch) + valid_indices.append(idx) + + if valid_indices: + rois_formatted[roi_name] = valid_indices + else: + logger.warning(f"No channels from ROI '{roi_name}' found in raw_haemo.") + + except Exception as e: + logger.error(f"Failed to load or parse ROI JSON: {e}") + # Fallback to All Channels (as indices) if the JSON fails + rois_formatted = dict(AllChannels=list(range(len(raw_haemo.ch_names)))) + + # 3. Calculate ROI results for all conditions + conditions = df_design_matrix.columns + + # Compute output metrics by custom parsed ROIs (now passing lists of integers) + df_roi = glm_est.to_dataframe_region_of_interest(rois_formatted, conditions) + df_roi["ID"] = file_path + + # rois = dict(AllChannels=range(len(raw_haemo.ch_names))) + # # Calculate ROI for all conditions + # conditions = design_matrix.columns + # # Compute output metrics by ROI + # df_roi = glm_est.to_dataframe_region_of_interest(rois, conditions) + # Step 18: Fold channels # fig_fold_data, fig_fold_legend = fold_channels(raw_haemo) # fig_individual.append(fig_fold_data) # fig_individual.append(fig_fold_legend) - print(design_matrix) + print(df_design_matrix) - contrast_matrix = np.eye(design_matrix.shape[1]) + contrast_matrix = np.eye(df_design_matrix.shape[1]) basic_conts = dict( - [(column, contrast_matrix[i]) for i, column in enumerate(design_matrix.columns)] + [(column, contrast_matrix[i]) for i, column in enumerate(df_design_matrix.columns)] ) if HRF_MODEL == "fir": - all_delay_cols = [col for col in design_matrix.columns if "_delay_" in col] + all_delay_cols = [col for col in df_design_matrix.columns if "_delay_" in col] all_conditions = sorted({col.split("_delay_")[0] for col in all_delay_cols}) if not all_conditions: @@ -4630,7 +4489,7 @@ def process_participant(file_path, progress_callback=None): logger.info("26") # Step 23: Compute Contrast Results - contrast_results = {} + contrast_results_dict = {} # 0 is NOT a baseline. # AI Explaination: @@ -4645,40 +4504,40 @@ def process_participant(file_path, progress_callback=None): contrast = glm_est.compute_contrast(contrast_vector) # type: ignore df = contrast.to_dataframe() df["ID"] = file_path - contrast_results[cond] = df + contrast_results_dict[cond] = df else: exclude_list = ["drift", "constant", "short"] - task_cols = [c for c in design_matrix.columns if not any(ex in c.lower() for ex in exclude_list)] + task_cols = [c for c in df_design_matrix.columns if not any(ex in c.lower() for ex in exclude_list)] # Dictionary to hold all our contrast results - contrast_results = {} + contrast_results_dict = {} # 2. Loop through every task to create "Simple" contrasts (Task vs 0) for cond in task_cols: # Create a vector of zeros - vec = np.zeros(len(design_matrix.columns)) + vec = np.zeros(len(df_design_matrix.columns)) # Set the index of our condition to 1 - vec[list(design_matrix.columns).index(cond)] = 1 + vec[list(df_design_matrix.columns).index(cond)] = 1 contrast = glm_est.compute_contrast(vec) df = contrast.to_dataframe() df["ID"] = file_path - contrast_results[f"{cond}_vs_Zero"] = df + contrast_results_dict[f"{cond}_vs_Zero"] = df # 3. Loop through all combinations to create "Differential" contrasts (Task A vs Task B) # This uses itertools.combinations to get every unique pair for cond_a, cond_b in itertools.combinations(task_cols, 2): - vec = np.zeros(len(design_matrix.columns)) - vec[list(design_matrix.columns).index(cond_a)] = 1 - vec[list(design_matrix.columns).index(cond_b)] = -1 + vec = np.zeros(len(df_design_matrix.columns)) + vec[list(df_design_matrix.columns).index(cond_a)] = 1 + vec[list(df_design_matrix.columns).index(cond_b)] = -1 contrast = glm_est.compute_contrast(vec) df = contrast.to_dataframe() df["ID"] = file_path # Name the key nicely, e.g., "Tapping_Right_vs_Tapping_Left" - contrast_results[f"{cond_a}_vs_{cond_b}"] = df + contrast_results_dict[f"{cond_a}_vs_{cond_b}"] = df #NOTE: Temporary export_dir = os.path.join(os.path.dirname(file_path), "exported_contrasts") @@ -4686,7 +4545,7 @@ def process_participant(file_path, progress_callback=None): base = os.path.splitext(os.path.basename(file_path))[0] combined_contrasts = [] - for contrast_name, df in contrast_results.items(): + for contrast_name, df in contrast_results_dict.items(): df = df.copy() df["contrast_name"] = contrast_name combined_contrasts.append(df) @@ -4696,17 +4555,17 @@ def process_participant(file_path, progress_callback=None): os.path.join(export_dir, f"{base}_contrasts.csv"), index=False ) - df_ind.to_csv(os.path.join(export_dir, f"{base}_df_ind.csv"), index=False) - design_matrix.to_csv(os.path.join(export_dir, f"{base}_design_matrix.csv")) + df_roi.to_csv(os.path.join(export_dir, f"{base}_df_ind.csv"), index=False) + df_design_matrix.to_csv(os.path.join(export_dir, f"{base}_design_matrix.csv")) logger.info(f"Exported contrast/ROI/design-matrix CSVs to {export_dir}") - cha["ID"] = file_path + df_cha["ID"] = file_path if progress_callback: progress_callback(27) logger.info("27") # Step 24: Finishing Up - fig_bytes = convert_fig_dict_to_png_bytes(fig_individual) + fig_bytes_dict = convert_fig_dict_to_png_bytes(fig_individual) if FOLDING_BYP: epochs = None @@ -4717,15 +4576,17 @@ def process_participant(file_path, progress_callback=None): # TODO: Tidy up # Extract the parameters this file was ran with. No need to return age, gender, group? - config = { + config_dict = { k: globals()[k] for k in __annotations__ if k in globals() and k != "REQUIRED_KEYS" } - print(config) + print(config_dict) + + return raw_haemo, epochs, df_cha, df_roi, df_design_matrix, config_dict, fig_bytes_dict, contrast_results_dict, True + - return raw_haemo, config, epochs, fig_bytes, cha, contrast_results, df_ind, design_matrix, True def sanitize_paths_for_pickle(raw_haemo, epochs): diff --git a/main.py b/main.py index ec31065..76bd61e 100644 --- a/main.py +++ b/main.py @@ -268,7 +268,13 @@ SECTIONS = [ {"name": "N_JOBS", "default": 1, "type": int, "help": "The number of CPUs to use to do the GLM computation. -1 means 'all CPUs'."}, ] }, - { + { + "title": "Region of Interest", + "params": [ + {"name": "JSON_LOCATION", "default": "", "type": str, "help": "Location of the JSON file containing region of interest results for significance calculations."}, + ] + }, + { "title": "Finishing Touches", "params": [ # Intentionally empty (TODO) @@ -287,6 +293,19 @@ SECTIONS = [ +DATA_SCHEMA = [ + {"key": "raw_haemo_dict", "help": "Dict[file_path, MNE RawArray]: Haemodynamic raw data"}, + {"key": "epochs_dict", "help": "Dict[file_path, MNE Epochs]: Time-locked epoch data"}, + {"key": "cha_dict", "help": "Dict[file_path, DataFrame]: Channel analysis results"}, + {"key": "df_ind_dict", "help": "Dict[file_path, DataFrame]: Individual-level data/ROI results"}, + {"key": "design_matrix_dict", "help": "Dict[file_path, DataFrame]: GLM design matrices"}, + {"key": "config_dict", "help": "Dict[file_path, dict]: Processing configuration parameters"}, + {"key": "fig_bytes_dict", "help": "Dict[file_path, dict]: Serialized figure data"}, + {"key": "contrast_results_dict", "help": "Dict[file_path, dict]: Calculated contrast statistical results"}, + {"key": "valid_dict", "help": "Dict[file_path, bool]: Boolean validity status per file"} +] + + @@ -488,15 +507,9 @@ class MainApplication(QMainWindow): # Initialization to ensure that saving can occur - self.raw_haemo_dict = {} # Processed Hemodynamic data - self.config_dict = {} # Analysis parameters/settings - self.epochs_dict = {} # Timing/Event data - self.cha_dict = {} # Channel configurations - self.contrast_results_dict = {} # Statistical results - self.df_ind_dict = {} # Individual dataframes - self.design_matrix_dict = {} # GLM Design matrices - self.valid_dict = {} # Quality control/Validity flags - self.fig_bytes_dict = {} # Cached plot images (serialized) + for item in DATA_SCHEMA: + setattr(self, item["key"], {}) + self.file_metadata = {} # AGE, GENDER, GROUP self.metadata_cache = {} # Internal file/path information metadata cache self.bubble_widgets = {} # References to the UI "Bubble" objects @@ -878,15 +891,8 @@ class MainApplication(QMainWindow): self.files_done = set() self.files_failed = set() - self.raw_haemo_dict = {} - self.config_dict = {} - self.epochs_dict = {} - self.fig_bytes_dict = {} - self.cha_dict = {} - self.contrast_results_dict = {} - self.df_ind_dict = {} - self.design_matrix_dict = {} - self.valid_dict = {} + for item in DATA_SCHEMA: + setattr(self, item["key"], {}) self.metadata_cache = {} @@ -1045,7 +1051,22 @@ class MainApplication(QMainWindow): def open_launcher_window(self): - self.launcher_window = ViewerLauncherWidget(self.raw_haemo_dict, self.config_dict, self.fig_bytes_dict, self.cha_dict, self.contrast_results_dict, self.df_ind_dict, self.design_matrix_dict, self.epochs_dict, self.folding_bypass) + data_map = {item["key"]: getattr(self, item["key"]) for item in DATA_SCHEMA} + + # 2. Extract values in the specific order the widget constructor expects + args = [ + data_map["raw_haemo_dict"], + data_map["epochs_dict"], + data_map["cha_dict"], + data_map["df_ind_dict"], + data_map["design_matrix_dict"], + data_map["config_dict"], + data_map["fig_bytes_dict"], + data_map["contrast_results_dict"], + self.folding_bypass + ] + + self.launcher_window = ViewerLauncherWidget(*args) self.launcher_window.show() def copy_text(self): @@ -1188,7 +1209,7 @@ class MainApplication(QMainWindow): def open_folder_dialog(self): folder_path = QFileDialog.getExistingDirectory(self, "Select Folder", "") if folder_path: - snirf_files = [os.path.normpath(str(f)) for f in Path(folder_path).glob("*.snirf")] + snirf_files = [os.path.normpath(str(f)) for f in Path(folder_path).rglob("*.snirf")] self._load_files_into_pipeline(snirf_files) @@ -1297,7 +1318,10 @@ class MainApplication(QMainWindow): has_param_changes = any(section.has_any_changes() for section in self.param_sections) # Check if there is processed data - has_processed_data = bool(getattr(self, 'raw_haemo_dict', None)) + has_processed_data = any( + len(getattr(self, item["key"], {})) > 0 + for item in DATA_SCHEMA + ) if not (has_processed_data or has_metadata or has_param_changes): if not onCrash: # Don't show popups during a crash/autosave @@ -1368,23 +1392,17 @@ class MainApplication(QMainWindow): current_params = self.config_dict[first_file] version = CURRENT_VERSION - project_data = { + + project_data = {item["key"]: getattr(self, item["key"]) for item in DATA_SCHEMA} + + project_data.update({ "version": version, "file_list": file_list, "progress_states": progress_states, - "raw_haemo_dict": self.raw_haemo_dict, "file_metadata": rel_metadata, "file_parameters": rel_file_params, - "config_dict": self.config_dict, - "epochs_dict": self.epochs_dict, - "fig_bytes_dict": self.fig_bytes_dict, - "cha_dict": self.cha_dict, "current_ui_params": current_params, - "contrast_results_dict": self.contrast_results_dict, - "df_ind_dict": self.df_ind_dict, - "design_matrix_dict": self.design_matrix_dict, - "valid_dict": self.valid_dict, - } + }) def sanitize(obj): if isinstance(obj, Path): @@ -1472,15 +1490,9 @@ class MainApplication(QMainWindow): return - self.raw_haemo_dict = data.get("raw_haemo_dict", {}) - self.config_dict = data.get("config_dict", {}) - self.epochs_dict = data.get("epochs_dict", {}) - self.fig_bytes_dict = data.get("fig_bytes_dict", {}) - self.cha_dict = data.get("cha_dict", {}) - self.contrast_results_dict = data.get("contrast_results_dict", {}) - self.df_ind_dict = data.get("df_ind_dict", {}) - self.design_matrix_dict = data.get("design_matrix_dict", {}) - self.valid_dict = data.get("valid_dict", {}) + for item in DATA_SCHEMA: + key = item["key"] + setattr(self, key, data.get(key, {})) project_dir = Path(filename).parent @@ -1530,7 +1542,7 @@ class MainApplication(QMainWindow): first_file = next(iter(self.config_dict.keys())) self.restore_sections_from_config(self.config_dict[first_file]) - has_data = bool(self.raw_haemo_dict) + has_data = any(len(getattr(self, item["key"], {})) > 0 for item in DATA_SCHEMA) self.button1.setVisible(not has_data) self.button3.setVisible(has_data) @@ -1955,15 +1967,8 @@ class MainApplication(QMainWindow): self.button3.setVisible(False) - self.raw_haemo_dict = {} - self.config_dict = {} - self.epochs_dict = {} - self.fig_bytes_dict = {} - self.cha_dict = {} - self.contrast_results_dict = {} - self.df_ind_dict = {} - self.design_matrix_dict = {} - self.valid_dict = {} + for item in DATA_SCHEMA: + setattr(self, item["key"], {}) self.button1.clicked.disconnect(self.on_run_task) self.button1.setText("Cancel") @@ -2091,27 +2096,13 @@ class MainApplication(QMainWindow): # print(f"[DEBUG] Progress: {len(self.files_done)} / {self.files_total}") if msg.get("success"): - # Unpack the massive tuple - raw_haemo, config, epochs, fig_bytes, cha, contrast, df_ind, design, valid = msg["result"] - - # Initialize dictionaries once if needed - if not hasattr(self, 'raw_haemo_dict') or self.raw_haemo_dict is None: - attrs = ['raw_haemo_dict', 'config_dict', 'epochs_dict', 'fig_bytes_dict', - 'cha_dict', 'contrast_results_dict', 'df_ind_dict', - 'design_matrix_dict', 'valid_dict'] - for attr in attrs: - setattr(self, attr, {}) - self.files_results[file_path] = msg["result"] - self.raw_haemo_dict[file_path] = raw_haemo - self.config_dict[file_path] = config - self.epochs_dict[file_path] = epochs - self.fig_bytes_dict[file_path] = fig_bytes - self.cha_dict[file_path] = cha - self.contrast_results_dict[file_path] = contrast - self.df_ind_dict[file_path] = df_ind - self.design_matrix_dict[file_path] = design - self.valid_dict[file_path] = valid + results = msg["result"] + self.files_results[file_path] = results + + # Simple, clean assignment + for item, value in zip(DATA_SCHEMA, results): + getattr(self, item["key"])[file_path] = value self.statusbar.showMessage(f"Processed: {os.path.basename(file_path)}") diff --git a/src/analysis/crossgroupbrainimage.py b/src/analysis/crossgroupbrainimage.py new file mode 100644 index 0000000..17d0eca --- /dev/null +++ b/src/analysis/crossgroupbrainimage.py @@ -0,0 +1,137 @@ +""" +Filename: crossgroupbrainimage.py +Description: Logic for the Cross-Group Brain & Image analysis window + +Author: Tyler de Zeeuw +License: GPL-3.0 +""" + +# External library imports +import pandas as pd + +from flares import aggregate_fnirs_group_geometry, plot_2d_3d_contrasts_between_groups +from src.shared.flaresbasewidget import CrossGroupUIMixin, FlaresBaseWidget +from src.shared.shareddata import APP_NAME + + +PARAMETERIZED_INDEXES = { + 0: [ + { + "key": "show_optodes", + "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", + "default": "all", + "type": str, + }, + { + "key": "t_or_theta", + "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", + "default": "theta", + "type": str, + }, + { + "key": "show_text", + "label": "Display informative text on the top left corner about the contrast.", + "default": "True", + "type": bool, + }, + { + "key": "brain_bounds", + "label": "Graph Upper/Lower Limit", + "default": "1.0", + "type": float, + }, + { + "key": "is_3d", + "label": "Should we display the results in a 3D interactive window?", + "default": "True", + "type": bool, + } + ], +} + + + +class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget): + def __init__(self, haemo_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict): + super().__init__("CrossGroupBrainImage") + self.setWindowTitle(f"Cross-Group Brain & Image Viewer - {APP_NAME.upper()}") + self.haemo_dict = haemo_dict + self.df_ind_dict = df_ind_dict + self.design_matrix_dict = design_matrix_dict + self.contrast_results_dict = contrast_results_dict + self.group_dict = group_dict + + self.setup_cross_group_ui(["0 (Contrast Image)"]) + + + def proccess_request(self): + + request = self.get_common_request_data(PARAMETERIZED_INDEXES) + if request is None: + return + + (selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = request + + + # Build group-level contrast DataFrames + def concat_group_contrasts(file_paths: list[str], event: str | None) -> pd.DataFrame: + group_df = pd.DataFrame() + for fp in file_paths: + print(f"Looking up contrast for: {fp}") + event_con_dict = self.contrast_results_dict.get(fp, {}) + print("Available events for this file:", list(event_con_dict.keys())) + if event and event in event_con_dict: + df = event_con_dict[event] + print(f"Appending contrast df for event: {event}") + group_df = pd.concat([group_df, df], ignore_index=True) + else: + print(f"Event '{event}' not found for {fp}") + return group_df + + print("Selected event:", selected_event) + print("File paths A:", file_paths_a) + print("File paths B:", file_paths_b) + + contrast_df_a = concat_group_contrasts(file_paths_a, selected_event) + contrast_df_b = concat_group_contrasts(file_paths_b, selected_event) + + print("contrast_df_a empty?", contrast_df_a.empty) + print("contrast_df_b empty?", contrast_df_b.empty) + + all_raw_objs = [self.haemo_dict.get(fp) for fp in all_selected_paths if self.haemo_dict.get(fp)] + + if len(all_raw_objs) > 1: + processed_raw = aggregate_fnirs_group_geometry(all_raw_objs) + else: + processed_raw = all_raw_objs[0].copy().pick(picks="hbo") + + # Visualizations + for idx in selected_indexes: + if idx == 0: + params = param_values.get(idx, {}) + show_optodes = params.get("show_optodes", None) + t_or_theta = params.get("t_or_theta", None) + show_text = params.get("show_text", None) + brain_bounds = params.get("brain_bounds", None) + is_3d = params.get("is_3d", None) + + if show_optodes is None or t_or_theta is None or show_text is None or brain_bounds is None or is_3d is None: + print(f"Missing parameters for index {idx}, skipping.") + continue + + if not contrast_df_a.empty and not contrast_df_b.empty and processed_raw: + + plot_2d_3d_contrasts_between_groups( + contrast_df_a, + contrast_df_b, + raw_haemo=processed_raw, + group_a_name=self.group_a_dropdown.currentText(), + group_b_name=self.group_b_dropdown.currentText(), + is_3d=is_3d, + t_or_theta=t_or_theta, + show_optodes=show_optodes, + show_text=show_text, + brain_bounds=brain_bounds + ) + else: + print(f"No method defined for index {idx}") \ No newline at end of file diff --git a/src/analysis/crossgroupstats.py b/src/analysis/crossgroupstats.py new file mode 100644 index 0000000..6cff112 --- /dev/null +++ b/src/analysis/crossgroupstats.py @@ -0,0 +1,114 @@ +""" +Filename: crossgroupstats.py +Description: Cross-Group stats analysis window + +Author: Tyler de Zeeuw +License: GPL-3.0 +""" + +# External library imports +import pandas as pd + +from flares import run_cross_group_second_level_analysis +from src.shared.flaresbasewidget import CrossGroupUIMixin, FlaresBaseWidget +from src.shared.shareddata import APP_NAME + + +PARAMETERIZED_INDEXES = { + 0: [ + { + "key": "show_optodes", + "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", + "default": "all", + "type": str, + }, + { + "key": "t_or_theta", + "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", + "default": "theta", + "type": str, + }, + { + "key": "show_text", + "label": "Display informative text on the top left corner about the contrast.", + "default": "True", + "type": bool, + }, + { + "key": "brain_bounds", + "label": "Graph Upper/Lower Limit", + "default": "1.0", + "type": float, + }, + { + "key": "is_3d", + "label": "Should we display the results in a 3D interactive window?", + "default": "True", + "type": bool, + } + ], +} + + + +class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): + def __init__(self, haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict): + super().__init__("CrossGroupStats") + self.setWindowTitle(f"Cross-Group Stats Viewer - {APP_NAME.upper()}") + self.haemo_dict = haemo_dict + self.cha_dict = cha_dict + self.df_ind_dict = df_ind_dict + self.design_matrix_dict = design_matrix_dict + self.contrast_results_dict = contrast_results_dict + self.group_dict = group_dict + + self.setup_cross_group_ui(["0 (Compute Statistics)"]) + + + def process_request(self): + request = self.get_common_request_data(PARAMETERIZED_INDEXES) + if request is None: + return + + (selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = request + + if isinstance(self.df_ind_dict, dict): + # Filter out empty entries and concatenate + valid_dfs = [df for df in self.df_ind_dict.values() if isinstance(df, pd.DataFrame) and not df.empty] + if valid_dfs: + df_ind_combined = pd.concat(valid_dfs, ignore_index=True) + else: + df_ind_combined = pd.DataFrame() + else: + df_ind_combined = self.df_ind_dict + + if isinstance(self.cha_dict, dict): + valid_chas = [df for df in self.cha_dict.values() if isinstance(df, pd.DataFrame) and not df.empty] + cha_combined = pd.concat(valid_chas, ignore_index=True) if valid_chas else pd.DataFrame() + else: + cha_combined = self.cha_dict + + sample_path = file_paths_a[0] + p_haemo = self.haemo_dict.get(sample_path) + + # Visualizations + for idx in selected_indexes: + if idx == 0: + run_cross_group_second_level_analysis( + df_roi_all=df_ind_combined, # Individual stats dataframe + file_paths_a=file_paths_a, + file_paths_b=file_paths_b, + group_a_name=self.group_a_dropdown.currentText(), + group_b_name=self.group_b_dropdown.currentText(), + df_cha_all=cha_combined, + raw_haemo=p_haemo, + p_threshold=0.05, + min_subjects=3, + correction_method='fdr_bh', + target_chroma='hbo', + selected_event=selected_event, + roi_config=r"C:\Users\tyler\Desktop\research\flares\regions.json", + threshold_topo=False # Shows the raw difference map (Unthresholded) + ) + else: + print("no") \ No newline at end of file diff --git a/src/analysis/exportcsv.py b/src/analysis/exporttocsv.py similarity index 63% rename from src/analysis/exportcsv.py rename to src/analysis/exporttocsv.py index 18c2c81..e1812b6 100644 --- a/src/analysis/exportcsv.py +++ b/src/analysis/exporttocsv.py @@ -1,27 +1,28 @@ """ -Filename: exportcsv.py -Description: Export data as csv analysis window for FLARES +Filename: exporttocsv.py +Description: Logic for the Export To CSV analysis window Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in imports import os +# External library imports import numpy as np import pandas as pd -from PySide6.QtWidgets import QFileDialog, QGridLayout, QHBoxLayout, QMessageBox, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel -from PySide6.QtCore import QSize +from PySide6.QtWidgets import QFileDialog, QMessageBox -from src.shared.flaresbasewidget import FlaresBaseWidget +from src.shared.flaresbasewidget import CSVUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -class ExportDataAsCSVViewerWidget(FlaresBaseWidget): +class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget): def __init__(self, haemo_dict, cha_dict, df_ind, design_matrix, group, contrast_results_dict): - super().__init__("ExportDataAsCSVViewer") - self.setWindowTitle(f"Export Data As CSV Viewer - {APP_NAME.upper()}") + super().__init__("ExportToCSV") + self.setWindowTitle(f"Export To CSV Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict self.cha_dict = cha_dict self.df_ind = df_ind @@ -29,55 +30,11 @@ class ExportDataAsCSVViewerWidget(FlaresBaseWidget): self.group = group self.contrast_results_dict = contrast_results_dict - # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" - self.participant_dropdown_items = [] # "Participant 1 (filename)" - - for i, file_path in enumerate(self.haemo_dict.keys(), start=1): - short_label = f"Participant {i}" - display_label = f"{short_label} ({os.path.basename(file_path)})" - self.participant_map[file_path] = short_label - self.participant_dropdown_items.append(display_label) - - self.layout = QVBoxLayout(self) - self.top_bar = QHBoxLayout() - self.layout.addLayout(self.top_bar) - - self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) - self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) - - self.index_texts = [ - "0 (Export Data to CSV)", - "1 (CSV for SPARKS)", - # "2 (third image)", - # "3 (fourth image)", - ] - - self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) - self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) - - self.submit_button = QPushButton("Submit") - self.submit_button.clicked.connect(self.generate_and_save_csv) - - self.top_bar.addWidget(QLabel("Participants:")) - self.top_bar.addWidget(self.participant_dropdown) - self.top_bar.addWidget(QLabel("Export Type:")) - self.top_bar.addWidget(self.image_index_dropdown) - self.top_bar.addWidget(self.submit_button) - - self.scroll = QScrollArea() - self.scroll.setWidgetResizable(True) - self.scroll_content = QWidget() - self.grid_layout = QGridLayout(self.scroll_content) - self.scroll.setWidget(self.scroll_content) - self.layout.addWidget(self.scroll) - - self.thumb_size = QSize(280, 180) - self.showMaximized() - - - def generate_and_save_csv(self): + self.setup_csv_ui(["0 (Export Data to CSV)", "1 (CSV for SPARKS)",]) + + def process_request(self): + # TODO: Move this into flares for the call? selected_display_names = self._get_checked_items(self.participant_dropdown) selected_file_paths = [] for display_name in selected_display_names: diff --git a/src/analysis/group.py b/src/analysis/group.py deleted file mode 100644 index 0ac708c..0000000 --- a/src/analysis/group.py +++ /dev/null @@ -1,306 +0,0 @@ -""" -Filename: group.py -Description: Group analysis window for FLARES - -Author: Tyler de Zeeuw -License: GPL-3.0 -""" - -import os - -import pandas as pd - -from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel -from PySide6.QtCore import QSize - -from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog -from src.shared.shareddata import APP_NAME - - -class GroupViewerWidget(FlaresBaseWidget): - def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group): - super().__init__("GroupViewer") - self.setWindowTitle(f"Group Viewer - {APP_NAME.upper()}") - self.haemo_dict = haemo_dict - self.cha = cha - self.df_ind = df_ind - self.design_matrix = design_matrix - self.contrast_results = contrast_results - self.group = group - self.show_all_events = True - self._updating_checkstates = False - - # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" - self.participant_dropdown_items = [] # "Participant 1 (filename)" - - for i, file_path in enumerate(self.haemo_dict.keys(), start=1): - short_label = f"Participant {i}" - display_label = f"{short_label} ({os.path.basename(file_path)})" - self.participant_map[file_path] = short_label - self.participant_dropdown_items.append(display_label) - - self.layout = QVBoxLayout(self) - self.top_bar = QHBoxLayout() - self.layout.addLayout(self.top_bar) - - self.group_to_paths = {} - for file_path, group_name in self.group.items(): - self.group_to_paths.setdefault(group_name, []).append(file_path) - - self.group_names = sorted(self.group_to_paths.keys()) - - self.group_dropdown = QComboBox() - self.group_dropdown.addItem("") - self.group_dropdown.addItems(self.group_names) - self.group_dropdown.setCurrentIndex(0) - self.group_dropdown.currentIndexChanged.connect(self.update_participant_list_for_group) - - self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) - self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) - self.participant_dropdown.setEnabled(False) - - self.event_dropdown = QComboBox() - self.event_dropdown.addItem("") - - self.index_texts = [ - "0 (GLM Results)", - "1 (Significance)", - "2 (Brain Activity Visualization)", - # "3 (fourth image)", - ] - - self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) - self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) - - self.submit_button = QPushButton("Submit") - self.submit_button.clicked.connect(self.show_brain_images) - - self.top_bar.addWidget(QLabel("Group:")) - self.top_bar.addWidget(self.group_dropdown) - self.top_bar.addWidget(QLabel("Participants:")) - self.top_bar.addWidget(self.participant_dropdown) - self.top_bar.addWidget(QLabel("Event:")) - self.top_bar.addWidget(self.event_dropdown) - self.top_bar.addWidget(QLabel("Image Indexes:")) - self.top_bar.addWidget(self.image_index_dropdown) - self.top_bar.addWidget(self.submit_button) - - self.scroll = QScrollArea() - self.scroll.setWidgetResizable(True) - self.scroll_content = QWidget() - self.grid_layout = QGridLayout(self.scroll_content) - self.scroll.setWidget(self.scroll_content) - self.layout.addWidget(self.scroll) - - self.thumb_size = QSize(280, 180) - self.showMaximized() - - - - def show_brain_images(self): - import flares as flares - - selected_event = self.event_dropdown.currentText() - if selected_event == "": - selected_event = None - - selected_display_names = self._get_checked_items(self.participant_dropdown) - selected_file_paths = [] - for display_name in selected_display_names: - for fp, short_label in self.participant_map.items(): - expected_display = f"{short_label} ({os.path.basename(fp)})" - if display_name == expected_display: - selected_file_paths.append(fp) - break - - if selected_event: - valid_paths = [] - for fp in selected_file_paths: - raw = self.haemo_dict.get(fp) - # Check if this participant actually has the event in their annotations - if raw is not None and hasattr(raw, "annotations"): - if selected_event in raw.annotations.description: - valid_paths.append(fp) - - selected_file_paths = valid_paths - - selected_indexes = [ - int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) - ] - - if not selected_file_paths: - print("No participants selected.") - return - - # Only keep indexes 0 and 1 that need parameters - parameterized_indexes = { - 0: [ - { - "key": "lower_bound", - "label": "Lower bound + ", - "default": "-0.3", - "type": float, # specify int here - }, - { - "key": "upper_bound", - "label": "Upper bound + ", - "default": "0.8", - "type": float, # specify int here - } - ], - 1: [ - { - "key": "p_value", - "label": "Significance threshold P-value (e.g. 0.05)", - "default": "0.05", - "type": float, - }, - { - "key": "graph_bounds", - "label": "Graph Upper/Lower Limit", - "default": "3.0", - "type": float, - } - ], - 2: [ - { - "key": "show_optodes", - "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", - "default": "all", - "type": str, - }, - { - "key": "t_or_theta", - "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", - "default": "theta", - "type": str, - }, - { - "key": "show_text", - "label": "Display informative text on the top left corner. THIS DOES NOT WORK AND SHOULD BE LEFT AT FALSE", - "default": "False", - "type": bool, - }, - { - "key": "brain_bounds", - "label": "Graph Upper/Lower Limit", - "default": "1.0", - "type": float, - } - ], - } - - # Inject full_text from index_texts - for idx, params_list in parameterized_indexes.items(): - full_text = self.index_texts[idx] if idx < len(self.index_texts) else f"{idx} (No label found)" - for param_info in params_list: - param_info["full_text"] = full_text - - indexes_needing_params = {idx: parameterized_indexes[idx] for idx in selected_indexes if idx in parameterized_indexes} - - param_values = {} - if indexes_needing_params: - dialog = ParameterInputDialog(indexes_needing_params, parent=self) - if dialog.exec_() == QDialog.Accepted: - param_values = dialog.get_values() - if param_values is None: - return - else: - return - - - all_cha = pd.DataFrame() - for file_path in selected_file_paths: - haemo_obj = self.haemo_dict.get(file_path) - - if selected_event: - participant_events = set(haemo_obj.annotations.description) - if selected_event not in participant_events: - print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") - continue - - if haemo_obj is None: - continue - - cha_df = self.cha.get(file_path) - if cha_df is not None: - all_cha = pd.concat([all_cha, cha_df], ignore_index=True) - - # Pass the necessary arguments to each method - file_path = selected_file_paths[0] - p_haemo = self.haemo_dict.get(file_path) - p_design_matrix = self.design_matrix.get(file_path) - - df_group = pd.DataFrame() - - if selected_file_paths: - for file_path in selected_file_paths: - df = self.df_ind.get(file_path) - if df is not None: - df_group = pd.concat([df_group, df], ignore_index=True) - - - for idx in selected_indexes: - if idx == 0: - params = param_values.get(idx, {}) - lower_bound = params.get("lower_bound", None) - upper_bound = params.get("upper_bound", None) - - if lower_bound is None or upper_bound is None: - print(f"Missing parameters for index {idx}, skipping.") - continue - - - flares.plot_fir_model_results(df_group, p_haemo, p_design_matrix, selected_event, lower_bound, upper_bound) - - elif idx == 1: - params = param_values.get(idx, {}) - p_val = params.get("p_value", None) - graph_bounds = params.get("graph_bounds", None) - - if p_val is None or graph_bounds is None: - print(f"Missing parameters for index {idx}, skipping.") - continue - - all_contrasts = [] - for fp in selected_file_paths: - condition_dfs = self.contrast_results.get(fp, {}) - if selected_event in condition_dfs: - df = condition_dfs[selected_event].copy() - df["ID"] = fp - all_contrasts.append(df) - - if not all_contrasts: - print("No contrast data found for selected participants and event.") - return - - df_contrasts = pd.concat(all_contrasts, ignore_index=True) - flares.run_second_level_analysis(df_contrasts, p_haemo, p_val, graph_bounds) - - elif idx == 2: - params = param_values.get(idx, {}) - show_optodes = params.get("show_optodes", None) - t_or_theta = params.get("t_or_theta", None) - show_text = params.get("show_text", None) - brain_bounds = params.get("brain_bounds", None) - - if show_optodes is None or t_or_theta is None or show_text is None or brain_bounds is None: - print(f"Missing parameters for index {idx}, skipping.") - continue - - raw_list = [self.haemo_dict.get(fp) for fp in selected_file_paths] - - if len(selected_file_paths) > 1: - print(f"Aggregating geometry for {len(selected_file_paths)} participants...") - processed_raw = flares.aggregate_fnirs_group_geometry(raw_list) - else: - processed_raw = raw_list[0].copy().pick(picks="hbo") - - flares.brain_3d_visualization(processed_raw, all_cha, selected_event, t_or_theta=t_or_theta, show_optodes=show_optodes, show_text=show_text, brain_bounds=brain_bounds) - - elif idx == 3: - pass - - else: - print(f"No method defined for index {idx}") diff --git a/src/analysis/groupbrain.py b/src/analysis/groupbrain.py deleted file mode 100644 index 7dd9f06..0000000 --- a/src/analysis/groupbrain.py +++ /dev/null @@ -1,311 +0,0 @@ -""" -Filename: groupbrain.py -Description: Group brain analysis window for FLARES - -Author: Tyler de Zeeuw -License: GPL-3.0 -""" - -import os - -import pandas as pd - -from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel -from PySide6.QtCore import QSize - -from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog -from src.shared.shareddata import APP_NAME - - -class GroupBrainViewerWidget(FlaresBaseWidget): - def __init__(self, haemo_dict, df_ind, design_matrix, group, contrast_results_dict): - super().__init__("GroupBrainViewer") - self.setWindowTitle(f"Group Brain Viewer - {APP_NAME.upper()}") - self.haemo_dict = haemo_dict - self.df_ind = df_ind - self.design_matrix = design_matrix - self.group = group - self.contrast_results_dict = contrast_results_dict - - self.group_to_paths = {} - for file_path, group_name in self.group.items(): - self.group_to_paths.setdefault(group_name, []).append(file_path) - - self.group_names = sorted(self.group_to_paths.keys()) - - self.layout = QVBoxLayout(self) - self.top_bar = QHBoxLayout() - self.layout.addLayout(self.top_bar) - - - self.group_a_dropdown = QComboBox() - self.group_a_dropdown.addItem("") - self.group_a_dropdown.addItems(self.group_names) - self.group_a_dropdown.currentIndexChanged.connect(self._update_group_a_options) - - - self.group_b_dropdown = QComboBox() - self.group_b_dropdown.addItem("") - self.group_b_dropdown.addItems(self.group_names) - self.group_b_dropdown.currentIndexChanged.connect(self._update_group_b_options) - - - self.event_dropdown = QComboBox() - self.event_dropdown.addItem("") - - self.participant_dropdown_a = self._create_multiselect_dropdown([]) - self.participant_dropdown_a.lineEdit().setPlaceholderText("Select participants (Group A)") - self.participant_dropdown_a.model().itemChanged.connect(self._on_participants_changed) - - - self.participant_dropdown_b = self._create_multiselect_dropdown([]) - self.participant_dropdown_b.lineEdit().setPlaceholderText("Select participants (Group B)") - self.participant_dropdown_b.model().itemChanged.connect(self._on_participants_changed) - - - self.index_texts = [ - "0 (Contrast Image)", - # "1 (3D Brain Contrast)", - # "2 (third image)", - # "3 (fourth image)", - ] - self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) - self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) - - - self.submit_button = QPushButton("Submit") - self.submit_button.clicked.connect(self.show_brain_images) - - - self.top_bar.addWidget(QLabel("Group A:")) - self.top_bar.addWidget(self.group_a_dropdown) - self.top_bar.addWidget(QLabel("Participants (Group A):")) - self.top_bar.addWidget(self.participant_dropdown_a) - self.top_bar.addWidget(QLabel("Group B:")) - self.top_bar.addWidget(self.group_b_dropdown) - self.top_bar.addWidget(QLabel("Participants (Group B):")) - self.top_bar.addWidget(self.participant_dropdown_b) - self.top_bar.addWidget(QLabel("Event:")) - self.top_bar.addWidget(self.event_dropdown) - self.top_bar.addWidget(QLabel("Image Indexes:")) - self.top_bar.addWidget(self.image_index_dropdown) - self.top_bar.addWidget(self.submit_button) - - self.scroll = QScrollArea() - self.scroll.setWidgetResizable(True) - self.scroll_content = QWidget() - self.grid_layout = QGridLayout(self.scroll_content) - self.scroll.setWidget(self.scroll_content) - self.layout.addWidget(self.scroll) - - self.thumb_size = QSize(280, 180) - self.showMaximized() - - def _update_group_b_options(self): - """Triggered when Group B changes: Update Group A to exclude B's choice""" - selected_b = self.group_b_dropdown.currentText() - - # Refresh Group A and exclude what was just picked in Group B - self._refresh_group_dropdown(self.group_a_dropdown, exclude=selected_b) - - # Update the participants for Group B - self.update_participant_list_for_group(selected_b, self.participant_dropdown_b) - self._update_event_dropdown() - - def _update_group_a_options(self): - """Triggered when Group A changes: Update Group B to exclude A's choice""" - selected_a = self.group_a_dropdown.currentText() - - # Refresh Group B and exclude what was just picked in Group A - self._refresh_group_dropdown(self.group_b_dropdown, exclude=selected_a) - - # Update the participants for Group A - self.update_participant_list_for_group(selected_a, self.participant_dropdown_a) - self._update_event_dropdown() - - def _on_participants_changed(self, item=None): - self._update_event_dropdown() - - - def _refresh_group_dropdown(self, dropdown, exclude): - current = dropdown.currentText() - dropdown.blockSignals(True) - dropdown.clear() - dropdown.addItem("") - for group in self.group_names: - if group != exclude: - dropdown.addItem(group) - # Restore previous selection if still valid - if current != "" and current != exclude and dropdown.findText(current) != -1: - dropdown.setCurrentText(current) - else: - dropdown.setCurrentIndex(0) # Reset to "" - dropdown.blockSignals(False) - - - def _get_file_paths_from_labels(self, labels, group_name): - file_paths = [] - - if group_name == self.group_a_dropdown.currentText(): - participant_map = self.participant_map_a - elif group_name == self.group_b_dropdown.currentText(): - participant_map = self.participant_map_b - else: - return [] - - # Reverse map: display label -> file path - reverse_map = { - f"{label} ({os.path.basename(fp)})": fp - for fp, label in participant_map.items() - } - - for label in labels: - file_path = reverse_map.get(label) - if file_path: - file_paths.append(file_path) - - return file_paths - - def show_brain_images(self): - import flares as flares - - selected_event = self.event_dropdown.currentText() - if selected_event == "": - selected_event = None - - # Group A - participants_a = self._get_checked_items(self.participant_dropdown_a) - file_paths_a = self._get_file_paths_from_labels(participants_a, self.group_a_dropdown.currentText()) - - # Group B - participants_b = self._get_checked_items(self.participant_dropdown_b) - file_paths_b = self._get_file_paths_from_labels(participants_b, self.group_b_dropdown.currentText()) - - selected_indexes = [ - int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) - ] - - all_selected_paths = list(set(file_paths_a + file_paths_b)) - - if not all_selected_paths: - print("No participants selected.") - return - - parameterized_indexes = { - 0: [ - { - "key": "show_optodes", - "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", - "default": "all", - "type": str, - }, - { - "key": "t_or_theta", - "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", - "default": "theta", - "type": str, - }, - { - "key": "show_text", - "label": "Display informative text on the top left corner about the contrast.", - "default": "True", - "type": bool, - }, - { - "key": "brain_bounds", - "label": "Graph Upper/Lower Limit", - "default": "1.0", - "type": float, - }, - { - "key": "is_3d", - "label": "Should we display the results in a 3D interactive window?", - "default": "True", - "type": bool, - } - ], - } - - - # Inject full_text from index_texts - for idx, params_list in parameterized_indexes.items(): - full_text = self.index_texts[idx] if idx < len(self.index_texts) else f"{idx} (No label found)" - for param_info in params_list: - param_info["full_text"] = full_text - - indexes_needing_params = {idx: parameterized_indexes[idx] for idx in selected_indexes if idx in parameterized_indexes} - - param_values = {} - if indexes_needing_params: - dialog = ParameterInputDialog(indexes_needing_params, parent=self) - if dialog.exec_() == QDialog.Accepted: - param_values = dialog.get_values() - if param_values is None: - return - else: - return - - # Build group-level contrast DataFrames - def concat_group_contrasts(file_paths: list[str], event: str | None) -> pd.DataFrame: - group_df = pd.DataFrame() - for fp in file_paths: - print(f"Looking up contrast for: {fp}") - event_con_dict = self.contrast_results_dict.get(fp, {}) - print("Available events for this file:", list(event_con_dict.keys())) - if event and event in event_con_dict: - df = event_con_dict[event] - print(f"Appending contrast df for event: {event}") - group_df = pd.concat([group_df, df], ignore_index=True) - else: - print(f"Event '{event}' not found for {fp}") - return group_df - - print("Selected event:", selected_event) - print("File paths A:", file_paths_a) - print("File paths B:", file_paths_b) - - contrast_df_a = concat_group_contrasts(file_paths_a, selected_event) - contrast_df_b = concat_group_contrasts(file_paths_b, selected_event) - - print("contrast_df_a empty?", contrast_df_a.empty) - print("contrast_df_b empty?", contrast_df_b.empty) - - all_raw_objs = [self.haemo_dict.get(fp) for fp in all_selected_paths if self.haemo_dict.get(fp)] - - if len(all_raw_objs) > 1: - processed_raw = flares.aggregate_fnirs_group_geometry(all_raw_objs) - else: - processed_raw = all_raw_objs[0].copy().pick(picks="hbo") - - # Visualizations - for idx in selected_indexes: - if idx == 0: - params = param_values.get(idx, {}) - show_optodes = params.get("show_optodes", None) - t_or_theta = params.get("t_or_theta", None) - show_text = params.get("show_text", None) - brain_bounds = params.get("brain_bounds", None) - is_3d = params.get("is_3d", None) - - if show_optodes is None or t_or_theta is None or show_text is None or brain_bounds is None or is_3d is None: - print(f"Missing parameters for index {idx}, skipping.") - continue - - if not contrast_df_a.empty and not contrast_df_b.empty and processed_raw: - - flares.plot_2d_3d_contrasts_between_groups( - contrast_df_a, - contrast_df_b, - raw_haemo=processed_raw, - group_a_name=self.group_a_dropdown.currentText(), - group_b_name=self.group_b_dropdown.currentText(), - is_3d=is_3d, - t_or_theta=t_or_theta, - show_optodes=show_optodes, - show_text=show_text, - brain_bounds=brain_bounds - ) - else: - print("no") - - diff --git a/src/analysis/intergroupbrainimage.py b/src/analysis/intergroupbrainimage.py new file mode 100644 index 0000000..193a96e --- /dev/null +++ b/src/analysis/intergroupbrainimage.py @@ -0,0 +1,190 @@ +""" +Filename: intergroupbrainimage.py +Description: Logic for the Inter-Group Brain & Image analysis window + +Author: Tyler de Zeeuw +License: GPL-3.0 +""" + +# External library imports +import pandas as pd + +from flares import aggregate_fnirs_group_geometry, plot_fir_model_results, brain_3d_visualization +from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget +from src.shared.shareddata import APP_NAME + + +PARAMETERIZED_INDEXES = { + 0: [ + { + "key": "lower_bound", + "label": "Lower bound + ", + "default": "-0.3", + "type": float, # specify int here + }, + { + "key": "upper_bound", + "label": "Upper bound + ", + "default": "0.8", + "type": float, # specify int here + } + ], + 1: [ + { + "key": "p_value", + "label": "Significance threshold P-value (e.g. 0.05)", + "default": "0.05", + "type": float, + }, + { + "key": "graph_bounds", + "label": "Graph Upper/Lower Limit", + "default": "3.0", + "type": float, + } + ], + 2: [ + { + "key": "show_optodes", + "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", + "default": "all", + "type": str, + }, + { + "key": "t_or_theta", + "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", + "default": "theta", + "type": str, + }, + { + "key": "show_text", + "label": "Display informative text on the top left corner. THIS DOES NOT WORK AND SHOULD BE LEFT AT FALSE", + "default": "False", + "type": bool, + }, + { + "key": "brain_bounds", + "label": "Graph Upper/Lower Limit", + "default": "1.0", + "type": float, + } + ], +} + + + +class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): + def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group): + super().__init__("InterGroupBrainImage") + self.setWindowTitle(f"Inter-Group Brain & Image Viewer - {APP_NAME.upper()}") + self.haemo_dict = haemo_dict + self.cha = cha + self.df_ind = df_ind + self.design_matrix = design_matrix + self.contrast_results = contrast_results + self.group = group + + self.setup_inter_group_ui(["0 (GLM Results)", "1 (Significance)", "2 (Brain Activity Visualization)",]) + + + def process_request(self): + request = self.get_common_request_data(PARAMETERIZED_INDEXES) + if request is None: + return + + (selected_event, selected_file_paths, selected_indexes, param_values,) = request + + all_cha = pd.DataFrame() + for file_path in selected_file_paths: + haemo_obj = self.haemo_dict.get(file_path) + + if selected_event: + participant_events = set(haemo_obj.annotations.description) + if selected_event not in participant_events: + print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") + continue + + if haemo_obj is None: + continue + + cha_df = self.cha.get(file_path) + if cha_df is not None: + all_cha = pd.concat([all_cha, cha_df], ignore_index=True) + + # Pass the necessary arguments to each method + file_path = selected_file_paths[0] + p_haemo = self.haemo_dict.get(file_path) + p_design_matrix = self.design_matrix.get(file_path) + + df_group = pd.DataFrame() + + if selected_file_paths: + for file_path in selected_file_paths: + df = self.df_ind.get(file_path) + if df is not None: + df_group = pd.concat([df_group, df], ignore_index=True) + + + for idx in selected_indexes: + if idx == 0: + params = param_values.get(idx, {}) + lower_bound = params.get("lower_bound", None) + upper_bound = params.get("upper_bound", None) + + if lower_bound is None or upper_bound is None: + print(f"Missing parameters for index {idx}, skipping.") + continue + + + plot_fir_model_results(df_group, p_haemo, p_design_matrix, selected_event, lower_bound, upper_bound) + + elif idx == 1: + params = param_values.get(idx, {}) + p_val = params.get("p_value", None) + graph_bounds = params.get("graph_bounds", None) + + if p_val is None or graph_bounds is None: + print(f"Missing parameters for index {idx}, skipping.") + continue + + all_contrasts = [] + for fp in selected_file_paths: + condition_dfs = self.contrast_results.get(fp, {}) + if selected_event in condition_dfs: + df = condition_dfs[selected_event].copy() + df["ID"] = fp + all_contrasts.append(df) + + if not all_contrasts: + print("No contrast data found for selected participants and event.") + return + + df_contrasts = pd.concat(all_contrasts, ignore_index=True) + #flares.run_second_level_analysis(df_contrasts, p_haemo, p_val, graph_bounds) + + elif idx == 2: + params = param_values.get(idx, {}) + show_optodes = params.get("show_optodes", None) + t_or_theta = params.get("t_or_theta", None) + show_text = params.get("show_text", None) + brain_bounds = params.get("brain_bounds", None) + + if show_optodes is None or t_or_theta is None or show_text is None or brain_bounds is None: + print(f"Missing parameters for index {idx}, skipping.") + continue + + raw_list = [self.haemo_dict.get(fp) for fp in selected_file_paths] + + if len(selected_file_paths) > 1: + print(f"Aggregating geometry for {len(selected_file_paths)} participants...") + processed_raw = aggregate_fnirs_group_geometry(raw_list) + else: + processed_raw = raw_list[0].copy().pick(picks="hbo") + + brain_3d_visualization(processed_raw, all_cha, selected_event, t_or_theta=t_or_theta, show_optodes=show_optodes, show_text=show_text, brain_bounds=brain_bounds) + + elif idx == 3: + pass + + else: + print(f"No method defined for index {idx}") \ No newline at end of file diff --git a/src/analysis/groupfunctionalconnectivity.py b/src/analysis/intergroupfunctionalconnectivity.py similarity index 100% rename from src/analysis/groupfunctionalconnectivity.py rename to src/analysis/intergroupfunctionalconnectivity.py diff --git a/src/analysis/intergroupstats.py b/src/analysis/intergroupstats.py new file mode 100644 index 0000000..f7d1471 --- /dev/null +++ b/src/analysis/intergroupstats.py @@ -0,0 +1,130 @@ +""" +Filename: intergroupstats.py +Description: Logic for the Inter-Group Stats analysis window + +Author: Tyler de Zeeuw +License: GPL-3.0 +""" + +# External library imports +import pandas as pd + +from flares import run_roi_second_level_analysis +from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget +from src.shared.shareddata import APP_NAME + + +PARAMETERIZED_INDEXES = { + 0: [ + { + "key": "p_value", + "label": "Significance threshold P-value (e.g. 0.05)", + "default": "0.05", + "type": float, + }, + { + "key": "graph_bounds", + "label": "Graph Y-Limit (Optional, e.g. 1e-5)", + "default": "0.0", # Set to 0.0 to auto-scale + "type": float, + } + ], +} + + + +class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget): + def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group): + super().__init__("InterGroupStats") + self.setWindowTitle(f"Inter-Group Stats Viewer - {APP_NAME.upper()}") + self.haemo_dict = haemo_dict + self.cha = cha + self.df_ind = df_ind + self.design_matrix = design_matrix + self.contrast_results = contrast_results + self.group = group + + self.setup_inter_group_ui(["0 (Significance)",]) + + + def process_request(self): + request = self.get_common_request_data(PARAMETERIZED_INDEXES) + if request is None: + return + + (selected_event, selected_file_paths, selected_indexes, param_values,) = request + + all_cha = pd.DataFrame() + for file_path in selected_file_paths: + haemo_obj = self.haemo_dict.get(file_path) + + if selected_event: + participant_events = set(haemo_obj.annotations.description) + if selected_event not in participant_events: + print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") + continue + + if haemo_obj is None: + continue + + cha_df = self.cha.get(file_path) + if cha_df is not None: + all_cha = pd.concat([all_cha, cha_df], ignore_index=True) + + file_path = selected_file_paths[0] + p_haemo = self.haemo_dict.get(file_path) + + # Concatenate individual ROI stats (df_ind) for all chosen subjects + df_group = pd.DataFrame() + if selected_file_paths: + for file_path in selected_file_paths: + df = self.df_ind.get(file_path) + if df is not None: + df_group = pd.concat([df_group, df], ignore_index=True) + + for idx in selected_indexes: + if idx == 0: + params = param_values.get(idx, {}) + p_val = params.get("p_value", 0.05) + graph_bounds = params.get("graph_bounds", 0.0) + + if df_group.empty: + print("No ROI data (df_ind) found for selected participants.") + continue + + # Filter down to the selected experimental event/condition + if selected_event: + if 'Condition' in df_group.columns: + df_filtered = df_group[df_group['Condition'] == selected_event] + else: + print("Warning: 'Condition' column not found in ROI data.") + df_filtered = df_group + else: + df_filtered = df_group + + if df_filtered.empty: + print(f"No ROI data matches the condition '{selected_event}'.") + continue + + all_cha_filtered = pd.DataFrame() + if not all_cha.empty: + if selected_event and 'Condition' in all_cha.columns: + all_cha_filtered = all_cha[all_cha['Condition'] == selected_event] + else: + all_cha_filtered = all_cha + + # Call your new custom group ROI method! + run_roi_second_level_analysis( + df_roi_all=df_filtered, + df_cha_all=all_cha_filtered, + raw_haemo=p_haemo, + p_threshold=p_val, + min_subjects=len(selected_file_paths), + correction_method='fdr_bh', + target_chroma='hbo', + graph_bounds=graph_bounds if graph_bounds > 0.0 else None, + roi_config=r"C:\Users\tyler\Desktop\research\flares\regions.json" + ) + + else: + print(f"No method defined for index {idx}") \ No newline at end of file diff --git a/src/analysis/participantbrain.py b/src/analysis/participantbrain.py index 93c196b..4124b52 100644 --- a/src/analysis/participantbrain.py +++ b/src/analysis/participantbrain.py @@ -1,161 +1,78 @@ """ Filename: participantbrain.py -Description: Participant brain analysis window for FLARES +Description: Logic for the Participant Brain analysis window Author: Tyler de Zeeuw License: GPL-3.0 """ -import os - -from PySide6.QtWidgets import QComboBox, QDialog, QGridLayout, QHBoxLayout, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel -from PySide6.QtCore import QSize - -from src.shared.flaresbasewidget import FlaresBaseWidget, ParameterInputDialog +# External library imports +from flares import brain_3d_visualization, brain_landmarks_3d +from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -class ParticipantBrainViewerWidget(FlaresBaseWidget): +PARAMETERIZED_INDEXES = { + 0: [ + { + "key": "show_optodes", + "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", + "default": "all", + "type": str, + }, + { + "key": "show_brodmann", + "label": "Show common brodmann areas on the brain.", + "default": "True", + "type": bool, + } + ], + 1: [ + { + "key": "show_optodes", + "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", + "default": "all", + "type": str, + }, + { + "key": "t_or_theta", + "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", + "default": "theta", + "type": str, + }, + { + "key": "show_text", + "label": "Display informative text on the top left corner. THIS DOES NOT WORK AND SHOULD BE LEFT AT FALSE", + "default": "False", + "type": bool, + }, + { + "key": "brain_bounds", + "label": "Graph Upper/Lower Limit", + "default": "1.0", + "type": float, + } + ], +} + + +class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget): def __init__(self, haemo_dict, cha_dict): super().__init__("ParticipantBrainViewer") self.setWindowTitle(f"Participant Brain Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict self.cha_dict = cha_dict - # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" - self.participant_dropdown_items = [] # "Participant 1 (filename)" - - for i, file_path in enumerate(self.haemo_dict.keys(), start=1): - short_label = f"Participant {i}" - display_label = f"{short_label} ({os.path.basename(file_path)})" - self.participant_map[file_path] = short_label - self.participant_dropdown_items.append(display_label) - - self.layout = QVBoxLayout(self) - self.top_bar = QHBoxLayout() - self.layout.addLayout(self.top_bar) - - self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) - self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) - - self.event_dropdown = QComboBox() - self.event_dropdown.addItem("") - - - self.index_texts = [ - "0 (Brain Landmarks)", - "1 (Brain Activity Visualization)", - # "2 (third image)", - # "3 (fourth image)", - ] - - self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) - self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) - - self.submit_button = QPushButton("Submit") - self.submit_button.clicked.connect(self.show_brain_images) - - self.top_bar.addWidget(QLabel("Participants:")) - self.top_bar.addWidget(self.participant_dropdown) - self.top_bar.addWidget(QLabel("Event:")) - self.top_bar.addWidget(self.event_dropdown) - self.top_bar.addWidget(QLabel("Image Indexes:")) - self.top_bar.addWidget(self.image_index_dropdown) - self.top_bar.addWidget(self.submit_button) - - self.scroll = QScrollArea() - self.scroll.setWidgetResizable(True) - self.scroll_content = QWidget() - self.grid_layout = QGridLayout(self.scroll_content) - self.scroll.setWidget(self.scroll_content) - self.layout.addWidget(self.scroll) - - self.thumb_size = QSize(280, 180) - self.showMaximized() - - - def show_brain_images(self): - import flares as flares - - selected_event = self.event_dropdown.currentText() - if selected_event == "": - selected_event = None - - selected_display_names = self._get_checked_items(self.participant_dropdown) - selected_file_paths = [] - for display_name in selected_display_names: - for fp, short_label in self.participant_map.items(): - expected_display = f"{short_label} ({os.path.basename(fp)})" - if display_name == expected_display: - selected_file_paths.append(fp) - break - - selected_indexes = [ - int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) - ] + self.setup_participant_ui(["0 (Brain Landmarks)", "1 (Brain Activity Visualization)",]) - parameterized_indexes = { - 0: [ - { - "key": "show_optodes", - "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", - "default": "all", - "type": str, - }, - { - "key": "show_brodmann", - "label": "Show common brodmann areas on the brain.", - "default": "True", - "type": bool, - } - ], - 1: [ - { - "key": "show_optodes", - "label": "Determine what is rendered above the brain. Valid values are 'sensors', 'labels', 'none', 'all'.", - "default": "all", - "type": str, - }, - { - "key": "t_or_theta", - "label": "Specify if t values or theta values should be plotted. Valid values are 't', 'theta'", - "default": "theta", - "type": str, - }, - { - "key": "show_text", - "label": "Display informative text on the top left corner. THIS DOES NOT WORK AND SHOULD BE LEFT AT FALSE", - "default": "False", - "type": bool, - }, - { - "key": "brain_bounds", - "label": "Graph Upper/Lower Limit", - "default": "1.0", - "type": float, - } - ], - } + def process_request(self): - # Inject full_text from index_texts - for idx, params_list in parameterized_indexes.items(): - full_text = self.index_texts[idx] if idx < len(self.index_texts) else f"{idx} (No label found)" - for param_info in params_list: - param_info["full_text"] = full_text + request = self.get_common_request_data(PARAMETERIZED_INDEXES) + if request is None: + return - indexes_needing_params = {idx: parameterized_indexes[idx] for idx in selected_indexes if idx in parameterized_indexes} - - param_values = {} - if indexes_needing_params: - dialog = ParameterInputDialog(indexes_needing_params, parent=self) - if dialog.exec_() == QDialog.Accepted: - param_values = dialog.get_values() - if param_values is None: - return - else: - return + (selected_event, selected_file_paths, selected_indexes, param_values,) = request # Pass the necessary arguments to each method for file_path in selected_file_paths: @@ -183,7 +100,7 @@ class ParticipantBrainViewerWidget(FlaresBaseWidget): print(f"Missing parameters for index {idx}, skipping.") continue - flares.brain_landmarks_3d(haemo_obj, show_optodes, show_brodmann) + brain_landmarks_3d(haemo_obj, show_optodes, show_brodmann) elif idx == 1: params = param_values.get(idx, {}) @@ -196,7 +113,7 @@ class ParticipantBrainViewerWidget(FlaresBaseWidget): print(f"Missing parameters for index {idx}, skipping.") continue - flares.brain_3d_visualization(haemo_obj, cha, selected_event, t_or_theta=t_or_theta, show_optodes=show_optodes, show_text=show_text, brain_bounds=brain_bounds) + brain_3d_visualization(haemo_obj, cha, selected_event, t_or_theta=t_or_theta, show_optodes=show_optodes, show_text=show_text, brain_bounds=brain_bounds) else: print(f"No method defined for index {idx}") \ No newline at end of file diff --git a/src/analysis/participant.py b/src/analysis/participantimage.py similarity index 98% rename from src/analysis/participant.py rename to src/analysis/participantimage.py index c3e40c8..292bf6a 100644 --- a/src/analysis/participant.py +++ b/src/analysis/participantimage.py @@ -18,9 +18,9 @@ from src.shared.flaresbasewidget import ClickableLabel, FlaresBaseWidget from src.shared.shareddata import APP_NAME -class ParticipantViewerWidget(FlaresBaseWidget): +class ParticipantImageViewerWidget(FlaresBaseWidget): def __init__(self, haemo_dict, fig_bytes_dict): - super().__init__("ParticipantViewer") + super().__init__("ParticipantImage") self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose) self.setWindowTitle(f"Participant Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict diff --git a/src/shared/flaresbasewidget.py b/src/shared/flaresbasewidget.py index e105e9c..00c292a 100644 --- a/src/shared/flaresbasewidget.py +++ b/src/shared/flaresbasewidget.py @@ -8,9 +8,9 @@ License: GPL-3.0 import os -from PySide6.QtWidgets import QApplication, QComboBox, QDialog, QHBoxLayout, QLabel, QLineEdit, QListView, QMessageBox, QPushButton, QVBoxLayout, QWidget, QFrame, QSpinBox +from PySide6.QtWidgets import QApplication, QComboBox, QDialog, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QListView, QMessageBox, QPushButton, QScrollArea, QVBoxLayout, QWidget, QFrame, QSpinBox from PySide6.QtGui import QStandardItemModel, QStandardItem, QPixmap, QIntValidator, QDoubleValidator -from PySide6.QtCore import QEvent, Qt +from PySide6.QtCore import QEvent, QSize, Qt from src.shared.shareddata import APP_NAME @@ -836,9 +836,9 @@ class FlaresBaseWidget(QWidget): # 3. Conditional trigger for event updates # We only update events if we aren't in one of the excluded viewers excluded_viewers = { - "ParticipantViewer", + "ParticipantImage", "ParticipantFoldChannels", - "ExportDataAsCSVViewer", + "ExportToCSV", } if getattr(self, "caller", None) not in excluded_viewers: @@ -1050,4 +1050,480 @@ class FlaresBaseWidget(QWidget): model.appendRow(item) self._connect_select_all_toggle(toggle_ref, model) - self.update_participant_dropdown_label(combo=target_combo) \ No newline at end of file + self.update_participant_dropdown_label(combo=target_combo) + + +class CrossGroupUIMixin: + + def setup_cross_group_ui(self, index_texts): + + self.group_to_paths = {} + for file_path, group_name in self.group_dict.items(): + self.group_to_paths.setdefault(group_name, []).append(file_path) + + self.group_names = sorted(self.group_to_paths.keys()) + + self.main_layout = QVBoxLayout(self) + self.top_bar = QHBoxLayout() + self.main_layout.addLayout(self.top_bar) + + + self.group_a_dropdown = QComboBox() + self.group_a_dropdown.addItem("") + self.group_a_dropdown.addItems(self.group_names) + self.group_a_dropdown.currentIndexChanged.connect(self._update_group_a_options) + + + self.group_b_dropdown = QComboBox() + self.group_b_dropdown.addItem("") + self.group_b_dropdown.addItems(self.group_names) + self.group_b_dropdown.currentIndexChanged.connect(self._update_group_b_options) + + + self.event_dropdown = QComboBox() + self.event_dropdown.addItem("") + + + self.participant_dropdown_a = self._create_multiselect_dropdown([]) + line_edit = self.participant_dropdown_a.lineEdit() + assert line_edit is not None, "Dropdown A must be editable to have a lineEdit" + line_edit.setPlaceholderText("Select participants (Group A)") + model = self.participant_dropdown_a.model() + assert isinstance(model, QStandardItemModel), "Model must be QStandardItemModel" + model.itemChanged.connect(self._on_participants_changed) + + + self.participant_dropdown_b = self._create_multiselect_dropdown([]) + line_edit = self.participant_dropdown_b.lineEdit() + assert line_edit is not None, "Dropdown B must be editable to have a lineEdit" + line_edit.setPlaceholderText("Select participants (Group B)") + model = self.participant_dropdown_b.model() + assert isinstance(model, QStandardItemModel), "Model must be QStandardItemModel" + model.itemChanged.connect(self._on_participants_changed) + + + self.index_texts = index_texts + self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) + self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) + + + self.submit_button = QPushButton("Submit") + self.submit_button.clicked.connect(self.proccess_request) + + + self.top_bar.addWidget(QLabel("Group A:")) + self.top_bar.addWidget(self.group_a_dropdown) + self.top_bar.addWidget(QLabel("Participants (Group A):")) + self.top_bar.addWidget(self.participant_dropdown_a) + self.top_bar.addWidget(QLabel("Group B:")) + self.top_bar.addWidget(self.group_b_dropdown) + self.top_bar.addWidget(QLabel("Participants (Group B):")) + self.top_bar.addWidget(self.participant_dropdown_b) + self.top_bar.addWidget(QLabel("Event:")) + self.top_bar.addWidget(self.event_dropdown) + self.top_bar.addWidget(QLabel("Image Indexes:")) + self.top_bar.addWidget(self.image_index_dropdown) + self.top_bar.addWidget(self.submit_button) + + self.scroll_area = QScrollArea() + self.scroll_area.setWidgetResizable(True) + self.scroll_content = QWidget() + self.grid_layout = QGridLayout(self.scroll_content) + self.scroll_area.setWidget(self.scroll_content) + self.main_layout.addWidget(self.scroll_area) + + self.thumb_size = QSize(280, 180) + self.showMaximized() + + def _update_group_b_options(self): + """Triggered when Group B changes: Update Group A to exclude B's choice""" + selected_b = self.group_b_dropdown.currentText() + + # Refresh Group A and exclude what was just picked in Group B + self._refresh_group_dropdown(self.group_a_dropdown, exclude=selected_b) + + # Update the participants for Group B + self.update_participant_list_for_group(selected_b, self.participant_dropdown_b) + self._update_event_dropdown() + + def _update_group_a_options(self): + """Triggered when Group A changes: Update Group B to exclude A's choice""" + selected_a = self.group_a_dropdown.currentText() + + # Refresh Group B and exclude what was just picked in Group A + self._refresh_group_dropdown(self.group_b_dropdown, exclude=selected_a) + + # Update the participants for Group A + self.update_participant_list_for_group(selected_a, self.participant_dropdown_a) + self._update_event_dropdown() + + def _on_participants_changed(self, item=None): + self._update_event_dropdown() + + + def _refresh_group_dropdown(self, dropdown, exclude): + current = dropdown.currentText() + dropdown.blockSignals(True) + dropdown.clear() + dropdown.addItem("") + for group in self.group_names: + if group != exclude: + dropdown.addItem(group) + # Restore previous selection if still valid + if current != "" and current != exclude and dropdown.findText(current) != -1: + dropdown.setCurrentText(current) + else: + dropdown.setCurrentIndex(0) # Reset to "" + dropdown.blockSignals(False) + + + + + def _get_file_paths_from_labels(self, labels, group_name): + file_paths = [] + + if group_name == self.group_a_dropdown.currentText(): + participant_map = self.participant_map_a + elif group_name == self.group_b_dropdown.currentText(): + participant_map = self.participant_map_b + else: + return [] + + # Reverse map: display label -> file path + reverse_map = { + f"{label} ({os.path.basename(fp)})": fp + for fp, label in participant_map.items() + } + + for label in labels: + file_path = reverse_map.get(label) + if file_path: + file_paths.append(file_path) + + return file_paths + + def get_common_request_data(self, parameterized_indexes): + selected_event = self.event_dropdown.currentText() + if selected_event == "": + selected_event = None + + participants_a = self._get_checked_items(self.participant_dropdown_a) + file_paths_a = self._get_file_paths_from_labels( + participants_a, self.group_a_dropdown.currentText() + ) + + participants_b = self._get_checked_items(self.participant_dropdown_b) + file_paths_b = self._get_file_paths_from_labels( + participants_b, self.group_b_dropdown.currentText() + ) + + selected_indexes = [ + int(s.split(" ")[0]) + for s in self._get_checked_items(self.image_index_dropdown) + ] + + all_selected_paths = list(set(file_paths_a + file_paths_b)) + + if not all_selected_paths: + print("No participants selected.") + return None + + # Inject full_text + for idx, params_list in parameterized_indexes.items(): + full_text = self.index_texts[idx] + for param in params_list: + param["full_text"] = full_text + + indexes_needing_params = { + idx: parameterized_indexes[idx] + for idx in selected_indexes + if idx in parameterized_indexes + } + + param_values = {} + if indexes_needing_params: + dialog = ParameterInputDialog(indexes_needing_params, parent=self) + if dialog.exec() != QDialog.DialogCode.Accepted: + return None + + param_values = dialog.get_values() + if param_values is None: + return None + + return ( + selected_event, + file_paths_a, + file_paths_b, + all_selected_paths, + selected_indexes, + param_values, + ) + + +class CSVUIMixin: + + def setup_csv_ui(self, index_texts): + + # Create mappings: file_path -> participant label and dropdown display text + self.participant_map = {} # file_path -> "Participant 1" + self.participant_dropdown_items = [] # "Participant 1 (filename)" + + for i, file_path in enumerate(self.haemo_dict.keys(), start=1): + short_label = f"Participant {i}" + display_label = f"{short_label} ({os.path.basename(file_path)})" + self.participant_map[file_path] = short_label + self.participant_dropdown_items.append(display_label) + + self.layout = QVBoxLayout(self) + self.top_bar = QHBoxLayout() + self.layout.addLayout(self.top_bar) + + self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) + self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) + + self.index_texts = index_texts + + self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) + self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) + + self.submit_button = QPushButton("Submit") + self.submit_button.clicked.connect(self.process_request) + + self.top_bar.addWidget(QLabel("Participants:")) + self.top_bar.addWidget(self.participant_dropdown) + self.top_bar.addWidget(QLabel("Export Type:")) + self.top_bar.addWidget(self.image_index_dropdown) + self.top_bar.addWidget(self.submit_button) + + self.scroll = QScrollArea() + self.scroll.setWidgetResizable(True) + self.scroll_content = QWidget() + self.grid_layout = QGridLayout(self.scroll_content) + self.scroll.setWidget(self.scroll_content) + self.layout.addWidget(self.scroll) + + self.thumb_size = QSize(280, 180) + self.showMaximized() + + +class InterGroupUIMixin: + def setup_inter_group_ui(self, index_texts): + self.show_all_events = True + self._updating_checkstates = False + + # Create mappings: file_path -> participant label and dropdown display text + self.participant_map = {} # file_path -> "Participant 1" + self.participant_dropdown_items = [] # "Participant 1 (filename)" + + for i, file_path in enumerate(self.haemo_dict.keys(), start=1): + short_label = f"Participant {i}" + display_label = f"{short_label} ({os.path.basename(file_path)})" + self.participant_map[file_path] = short_label + self.participant_dropdown_items.append(display_label) + + self.layout = QVBoxLayout(self) + self.top_bar = QHBoxLayout() + self.layout.addLayout(self.top_bar) + + self.group_to_paths = {} + for file_path, group_name in self.group.items(): + self.group_to_paths.setdefault(group_name, []).append(file_path) + + self.group_names = sorted(self.group_to_paths.keys()) + + self.group_dropdown = QComboBox() + self.group_dropdown.addItem("") + self.group_dropdown.addItems(self.group_names) + self.group_dropdown.setCurrentIndex(0) + self.group_dropdown.currentIndexChanged.connect(self.update_participant_list_for_group) + + self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) + self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) + self.participant_dropdown.setEnabled(False) + + self.event_dropdown = QComboBox() + self.event_dropdown.addItem("") + + self.index_texts = index_texts + self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) + self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) + + self.submit_button = QPushButton("Submit") + self.submit_button.clicked.connect(self.process_request) + + self.top_bar.addWidget(QLabel("Group:")) + self.top_bar.addWidget(self.group_dropdown) + self.top_bar.addWidget(QLabel("Participants:")) + self.top_bar.addWidget(self.participant_dropdown) + self.top_bar.addWidget(QLabel("Event:")) + self.top_bar.addWidget(self.event_dropdown) + self.top_bar.addWidget(QLabel("Image Indexes:")) + self.top_bar.addWidget(self.image_index_dropdown) + self.top_bar.addWidget(self.submit_button) + + self.scroll = QScrollArea() + self.scroll.setWidgetResizable(True) + self.scroll_content = QWidget() + self.grid_layout = QGridLayout(self.scroll_content) + self.scroll.setWidget(self.scroll_content) + self.layout.addWidget(self.scroll) + + self.thumb_size = QSize(280, 180) + self.showMaximized() + + def get_common_request_data(self, parameterized_indexes): + selected_event = self.event_dropdown.currentText() + if selected_event == "": + selected_event = None + + selected_display_names = self._get_checked_items(self.participant_dropdown) + selected_file_paths = [] + for display_name in selected_display_names: + for fp, short_label in self.participant_map.items(): + expected_display = f"{short_label} ({os.path.basename(fp)})" + if display_name == expected_display: + selected_file_paths.append(fp) + break + + if selected_event: + valid_paths = [] + for fp in selected_file_paths: + raw = self.haemo_dict.get(fp) + # Check if this participant actually has the event in their annotations + if raw is not None and hasattr(raw, "annotations"): + if selected_event in raw.annotations.description: + valid_paths.append(fp) + + selected_file_paths = valid_paths + + selected_indexes = [ + int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) + ] + + if not selected_file_paths: + print("No participants selected.") + return + + # Only keep indexes 0 and 1 that need parameters + + + # Inject full_text from index_texts + for idx, params_list in parameterized_indexes.items(): + full_text = self.index_texts[idx] if idx < len(self.index_texts) else f"{idx} (No label found)" + for param_info in params_list: + param_info["full_text"] = full_text + + indexes_needing_params = {idx: parameterized_indexes[idx] for idx in selected_indexes if idx in parameterized_indexes} + + param_values = {} + if indexes_needing_params: + dialog = ParameterInputDialog(indexes_needing_params, parent=self) + if dialog.exec_() == QDialog.Accepted: + param_values = dialog.get_values() + if param_values is None: + return + else: + return + + return ( + selected_event, + selected_file_paths, + selected_indexes, + param_values, + ) + +class ParticipantUIMixin: + def setup_participant_ui(self, index_texts): + # Create mappings: file_path -> participant label and dropdown display text + self.participant_map = {} # file_path -> "Participant 1" + self.participant_dropdown_items = [] # "Participant 1 (filename)" + + for i, file_path in enumerate(self.haemo_dict.keys(), start=1): + short_label = f"Participant {i}" + display_label = f"{short_label} ({os.path.basename(file_path)})" + self.participant_map[file_path] = short_label + self.participant_dropdown_items.append(display_label) + + self.layout = QVBoxLayout(self) + self.top_bar = QHBoxLayout() + self.layout.addLayout(self.top_bar) + + self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) + self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) + + self.event_dropdown = QComboBox() + self.event_dropdown.addItem("") + + + self.index_texts = index_texts + + self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) + self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) + + self.submit_button = QPushButton("Submit") + self.submit_button.clicked.connect(self.process_request) + + self.top_bar.addWidget(QLabel("Participants:")) + self.top_bar.addWidget(self.participant_dropdown) + self.top_bar.addWidget(QLabel("Event:")) + self.top_bar.addWidget(self.event_dropdown) + self.top_bar.addWidget(QLabel("Image Indexes:")) + self.top_bar.addWidget(self.image_index_dropdown) + self.top_bar.addWidget(self.submit_button) + + self.scroll = QScrollArea() + self.scroll.setWidgetResizable(True) + self.scroll_content = QWidget() + self.grid_layout = QGridLayout(self.scroll_content) + self.scroll.setWidget(self.scroll_content) + self.layout.addWidget(self.scroll) + + self.thumb_size = QSize(280, 180) + self.showMaximized() + + + def get_common_request_data(self, parameterized_indexes): + selected_event = self.event_dropdown.currentText() + if selected_event == "": + selected_event = None + + selected_display_names = self._get_checked_items(self.participant_dropdown) + selected_file_paths = [] + for display_name in selected_display_names: + for fp, short_label in self.participant_map.items(): + expected_display = f"{short_label} ({os.path.basename(fp)})" + if display_name == expected_display: + selected_file_paths.append(fp) + break + + selected_indexes = [ + int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) + ] + + + + + # Inject full_text from index_texts + for idx, params_list in parameterized_indexes.items(): + full_text = self.index_texts[idx] if idx < len(self.index_texts) else f"{idx} (No label found)" + for param_info in params_list: + param_info["full_text"] = full_text + + indexes_needing_params = {idx: parameterized_indexes[idx] for idx in selected_indexes if idx in parameterized_indexes} + + param_values = {} + if indexes_needing_params: + dialog = ParameterInputDialog(indexes_needing_params, parent=self) + if dialog.exec_() == QDialog.Accepted: + param_values = dialog.get_values() + if param_values is None: + return + else: + return + + return ( + selected_event, + selected_file_paths, + selected_indexes, + param_values, + ) \ No newline at end of file diff --git a/src/window/viewerlauncher.py b/src/window/viewerlauncher.py index de25ef0..b7a3c5c 100644 --- a/src/window/viewerlauncher.py +++ b/src/window/viewerlauncher.py @@ -1,19 +1,22 @@ """ Filename: viewerlauncher.py -Description: Analysis options launcher for FLARES +Description: Viewer launcher window Author: Tyler de Zeeuw License: GPL-3.0 """ +# External library imports from PySide6.QtWidgets import QPushButton, QWidget, QVBoxLayout from PySide6.QtCore import QTimer -from src.analysis.exportcsv import ExportDataAsCSVViewerWidget -from src.analysis.group import GroupViewerWidget -from src.analysis.groupbrain import GroupBrainViewerWidget -from src.analysis.groupfunctionalconnectivity import GroupFunctionalConnectivityWidget -from src.analysis.participant import ParticipantViewerWidget +from src.analysis.exporttocsv import ExportToCSVWidget +from src.analysis.intergroupbrainimage import InterGroupBrainImageWidget +from src.analysis.crossgroupbrainimage import CrossGroupBrainImageWidget +from src.analysis.intergroupfunctionalconnectivity import InterGroupFunctionalConnectivityWidget +from src.analysis.intergroupstats import InterGroupStatsWidget +from src.analysis.crossgroupstats import CrossGroupStatsWidget +from src.analysis.participantimage import ParticipantImageViewerWidget from src.analysis.participantbrain import ParticipantBrainViewerWidget from src.analysis.participantfoldchannels import ParticipantFoldChannelsWidget from src.analysis.participantfunctionalconnectivity import ParticipantFunctionalConnectivityWidget @@ -21,92 +24,42 @@ from src.shared.shareddata import APP_NAME class ViewerLauncherWidget(QWidget): - def __init__(self, haemo_dict, config_dict, fig_bytes_dict, cha_dict, contrast_results_dict, df_ind, design_matrix, epochs_dict, folding_bypass): + def __init__(self, haemo_dict, epochs_dict, cha_dict, df_ind_dict, design_matrix_dict, config_dict, fig_bytes_dict, contrast_results_dict, folding_bypass): super().__init__() self.setWindowTitle(f"Viewer Launcher - {APP_NAME.upper()}") - group_dict = { - file_path: config.get("GROUP", "Unknown") - for file_path, config in config_dict.items() - } - - def launch(func, btn, *args): - func(*args) - self._trigger_success(btn) + group_dict = {f: c.get("GROUP", "Unknown") for f, c in config_dict.items()} + + btn_data = [ + ("Participant Image Viewer", ParticipantImageViewerWidget, [haemo_dict, fig_bytes_dict], True), + ("Participant Brain Viewer", ParticipantBrainViewerWidget, [haemo_dict, cha_dict], True), + ("Participant Fold Channels Viewer", ParticipantFoldChannelsWidget, [haemo_dict, cha_dict], False), + ("Participant Functional Connectivity Viewer [BETA]", ParticipantFunctionalConnectivityWidget, [haemo_dict, epochs_dict], True), + ("Inter-Group Functional Connectivity Viewer [BETA]", InterGroupFunctionalConnectivityWidget, [haemo_dict, group_dict, config_dict], True), + ("Inter-Group Stats Viewer", InterGroupStatsWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), + ("Cross-Group Stats Viewer", CrossGroupStatsWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), + ("Inter-Group Brain & Image Viewer", InterGroupBrainImageWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), + ("Cross-Group Brain & Image Viewer", CrossGroupBrainImageWidget, [haemo_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), + ("Export To CSV Viewer", ExportToCSVWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, group_dict, contrast_results_dict], True) + ] layout = QVBoxLayout(self) + for label, widget_class, args, requires_bypass in btn_data: + btn = QPushButton(f"Open {label}") + # Connect directly to the generic opener + btn.clicked.connect(lambda _, c=widget_class, b=btn, a=args: self._open_viewer(c, b, *a)) + btn.setEnabled(not (requires_bypass and folding_bypass)) + layout.addWidget(btn) - btn1 = QPushButton("Open Participant Viewer") - btn1.clicked.connect(lambda: launch(self.open_participant_viewer, btn1, haemo_dict, fig_bytes_dict)) - btn1.setEnabled(not folding_bypass) + def _open_viewer(self, widget_class, btn, *args): + # Instantiate and show dynamically + self.active_viewer = widget_class(*args) + self.active_viewer.show() + self._trigger_success(btn) - btn2 = QPushButton("Open Participant Brain Viewer") - btn2.clicked.connect(lambda: launch(self.open_participant_brain_viewer, btn2, haemo_dict, cha_dict)) - btn2.setEnabled(not folding_bypass) - - btn3 = QPushButton("Open Participant Fold Channels Viewer") - btn3.clicked.connect(lambda: launch(self.open_participant_fold_channels_viewer, btn3, haemo_dict, cha_dict)) - - btn7 = QPushButton("Open Functional Connectivity Viewer [BETA]") - btn7.clicked.connect(lambda: launch(self.open_participant_functional_connectivity_viewer, btn7, haemo_dict, epochs_dict)) - btn7.setEnabled(not folding_bypass) - - btn8 = QPushButton("Open Group Functional Connectivity Viewer [BETA]") - btn8.clicked.connect(lambda: launch(self.open_group_functional_connectivity_viewer, btn8, haemo_dict, group_dict, config_dict)) - btn8.setEnabled(not folding_bypass) - - btn4 = QPushButton("Open Inter-Group Viewer") - btn4.clicked.connect(lambda: launch(self.open_group_viewer, btn4, haemo_dict, cha_dict, df_ind, design_matrix, contrast_results_dict, group_dict)) - btn4.setEnabled(not folding_bypass) - - btn5 = QPushButton("Open Cross Group Brain Viewer") - btn5.clicked.connect(lambda: launch(self.open_group_brain_viewer, btn5, haemo_dict, df_ind, design_matrix, group_dict, contrast_results_dict)) - btn5.setEnabled(not folding_bypass) - - btn6 = QPushButton("Open Export Data As CSV Viewer") - btn6.clicked.connect(lambda: launch(self.open_export_data_as_csv_viewer, btn6, haemo_dict, cha_dict, df_ind, design_matrix, group_dict, contrast_results_dict)) - btn6.setEnabled(not folding_bypass) - - layout.addWidget(btn1) - layout.addWidget(btn2) - layout.addWidget(btn3) - layout.addWidget(btn7) - layout.addWidget(btn8) - layout.addWidget(btn4) - layout.addWidget(btn5) - layout.addWidget(btn6) - - def open_participant_viewer(self, haemo_dict, fig_bytes_dict): - self.participant_viewer = ParticipantViewerWidget(haemo_dict, fig_bytes_dict) - self.participant_viewer.show() - - def open_participant_brain_viewer(self, haemo_dict, cha_dict): - self.participant_brain_viewer = ParticipantBrainViewerWidget(haemo_dict, cha_dict) - self.participant_brain_viewer.show() - - def open_participant_fold_channels_viewer(self, haemo_dict, cha_dict): - self.participant_fold_channels_viewer = ParticipantFoldChannelsWidget(haemo_dict, cha_dict) - self.participant_fold_channels_viewer.show() - - def open_participant_functional_connectivity_viewer(self, haemo_dict, epochs_dict): - self.participant_brain_viewer = ParticipantFunctionalConnectivityWidget(haemo_dict, epochs_dict) - self.participant_brain_viewer.show() - - def open_group_functional_connectivity_viewer(self, haemo_dict, group, config_dict): - self.participant_brain_viewer = GroupFunctionalConnectivityWidget(haemo_dict, group, config_dict) - self.participant_brain_viewer.show() - - def open_group_viewer(self, haemo_dict, cha_dict, df_ind, design_matrix, contrast_results_dict, group): - self.participant_brain_viewer = GroupViewerWidget(haemo_dict, cha_dict, df_ind, design_matrix, contrast_results_dict, group) - self.participant_brain_viewer.show() - - def open_group_brain_viewer(self, haemo_dict, df_ind, design_matrix, group, contrast_results_dict): - self.participant_brain_viewer = GroupBrainViewerWidget(haemo_dict, df_ind, design_matrix, group, contrast_results_dict) - self.participant_brain_viewer.show() - - def open_export_data_as_csv_viewer(self, haemo_dict, cha_dict, df_ind, design_matrix, group, contrast_results_dict): - self.export_data_as_csv_viewer = ExportDataAsCSVViewerWidget(haemo_dict, cha_dict, df_ind, design_matrix, group, contrast_results_dict) - self.export_data_as_csv_viewer.show() + def _launch(self, func, btn, *args): + func(*args) + self._trigger_success(btn) def _trigger_success(self, button): """Temporarily adds a green checkmark to the button text."""