small rewrite to impose dry principles
This commit is contained in:
@@ -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}")
|
||||
@@ -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")
|
||||
@@ -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:
|
||||
@@ -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("<None Selected>")
|
||||
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("<None Selected>")
|
||||
|
||||
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 == "<None Selected>":
|
||||
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 + <description>",
|
||||
"default": "-0.3",
|
||||
"type": float, # specify int here
|
||||
},
|
||||
{
|
||||
"key": "upper_bound",
|
||||
"label": "Upper bound + <description>",
|
||||
"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}")
|
||||
@@ -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("<None Selected>")
|
||||
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("<None Selected>")
|
||||
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("<None Selected>")
|
||||
|
||||
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("<None Selected>")
|
||||
for group in self.group_names:
|
||||
if group != exclude:
|
||||
dropdown.addItem(group)
|
||||
# Restore previous selection if still valid
|
||||
if current != "<None Selected>" and current != exclude and dropdown.findText(current) != -1:
|
||||
dropdown.setCurrentText(current)
|
||||
else:
|
||||
dropdown.setCurrentIndex(0) # Reset to "<None Selected>"
|
||||
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 == "<None Selected>":
|
||||
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")
|
||||
|
||||
|
||||
@@ -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 + <description>",
|
||||
"default": "-0.3",
|
||||
"type": float, # specify int here
|
||||
},
|
||||
{
|
||||
"key": "upper_bound",
|
||||
"label": "Upper bound + <description>",
|
||||
"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}")
|
||||
@@ -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}")
|
||||
@@ -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("<None Selected>")
|
||||
|
||||
|
||||
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 == "<None Selected>":
|
||||
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}")
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user