220 lines
7.9 KiB
Python
220 lines
7.9 KiB
Python
"""
|
|
Filename: intergroupbrainimage.py
|
|
Description: Logic for the Inter-Group Brain & Image analysis window
|
|
Note: Compliant with pylance strict type checking
|
|
|
|
Author: Tyler de Zeeuw
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
# Built-in Imports
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
|
|
# External library imports
|
|
import pandas as pd
|
|
from pandas import DataFrame
|
|
|
|
from mne import Annotations
|
|
from mne.io.base import BaseRaw
|
|
|
|
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: dict[int, list[dict[str, Any]]] = {
|
|
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: dict[str | Path, BaseRaw],
|
|
cha_dict: dict[str, DataFrame],
|
|
df_ind_dict: dict[str, DataFrame],
|
|
design_matrix_dict: dict[str, DataFrame],
|
|
contrast_results_dict: dict[str, dict[str, Any]],
|
|
group_dict: dict[str, str]
|
|
) -> None:
|
|
|
|
super().__init__("InterGroupBrainImage")
|
|
self.setWindowTitle(f"Inter-Group Brain & Image 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_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, raw_params) = request
|
|
|
|
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
|
|
|
|
all_cha = pd.DataFrame()
|
|
for file_path in selected_file_paths:
|
|
haemo_obj = self.haemo_dict.get(file_path)
|
|
|
|
if haemo_obj is None:
|
|
continue
|
|
|
|
if selected_event:
|
|
raw_annotations = getattr(haemo_obj, "annotations", None)
|
|
|
|
if raw_annotations is not None:
|
|
annotations = cast(Annotations, raw_annotations)
|
|
descriptions = cast(list[str], list(annotations.description))
|
|
participant_events: set[str] = set(descriptions)
|
|
else:
|
|
participant_events: set[str] = set()
|
|
|
|
if selected_event not in participant_events:
|
|
print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.")
|
|
continue
|
|
|
|
cha_df = self.cha_dict.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_dict.get(file_path)
|
|
|
|
df_group = pd.DataFrame()
|
|
|
|
if selected_file_paths:
|
|
for file_path in selected_file_paths:
|
|
df = self.df_ind_dict.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: list[DataFrame] = []
|
|
for fp in selected_file_paths:
|
|
condition_dfs = self.contrast_results_dict.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
|
|
|
|
# TODO: look at intergroupstats and figure out what to do
|
|
_ = 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
|
|
|
|
all_raw_objs = [self.haemo_dict.get(fp) for fp in selected_file_paths if self.haemo_dict.get(fp)]
|
|
|
|
if len(all_raw_objs) > 1:
|
|
processed_raw = aggregate_fnirs_group_geometry(all_raw_objs)
|
|
elif len(all_raw_objs) == 1 and all_raw_objs[0] is not None:
|
|
processed_raw = all_raw_objs[0].copy()
|
|
processed_raw.pick(picks="hbo") # type: ignore
|
|
else:
|
|
processed_raw = None
|
|
|
|
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}") |