pylance things
This commit is contained in:
@@ -12,47 +12,18 @@ from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
# External library imports
|
||||
from mne.io.base import BaseRaw
|
||||
|
||||
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 flares import aggregate_fnirs_group_geometry, plot_2d_3d_contrasts_between_groups
|
||||
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'.",
|
||||
@@ -67,8 +38,8 @@ PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
|
||||
},
|
||||
{
|
||||
"key": "show_text",
|
||||
"label": "Display informative text on the top left corner. THIS DOES NOT WORK AND SHOULD BE LEFT AT FALSE",
|
||||
"default": "False",
|
||||
"label": "Display informative text on the top left corner about the contrast.",
|
||||
"default": "True",
|
||||
"type": bool,
|
||||
},
|
||||
{
|
||||
@@ -76,145 +47,111 @@ PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
|
||||
"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 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]
|
||||
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)",])
|
||||
|
||||
self.setup_inter_group_ui(["0 (Contrast Image)"])
|
||||
|
||||
|
||||
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
|
||||
(selected_event, file_paths_a, file_paths_b, all_selected_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)
|
||||
# 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:
|
||||
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)
|
||||
print(f"Event '{event}' not found for {fp}")
|
||||
return group_df
|
||||
|
||||
# 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)
|
||||
print("Selected event:", selected_event)
|
||||
print("File paths A:", file_paths_a)
|
||||
print("File paths B:", file_paths_b)
|
||||
|
||||
df_group = pd.DataFrame()
|
||||
contrast_df_a = concat_group_contrasts(file_paths_a, selected_event)
|
||||
contrast_df_b = concat_group_contrasts(file_paths_b, selected_event)
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
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
|
||||
|
||||
# Visualizations
|
||||
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)
|
||||
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:
|
||||
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
|
||||
|
||||
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
|
||||
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}")
|
||||
Reference in New Issue
Block a user