tyler/flares
2
0
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases 22 Wiki Activity

pylance standardization

Browse Source
This commit is contained in:
tyler
2026-07-21 00:54:51 -07:00
parent 2b019c1bc0
commit 8b017005c5
20 changed files with 608 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelog.md
+1
View File
@@ -31,6 +31,7 @@
- Fixed a crucial bug where short channels were not being processed and filtered the same way as long channels before being used as regressors
- Fixed a crucial bug where short channels were being presented to the design matrix as normal long channels
- Fixed a crucial bug where long channels could be interpolated from short channels. Short channels are still potentially interpolated from long channels. See [this link](https://git.research.dezeeuw.ca/tyler/flares/issues/80) for more information regarding this issue.
- Decreased unnecessary processing time when fOLDing channels by an order of magnitude
- Added a welcome message when the terminal is opened, resized the terminal, and added more commands
flares.py
+186 -87
View File
@@ -8,12 +8,13 @@ License: GPL-3.0
# Built-in imports
import os
from pathlib import Path
import sys
import platform
import threading
import logging
from io import BytesIO
from typing import Any, Optional, cast, Literal, Union
from typing import Any, Optional, Sequence, cast, Literal, Union
from itertools import compress
from copy import deepcopy
from multiprocessing import Queue, Pool
@@ -1765,29 +1766,6 @@ def _check_load_fold(fold_files, atlas):
def fold_channel_specificity_normal(raw, fold_files=None, atlas="Juelich", interpolate=False):
"""Return the landmarks and specificity a channel is sensitive to.
Parameters
""" # noqa: E501
_validate_type(raw, BaseRaw, "raw")
reference_locations = generate_montage_locations()
fold_tbl = _check_load_fold(fold_files, atlas)
chan_spec = list()
for cidx in range(len(raw.ch_names)):
tbl = _source_detector_fold_table(
raw, cidx, reference_locations, fold_tbl, interpolate
)
chan_spec.append(tbl.reset_index(drop=True))
return chan_spec
def resource_path(relative_path):
"""
Get absolute path to resource regardless of running directly or packaged using PyInstaller
@@ -1958,35 +1936,42 @@ def resource_path(relative_path):
def fold_channels(raw: BaseRaw, p_name: str, progress_queue=None) -> dict[str, list[dict[str, Any]]]:
def fold_channels(raw: BaseRaw, p_name: str, atlas: str='Brodmann', progress_queue=None) -> dict[str, list[dict[str, Any]]]:
"""Runs in background process.
Does only heavy math/lookup. Returns data instead of a static image.
"""
if getattr(sys, 'frozen', False):
set_config('MNE_NIRS_FOLD_PATH', resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary"))
fold_dir = resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary")
else:
path = os.path.expanduser("~") + "/mne_data/fOLD/fOLD-public-master/Supplementary"
set_config('MNE_NIRS_FOLD_PATH', resource_path(path))
fold_dir = resource_path(path)
set_config('MNE_NIRS_FOLD_PATH', fold_dir)
hbo_channel_names = cast(list[str], getattr(raw.copy().pick(picks='hbo'), "ch_names"))
# Store clean, picklable data lists instead of complex DataFrames
channel_results = {}
_validate_type(raw, BaseRaw, "raw")
reference_locations = generate_montage_locations()
fold_tbl = _check_load_fold(fold_files=fold_dir, atlas=atlas)
channel_results = {}
step_idx = 0
for channel_name in hbo_channel_names:
channel_data = raw.copy().pick(picks=channel_name)
output = cast(list[DataFrame], fold_channel_specificity_normal(channel_data, interpolate=True, atlas='Brodmann'))
for cidx, channel_name in enumerate(hbo_channel_names):
tbl = _source_detector_fold_table(
raw, cidx, reference_locations, fold_tbl, interpolate=True
)
channel_results[channel_name] = []
for df_data in output:
# Extract just raw primitive types so they transfer over process channels flawlessly
for _, row in df_data.iterrows():
channel_results[channel_name].append({
'Landmark': str(row['Landmark']),
'Specificity': float(row['Specificity'])
})
for _, row in tbl.iterrows():
channel_results[channel_name].append({
'Landmark': str(row['Landmark']),
'Specificity': float(row['Specificity'])
})
step_idx += 1
if progress_queue is not None:
progress_queue.put((p_name, step_idx))
@@ -2180,7 +2165,7 @@ def plot_3d_evoked_array(
return brain
def aggregate_fnirs_group_geometry(raw_list):
def aggregate_fnirs_group_geometry(raw_list: Sequence[BaseRaw | None]) -> BaseRaw:
"""
Averages fNIRS geometry across participants in two tiers:
1. Average by Channel Pairing (S_D).
@@ -2246,7 +2231,15 @@ def aggregate_fnirs_group_geometry(raw_list):
def brain_3d_visualization(raw_haemo, df_cha, selected_event, t_or_theta: Literal['t', 'theta'] = 'theta', show_optodes: Literal['sensors', 'labels', 'none', 'all'] = 'all', show_text: bool = True, brain_bounds: float = 1.0) -> None:
def brain_3d_visualization(
raw_haemo: BaseRaw | None,
df_cha: DataFrame | None,
selected_event: str | None,
t_or_theta: Literal["t", "theta"] = "theta",
show_optodes: Literal["sensors", "labels", "none", "all"] = "all",
show_text: bool = True,
brain_bounds: float | tuple[float, float] | Sequence[float] = 1.0,
) -> None:
clim = dict(kind="value", pos_lims=(0, brain_bounds/2, brain_bounds))
@@ -2582,7 +2575,14 @@ def plot_2d_3d_contrasts_between_groups(
def plot_fir_model_results(df, raw_haemo, dm, selected_event, l_bound, u_bound):
def plot_fir_model_results(
df: DataFrame,
raw_haemo: BaseRaw | None,
dm: DataFrame | None,
selected_event: str | None,
l_bound: float,
u_bound: float,
) -> None:
df["isActivity"] = [f"{selected_event}" in n for n in df["Condition"]]
@@ -2801,11 +2801,19 @@ def load_snirf(file_path: str) -> tuple[BaseRaw, Figure]:
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
def run_roi_second_level_analysis(
df_roi_all: DataFrame,
df_cha_all: DataFrame | None = None,
raw_haemo: BaseRaw | None = None,
p_threshold: float = 0.05,
min_subjects: int = 5,
correction_method: str | None = "fdr_bh",
target_chroma: str = "hbo",
graph_bounds: float | None = None,
roi_config: str | Path | None = None,
threshold_topo: bool = False,
) -> DataFrame:
"""
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.
@@ -3064,13 +3072,24 @@ def clean_subject_id(path_or_id):
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):
def run_cross_group_second_level_analysis(
df_roi_all: DataFrame,
file_paths_a: list[str],
file_paths_b: list[str],
group_a_name: str = "Group A",
group_b_name: str = "Group B",
df_cha_all: DataFrame | None = None,
raw_haemo: Any = None,
p_threshold: float = 0.05,
min_subjects: int = 3,
correction_method: str | None = "fdr_bh",
target_chroma: str = "hbo",
selected_event: str | None = None,
graph_bounds: tuple[float, float] | list[float] | None = None,
roi_config: Path | str | None = None,
threshold_topo: bool = False,
) -> DataFrame:
"""
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.
@@ -3331,11 +3350,21 @@ def run_cross_group_second_level_analysis(df_roi_all, file_paths_a, file_paths_b
def run_cross_group_laterality_analysis(df_roi_all_a, df_roi_all_b, roi_pairs, condition,
group_a_name="Group A", group_b_name="Group B",
target_chroma='hbo', min_subjects=3,
p_threshold=0.05, correction_method=None,
roi_contra_label=None, roi_ipsi_label=None):
def run_cross_group_laterality_analysis(
df_roi_all_a: DataFrame,
df_roi_all_b: DataFrame,
roi_pairs: tuple[str, str] | None,
condition: str | None,
group_a_name: str = "Group A",
group_b_name: str = "Group B",
target_chroma: str = "hbo",
min_subjects: int = 3,
p_threshold: float = 0.05,
correction_method: str | None = None,
roi_contra_label: str | None = None,
roi_ipsi_label: str | None = None,
) -> DataFrame:
"""
Compare LATERALITY between two independent groups of subjects (e.g. a
control group vs. a target group), using Welch's t-test on each
@@ -3582,11 +3611,20 @@ def run_cross_group_laterality_analysis(df_roi_all_a, df_roi_all_b, roi_pairs, c
def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_name, roi_json_path,
group_a_name="Group A", group_b_name="Group B",
target_chroma='hbo', min_subjects=3,
p_threshold=0.05, correction_method='fdr_bh',
weighted=True):
def run_cross_group_contrast_analysis(
df_contrasts_a: DataFrame,
df_contrasts_b: DataFrame,
contrast_name: str,
roi_json_path: str | Path | None,
group_a_name: str = "Group A",
group_b_name: str = "Group B",
target_chroma: str = "hbo",
min_subjects: int = 3,
p_threshold: float = 0.05,
correction_method: str = "fdr_bh",
weighted: bool = True,
) -> DataFrame:
"""
Compare a JOINT-FIT TASK CONTRAST (e.g. '2.0_vs_3.0'), aggregated to ROI
level, between two independent groups. This is the cross-group analog
@@ -3665,10 +3703,10 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n
if df_a_filt.empty:
print(f"[ERROR] Contrast '{contrast_name}' not found anywhere in {group_a_name}'s data.")
return pd.DataFrame()
return DataFrame()
if df_b_filt.empty:
print(f"[ERROR] Contrast '{contrast_name}' not found anywhere in {group_b_name}'s data.")
return pd.DataFrame()
return DataFrame()
roi_a = aggregate_channel_contrasts_to_roi(df_a_filt, roi_json_path, weighted=weighted)
roi_b = aggregate_channel_contrasts_to_roi(df_b_filt, roi_json_path, weighted=weighted)
@@ -3679,7 +3717,7 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n
if roi_a.empty or roi_b.empty:
print(f"[ERROR] No ROI-aggregated values produced for one or both groups "
f"(check regions.json channel names against this montage).")
return pd.DataFrame()
return DataFrame()
all_rois = sorted(set(roi_a['ROI'].unique()) | set(roi_b['ROI'].unique()))
results = []
@@ -3798,10 +3836,17 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n
def run_roi_paired_contrast_analysis(df_roi_all, roi_pairs, condition,
target_chroma='hbo', min_subjects=5,
p_threshold=0.05, correction_method=None,
roi_a_label=None, roi_b_label=None):
def run_roi_paired_contrast_analysis(
df_roi_all: DataFrame,
roi_pairs: Sequence[tuple[str, str]] | list[list[str]],
condition: str,
target_chroma: str = 'hbo',
min_subjects: int = 5,
p_threshold: float = 0.05,
correction_method: str | None = None,
roi_a_label: str | None = None,
roi_b_label: str | None = None,
) -> DataFrame:
"""
Paired within-subject ROI contrast (e.g. contralateral minus ipsilateral
motor ROI), as a companion to run_roi_second_level_analysis rather than a
@@ -4012,7 +4057,11 @@ def run_roi_paired_contrast_analysis(df_roi_all, roi_pairs, condition,
def aggregate_channel_contrasts_to_roi(df_contrasts, roi_json_path, weighted=True):
def aggregate_channel_contrasts_to_roi(
df_contrasts: DataFrame,
roi_json_path: str | Path | None,
weighted: bool = True
) -> DataFrame:
"""
Combine already-computed per-channel CONTRAST results (e.g. your
'2.0_vs_3.0' rows from contrasts.csv / contrast_results) into
@@ -5519,13 +5568,13 @@ def process_participant(file_path, progress_callback=None):
if BAD_CHANNELS_HANDLING != "None" and not FOLDING_BYP:
raw, fig_dropped, fig_raw_before, bad_channels = mark_bads(raw, bad_sci, bad_snr, bad_psp, bad_coeff_var, bad_amplitude_range, bad_noise, bad_disp)
if fig_dropped and fig_raw_before is not None:
fig_individual["fig2"] = fig_dropped
fig_individual["fig3"] = fig_raw_before
fig_individual["Bad Channels by Method"] = fig_dropped
fig_individual["Bad Channels Data"] = fig_raw_before
if bad_channels:
if BAD_CHANNELS_HANDLING == "Interpolate":
raw, fig_raw_after, fig_compare = interpolate_fNIRS_bads_weighted_average(raw, max_dist=MAX_DIST, min_neighbors=MIN_NEIGHBORS)
fig_individual["fig4"] = fig_raw_after
fig_individual["Compare"] = fig_compare
fig_individual["Data after Interpolating Bad Channels"] = fig_raw_after
fig_individual["Bad Channels Interpolation Results"] = fig_compare
elif BAD_CHANNELS_HANDLING == "Remove":
raw = remove_bad_channels(raw, bad_channels)
if progress_callback: progress_callback(13)
@@ -5542,7 +5591,7 @@ def process_participant(file_path, progress_callback=None):
if TDDR and not FOLDING_BYP:
raw_od = temporal_derivative_distribution_repair(raw_od)
fig_raw_od_tddr = raw_od.plot(duration=raw.times[-1], n_channels=raw.info['nchan'], title="After TDDR (Motion Correction)", show=False)
fig_individual["TDDR"] = fig_raw_od_tddr
fig_individual["Temporal Derivative Distribution Repair"] = fig_raw_od_tddr
if progress_callback: progress_callback(15)
logger.info("Step 15 Completed.")
@@ -5556,7 +5605,7 @@ def process_participant(file_path, progress_callback=None):
# Step 17: Haemoglobin Concentration
raw_haemo = beer_lambert_law(raw_od, ppf=calculate_dpf(file_path))
fig_raw_haemo_bll = raw_haemo.plot(duration=raw_haemo.times[-1], n_channels=raw_haemo.info['nchan'], title="HbO and HbR Signals", show=False)
fig_individual["BLL"] = fig_raw_haemo_bll
fig_individual["Modified Beer Lambert Law"] = fig_raw_haemo_bll
if progress_callback: progress_callback(17)
logger.info("Step 17 Completed.")
@@ -5564,15 +5613,15 @@ def process_participant(file_path, progress_callback=None):
if ENHANCE_NEGATIVE_CORRELATION and not FOLDING_BYP:
raw_haemo = enhance_negative_correlation(raw_haemo)
fig_raw_haemo_enc = raw_haemo.plot(duration=raw_haemo.times[-1], n_channels=raw_haemo.info['nchan'], title="Enhance Negative Correlation", show=False)
fig_individual["ENC"] = fig_raw_haemo_enc
fig_individual["Enhance Negative Correlation"] = fig_raw_haemo_enc
if progress_callback: progress_callback(18)
logger.info("Step 18 Completed.")
# Step 19: Filter
if FILTER and not FOLDING_BYP:
raw_haemo, fig_filter, fig_raw_haemo_filter = filter_the_data(raw_haemo)
fig_individual["filter1"] = fig_filter
fig_individual["filter2"] = fig_raw_haemo_filter
fig_individual["Filter_1"] = fig_filter
fig_individual["Filter_2"] = fig_raw_haemo_filter
if progress_callback: progress_callback(19)
logger.info("Step 19 Completed.")
@@ -5580,7 +5629,7 @@ def process_participant(file_path, progress_callback=None):
if not FOLDING_BYP:
events, event_dict = events_from_annotations(raw_haemo)
fig_events = plot_events(events, event_id=event_dict, sfreq=raw_haemo.info["sfreq"], show=False)
fig_individual["events"] = fig_events
fig_individual["Events"] = fig_events
if progress_callback: progress_callback(20)
logger.info("Step 20 Completed.")
@@ -5652,7 +5701,11 @@ def sanitize_paths_for_pickle(raw_haemo, epochs):
epochs._raw._filenames = [str(p) for p in epochs._raw._filenames]
def functional_connectivity_spectral_epochs(epochs, n_lines, vmin):
def functional_connectivity_spectral_epochs(
epochs: DataFrame | None,
n_lines: int,
vmin: float,
) -> None:
# will crash without this load
epochs.load_data()
@@ -5691,7 +5744,11 @@ def functional_connectivity_spectral_epochs(epochs, n_lines, vmin):
def functional_connectivity_spectral_time(epochs, n_lines, vmin):
def functional_connectivity_spectral_time(
epochs: DataFrame | None,
n_lines: int,
vmin: float,
) -> None:
# will crash without this load
epochs.load_data()
@@ -5735,7 +5792,12 @@ def functional_connectivity_spectral_time(epochs, n_lines, vmin):
def functional_connectivity_envelope(epochs, n_lines, vmin):
def functional_connectivity_envelope(
epochs: DataFrame | None,
n_lines: int,
vmin: float,
) -> None:
# will crash without this load
epochs.load_data()
@@ -5765,7 +5827,12 @@ def functional_connectivity_envelope(epochs, n_lines, vmin):
)
def functional_connectivity_betas(raw_hbo, n_lines, vmin, event_name=None):
def functional_connectivity_betas(
raw_hbo: BaseRaw,
n_lines: int,
vmin: float,
event_name: str | None = None,
) -> None:
raw_hbo = raw_hbo.copy().pick(picks="hbo")
onsets = raw_hbo.annotations.onset
@@ -5966,7 +6033,15 @@ def get_single_subject_beta_corr(raw_hbo, event_name=None, config=None):
return corr_matrix, raw_hbo.ch_names
def run_group_functional_connectivity(haemo_dict, config_dict, selected_paths, event_name, n_lines, vmin):
def run_group_functional_connectivity(
haemo_dict: dict[str | Path, BaseRaw],
config_dict: dict[str, Any],
selected_paths: list[str],
event_name: str | None,
n_lines: int,
vmin: float,
) -> None:
"""Aggregates multiple participants and triggers the plot."""
all_z_matrices = []
common_names = None
@@ -6068,3 +6143,27 @@ def run_group_functional_connectivity(haemo_dict, config_dict, selected_paths, e
title=f"Group Connectivity: {event_name if event_name else 'All Events'}",
vmin=vmin, vmax=1.0, colormap='hot'
)
def sparks_csv_export(
haemo_obj: BaseRaw,
save_path: str,
) -> None:
raw = haemo_obj
data, times = raw.get_data(return_times=True)
ann_col = np.full(times.shape, "", dtype=object)
if raw.annotations is not None and len(raw.annotations) > 0:
for onset, duration, desc in zip(
raw.annotations.onset,
raw.annotations.duration,
raw.annotations.description
):
mask = (times >= onset) & (times < onset + duration)
ann_col[mask] = desc
df = pd.DataFrame(data.T, columns=raw.ch_names)
df.insert(0, "annotation", ann_col)
df.insert(0, "time", times)
df.to_csv(save_path, index=False)
src/analysis/crossgroupbrainimage.py
+24 -4
View File
@@ -1,20 +1,28 @@
"""
Filename: crossgroupbrainimage.py
Description: Logic for the Cross-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
from mne.io.base import BaseRaw
import pandas as pd
from pandas import DataFrame
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 = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "show_optodes",
@@ -52,7 +60,15 @@ PARAMETERIZED_INDEXES = {
class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
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__("CrossGroupBrainImage")
self.setWindowTitle(f"Cross-Group Brain & Image Viewer - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
@@ -70,8 +86,9 @@ class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget):
if request is None:
return
(selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = 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)
# Build group-level contrast DataFrames
def concat_group_contrasts(file_paths: list[str], event: str | None) -> pd.DataFrame:
@@ -102,8 +119,11 @@ class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget):
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 = all_raw_objs[0].copy().pick(picks="hbo")
processed_raw = None
# Visualizations
for idx in selected_indexes:
src/analysis/crossgroupstats.py
+48 -24
View File
@@ -1,20 +1,28 @@
"""
Filename: crossgroupstats.py
Description: Cross-Group stats 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.io.base import BaseRaw
from flares import run_cross_group_contrast_analysis, run_cross_group_laterality_analysis, run_cross_group_second_level_analysis
from src.shared.flaresbasewidget import CrossGroupUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "p_threshold",
@@ -136,7 +144,18 @@ DESCRIPTION = """0. Raw ROI Comparison (run_cross_group_second_level_analysis)
class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict, json_location):
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],
json_location: str | Path
) -> None:
super().__init__("CrossGroupStats")
self.setWindowTitle(f"Cross-Group Stats Viewer - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
@@ -144,7 +163,7 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
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.group_dict = group_dict
self.json_location = json_location
self.setup_cross_group_ui(["0 (Raw ROI Comparison)", "1 (Laterality Comparison)", "2 (Contrast Comparison)",], placeholder_text=DESCRIPTION)
@@ -155,23 +174,18 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
if request is None:
return
(selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = request
(selected_event, file_paths_a, file_paths_b, _, selected_indexes, raw_params) = 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
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
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()
valid_dfs = [df for df in self.df_ind_dict.values() if not df.empty]
if valid_dfs:
df_ind_combined = pd.concat(valid_dfs, ignore_index=True)
else:
cha_combined = self.cha_dict
df_ind_combined = pd.DataFrame()
valid_chas = [df for df in self.cha_dict.values() if not df.empty]
cha_combined = pd.concat(valid_chas, ignore_index=True) if valid_chas else pd.DataFrame()
sample_path = file_paths_a[0]
p_haemo = self.haemo_dict.get(sample_path)
@@ -213,8 +227,8 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
min_subjects = params.get("min_subjects", 3)
correction_method = params.get("correction_method", "None")
target_chroma = params.get("target_chroma", "hbo")
roi_a = params.get("roi_a", "").strip()
roi_b = params.get("roi_b", "").strip()
roi_a: str = params.get("roi_a", "").strip()
roi_b: str = params.get("roi_b", "").strip()
if not roi_a or not roi_b:
print("Both a contralateral and ipsilateral ROI name must be specified.")
@@ -225,14 +239,19 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
# Build each group's dataframe directly from the dict using
# the file-path lists as keys - no ID cleaning/matching needed.
def _build_group_df(file_paths, dict_source):
def _build_group_df(
file_paths: list[str],
dict_source: dict[str, DataFrame]
) -> DataFrame:
valid_dfs = [
dict_source[fp] for fp in file_paths
if fp in dict_source and isinstance(dict_source[fp], pd.DataFrame)
and not dict_source[fp].empty
if fp in dict_source and not dict_source[fp].empty
]
return pd.concat(valid_dfs, ignore_index=True) if valid_dfs else pd.DataFrame()
df_roi_a = _build_group_df(file_paths_a, self.df_ind_dict)
df_roi_b = _build_group_df(file_paths_b, self.df_ind_dict)
@@ -271,8 +290,13 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget):
# directly from contrast_results_dict, keyed by file path -
# same dict-key approach as the laterality patch, avoids
# any ID-string matching.
def _build_group_contrast_df(file_paths, contrast_dict, name):
all_rows = []
def _build_group_contrast_df(
file_paths: list[str],
contrast_dict: dict[str, dict[str, pd.DataFrame]],
name: str,
) -> pd.DataFrame:
all_rows: list[DataFrame] = []
for fp in file_paths:
condition_dfs = contrast_dict.get(fp)
if condition_dfs is None:
src/analysis/exporttocsv.py
+24 -31
View File
@@ -1,6 +1,7 @@
"""
Filename: exporttocsv.py
Description: Logic for the Export To CSV analysis window
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
@@ -8,35 +9,47 @@ License: GPL-3.0
# Built-in imports
import os
from pathlib import Path
from typing import Any
# External library imports
import numpy as np
import pandas as pd
from pandas import DataFrame
from mne.io.base import BaseRaw
from PySide6.QtWidgets import QFileDialog, QMessageBox
from flares import sparks_csv_export
from src.shared.flaresbasewidget import CSVUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, cha_dict, df_ind, design_matrix, group, contrast_results_dict):
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__("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
self.design_matrix = design_matrix
self.group = group
self.contrast_results_dict = contrast_results_dict
# self.df_ind = df_ind_dict
# self.design_matrix = design_matrix_dict
# self.contrast_results_dict = contrast_results_dict
# self.group = group_dict
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 = []
selected_file_paths: list[str] = []
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)})"
@@ -52,7 +65,6 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget):
QMessageBox.warning(self, "Selection Missing", "Please select at least one participant and one export type.")
return
# 2. ASK ONCE: Select Output Directory
output_dir = QFileDialog.getExistingDirectory(self, "Select Output Folder for CSV Exports")
if not output_dir:
@@ -78,29 +90,11 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget):
cha.to_csv(save_path)
success_count += 1
elif idx == 1:
# SPARKS Export
save_path = os.path.join(output_dir, f"{base_filename}_sparks.csv")
if haemo_obj is not None:
raw = haemo_obj
data, times = raw.get_data(return_times=True)
ann_col = np.full(times.shape, "", dtype=object)
if raw.annotations is not None and len(raw.annotations) > 0:
for onset, duration, desc in zip(
raw.annotations.onset,
raw.annotations.duration,
raw.annotations.description
):
mask = (times >= onset) & (times < onset + duration)
ann_col[mask] = desc
df = pd.DataFrame(data.T, columns=raw.ch_names)
df.insert(0, "annotation", ann_col)
df.insert(0, "time", times)
df.to_csv(save_path, index=False)
success_count += 1
sparks_csv_export(haemo_obj, save_path)
success_count += 1
else:
print(f"No method defined for index {idx}")
@@ -120,4 +114,3 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget):
# caller="Video Alignment Tool"
# )
# win.show()
src/analysis/intergroupbrainimage.py
+55 -24
View File
@@ -1,20 +1,29 @@
"""
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
from mne.io import BaseRaw
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "lower_bound",
@@ -74,15 +83,24 @@ PARAMETERIZED_INDEXES = {
class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group):
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 = cha
self.df_ind = df_ind
self.design_matrix = design_matrix
self.contrast_results = contrast_results
self.group = group
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)",])
@@ -92,35 +110,45 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, param_values,) = request
(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:
participant_events = set(haemo_obj.annotations.description)
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
if haemo_obj is None:
continue
cha_df = self.cha.get(file_path)
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.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.get(file_path)
df = self.df_ind_dict.get(file_path)
if df is not None:
df_group = pd.concat([df_group, df], ignore_index=True)
@@ -147,9 +175,9 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
print(f"Missing parameters for index {idx}, skipping.")
continue
all_contrasts = []
all_contrasts: list[DataFrame] = []
for fp in selected_file_paths:
condition_dfs = self.contrast_results.get(fp, {})
condition_dfs = self.contrast_results_dict.get(fp, {})
if selected_event in condition_dfs:
df = condition_dfs[selected_event].copy()
df["ID"] = fp
@@ -159,7 +187,8 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
print("No contrast data found for selected participants and event.")
return
df_contrasts = pd.concat(all_contrasts, ignore_index=True)
# 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:
@@ -173,13 +202,15 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
print(f"Missing parameters for index {idx}, skipping.")
continue
raw_list = [self.haemo_dict.get(fp) for fp in selected_file_paths]
all_raw_objs = [self.haemo_dict.get(fp) for fp in selected_file_paths if self.haemo_dict.get(fp)]
if len(selected_file_paths) > 1:
print(f"Aggregating geometry for {len(selected_file_paths)} participants...")
processed_raw = aggregate_fnirs_group_geometry(raw_list)
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 = raw_list[0].copy().pick(picks="hbo")
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)
src/analysis/intergroupfunctionalconnectivity.py
+19 -4
View File
@@ -1,20 +1,27 @@
"""
Filename: intergroupfunctionalconnectivity.py
Description: Logic for the Inter-Group Functional Connectivity 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
from PySide6.QtWidgets import QMessageBox
from mne.io.base import BaseRaw
from flares import run_group_functional_connectivity
from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "n_lines",
@@ -34,11 +41,17 @@ PARAMETERIZED_INDEXES = {
class InterGroupFunctionalConnectivityWidget(InterGroupUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, group, config_dict):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
group_dict: dict[str, str],
config_dict: dict[str, str]
) -> None:
super().__init__("InterGroupFunctionalConnectivity")
self.setWindowTitle(f"Inter-Group Functional Connectivity Viewer [BETA] - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
self.group = group
#self.group_dict = group_dict
self.config_dict = config_dict
QMessageBox.warning(self, f"Warning - {APP_NAME.upper()}", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. "
@@ -52,7 +65,9 @@ class InterGroupFunctionalConnectivityWidget(InterGroupUIMixin, FlaresBaseWidget
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, param_values,) = request
(selected_event, selected_file_paths, selected_indexes, raw_params) = request
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
for idx in selected_indexes:
if idx == 0:
src/analysis/intergroupstats.py
+55 -22
View File
@@ -1,20 +1,29 @@
"""
Filename: intergroupstats.py
Description: Logic for the Inter-Group Stats 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 run_roi_paired_contrast_analysis, run_roi_second_level_analysis, aggregate_channel_contrasts_to_roi
from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "p_threshold",
@@ -148,40 +157,64 @@ DESCRIPTION = """0. ROI vs. Zero (run_roi_second_level_analysis)
class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group, json_location):
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],
json_location: str | Path
) -> None:
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.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.json_location = json_location
self.setup_inter_group_ui(["0 (ROI vs. Zero)", "1 (Paired ROI Contrast)", "2 (Joint Contrast, ROI-Aggregated)"], placeholder_text=DESCRIPTION)
def process_request(self):
request = self.get_common_request_data(PARAMETERIZED_INDEXES, self.json_location, self.contrast_results)
request = self.get_common_request_data(PARAMETERIZED_INDEXES, self.json_location, self.contrast_results_dict)
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, param_values,) = request
(selected_event, selected_file_paths, selected_indexes, raw_params) = request
all_cha = pd.DataFrame()
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
all_cha = 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 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)
@@ -189,10 +222,10 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget):
p_haemo = self.haemo_dict.get(file_path)
# Concatenate individual ROI stats (df_ind) for all chosen subjects
df_group = pd.DataFrame()
df_group = DataFrame()
if selected_file_paths:
for file_path in selected_file_paths:
df = self.df_ind.get(file_path)
df = self.df_ind_dict.get(file_path)
if df is not None:
df_group = pd.concat([df_group, df], ignore_index=True)
@@ -226,7 +259,7 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget):
print(f"No ROI data matches the condition '{selected_event}'.")
continue
all_cha_filtered = pd.DataFrame()
all_cha_filtered = 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]
@@ -304,9 +337,9 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget):
continue
all_contrasts = []
all_contrasts: list[DataFrame] = []
for fp in selected_file_paths:
condition_dfs = self.contrast_results.get(fp)
condition_dfs = self.contrast_results_dict.get(fp)
if condition_dfs is None:
print(f" [MISSING] '{fp}' not found in contrast_results.")
continue
src/analysis/participantbrain.py
+32 -7
View File
@@ -1,18 +1,28 @@
"""
Filename: participantbrain.py
Description: Logic for the Participant Brain 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
from mne import Annotations
from pandas import DataFrame
from mne.io.base import BaseRaw
from flares import brain_3d_visualization, brain_landmarks_3d
from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "show_optodes",
@@ -57,7 +67,12 @@ PARAMETERIZED_INDEXES = {
class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, cha_dict):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
cha_dict: dict[str, DataFrame],
) -> None:
super().__init__("ParticipantBrain")
self.setWindowTitle(f"Participant Brain Viewer - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
@@ -72,21 +87,31 @@ class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget):
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, param_values,) = request
(selected_event, selected_file_paths, selected_indexes, raw_params) = request
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
# Pass the necessary arguments to each method
for file_path in selected_file_paths:
haemo_obj = self.haemo_dict.get(file_path)
if haemo_obj is None:
continue
if selected_event:
participant_events = set(haemo_obj.annotations.description)
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
if haemo_obj is None:
raise Exception("How did we get here?")
cha = self.cha_dict.get(file_path)
for idx in selected_indexes:
src/analysis/participantfoldchannels.py
+1 -1
View File
@@ -71,7 +71,7 @@ def single_participant_worker(file_path, raw_data, result_queue, progress_queue)
try:
from flares import fold_channels
# Perform the heavy fold_channels logic
channel_results = fold_channels(raw_data, p_name, progress_queue)
channel_results = fold_channels(raw=raw_data, p_name=p_name, progress_queue=progress_queue)
# Hand back results and signal completion
result_queue.put({file_path: channel_results})
src/analysis/participantfunctionalconnectivity.py
+32 -7
View File
@@ -1,20 +1,30 @@
"""
Filename: participantfunctionalconnectivity.py
Description: Logic for the Participant Functional Connectivity 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
from PySide6.QtWidgets import QMessageBox
from pandas import DataFrame
from mne import Annotations
from mne.io.base import BaseRaw
from flares import functional_connectivity_betas, functional_connectivity_envelope, functional_connectivity_spectral_epochs, functional_connectivity_spectral_time
from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
PARAMETERIZED_INDEXES = {
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [
{
"key": "n_lines",
@@ -79,7 +89,12 @@ PARAMETERIZED_INDEXES = {
class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidget):
def __init__(self, haemo_dict, epochs_dict):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
epochs_dict: dict[str, DataFrame],
) -> None:
super().__init__("ParticipantFunctionalConnectivity")
self.setWindowTitle(f"Participant Functional Connectivity Viewer [BETA] - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
@@ -97,22 +112,32 @@ class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidg
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, param_values,) = request
(selected_event, selected_file_paths, selected_indexes, raw_params) = request
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
# Pass the necessary arguments to each method
for file_path in selected_file_paths:
haemo_obj = self.haemo_dict.get(file_path)
epochs_obj = self.epochs_dict.get(file_path)
if haemo_obj is None:
continue
if selected_event:
participant_events = set(haemo_obj.annotations.description)
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
if haemo_obj is None:
raise Exception("How did we get here?")
for idx in selected_indexes:
if idx == 0:
src/analysis/participantimage.py
+23 -13
View File
@@ -1,17 +1,20 @@
"""
Filename: participantimage.py
Description: Logic for the Participant Image analysis window
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in Imports
import os
import os.path as op
from pathlib import Path
from datetime import datetime
# External library imports
from mne.io.base import BaseRaw
from PySide6.QtWidgets import QGridLayout, QHBoxLayout, QMessageBox, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import Qt, QSize
from PySide6.QtGui import QPixmap
@@ -21,7 +24,13 @@ from src.shared.shareddata import APP_NAME
class ParticipantImageViewerWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, fig_bytes_dict):
def __init__(
self,
haemo_dict: dict[str, BaseRaw],
fig_bytes_dict: dict[str, dict[str, bytes]]
) -> None:
super().__init__("ParticipantImage")
self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
self.setWindowTitle(f"Participant Image Viewer - {APP_NAME.upper()}")
@@ -29,12 +38,12 @@ class ParticipantImageViewerWidget(FlaresBaseWidget):
self.fig_bytes_dict = fig_bytes_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)"
self.participant_map: dict[str, str] = {}
self.participant_dropdown_items: list[str] = []
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)})"
display_label = f"{short_label} ({op.basename(file_path)})"
self.participant_map[file_path] = short_label
self.participant_dropdown_items.append(display_label)
@@ -87,23 +96,24 @@ class ParticipantImageViewerWidget(FlaresBaseWidget):
selected_display_names = self._get_checked_items(self.participant_dropdown)
# Map from display names back to file paths
selected_file_paths = []
selected_file_paths: list[str] = []
for display_name in selected_display_names:
# Find file_path by matching display name
for fp, short_label in self.participant_map.items():
expected_display = f"{short_label} ({os.path.basename(fp)})"
expected_display = f"{short_label} ({Path(fp).name})"
if display_name == expected_display:
selected_file_paths.append(fp)
selected_file_paths.append(str(fp))
break
selected_labels = self._get_checked_items(self.image_index_dropdown)
row, col = 0, 0
for file_path in selected_file_paths:
fig_list = self.fig_bytes_dict.get(file_path, [])
participant_label = self.participant_map[file_path]
fig_map: dict[str, bytes] = self.fig_bytes_dict.get(file_path, {})
participant_label: str = self.participant_map.get(file_path, "Unknown")
for label in selected_labels:
fig_bytes = fig_list.get(label)
fig_bytes: bytes | None = fig_map.get(label)
if not fig_bytes:
continue
@@ -149,7 +159,7 @@ class ParticipantImageViewerWidget(FlaresBaseWidget):
for display_name in selected_display_names:
# Match display name to file path
for file_path, short_label in self.participant_map.items():
expected_display = f"{short_label} ({os.path.basename(file_path)})"
expected_display = f"{short_label} ({op.basename(file_path)})"
if display_name == expected_display:
fig_dict = self.fig_bytes_dict.get(file_path, {})
for label in selected_image_labels:
@@ -157,7 +167,7 @@ class ParticipantImageViewerWidget(FlaresBaseWidget):
continue
fig_bytes = fig_dict[label]
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{os.path.basename(file_path)}_{label}_{timestamp}.png"
filename = f"{op.basename(file_path)}_{label}_{timestamp}.png"
output_path = save_dir / filename
with open(output_path, "wb") as f:
f.write(fig_bytes)
src/shared/flaresbasewidget.py
+66 -17
View File
@@ -9,6 +9,8 @@ License: GPL-3.0
import os
import json
from pathlib import Path
from typing import Sequence, Any
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, QSize, Qt
@@ -807,7 +809,11 @@ class FlaresBaseWidget(QWidget):
self.image_index_dropdown = None
def _create_multiselect_dropdown(self, items):
def _create_multiselect_dropdown(
self,
items: Sequence[str]
) -> FullClickComboBox:
combo = FullClickComboBox()
combo.setView(QListView())
model = QStandardItemModel()
@@ -874,7 +880,11 @@ class FlaresBaseWidget(QWidget):
# checked.append(item.text())
# return checked
def _get_checked_items(self, combo=None):
def _get_checked_items(
self,
combo: QComboBox | None = None
) -> list[str]:
target = combo if combo is not None else getattr(self, 'participant_dropdown', None)
if target is None or target.model() is None:
@@ -897,7 +907,10 @@ class FlaresBaseWidget(QWidget):
return checked_items
def update_participant_dropdown_label(self, combo=None):
def update_participant_dropdown_label(
self,
combo: QComboBox | int | None = None
) -> None:
"""
Handles label updates for ANY participant dropdown.
If 'combo' is None, it defaults to the standard self.participant_dropdown.
@@ -1142,7 +1155,13 @@ class FlaresBaseWidget(QWidget):
class CrossGroupUIMixin:
def setup_cross_group_ui(self, index_texts, placeholder_text=""):
participant_map: dict[str, str]
def setup_cross_group_ui(
self,
index_texts: Sequence[str],
placeholder_text: str = ""
) -> None:
self.group_to_paths = {}
for file_path, group_name in self.group_dict.items():
@@ -1293,7 +1312,13 @@ class CrossGroupUIMixin:
return file_paths
def get_common_request_data(self, parameterized_indexes, json_location=None, contrast_dfs=None):
def get_common_request_data(
self,
parameterized_indexes: dict[int, list[dict[str, Any]]],
json_location: str | Path | None = None,
contrast_dfs: dict[str, dict[str, Any]] | None = None,
) -> tuple[str | None, list[str], list[str], list[str], list[int], dict[str, Any]] | None:
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
selected_event = None
@@ -1412,11 +1437,14 @@ class CrossGroupUIMixin:
class CSVUIMixin:
def setup_csv_ui(self, index_texts):
def setup_csv_ui(
self,
index_texts: Sequence[str]
) -> None:
# Create mappings: file_path -> participant label and dropdown display text
self.participant_map = {} # file_path -> "Participant 1"
self.participant_dropdown_items = [] # "Participant 1 (filename)"
self.participant_map: dict[str, str] = {} # file_path -> "Participant 1"
self.participant_dropdown_items: list[str] = [] # "Participant 1 (filename)"
for i, file_path in enumerate(self.haemo_dict.keys(), start=1):
short_label = f"Participant {i}"
@@ -1428,12 +1456,12 @@ class CSVUIMixin:
self.top_bar = QHBoxLayout()
self.layout.addLayout(self.top_bar)
self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items)
self.participant_dropdown: FullClickComboBox = 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: FullClickComboBox = self._create_multiselect_dropdown(self.index_texts)
self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label)
self.submit_button = QPushButton("Submit")
@@ -1456,13 +1484,20 @@ class CSVUIMixin:
self.showMaximized()
class InterGroupUIMixin:
def setup_inter_group_ui(self, index_texts, placeholder_text=""):
def setup_inter_group_ui(
self,
index_texts: Sequence[str],
placeholder_text: str = ""
) -> None:
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_map: dict[str, str] = {} # file_path -> "Participant 1"
self.participant_dropdown_items = [] # "Participant 1 (filename)"
for i, file_path in enumerate(self.haemo_dict.keys(), start=1):
@@ -1525,7 +1560,13 @@ class InterGroupUIMixin:
self.thumb_size = QSize(280, 180)
self.showMaximized()
def get_common_request_data(self, parameterized_indexes, json_location=None, contrast_dfs=None):
def get_common_request_data(
self,
parameterized_indexes: dict[int, list[dict[str, Any]]],
json_location: str | Path | None = None,
contrast_dfs: dict[str, dict[str, Any]] | None = None,
) -> tuple[str | None, list[str], list[int], dict[str, Any]] | None:
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
selected_event = None
@@ -1570,7 +1611,7 @@ class InterGroupUIMixin:
dynamic_rois = []
# 1. Check for the JSON file and parse ROI names
if os.path.exists(json_location):
if json_location is not None and os.path.exists(json_location):
try:
with open(json_location, 'r', encoding='utf-8') as f:
regions_data = json.load(f)
@@ -1645,9 +1686,13 @@ class InterGroupUIMixin:
)
class ParticipantUIMixin:
def setup_participant_ui(self, index_texts):
def setup_participant_ui(
self,
index_texts: Sequence[str]
) -> None:
# Create mappings: file_path -> participant label and dropdown display text
self.participant_map = {} # file_path -> "Participant 1"
self.participant_map: dict[str, str] = {} # file_path -> "Participant 1"
self.participant_dropdown_items = [] # "Participant 1 (filename)"
for i, file_path in enumerate(self.haemo_dict.keys(), start=1):
@@ -1694,7 +1739,11 @@ class ParticipantUIMixin:
self.showMaximized()
def get_common_request_data(self, parameterized_indexes):
def get_common_request_data(
self,
parameterized_indexes: dict[int, list[dict[str, Any]]]
) -> tuple[str | None, list[str], list[int], dict[str, Any]] | None:
selected_event = self.event_dropdown.currentText()
if selected_event == "<None Selected>":
selected_event = None
src/shared/shareddata.py
+12 -11
View File
@@ -1,22 +1,27 @@
"""
Filename: shareddata.py
Description: Shared constants and methods for FLARES
Description: Shared constants and methods other files depend on
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
"""
import sys
# Built-in imports
import os
import sys
import platform
CURRENT_VERSION = "1.5.0"
APP_NAME = "flares"
APP_NAME_EXPANDED = "fNIRS Lightweight Analysis, Research, & Evaluation Suite"
API_URL = f"https://git.research.dezeeuw.ca/api/v1/repos/tyler/{APP_NAME}/releases"
API_URL_SECONDARY = f"https://git.research2.dezeeuw.ca/api/v1/repos/tyler/{APP_NAME}/releases"
PLATFORM_NAME = platform.system().lower()
CHANGELOG_URL = "https://git.research.dezeeuw.ca/tyler/flares/raw/branch/main/changelog_major.md"
WIKI_URL = "https://git.research.dezeeuw.ca/tyler/flares/wiki"
CHANGELOG_URL = f"https://git.research.dezeeuw.ca/tyler/{APP_NAME}/raw/branch/main/changelog_major.md"
WIKI_URL = f"https://git.research.dezeeuw.ca/tyler/{APP_NAME}/wiki"
PIPELINE_STAGES = [
"Preprocessing",
@@ -49,15 +54,11 @@ PIPELINE_STAGES = [
"Finishing Up"
]
def resource_path(relative_path):
def resource_path(relative_path: str) -> str:
"""
Get absolute path to resource regardless of running directly or packaged using PyInstaller
"""
if hasattr(sys, '_MEIPASS'):
# PyInstaller bundle path
base_path = sys._MEIPASS
else:
base_path = os.path.abspath(".")
base_path = getattr(sys, "_MEIPASS", os.path.abspath("."))
return os.path.join(base_path, relative_path)
src/window/about.py
+5 -4
View File
@@ -1,6 +1,7 @@
"""
Filename: about.py
Description: About window for FLARES
Description: About window
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
@@ -9,7 +10,7 @@ License: GPL-3.0
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel
from PySide6.QtCore import Qt
from src.shared.shareddata import APP_NAME, CURRENT_VERSION
from src.shared.shareddata import APP_NAME, APP_NAME_EXPANDED, CURRENT_VERSION
class AboutWindow(QWidget):
"""
@@ -19,14 +20,14 @@ class AboutWindow(QWidget):
parent (QWidget, optional): Parent widget of this window. Defaults to None.
"""
def __init__(self, parent=None):
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent, Qt.WindowType.Window)
self.setWindowTitle(f"About {APP_NAME.upper()}")
self.resize(250, 100)
layout = QVBoxLayout()
label = QLabel(f"About {APP_NAME.upper()}", self)
label2 = QLabel("fNIRS Lightweight Analysis, Research, & Evaluation Suite", self)
label2 = QLabel(f"{APP_NAME_EXPANDED}", self)
label3 = QLabel(f"{APP_NAME.upper()} is licensed under the GPL-3.0 licence. For more information, visit https://www.gnu.org/licenses/gpl-3.0.en.html", self)
label4 = QLabel(f"Version v{CURRENT_VERSION}")
src/window/terminal.py
+13 -10
View File
@@ -1,21 +1,24 @@
"""
Filename: terminal.py
Description: Terminal window for FLARES
Description: Terminal window
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
"""
from typing import Any, Callable
from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit
from PySide6.QtCore import Qt
from src.shared.shareddata import API_URL, API_URL_SECONDARY, APP_NAME, CURRENT_VERSION, PLATFORM_NAME
from src.window.about import AboutWindow
from updater import LocalPendingUpdateCheckThread, UpdateManager
from updater import UpdateManager
class TerminalWindow(QWidget):
def __init__(self, parent=None):
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent, Qt.WindowType.Window)
self.setWindowTitle(f"Terminal - {APP_NAME.upper()}")
self.resize(320, 180)
@@ -30,7 +33,7 @@ class TerminalWindow(QWidget):
layout.addWidget(self.input_line)
self.setLayout(layout)
self.commands = {
self.commands: dict[str, Callable[..., Any]] = {
"hello": self.cmd_hello,
"help": self.cmd_help,
"version": self.cmd_version,
@@ -68,22 +71,22 @@ class TerminalWindow(QWidget):
self.output_area.append(f"[Unknown command] '{command_name}'")
def cmd_hello(self, *args):
def cmd_hello(self, *args: Any) -> str:
return "Hello from the terminal!"
def cmd_help(self, *args):
def cmd_help(self, *args: Any) -> str:
return f"Available commands: {', '.join(self.commands.keys())}"
def cmd_version(self, *args):
def cmd_version(self, *args: Any) -> str:
return f"{APP_NAME.upper()} is running version {CURRENT_VERSION}."
def cmd_about(self, *args):
def cmd_about(self, *args: Any) -> None:
self.about = AboutWindow(self)
self.about.show()
def cmd_update(self, *args):
def cmd_update(self, *args: Any) -> str:
main_win = self.parent()
if main_win is None:
if not isinstance(main_win, QWidget):
return "[Error] Main window context not found."
self.updater = UpdateManager(
src/window/updateoptodes.py
+3 -3
View File
@@ -15,9 +15,9 @@ import numpy as np
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QHBoxLayout, QMessageBox, QLineEdit, QPushButton, QFileDialog
from PySide6.QtCore import Qt
from mne.io import read_raw_snirf
from mne_nirs.io import write_raw_snirf
from mne.channels import make_dig_montage
from mne.io import read_raw_snirf #type: ignore
from mne_nirs.io import write_raw_snirf #type: ignore
from mne.channels import make_dig_montage #type: ignore
from src.shared.shareddata import APP_NAME
src/window/userguide.py
+1 -1
View File
@@ -20,7 +20,7 @@ class UserGuideWindow(QWidget):
parent (QWidget, optional): Parent widget of this window. Defaults to None.
"""
def __init__(self, parent=None):
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent, Qt.WindowType.Window)
self.setWindowTitle(f"User Guide - {APP_NAME.upper()}")
self.resize(250, 100)
src/window/viewerlauncher.py
+1 -1
View File
@@ -40,7 +40,7 @@ class ViewerLauncherWidget(QWidget):
("Cross-Group Stats Viewer", CrossGroupStatsWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict, json_location], True),
("Inter-Group Brain and Image Viewer", InterGroupBrainImageWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True),
("Cross-Group Brain and 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)
("Export To CSV Viewer", ExportToCSVWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True)
]
layout = QVBoxLayout(self)
src/window/welcome.py
+5 -4
View File
@@ -1,6 +1,7 @@
"""
Filename: welcome.py
Description: Welcome dialog for FLARES
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
@@ -9,13 +10,13 @@ License: GPL-3.0
from PySide6.QtWidgets import QTextBrowser, QVBoxLayout, QLabel, QDialog, QHBoxLayout, QPushButton
from PySide6.QtGui import QDesktopServices, QIcon
from PySide6.QtCore import QUrl
from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest
from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply
from src.shared.shareddata import APP_NAME, CURRENT_VERSION, CHANGELOG_URL, resource_path
class WelcomeDialog(QDialog):
def __init__(self, parent=None, direct=True, first=False):
def __init__(self, parent: QDialog | None = None, direct: bool = True, first: bool = False):
super().__init__(parent)
self.setWindowTitle(f"What's New - {APP_NAME.upper()}")
self.setMinimumSize(550, 450)
@@ -64,10 +65,10 @@ class WelcomeDialog(QDialog):
self.network_manager.get(QNetworkRequest(QUrl(CHANGELOG_URL)))
def _on_download_complete(self, reply):
def _on_download_complete(self, reply: QNetworkReply) -> None:
"""Processes the downloaded markdown and drops it into the view frame."""
if reply.error() == reply.NetworkError.NoError:
raw_bytes = reply.readAll()
raw_bytes = reply.readAll().data()
# Convert raw bytes to standard text string
markdown_text = str(raw_bytes, encoding='utf-8')