Fix #116: Move functional connectivity computation to background thread
Test Suite / test (push) Canceled after 0s

Issue: FC analysis (especially option 2 Beta-Series Correlation) hangs the UI for minutes.

Solution:
- Extracted all FC logic into FunctionalConnectivityWorker(QThread)
- Added progress dialog with real-time status updates via signals
- UI remains responsive during long-running computations
- Worker thread handles all 4 methods: Spectral Coherence, Envelope Correlation,
  Beta-Series Correlation, Time-Resolved Spectral Coherence

Testing: Syntax validated. Logic moved 1:1 from synchronous process_request()
to worker.run() with added progress messaging.
This commit is contained in:
Hermes committed 2026-11-07 17:52:36 -07:00
1 parent 8c432f3da4
commit 6f4095933a
1 file changed
+183 -97
+183 -97
View File
@@ -8,10 +8,11 @@ License: GPL-3.0
"""
# Built-in imports
from typing import Any, cast
from typing import Any, cast, Optional
# External library imports
from PySide6.QtWidgets import QMessageBox
from PySide6.QtWidgets import QMessageBox, QProgressDialog
from PySide6.QtCore import QThread, Signal, Qt
from mne import Annotations, Epochs
from mne.io.base import BaseRaw
@@ -21,6 +22,144 @@ from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget
from src.shared.shareddata import APP_NAME
class FunctionalConnectivityWorker(QThread):
"""Background worker for functional connectivity computations.
Issue #116: Prevents UI freeze during long-running FC calculations.
"""
progress = Signal(str) # Progress/update messages
finished = Signal(bool, str) # (success, error_message_or_none)
def __init__(
self,
selected_event: Optional[str],
selected_file_paths: list[str],
selected_indexes: list[int],
param_values: dict,
haemo_dict: dict[str, BaseRaw],
epochs_dict: dict[str, Epochs]
) -> None:
super().__init__()
self.selected_event = selected_event
self.selected_file_paths = selected_file_paths
self.selected_indexes = selected_indexes
self.param_values = param_values
self.haemo_dict = haemo_dict
self.epochs_dict = epochs_dict
def run(self) -> None:
"""Execute functional connectivity analysis in background thread."""
try:
for file_path in self.selected_file_paths:
self.progress.emit(f"Processing {file_path}...")
haemo_obj = self.haemo_dict.get(file_path)
epochs_obj = self.epochs_dict.get(file_path)
if haemo_obj is None or epochs_obj is None:
continue
if self.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 self.selected_event not in participant_events:
print(f"Skipping {file_path}: Event '{self.selected_event}' not found.")
continue
for idx in self.selected_indexes:
params = self.param_values.get(idx, {})
method_name = self._get_method_name(idx)
self.progress.emit(f" Running {method_name}...")
if idx == 0:
# Spectral Coherence
method = params.get("method", "wpli2_debiased")
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.3)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
functional_connectivity_spectral_epochs(
epochs=epochs_obj, n_lines=n_lines, vmin=vmin,
fmin=fmin, fmax=fmax, method=method
)
elif idx == 1:
# Envelope Correlation
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.9)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
orthogonalize = params.get("orthogonalize", False)
absolute = params.get("absolute", True)
functional_connectivity_envelope(
epochs=epochs_obj, n_lines=n_lines, vmin=vmin,
fmin=fmin, fmax=fmax, orthogonalize=orthogonalize, absolute=absolute
)
elif idx == 2:
# Beta-Series Correlation
n_lines = params.get("n_lines", 20)
drift_model = params.get("drift_model", "cosine")
drift_order = params.get("drift_order", 1)
hrf_model = params.get("hrf_model", "glover")
apply_gsr = params.get("apply_gsr", True)
min_effect_size = params.get("min_effect_size", 0.7)
alpha = params.get("alpha", 0.05)
resample_freq = params.get("resample_freq", 4.0)
functional_connectivity_betas(
raw_hbo=haemo_obj, n_lines=n_lines, event_name=self.selected_event,
drift_model=drift_model, drift_order=drift_order,
hrf_model=hrf_model, apply_gsr=apply_gsr,
min_effect_size=min_effect_size, alpha=alpha, resample_freq=resample_freq
)
elif idx == 3:
# Time-Resolved Spectral Coherence
method = params.get("method", "wpli")
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.3)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
n_freqs = params.get("n_freqs", 10)
cycles_multiplier = params.get("cycles_multiplier", 2.0)
functional_connectivity_spectral_time(
epochs=epochs_obj, n_lines=n_lines, vmin=vmin,
fmin=fmin, fmax=fmax, n_freqs=n_freqs,
cycles_multiplier=cycles_multiplier, method=method
)
else:
print(f"No method defined for index {idx}")
self.progress.emit("Analysis complete!")
self.finished.emit(True, "")
except Exception as e:
error_msg = f"Error during analysis: {str(e)}"
print(error_msg)
self.finished.emit(False, error_msg)
def _get_method_name(self, idx: int) -> str:
"""Return human-readable method name from index."""
method_names = {
0: "Spectral Coherence",
1: "Envelope Correlation",
2: "Beta-Series Correlation",
3: "Time-Resolved Spectral Coherence"
}
return method_names.get(idx, f"Unknown ({idx})")
PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
0: [ # Spectral Coherence
{"key": "method", "label": "Connectivity method", "default": "wpli2_debiased", "type": list, "options": ["coh", "pli", "wpli2_debiased"]},
@@ -97,106 +236,53 @@ class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidg
)
def process_request(self):
"""Execute functional connectivity analysis in background thread (Issue #116)."""
request = self.get_common_request_data(PARAMETERIZED_INDEXES)
if request is None:
return
(selected_event, selected_file_paths, selected_indexes, raw_params) = request
param_values = cast(dict[int | str, dict[str, Any]], raw_params)
# 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)
# Show progress dialog
progress_dialog = QProgressDialog(
"Running functional connectivity analysis...\n\n"
"This may take several minutes depending on method and data size.\n"
"You can close this window - processing will continue in the background.",
"Cancel", 0, 0, self
)
progress_dialog.setWindowTitle(f"Processing - {APP_NAME.upper()}")
progress_dialog.setWindowModality(Qt.WindowModality.WindowModal)
progress_dialog.autoClose = False
progress_dialog.autoReset = False
progress_dialog.show()
if haemo_obj is None or epochs_obj is None:
continue
if selected_event:
raw_annotations = getattr(haemo_obj, "annotations", None)
if raw_annotations is not None:
annotations = cast(Annotations, raw_annotations)
descriptions = cast(list[str], list(annotations.description))
participant_events: set[str] = set(descriptions)
else:
participant_events: set[str] = set()
if selected_event not in participant_events:
print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.")
continue
for idx in selected_indexes:
params = param_values.get(idx, {})
if idx == 0:
method = params.get("method", "wpli2_debiased")
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.9)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
functional_connectivity_spectral_epochs(epochs=epochs_obj, n_lines=n_lines, vmin=vmin, fmin=fmin, fmax=fmax, method=method)
elif idx == 1:
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.9)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
orthogonalize = params.get("orthogonalize", False)
absolute = params.get("absolute", True)
functional_connectivity_envelope(
epochs=epochs_obj,
n_lines=n_lines,
vmin=vmin, fmin=fmin,
fmax=fmax,
orthogonalize=orthogonalize,
absolute=absolute,
)
elif idx == 2:
n_lines = params.get("n_lines", 20)
drift_model = params.get("drift_model", "cosine")
drift_order = params.get("drift_order", 1)
hrf_model = params.get("hrf_model", "glover")
apply_gsr = params.get("apply_gsr", True)
min_effect_size = params.get("min_effect_size", 0.7)
alpha = params.get("alpha", 0.05)
resample_freq = params.get("resample_freq", 4.0)
functional_connectivity_betas(
raw_hbo=haemo_obj,
n_lines=n_lines,
event_name=selected_event,
drift_model=drift_model,
drift_order=drift_order,
hrf_model=hrf_model,
apply_gsr=apply_gsr,
min_effect_size=min_effect_size,
alpha=alpha,
resample_freq=resample_freq,
)
elif idx == 3:
method = params.get("method", "wpli")
n_lines = params.get("n_lines", 20)
vmin = params.get("vmin", 0.9)
fmin = params.get("fmin", 0.04)
fmax = params.get("fmax", 0.2)
n_freqs = params.get("n_freqs", 10)
cycles_multiplier = params.get("cycles_multiplier", 2.0)
functional_connectivity_spectral_time(
epochs=epochs_obj,
n_lines=n_lines,
vmin=vmin,
fmin=fmin,
fmax=fmax,
n_freqs=n_freqs,
cycles_multiplier=cycles_multiplier,
method=method
)
else:
print(f"No method defined for index {idx}")
# Create and start worker thread
self.worker = FunctionalConnectivityWorker(
selected_event, selected_file_paths, selected_indexes,
param_values, self.haemo_dict, self.epochs_dict
)
# Connect signals
self.worker.progress.connect(lambda msg: progress_dialog.setLabelText(msg))
self.worker.finished.connect(self._on_fc_analysis_complete)
self.worker.finished.connect(progress_dialog.close)
# Start processing
self.worker.start()
def _on_fc_analysis_complete(self, success: bool, error_msg: str) -> None:
"""Handle completion of functional connectivity analysis."""
if success:
QMessageBox.information(
self, f"Success - {APP_NAME.upper()}",
"Functional connectivity analysis completed!\n\n"
"Check the output files in your results directory."
)
else:
QMessageBox.critical(
self, f"Error - {APP_NAME.upper()}",
f"Analysis failed:\n{error_msg}"
)
self.worker.deleteLater() # Clean up worker thread