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

improvements for 1.5.2

Browse Source
This commit is contained in:
tyler
2026-07-29 17:20:23 -07:00
parent 8b017005c5
commit f15a5d9433
8 changed files with 750 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flares.py
+198 -135
View File
@@ -8,24 +8,26 @@ 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, Sequence, cast, Literal, Union
from itertools import compress
from copy import deepcopy
from multiprocessing import Queue, Pool
import os.path as op
import gc
import re
import traceback
from concurrent.futures import ProcessPoolExecutor, as_completed
from queue import Empty
import sys
import json
import time
import multiprocessing as mp
import logging
import platform
import warnings
import threading
import traceback
import itertools
import os.path as op
from io import BytesIO
from queue import Empty
from pathlib import Path
from copy import deepcopy
import multiprocessing as mp
from itertools import compress
from multiprocessing import Queue
from typing import Any, Optional, Sequence, cast, Literal, Union
# External library imports
import matplotlib.pyplot as plt
@@ -50,10 +52,11 @@ from nilearn.glm.regression import OLSModel
import statsmodels.formula.api as smf # type: ignore
from statsmodels.stats.multitest import multipletests
from statsmodels.tools.sm_exceptions import ConvergenceWarning
from scipy.spatial.distance import cdist
from scipy.signal import welch, butter, filtfilt # type: ignore
from scipy.stats import pearsonr, zscore, t
from scipy.stats import pearsonr, zscore, ttest_1samp, ttest_ind, sem
import pywt # type: ignore
import neurokit2 as nk # type: ignore
@@ -67,7 +70,7 @@ import xlrd
# External library imports for mne
from mne import (
EvokedArray, SourceEstimate, Info, Epochs, Label, Annotations,
events_from_annotations, read_source_spaces,
events_from_annotations, read_source_spaces, create_info,
stc_near_sensors, pick_types, grand_average, get_config, set_config, read_labels_from_annot
) # type: ignore
from mne.source_space import SourceSpaces
@@ -99,21 +102,6 @@ from mne_connectivity.viz import plot_connectivity_circle
from mne_connectivity import envelope_correlation, spectral_connectivity_epochs, spectral_connectivity_time
import os
import json
import warnings
import numpy as np
import pandas as pd
import scipy.stats as stats
import matplotlib.pyplot as plt
import seaborn as sns
import statsmodels.formula.api as smf
from statsmodels.stats.multitest import multipletests
from statsmodels.tools.sm_exceptions import ConvergenceWarning
from mne_nirs.statistics import statsmodels_to_results
from mne_nirs.visualisation import plot_glm_group_topo
import logging
# Needs to be set for mne
os.environ["SUBJECTS_DIR"] = str(data_path()) + "/subjects" # type: ignore
@@ -383,6 +371,7 @@ def set_metadata(file_path, metadata: dict[str, Any]) -> None:
globals()[key] = val
def gui_entry(config: dict[str, Any], gui_queue: Queue, progress_queue: Queue, ack_queue: Queue) -> None:
start_time = time.time()
try:
file_paths = config['SNIRF_FILES']
file_params = config['PARAMS']
@@ -393,7 +382,24 @@ def gui_entry(config: dict[str, Any], gui_queue: Queue, progress_queue: Queue, a
file_paths, file_params, file_metadata, progress_queue, gui_queue, max_workers
)
gui_queue.put({"type": "FINISHED_SUCCESSFULLY", "success": True})
elapsed = time.time() - start_time
success_count = getattr(process_multiple_participants, "_success_count", 0)
total_duration = getattr(process_multiple_participants, "_duration_total", 0.0)
failed_stages = getattr(process_multiple_participants, "_failed_stages", [])
speedup = None
if success_count > 0 and elapsed > 0:
avg_success_duration = total_duration / success_count
failed_credit = sum(stage * avg_success_duration for stage in failed_stages)
naive_serial_estimate = total_duration + failed_credit
speedup = min(naive_serial_estimate / elapsed, max_workers)
gui_queue.put({
"type": "FINISHED_SUCCESSFULLY",
"success": True,
"elapsed": elapsed,
"speedup": speedup,
})
try:
print("CHILD: Waiting for GUI acknowledgment...")
@@ -406,122 +412,186 @@ def gui_entry(config: dict[str, Any], gui_queue: Queue, progress_queue: Queue, a
"type": "FINISHED_SUCCESSFULLY",
"success": False,
"error": str(e),
"traceback": traceback.format_exc()
"traceback": traceback.format_exc(),
"elapsed": time.time() - start_time
})
finally:
pass
def process_participant_worker(file_path, file_params, file_metadata, result_queue, progress_queue):
file_name = os.path.basename(file_path)
file_start = time.time()
stage_tracker = {"value": 0.0}
try:
# 1. Setup
set_config_me(file_params)
set_metadata(file_path, file_metadata)
def progress_callback(step_idx):
stage_tracker["value"] = min(step_idx / 28, 1.0)
if progress_queue:
# We use put_nowait to prevent the worker from hanging on a full queue
try:
progress_queue.put_nowait(('progress', file_path, step_idx))
except: pass
except Exception:
pass
# 2. Process
result = process_participant(file_path, progress_callback=progress_callback)
# 3. Report Success
result_queue.put((file_path, result, None))
duration = time.time() - file_start
result_queue.put((file_path, result, None, duration, 1.0))
except Exception as e:
duration = time.time() - file_start
try:
result_queue.put((file_path, None, str(e)))
except: pass
result_queue.put((file_path, None, f"{e}\n{traceback.format_exc()}", duration, stage_tracker["value"]))
except Exception:
pass
finally:
try:
sys.stdout.flush()
sys.stderr.flush()
# Give the queue thread a moment to send the data before we vanish
time.sleep(0.2)
except: pass
sys.exit(0)
plt.close('all')
gc.collect()
except Exception:
pass
def process_multiple_participants(file_paths, file_params, file_metadata, progress_queue=None, gui_queue=None, max_workers=6):
#audit_log.info(f"--- SESSION START: {len(file_paths)} files ---")
def process_multiple_participants(file_paths, file_params, file_metadata,
progress_queue=None, gui_queue=None, max_workers=6):
ctx = mp.get_context("spawn")
result_queue = ctx.Queue()
pending_files = list(file_paths)
active_processes = []
pending_lock = threading.Lock()
results_by_file = {}
manager = mp.Manager()
result_queue = manager.Queue()
results_lock = threading.Lock()
stop_event = threading.Event()
active_lock = threading.Lock()
active_processes = [] # tracked only for emergency cleanup on error
start_time = time.time()
duration_total = {"value": 0.0}
success_count = {"value": 0}
failed_stages = {"value": []}
def elapsed_heartbeat():
# Ticks once a second so the GUI can show a live-updating timer,
# independent of when files actually finish.
while not stop_event.is_set():
if gui_queue:
try:
gui_queue.put({"type": "elapsed", "seconds": time.time() - start_time})
except Exception:
pass
stop_event.wait(1.0)
def relay_progress():
while not stop_event.is_set():
try:
msg = progress_queue.get(timeout=0.05)
except Empty:
continue
except (EOFError, OSError):
break
if gui_queue:
try:
gui_queue.put(msg)
except Exception:
pass
def result_collector():
# Runs continuously in its own thread, fully decoupled from spawning.
while not stop_event.is_set():
try:
res_path, result, error, duration, stage = result_queue.get(timeout=0.05)
except Empty:
continue
except (EOFError, OSError):
break
if error is None:
duration_total["value"] += duration
success_count["value"] += 1
else:
failed_stages["value"].append(stage)
if gui_queue:
try:
gui_queue.put({
"type": "file_done",
"file": res_path,
"success": error is None,
"result": result if error is None else None,
"error": error,
})
except Exception:
pass
else:
with results_lock:
results_by_file[res_path] = result
def slot_worker():
# Each slot independently: grab a file, spawn+join, repeat.
# Replacement happens immediately on exit -- no shared polling loop.
while not stop_event.is_set():
with pending_lock:
if not pending_files:
return
file_path = pending_files.pop(0)
p = ctx.Process(
target=process_participant_worker,
args=(file_path, file_params, file_metadata, result_queue, progress_queue)
)
p.daemon = False
p.start()
with active_lock:
active_processes.append(p)
p.join()
with active_lock:
if p in active_processes:
active_processes.remove(p)
relay_thread = None
if progress_queue is not None:
relay_thread = threading.Thread(target=relay_progress, daemon=True)
relay_thread.start()
collector_thread = threading.Thread(target=result_collector, daemon=True)
collector_thread.start()
heartbeat_thread = threading.Thread(target=elapsed_heartbeat, daemon=True)
heartbeat_thread.start()
n_slots = max(1, min(max_workers, len(pending_files)))
slot_threads = [threading.Thread(target=slot_worker, daemon=True) for _ in range(n_slots)]
try:
while pending_files or active_processes:
for p, f_path in active_processes[:]:
if not p.is_alive():
p.join(timeout=0.1)
active_processes.remove((p, f_path))
while len(active_processes) < max_workers and pending_files:
file_path = pending_files.pop(0)
p = mp.Process(
target=process_participant_worker,
args=(file_path, file_params, file_metadata, result_queue, progress_queue)
)
p.start()
active_processes.append((p, file_path))
if progress_queue:
while not progress_queue.empty():
try:
prog_msg = progress_queue.get_nowait()
if gui_queue:
# Forward straight to GUI
gui_queue.put(prog_msg, timeout=0.1)
except: break
while not result_queue.empty():
try:
res_path, result, error = result_queue.get(timeout=0.01)
if gui_queue:
gui_queue.put({
"type": "file_done",
"file": res_path,
"success": error is None,
"result": result if error is None else None,
"error": error if error else None
}, timeout=2)
else:
results_by_file[res_path] = result
except:
break
if not pending_files and not active_processes:
print("DEBUG: Loop finished naturally.")
break
# should no longer hang because pipes are being drained
time.sleep(0.1)
for t in slot_threads:
t.start() # all slots fire their first spawn essentially at once
for t in slot_threads:
t.join()
except Exception as e:
print(f"MAIN LOOP ERROR: {e}")
finally:
# Cleanup
for p, _ in active_processes:
try:
if p.is_alive():
p.terminate()
p.join(timeout=0.1)
except: pass
stop_event.set()
with active_lock:
for p in active_processes:
try:
if p.is_alive():
p.terminate()
p.join(timeout=1)
except Exception:
pass
if relay_thread:
relay_thread.join(timeout=2)
collector_thread.join(timeout=2)
heartbeat_thread.join(timeout=2)
if manager is not None:
try:
manager.shutdown()
except: pass
process_multiple_participants._duration_total = duration_total ["value"]
process_multiple_participants._success_count = success_count["value"]
process_multiple_participants._failed_stages = failed_stages["value"]
return results_by_file
@@ -2172,8 +2242,6 @@ def aggregate_fnirs_group_geometry(raw_list: Sequence[BaseRaw | None]) -> BaseRa
2. Average by Individual Optode (S, D) across all averaged pairings.
Returns a unified MNE Raw object with exactly one dot per optode.
"""
import mne
import numpy as np
channel_locs = {}
all_ch_names = []
@@ -2223,11 +2291,11 @@ def aggregate_fnirs_group_geometry(raw_list: Sequence[BaseRaw | None]) -> BaseRa
final_chs.append(new_ch)
# Create the final MNE Info
fake_info = mne.create_info(ch_names=all_ch_names, sfreq=ref_raw.info['sfreq'], ch_types='hbo')
fake_info = create_info(ch_names=all_ch_names, sfreq=ref_raw.info['sfreq'], ch_types='hbo')
with fake_info._unlock():
fake_info['chs'] = final_chs
return mne.io.RawArray(np.zeros((len(all_ch_names), 1)), fake_info)
return RawArray(np.zeros((len(all_ch_names), 1)), fake_info)
@@ -2840,9 +2908,9 @@ def run_roi_second_level_analysis(
continue
Y = sub_data['theta'].values
t_val, p_val = stats.ttest_1samp(Y, 0)
t_val, p_val = ttest_1samp(Y, 0)
mean_beta = np.mean(Y)
std_err = stats.sem(Y)
std_err = sem(Y)
group_results.append({
'ROI': roi,
@@ -3136,7 +3204,7 @@ def run_cross_group_second_level_analysis(
continue
# Welch's t-test (assumes unequal variances)
t_val, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False)
t_val, p_val = ttest_ind(vals_a, vals_b, equal_var=False)
mean_a, mean_b = np.mean(vals_a), np.mean(vals_b)
diff_val = mean_a - mean_b
@@ -3274,7 +3342,7 @@ def run_cross_group_second_level_analysis(
vals_b = ch_b[val_col].dropna().values
if len(vals_a) >= min_subjects and len(vals_b) >= min_subjects:
t_stat, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False)
t_stat, p_val = ttest_ind(vals_a, vals_b, equal_var=False)
mean_diff = np.mean(vals_a) - np.mean(vals_b)
else:
t_stat, p_val, mean_diff = 0.0, 1.0, 0.0
@@ -3511,7 +3579,7 @@ def run_cross_group_laterality_analysis(
vals_a = lat_a['laterality'].values
vals_b = lat_b['laterality'].values
t_val, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False)
t_val, p_val = ttest_ind(vals_a, vals_b, equal_var=False)
mean_a, mean_b = np.mean(vals_a), np.mean(vals_b)
mean_diff = mean_a - mean_b
@@ -3735,7 +3803,7 @@ def run_cross_group_contrast_analysis(
)
continue
t_val, p_val = stats.ttest_ind(vals_a, vals_b, equal_var=False)
t_val, p_val = ttest_ind(vals_a, vals_b, equal_var=False)
mean_a, mean_b = np.mean(vals_a), np.mean(vals_b)
mean_diff = mean_a - mean_b
@@ -3958,7 +4026,7 @@ def run_roi_paired_contrast_analysis(
continue
Y = merged['diff'].values
t_val, p_val = stats.ttest_1samp(Y, 0)
t_val, p_val = ttest_1samp(Y, 0)
mean_diff = np.mean(Y)
results.append({
@@ -4750,8 +4818,6 @@ def plot_heart_rate(
return fig1, fig2
# import numpy as np
# def mark_bads_by_db_threshold(raw, db_limit=-60):
# """
# Converts a dB threshold to absolute power and marks channels
@@ -4914,8 +4980,6 @@ def find_flatline_at_end(raw, threshold_ratio=0.05):
# plt.show(block=True)
# import numpy as np
def detect_sensor_dropout(raw, threshold_ratio=0.05):
@@ -6060,7 +6124,6 @@ def run_group_functional_connectivity(
all_z_matrices.append(z_mat)
common_names = names
from scipy.stats import ttest_1samp
# 1. Convert list to 3D array: (Participants, Channels, Channels)
group_z_data = np.array(all_z_matrices)