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

massive changes for 1.5.0

Browse Source
This commit is contained in:
tyler
2026-06-28 08:17:36 -07:00
parent 57b7082564
commit 766bf75dd7
11 changed files with 1603 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flares.py
+185 -142
View File
@@ -213,6 +213,8 @@ AGE: int = 25 # Assume 25 if not set from the GUI. This will result in a reason
GENDER: str = ""
GROUP: str = "Default"
FOLDING_BYP: bool = False
# These are parameters that are required for the analysis
REQUIRED_KEYS: dict[str, Any] = {
@@ -1487,7 +1489,7 @@ def make_design_matrix(raw_haemo, short_chans):
pass
# 2) Create design matrix
if SHORT_CHANNEL_REGRESSION:
if SHORT_CHANNEL_REGRESSION and not FOLDING_BYP:
design_matrix = make_first_level_design_matrix(
raw=raw_haemo,
stim_dur=STIM_DUR,
@@ -1739,161 +1741,197 @@ def resource_path(relative_path):
def fold_channels(raw: BaseRaw) -> None:
# def fold_channels(raw: BaseRaw) -> None:
# Locate the fOLD excel files
if getattr(sys, 'frozen', False):
set_config('MNE_NIRS_FOLD_PATH', resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary")) # type: ignore
else:
path = os.path.expanduser("~") + "/mne_data/fOLD/fOLD-public-master/Supplementary"
set_config('MNE_NIRS_FOLD_PATH', resource_path(path)) # type: ignore
# # Locate the fOLD excel files
# if getattr(sys, 'frozen', False):
# set_config('MNE_NIRS_FOLD_PATH', resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary")) # type: ignore
# else:
# path = os.path.expanduser("~") + "/mne_data/fOLD/fOLD-public-master/Supplementary"
# set_config('MNE_NIRS_FOLD_PATH', resource_path(path)) # type: ignore
output = None
# output = None
# List to store the results
landmark_specificity_data: list[dict[str, Any]] = []
# # List to store the results
# landmark_specificity_data: list[dict[str, Any]] = []
# Filter the data to only what we want
hbo_channel_names = cast(list[str], getattr(raw.copy().pick(picks='hbo'), "ch_names")) # type: ignore
# # Filter the data to only what we want
# hbo_channel_names = cast(list[str], getattr(raw.copy().pick(picks='hbo'), "ch_names")) # type: ignore
# Format the output to make it slightly easier to read
# # Format the output to make it slightly easier to read
if True:
num_channels = len(hbo_channel_names)
rows, cols = 4, 7 # 6 rows and 4 columns of pie charts
fig, axes = plt.subplots(rows, cols, figsize=(16, 10), constrained_layout=True)
axes = axes.flatten() # Flatten the axes array for easier indexing
# if True:
# num_channels = len(hbo_channel_names)
# rows, cols = 4, 7 # 6 rows and 4 columns of pie charts
# fig, axes = plt.subplots(rows, cols, figsize=(16, 10), constrained_layout=True)
# axes = axes.flatten() # Flatten the axes array for easier indexing
# If more pie charts than subplots, create extra subplots
if num_channels > rows * cols:
fig, axes = plt.subplots((num_channels // cols) + 1, cols, figsize=(16, 10), constrained_layout=True)
axes = axes.flatten()
# # If more pie charts than subplots, create extra subplots
# if num_channels > rows * cols:
# fig, axes = plt.subplots((num_channels // cols) + 1, cols, figsize=(16, 10), constrained_layout=True)
# axes = axes.flatten()
# Create a list for consistent color mapping
landmarks = [
"1 - Primary Somatosensory Cortex",
"2 - Primary Somatosensory Cortex",
"3 - Primary Somatosensory Cortex",
"4 - Primary Motor Cortex",
"5 - Somatosensory Association Cortex",
"6 - Pre-Motor and Supplementary Motor Cortex",
"7 - Somatosensory Association Cortex",
"8 - Includes Frontal eye fields",
"9 - Dorsolateral prefrontal cortex",
"10 - Frontopolar area",
"11 - Orbitofrontal area",
"17 - Primary Visual Cortex (V1)",
"18 - Visual Association Cortex (V2)",
"19 - V3",
"20 - Inferior Temporal gyrus",
"21 - Middle Temporal gyrus",
"22 - Superior Temporal Gyrus",
"23 - Ventral Posterior cingulate cortex",
"24 - Ventral Anterior cingulate cortex",
"25 - Subgenual cortex",
"32 - Dorsal anterior cingulate cortex",
"37 - Fusiform gyrus",
"38 - Temporopolar area",
"39 - Angular gyrus, part of Wernicke's area",
"40 - Supramarginal gyrus part of Wernicke's area",
"41 - Primary and Auditory Association Cortex",
"42 - Primary and Auditory Association Cortex",
"43 - Subcentral area",
"44 - pars opercularis, part of Broca's area",
"45 - pars triangularis Broca's area",
"46 - Dorsolateral prefrontal cortex",
"47 - Inferior prefrontal gyrus",
"48 - Retrosubicular area",
"Brain_Outside",
]
# # Create a list for consistent color mapping
# landmarks = [
# "1 - Primary Somatosensory Cortex",
# "2 - Primary Somatosensory Cortex",
# "3 - Primary Somatosensory Cortex",
# "4 - Primary Motor Cortex",
# "5 - Somatosensory Association Cortex",
# "6 - Pre-Motor and Supplementary Motor Cortex",
# "7 - Somatosensory Association Cortex",
# "8 - Includes Frontal eye fields",
# "9 - Dorsolateral prefrontal cortex",
# "10 - Frontopolar area",
# "11 - Orbitofrontal area",
# "17 - Primary Visual Cortex (V1)",
# "18 - Visual Association Cortex (V2)",
# "19 - V3",
# "20 - Inferior Temporal gyrus",
# "21 - Middle Temporal gyrus",
# "22 - Superior Temporal Gyrus",
# "23 - Ventral Posterior cingulate cortex",
# "24 - Ventral Anterior cingulate cortex",
# "25 - Subgenual cortex",
# "32 - Dorsal anterior cingulate cortex",
# "37 - Fusiform gyrus",
# "38 - Temporopolar area",
# "39 - Angular gyrus, part of Wernicke's area",
# "40 - Supramarginal gyrus part of Wernicke's area",
# "41 - Primary and Auditory Association Cortex",
# "42 - Primary and Auditory Association Cortex",
# "43 - Subcentral area",
# "44 - pars opercularis, part of Broca's area",
# "45 - pars triangularis Broca's area",
# "46 - Dorsolateral prefrontal cortex",
# "47 - Inferior prefrontal gyrus",
# "48 - Retrosubicular area",
# "Brain_Outside",
# ]
cmap1 = plt.get_cmap('tab20') # First 20 colors
cmap2 = plt.get_cmap('tab20b') # Next 20 colors
# cmap1 = plt.get_cmap('tab20') # First 20 colors
# cmap2 = plt.get_cmap('tab20b') # Next 20 colors
# Combine the colors from both colormaps
colors = [cmap1(i) for i in range(20)] + [cmap2(i) for i in range(20)] # Total 40 colors
# # Combine the colors from both colormaps
# colors = [cmap1(i) for i in range(20)] + [cmap2(i) for i in range(20)] # Total 40 colors
landmarks.sort(key=lambda x: (int(x.split(" - ")[0]) if x.split(" - ")[0].isdigit() else float('inf')))
# landmarks.sort(key=lambda x: (int(x.split(" - ")[0]) if x.split(" - ")[0].isdigit() else float('inf')))
landmark_color_map = {landmark: colors[i % len(colors)] for i, landmark in enumerate(landmarks)}
# landmark_color_map = {landmark: colors[i % len(colors)] for i, landmark in enumerate(landmarks)}
# Iterate over each channel
print(len(hbo_channel_names))
# # Iterate over each channel
# print(len(hbo_channel_names))
for idx, channel_name in enumerate(hbo_channel_names):
# for idx, channel_name in enumerate(hbo_channel_names):
print(idx, channel_name)
# Run the fOLD on the selected channel
channel_data = raw.copy().pick(picks=channel_name) # type: ignore
# print(idx, channel_name)
# # Run the fOLD on the selected channel
# channel_data = raw.copy().pick(picks=channel_name) # type: ignore
output = cast(list[DataFrame], fold_channel_specificity_normal(channel_data, interpolate=True, atlas='Brodmann'))
# output = cast(list[DataFrame], fold_channel_specificity_normal(channel_data, interpolate=True, atlas='Brodmann'))
# Process each DataFrame that fold_channel_specificity returns
for df_data in output:
# # Process each DataFrame that fold_channel_specificity returns
# for df_data in output:
# Extract the relevant columns
useful_data = df_data[['Landmark', 'Specificity']]
# # Extract the relevant columns
# useful_data = df_data[['Landmark', 'Specificity']]
# Store the results
landmark_specificity_data.append({
'Channel': channel_name,
'Data': useful_data,
})
# # Store the results
# landmark_specificity_data.append({
# 'Channel': channel_name,
# 'Data': useful_data,
# })
# Plot the results
# TODO: Fix this
if True:
unique_landmarks = sorted(useful_data['Landmark'].unique())
color_list = [landmark_color_map[landmark] for landmark in useful_data['Landmark']]
# # Plot the results
# # TODO: Fix this
# if True:
# unique_landmarks = sorted(useful_data['Landmark'].unique())
# color_list = [landmark_color_map[landmark] for landmark in useful_data['Landmark']]
# Plot specificity for each channel
ax = axes[idx]
# # Plot specificity for each channel
# ax = axes[idx]
labels = [f'{landmark.split(" - ")[0]}' if landmark != 'Brain_Outside' else 'B' for landmark in useful_data['Landmark']]
# labels = [f'{landmark.split(" - ")[0]}' if landmark != 'Brain_Outside' else 'B' for landmark in useful_data['Landmark']]
wedges, texts, autotexts = ax.pie(
useful_data['Specificity'],
autopct='%1.1f%%',
startangle=90,
labels=labels,
labeldistance=1.05,
colors=color_list)
# wedges, texts, autotexts = ax.pie(
# useful_data['Specificity'],
# autopct='%1.1f%%',
# startangle=90,
# labels=labels,
# labeldistance=1.05,
# colors=color_list)
ax.set_title(f'{channel_name}')
ax.axis('equal')
# ax.set_title(f'{channel_name}')
# ax.axis('equal')
landmark_specificity_data = []
# landmark_specificity_data = []
# TODO: Fix this
if True:
handles = [
plt.Line2D([0], [0], marker='o', color='w', label=landmark, markersize=10,
markerfacecolor=landmark_color_map[landmark])
for landmark in landmarks
]
n_landmarks = len(landmarks)
# # TODO: Fix this
# if True:
# handles = [
# plt.Line2D([0], [0], marker='o', color='w', label=landmark, markersize=10,
# markerfacecolor=landmark_color_map[landmark])
# for landmark in landmarks
# ]
# n_landmarks = len(landmarks)
# Calculate the figure size based on number of rows and columns
fig_width = 5
fig_height = n_landmarks / 4
# # Calculate the figure size based on number of rows and columns
# fig_width = 5
# fig_height = n_landmarks / 4
# Create a new figure window for the legend
legend_fig = plt.figure(figsize=(fig_width, fig_height))
legend_axes = legend_fig.add_subplot(111)
legend_axes.axis('off') # Turn off axis for the legend window
legend_axes.legend(handles=handles, loc='center', fontsize=10, title="Landmarks")
# # Create a new figure window for the legend
# legend_fig = plt.figure(figsize=(fig_width, fig_height))
# legend_axes = legend_fig.add_subplot(111)
# legend_axes.axis('off') # Turn off axis for the legend window
# legend_axes.legend(handles=handles, loc='center', fontsize=10, title="Landmarks")
for ax in axes[len(hbo_channel_names):]:
ax.axis('off')
# for ax in axes[len(hbo_channel_names):]:
# ax.axis('off')
#plt.show()
fig_dict = {"main": fig, "legend": legend_fig}
return convert_fig_dict_to_png_bytes(fig_dict)
# #plt.show()
# fig_dict = {"main": fig, "legend": legend_fig}
# return convert_fig_dict_to_png_bytes(fig_dict)
def fold_channels(raw: BaseRaw, p_name: str, 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"))
else:
path = os.path.expanduser("~") + "/mne_data/fOLD/fOLD-public-master/Supplementary"
set_config('MNE_NIRS_FOLD_PATH', resource_path(path))
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 = {}
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'))
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'])
})
step_idx += 1
if progress_queue is not None:
progress_queue.put((p_name, step_idx))
# Return raw data dictionary to the result_queue
return channel_results
def individual_significance(raw_haemo, glm_est):
@@ -3939,6 +3977,7 @@ def hr_calc(raw):
def process_participant(file_path, progress_callback=None):
fig_individual: dict[str, Figure] = {}
logger.info(f"Folding Bypass: {FOLDING_BYP}")
# Step 1: Preprocessing
raw = load_snirf(file_path)
@@ -3949,7 +3988,7 @@ def process_participant(file_path, progress_callback=None):
# Step 2: Trimming
# TODO: Clean this into a method
if TRIM:
if TRIM and not FOLDING_BYP:
if hasattr(raw, 'annotations') and len(raw.annotations) > 0:
# Get time of first event
first_event_time = raw.annotations.onset[0]
@@ -3985,7 +4024,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 3 Completed.")
# Step 4: Short/Long Channels
if SHORT_CHANNEL:
if SHORT_CHANNEL and not FOLDING_BYP:
short_chans = get_short_channels(raw, max_dist=SHORT_CHANNEL_THRESH)
fig_short_chans = short_chans.plot(duration=raw.times[-1], n_channels=raw.info['nchan'], title="Short Channels Only", show=False)
fig_individual["short"] = fig_short_chans
@@ -3996,7 +4035,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 4 Completed.")
# Step 5: Heart Rate
if HEART_RATE:
if HEART_RATE and not FOLDING_BYP:
fig, hr1, hr2, low, high = hr_calc(raw)
fig_individual["PSD"] = fig
fig_individual['HeartRate_PSD'] = hr1
@@ -4017,7 +4056,7 @@ def process_participant(file_path, progress_callback=None):
# Step 6: Scalp Coupling Index
bad_sci = []
if SCI:
if SCI and not FOLDING_BYP:
if HEART_RATE:
bad_sci, fig_sci_1, fig_sci_2 = calculate_scalp_coupling(raw, low, high)
else:
@@ -4029,7 +4068,7 @@ def process_participant(file_path, progress_callback=None):
# Step 7: Signal to Noise Ratio
bad_snr = []
if SNR:
if SNR and not FOLDING_BYP:
bad_snr, fig_snr = calculate_signal_noise_ratio(raw)
fig_individual["SNR1"] = fig_snr
if progress_callback: progress_callback(7)
@@ -4037,7 +4076,7 @@ def process_participant(file_path, progress_callback=None):
# Step 8: Peak Spectral Power
bad_psp = []
if PSP:
if PSP and not FOLDING_BYP:
bad_psp, fig_psp1, fig_psp2 = calculate_peak_power(raw)
fig_individual["PSP1"] = fig_psp1
fig_individual["PSP2"] = fig_psp2
@@ -4045,35 +4084,35 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 8 Completed.")
bad_cv = []
if CV:
if CV and not FOLDING_BYP:
bad_cv, fig_cv = find_bad_channels_cv(raw, cv_threshold=CV_THRESHOLD)
fig_individual['cv'] = fig_cv
if progress_callback: progress_callback(9)
logger.info("Step 9 Completed.")
bad_range = []
if MAD:
if MAD and not FOLDING_BYP:
bad_range, fig_range = find_bad_channels_range(raw, threshold=MAD_THRESHOLD)
fig_individual['range'] = fig_range
if progress_callback: progress_callback(10)
logger.info("Step 10 Completed.")
bad_noise = []
if PSD_NOISE:
if PSD_NOISE and not FOLDING_BYP:
bad_noise, fig_noise = detect_high_freq_noise(raw, db_limit=DB_LIMIT, freq_div=TARGET_FREQ_DIV)
fig_individual['psd_noise'] = fig_noise
if progress_callback: progress_callback(11)
logger.info("Step 11 Completed.")
bad_disp = []
if CHANNEL_VAR:
if CHANNEL_VAR and not FOLDING_BYP:
bad_disp, fig_disp = detect_sensor_displacement(raw, threshold_ratio=CHANNEL_THRESH)
fig_individual['displacement'] = fig_disp
if progress_callback: progress_callback(12)
logger.info("Step 12 Completed.")
# Step 9: Bad Channels Handling
if BAD_CHANNELS_HANDLING != "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_cv, bad_range, bad_noise, bad_disp)
if fig_dropped and fig_raw_before is not None:
fig_individual["fig2"] = fig_dropped
@@ -4108,7 +4147,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 14 Completed.")
# Step 11: Temporal Derivative Distribution Repair Filtering
if TDDR:
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
@@ -4116,7 +4155,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 15 Completed.")
# Step 12: Wavelet Filtering
if WAVELET:
if WAVELET and not FOLDING_BYP:
raw_od, fig = calculate_and_apply_wavelet(raw_od)
fig_individual["Wavelet"] = fig
if progress_callback: progress_callback(16)
@@ -4130,7 +4169,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 17 Completed.")
# Step 14: Enhance Negative Correlation
if ENHANCE_NEGATIVE_CORRELATION:
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
@@ -4138,7 +4177,7 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 18 Completed.")
# Step 15: Filter
if 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
@@ -4146,16 +4185,18 @@ def process_participant(file_path, progress_callback=None):
logger.info("Step 19 Completed.")
# Step 16: Extracting Events
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
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
if progress_callback: progress_callback(20)
logger.info("Step 20 Completed.")
# Step 17: Epoch Calculations
epochs, fig_epochs = epochs_calculations(raw_haemo, events, event_dict)
for name, fig in fig_epochs:
fig_individual[f"epochs_{name}"] = fig
if not FOLDING_BYP:
epochs, fig_epochs = epochs_calculations(raw_haemo, events, event_dict)
for name, fig in fig_epochs:
fig_individual[f"epochs_{name}"] = fig
if progress_callback: progress_callback(21)
logger.info("Step 21 Completed.")
@@ -4274,6 +4315,8 @@ def process_participant(file_path, progress_callback=None):
# Step 24: Finishing Up
fig_bytes = convert_fig_dict_to_png_bytes(fig_individual)
if FOLDING_BYP:
epochs = None
sanitize_paths_for_pickle(raw_haemo, epochs)
if progress_callback: progress_callback(28)