unit testing and others
This commit is contained in:
@@ -2,6 +2,12 @@
|
||||
|
||||
- Fixed an issue where file associations appeared to work but would not load the project on macOS
|
||||
- Fixed an issue where file associations would refuse to assosciate on macOS
|
||||
- Fixed an issue where certain parameters would not enable or disable depending on other parameters when they should've
|
||||
- Fixed an issue where not all widgets would close when attempting to close the application causing the application to crash
|
||||
- Revamped the Participant Functional Connectivity Viewer to contain descriptions of the methods similar to the Stats Viewers
|
||||
- Modified the Participant Functional Connectivity Analysis options to better perform their tasks. This remains as a BETA feature
|
||||
- Changed RESAMPLE to only apply where it is required to avoid having Functional Connectivity Analysis methods running on data that has been resampled
|
||||
- Added basic unit testing in an attempt to prevent any accidental processing changes from occuring in the future
|
||||
|
||||
|
||||
# Version 1.6.0
|
||||
|
||||
@@ -1565,19 +1565,20 @@ def make_design_matrix(
|
||||
else:
|
||||
short_chans = None
|
||||
|
||||
# Set the new annotations
|
||||
raw_haemo.set_annotations(new_annot)
|
||||
raw_haemo_dm = raw_haemo.copy()
|
||||
|
||||
if resample:
|
||||
raw_haemo.resample(resample_freq, npad="auto")
|
||||
raw_haemo._data = raw_haemo._data * 1e6
|
||||
raw_haemo_dm.resample(resample_freq, npad="auto")
|
||||
try:
|
||||
short_chans.resample(resample_freq)
|
||||
except:
|
||||
pass
|
||||
|
||||
raw_haemo_dm._data = raw_haemo_dm._data * 1e6
|
||||
|
||||
design_matrix = make_first_level_design_matrix(
|
||||
raw=raw_haemo,
|
||||
raw=raw_haemo_dm,
|
||||
stim_dur=stim_dur,
|
||||
hrf_model=hrf_model,
|
||||
drift_model=drift_model,
|
||||
@@ -1635,7 +1636,7 @@ def make_design_matrix(
|
||||
fig, ax1 = plt.subplots(figsize=(10, 6), constrained_layout=True)
|
||||
_ = plot_design_matrix(design_matrix, axes=ax1)
|
||||
|
||||
return raw_haemo, design_matrix, fig
|
||||
return raw_haemo, raw_haemo_dm, design_matrix, fig
|
||||
|
||||
|
||||
|
||||
@@ -5417,7 +5418,7 @@ def process_participant(file_path, file_start, progress_callback=None):
|
||||
step_start = lap(step_start, timings, "Step 21")
|
||||
|
||||
# Step 22: Design Matrix
|
||||
raw_haemo, df_design_matrix, fig_design_matrix = make_design_matrix(
|
||||
raw_haemo, raw_haemo_dm, df_design_matrix, fig_design_matrix = make_design_matrix(
|
||||
raw_haemo=raw_haemo,
|
||||
resample=RESAMPLE,
|
||||
resample_freq=RESAMPLE_FREQ,
|
||||
@@ -5442,7 +5443,7 @@ def process_participant(file_path, file_start, progress_callback=None):
|
||||
step_start = lap(step_start, timings, "Step 22")
|
||||
|
||||
# Step 23: General Linear Model
|
||||
glm_est, fig_glm_topo = make_and_run_glm(raw_haemo, df_design_matrix, noise_model=NOISE_MODEL, bins=BINS, n_jobs=N_JOBS, verbosity=VERBOSITY)
|
||||
glm_est, fig_glm_topo = make_and_run_glm(raw_haemo_dm, df_design_matrix, noise_model=NOISE_MODEL, bins=BINS, n_jobs=N_JOBS, verbosity=VERBOSITY)
|
||||
_enqueue("GLM Topography", fig_glm_topo, png_queue)
|
||||
if progress_callback: progress_callback(23)
|
||||
logger.info("23")
|
||||
@@ -5508,117 +5509,51 @@ def sanitize_paths_for_pickle(raw_haemo, epochs):
|
||||
|
||||
|
||||
def functional_connectivity_spectral_epochs(
|
||||
epochs: DataFrame | None,
|
||||
epochs: Epochs,
|
||||
n_lines: int,
|
||||
vmin: float,
|
||||
) -> None:
|
||||
|
||||
# will crash without this load
|
||||
epochs.load_data()
|
||||
hbo_epochs = epochs.copy().pick(picks="hbo")
|
||||
data = hbo_epochs.get_data()
|
||||
names = hbo_epochs.ch_names
|
||||
sfreq = hbo_epochs.info["sfreq"]
|
||||
con = spectral_connectivity_epochs(
|
||||
data,
|
||||
method=["coh", "plv"],
|
||||
|
||||
con_coh = spectral_connectivity_epochs(
|
||||
hbo_epochs,
|
||||
method="coh",
|
||||
mode="multitaper",
|
||||
sfreq=sfreq,
|
||||
sfreq=hbo_epochs.info["sfreq"],
|
||||
fmin=0.04,
|
||||
fmax=0.2,
|
||||
faverage=True,
|
||||
verbose=True
|
||||
)
|
||||
|
||||
con_coh, con_plv = con
|
||||
|
||||
coh = con_coh.get_data(output="dense").squeeze()
|
||||
plv = con_plv.get_data(output="dense").squeeze()
|
||||
|
||||
coh = np.squeeze(con_coh.get_data(output="dense"))
|
||||
np.fill_diagonal(coh, 0)
|
||||
np.fill_diagonal(plv, 0)
|
||||
|
||||
plot_connectivity_circle(
|
||||
coh,
|
||||
names,
|
||||
hbo_epochs.ch_names,
|
||||
title="fNIRS Functional Connectivity (HbO - Coherence)",
|
||||
n_lines=n_lines,
|
||||
vmin=vmin
|
||||
)
|
||||
|
||||
|
||||
|
||||
def functional_connectivity_spectral_time(
|
||||
epochs: DataFrame | None,
|
||||
n_lines: int,
|
||||
vmin: float,
|
||||
) -> None:
|
||||
|
||||
# will crash without this load
|
||||
epochs.load_data()
|
||||
hbo_epochs = epochs.copy().pick(picks="hbo")
|
||||
data = hbo_epochs.get_data()
|
||||
names = hbo_epochs.ch_names
|
||||
sfreq = hbo_epochs.info["sfreq"]
|
||||
|
||||
freqs = np.linspace(0.04, 0.2, 10)
|
||||
n_cycles = freqs * 2
|
||||
|
||||
con = spectral_connectivity_time(
|
||||
data,
|
||||
freqs=freqs,
|
||||
method=["coh", "plv"],
|
||||
mode="multitaper",
|
||||
sfreq=sfreq,
|
||||
fmin=0.04,
|
||||
fmax=0.2,
|
||||
n_cycles=n_cycles,
|
||||
faverage=True,
|
||||
verbose=True
|
||||
)
|
||||
|
||||
con_coh, con_plv = con
|
||||
|
||||
coh = con_coh.get_data(output="dense").squeeze()
|
||||
plv = con_plv.get_data(output="dense").squeeze()
|
||||
|
||||
np.fill_diagonal(coh, 0)
|
||||
np.fill_diagonal(plv, 0)
|
||||
|
||||
plot_connectivity_circle(
|
||||
coh,
|
||||
names,
|
||||
title="fNIRS Functional Connectivity (HbO - Coherence)",
|
||||
n_lines=n_lines,
|
||||
vmin=vmin
|
||||
)
|
||||
|
||||
|
||||
|
||||
def functional_connectivity_envelope(
|
||||
epochs: DataFrame | None,
|
||||
epochs: Epochs,
|
||||
n_lines: int,
|
||||
vmin: float,
|
||||
) -> None:
|
||||
|
||||
# will crash without this load
|
||||
|
||||
epochs.load_data()
|
||||
hbo_epochs = epochs.copy().pick(picks="hbo")
|
||||
data = hbo_epochs.get_data()
|
||||
|
||||
hbo_epochs.filter(l_freq=0.04, h_freq=0.2, verbose=True)
|
||||
|
||||
env = envelope_correlation(
|
||||
data,
|
||||
hbo_epochs.get_data(),
|
||||
orthogonalize=False,
|
||||
absolute=True
|
||||
)
|
||||
env_data = env.get_data(output="dense")
|
||||
|
||||
env_corr = env_data.mean(axis=0)
|
||||
|
||||
env_corr = np.mean(env.get_data(output="dense"), axis=0)
|
||||
env_corr = np.squeeze(env_corr)
|
||||
|
||||
np.fill_diagonal(env_corr, 0)
|
||||
|
||||
plot_connectivity_circle(
|
||||
@@ -5630,107 +5565,149 @@ def functional_connectivity_envelope(
|
||||
)
|
||||
|
||||
|
||||
def functional_connectivity_spectral_time(
|
||||
epochs: Epochs,
|
||||
n_lines: int,
|
||||
vmin: float,
|
||||
) -> None:
|
||||
epochs.load_data()
|
||||
hbo_epochs = epochs.copy().pick(picks="hbo")
|
||||
|
||||
freqs = np.linspace(0.04, 0.2, 10)
|
||||
n_cycles = freqs * 2
|
||||
|
||||
con_coh = spectral_connectivity_time(
|
||||
hbo_epochs.get_data(),
|
||||
freqs=freqs,
|
||||
method="coh",
|
||||
mode="multitaper",
|
||||
sfreq=hbo_epochs.info["sfreq"],
|
||||
fmin=0.04,
|
||||
fmax=0.2,
|
||||
n_cycles=n_cycles,
|
||||
faverage=True,
|
||||
verbose=True
|
||||
)
|
||||
coh = np.squeeze(con_coh.get_data(output="dense"))
|
||||
if coh.ndim == 3:
|
||||
coh = coh.mean(axis=0)
|
||||
np.fill_diagonal(coh, 0)
|
||||
|
||||
plot_connectivity_circle(
|
||||
coh,
|
||||
hbo_epochs.ch_names,
|
||||
title="fNIRS Functional Connectivity (HbO - Coherence, Time-Resolved)",
|
||||
n_lines=n_lines,
|
||||
vmin=vmin
|
||||
)
|
||||
|
||||
|
||||
def functional_connectivity_betas(
|
||||
raw_hbo: BaseRaw,
|
||||
n_lines: int,
|
||||
vmin: float,
|
||||
event_name: str | None = None,
|
||||
*,
|
||||
drift_model: str = "cosine",
|
||||
drift_order: int = 1,
|
||||
apply_gsr: bool = True,
|
||||
min_effect_size: float = 0.7,
|
||||
alpha: float = 0.05,
|
||||
) -> None:
|
||||
|
||||
raw_hbo = raw_hbo.copy().pick(picks="hbo")
|
||||
onsets = raw_hbo.annotations.onset
|
||||
|
||||
# CRITICAL: Update the Raw object's annotations so the GLM sees unique events
|
||||
ann = raw_hbo.annotations
|
||||
new_desc = []
|
||||
ann.description = np.array([
|
||||
f"{desc}__trial_{i:03d}" for i, desc in enumerate(ann.description)
|
||||
])
|
||||
|
||||
for i, desc in enumerate(ann.description):
|
||||
new_desc.append(f"{desc}__trial_{i:03d}")
|
||||
|
||||
ann.description = np.array(new_desc)
|
||||
|
||||
|
||||
# shoudl use user defiuned!!!!
|
||||
design_matrix = make_first_level_design_matrix(
|
||||
raw=raw_hbo,
|
||||
hrf_model='fir',
|
||||
hrf_model="fir",
|
||||
fir_delays=np.arange(0, 12, 1),
|
||||
drift_model='cosine',
|
||||
drift_order=1
|
||||
drift_model=drift_model,
|
||||
drift_order=drift_order,
|
||||
)
|
||||
|
||||
|
||||
# 3. Run GLM & Extract Betas
|
||||
glm_results = run_glm(raw_hbo, design_matrix)
|
||||
betas = np.array(glm_results.theta())
|
||||
if betas.ndim == 3 and betas.shape[-1] == 1:
|
||||
betas = betas.squeeze(axis=-1)
|
||||
|
||||
reg_names = list(design_matrix.columns)
|
||||
|
||||
|
||||
n_channels = betas.shape[0]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 5. Find unique trial tags (optionally filtered by event)
|
||||
# ------------------------------------------------------------------
|
||||
trial_tags = sorted({
|
||||
col.split("_delay")[0]
|
||||
for col in reg_names
|
||||
if (
|
||||
("__trial_" in col)
|
||||
and (event_name is None or col.startswith(event_name + "__"))
|
||||
)
|
||||
if ("__trial_" in col) and (event_name is None or col.startswith(event_name + "__"))
|
||||
})
|
||||
|
||||
if len(trial_tags) == 0:
|
||||
raise ValueError(f"No trials found for event_name={event_name}")
|
||||
if len(trial_tags) < 4:
|
||||
raise ValueError(
|
||||
f"Only {len(trial_tags)} trials found for event_name={event_name}; "
|
||||
"need at least 4 to compute correlation degrees of freedom."
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 6. Build beta series (average across FIR delays per trial)
|
||||
# ------------------------------------------------------------------
|
||||
beta_series = np.zeros((n_channels, len(trial_tags)))
|
||||
for t_idx, tag in enumerate(trial_tags):
|
||||
col_idx = [j for j, col in enumerate(reg_names) if col.split("_delay")[0] == tag]
|
||||
beta_series[:, t_idx] = betas[:, col_idx].mean(axis=1)
|
||||
|
||||
global_signal = np.mean(beta_series, axis=0)
|
||||
beta_series_clean = np.zeros_like(beta_series)
|
||||
for i in range(n_channels):
|
||||
slope, _ = np.polyfit(global_signal, beta_series[i, :], 1)
|
||||
beta_series_clean[i, :] = beta_series[i, :] - (slope * global_signal)
|
||||
# Vectorized Global Signal Regression (GSR)
|
||||
if apply_gsr:
|
||||
global_signal = np.mean(beta_series, axis=0)
|
||||
A = np.vstack([global_signal, np.ones(len(global_signal))]).T
|
||||
# Solve least squares for all channels simultaneously
|
||||
params, _, _, _ = np.linalg.lstsq(A, beta_series.T, rcond=None)
|
||||
beta_series_clean = (beta_series.T - A @ params).T
|
||||
else:
|
||||
beta_series_clean = beta_series
|
||||
|
||||
# --- Vectorized correlation + analytic p-values (replaces the nested
|
||||
# pearsonr loop below) ---
|
||||
n_trials = beta_series_clean.shape[1]
|
||||
corr_matrix = np.corrcoef(beta_series_clean)
|
||||
|
||||
with np.errstate(divide='ignore', invalid='ignore'):
|
||||
t_stats = corr_matrix * np.sqrt((n_trials - 2) / (1 - corr_matrix ** 2))
|
||||
# Safe t-statistic calculation avoiding division by zero on diagonal
|
||||
corr_clipped = np.clip(corr_matrix, -0.999999, 0.999999)
|
||||
t_stats = corr_clipped * np.sqrt((n_trials - 2) / (1 - corr_clipped ** 2))
|
||||
p_matrix = 2 * t_dist.sf(np.abs(t_stats), df=n_trials - 2)
|
||||
np.fill_diagonal(p_matrix, 1.0) # diagonal r=1 -> nan/inf guarded explicitly
|
||||
np.fill_diagonal(p_matrix, 1.0)
|
||||
|
||||
triu = np.triu_indices(n_channels, k=1)
|
||||
flat_p = p_matrix[triu]
|
||||
reject, _, _, _ = multipletests(flat_p, method="fdr_bh", alpha=alpha)
|
||||
|
||||
reject, _ = multipletests(flat_p, method='fdr_bh', alpha=0.05)[:2]
|
||||
sig_corr_matrix = np.zeros_like(corr_matrix)
|
||||
|
||||
for idx, is_sig in enumerate(reject):
|
||||
r_val = corr_matrix[triu[0][idx], triu[1][idx]]
|
||||
# Only keep the absolute strongest connections
|
||||
if is_sig and abs(r_val) > 0.7:
|
||||
if is_sig and abs(r_val) > min_effect_size:
|
||||
sig_corr_matrix[triu[0][idx], triu[1][idx]] = r_val
|
||||
sig_corr_matrix[triu[1][idx], triu[0][idx]] = r_val
|
||||
|
||||
# 6. Plot
|
||||
gsr_tag = "GSR" if apply_gsr else "no GSR"
|
||||
plot_connectivity_circle(
|
||||
sig_corr_matrix,
|
||||
raw_hbo.ch_names,
|
||||
title="Strictly Filtered Connectivity (TDDR + GSR + Z-Score)",
|
||||
n_lines=None,
|
||||
vmin=0.7,
|
||||
title=f"Beta-Series Connectivity (FDR q<{alpha}, |r|>{min_effect_size}, {gsr_tag})",
|
||||
n_lines=n_lines,
|
||||
vmin=min_effect_size,
|
||||
vmax=1.0,
|
||||
colormap='hot' # Use 'hot' to make positive connections pop
|
||||
colormap="hot",
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def get_single_subject_beta_corr(raw_hbo, event_name=None, config=None):
|
||||
"""Processes one participant and returns their correlation matrix."""
|
||||
raw_hbo = raw_hbo.copy().pick(picks="hbo")
|
||||
|
||||
@@ -242,10 +242,10 @@ SECTIONS = [
|
||||
{"name": "H_TRANS_BANDWIDTH", "default": 0.002, "type": float, "depends_on": "FILTER", "advanced": True, "help": "Width of the upper transition band to prevent abrupt filter cutoff."},
|
||||
# {"name": "IIR_TYPE", "default": ["butterworth"], "type": list, "options": ["butterworth", "chebyshev1", "chebyshev2", "elliptic", "bessel"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "IIR", "advanced": True, "help": "IIR filter design."},
|
||||
# {"name": "IIR_ORDER", "default": 4, "type": int, "depends_on": "FILTER_ALGORITHM", "depends_value": "IIR", "advanced": True, "help": "Order of the IIR filter."},
|
||||
{"name": "FILTER_LENGTH", "default": "auto", "type": str, "depends_on": "FILTER_ALGORITHM", "depends_value": "FIR", "advanced": True, "help": "Length of the FIR filter. 'auto' allows automatic selection."},
|
||||
{"name": "FILTER_PHASE", "default": ["zero"], "type": list, "options": ["zero", "zero-double", "minimum", "minimum-half", "linear"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "FIR", "advanced": True, "help": "Phase response of the FIR filter."},
|
||||
{"name": "FIR_WINDOW", "default": ["hamming"], "type": list, "options": ["hamming", "hann", "blackman"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "FIR", "advanced": True, "help": "Window function used when designing the FIR filter."},
|
||||
{"name": "FIR_DESIGN", "default": ["firwin"], "type": list, "options": ["firwin", "firwin2"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "FIR", "advanced": True, "help": "Method used to design the FIR filter."},
|
||||
{"name": "FILTER_LENGTH", "default": "auto", "type": str, "depends_on": "FILTER_ALGORITHM", "depends_value": "fir", "advanced": True, "help": "Length of the FIR filter. 'auto' allows automatic selection."},
|
||||
{"name": "FILTER_PHASE", "default": ["zero"], "type": list, "options": ["zero", "zero-double", "minimum", "minimum-half", "linear"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "fir", "advanced": True, "help": "Phase response of the FIR filter."},
|
||||
{"name": "FIR_WINDOW", "default": ["hamming"], "type": list, "options": ["hamming", "hann", "blackman"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "fir", "advanced": True, "help": "Window function used when designing the FIR filter."},
|
||||
{"name": "FIR_DESIGN", "default": ["firwin"], "type": list, "options": ["firwin", "firwin2"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "fir", "advanced": True, "help": "Method used to design the FIR filter."},
|
||||
# {"name": "IIR_OUTPUT", "default": ["sos"], "type": list, "options": ["sos", "ba", "zpk"], "exclusive": True, "depends_on": "FILTER_ALGORITHM", "depends_value": "IIR", "advanced": True, "help": "Representation used for IIR filter coefficients."},
|
||||
# {"name": "PASSBAND_RIPPLE", "default": 1.0, "type": float, "depends_on": "IIR_TYPE", "depends_value": ["chebyshev1", "elliptic"], "advanced": True, "help": "Maximum allowed ripple in the passband (dB)."},
|
||||
# {"name": "STOPBAND_ATTENUATION", "default": 40.0, "type": float, "depends_on": "IIR_TYPE", "depends_value": ["chebyshev2", "elliptic"], "advanced": True, "help": "Minimum attenuation in the stopband (dB)."},
|
||||
@@ -258,8 +258,8 @@ SECTIONS = [
|
||||
"title": "Extracting Events",
|
||||
"params": [
|
||||
{"name": "EVENTS", "default": True, "type": bool, "advanced": True, "help": "Extract events from annotations for visualization and downstream event-based analysis."},
|
||||
{"name": "EVENT_ID", "default": "auto", "type": str, "advanced": True, "help": "Controls how annotation descriptions are converted into event identifiers. Use 'auto' for automatic event detection."},
|
||||
{"name": "EVENT_REGEX", "default": r"^(?![Bb][Aa][Dd]|[Ee][Dd][Gg][Ee]).*$", "type": str, "advanced": True, "help": "Regular expression used to select which annotations are converted into events. By default, bad and edge annotations are ignored."},
|
||||
{"name": "EVENT_ID", "default": "auto", "type": str, "depends_on": "EVENTS", "advanced": True, "help": "Controls how annotation descriptions are converted into event identifiers. Use 'auto' for automatic event detection."},
|
||||
{"name": "EVENT_REGEX", "default": r"^(?![Bb][Aa][Dd]|[Ee][Dd][Gg][Ee]).*$", "type": str, "depends_on": "EVENTS", "advanced": True, "help": "Regular expression used to select which annotations are converted into events. By default, bad and edge annotations are ignored."},
|
||||
# {"name": "EVENT_CHUNK_DURATION", "default": 0.0, "type": float, "advanced": True, "help": "If provided, creates repeated events at this interval within longer annotations instead of only using annotation onset times."},
|
||||
]
|
||||
},
|
||||
@@ -642,7 +642,7 @@ class MainApplication(QMainWindow):
|
||||
self.left_v_splitter.setChildrenCollapsible(False)
|
||||
self.left_v_splitter.setMinimumWidth(460)
|
||||
|
||||
top_left_container = QGroupBox("File information")
|
||||
top_left_container = QGroupBox("File Information")
|
||||
top_left_container.setStyleSheet("QGroupBox { font-weight: bold; }")
|
||||
top_left_container.setMinimumHeight(240)
|
||||
top_left_layout = QHBoxLayout(top_left_container)
|
||||
@@ -2200,10 +2200,11 @@ class MainApplication(QMainWindow):
|
||||
return
|
||||
|
||||
for widget in list(QApplication.topLevelWidgets()):
|
||||
if widget is not self:
|
||||
if not widget.close():
|
||||
event.ignore()
|
||||
return
|
||||
if widget is not self and widget.isWindow() and not isinstance(widget, QMenu):
|
||||
try:
|
||||
widget.close()
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
if hasattr(self, 'loading_session_id'):
|
||||
self.loading_session_id += 1
|
||||
|
||||
@@ -0,0 +1,712 @@
|
||||
"""
|
||||
Filename: main_unit_tests.py
|
||||
Description: Unit tests for functionality validation
|
||||
|
||||
Author: Tyler de Zeeuw
|
||||
License: GPL-3.0
|
||||
"""
|
||||
|
||||
# Built-in imports
|
||||
import os
|
||||
import pickle
|
||||
import configparser
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
# External library imports
|
||||
import pytest
|
||||
from PySide6.QtWidgets import QApplication, QMenu
|
||||
|
||||
import main
|
||||
from updater import LocalPendingUpdateCheckThread, UpdateCheckThread
|
||||
|
||||
'''
|
||||
These test fluff currently. Very basic "Does the UI exist?" and not the functionality.
|
||||
main_test.py::test_save_project_actions_pass_correct_ask_parameter
|
||||
main_test.py::test_main_window_opens
|
||||
main_test.py::test_file_recent_submenus_exist
|
||||
main_test.py::test_view_reset_layout
|
||||
main_test.py::test_preferences_actions[2D Data Bypass-2d_data_bypass]
|
||||
main_test.py::test_preferences_actions[Incompatible Save Bypass-incompatible_save_bypass]
|
||||
main_test.py::test_preferences_actions[Missing Events Bypass-missing_events_bypass]
|
||||
main_test.py::test_preferences_actions[Analysis Clearing Bypass-analysis_clearing_bypass]
|
||||
main_test.py::test_preferences_actions[Folding Bypass-folding_bypass]
|
||||
main_test.py::test_preferences_actions[Show Advanced Parameters-advanced_parameters]
|
||||
'''
|
||||
|
||||
# ---------------------- HELPERS ----------------------
|
||||
def get_menu_by_title(menu_bar, title):
|
||||
"""Return the first QMenu with the given title, or None."""
|
||||
for menu in menu_bar.findChildren(QMenu):
|
||||
if menu.title() == title:
|
||||
return menu
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------- FIXTURES ----------------------
|
||||
@pytest.fixture(autouse=True)
|
||||
def disable_updater_threads():
|
||||
"""Stops updater threads from running asynchronously during qtbot teardown."""
|
||||
with patch.object(UpdateCheckThread, "start", return_value=None), \
|
||||
patch.object(LocalPendingUpdateCheckThread, "start", return_value=None):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_app_globals():
|
||||
"""Initializes global configuration objects that main.py expects at runtime."""
|
||||
main.cfg_path = os.path.join(os.getcwd(), f"{main.APP_NAME}.cfg")
|
||||
main.file_cfg = configparser.ConfigParser()
|
||||
main.ref_cfg = configparser.ConfigParser()
|
||||
|
||||
if hasattr(main, "config_init"):
|
||||
main.config_init()
|
||||
|
||||
|
||||
# ===================== FILE MENU =====================
|
||||
def test_main_window_opens(qtbot):
|
||||
"""Test 1: Verify MainApplication launches and becomes visible."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert window.isVisible()
|
||||
|
||||
|
||||
def test_open_file_dialog_with_mne_mock(qtbot, tmp_path):
|
||||
dummy_snirf = tmp_path / "test_data.snirf"
|
||||
dummy_snirf.write_text("dummy content")
|
||||
expected_path = os.path.normpath(str(dummy_snirf))
|
||||
|
||||
# Mock MNE Raw object returned by read_raw_snirf
|
||||
mock_raw = MagicMock()
|
||||
mock_raw.info = {"meas_date": "2026-01-01", "ch_names": ["S1_D1 760"], "dig": None}
|
||||
mock_raw.ch_names = ["S1_D1 760"]
|
||||
mock_raw.annotations = []
|
||||
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch("PySide6.QtWidgets.QFileDialog.getOpenFileName", return_value=(expected_path, "SNIRF Files (*.snirf)")), \
|
||||
patch("mne.io.snirf.read_raw_snirf", return_value=mock_raw), \
|
||||
patch("project_manager.source_detector_distances", return_value=[0.03]):
|
||||
|
||||
window.project_manager.open_file_dialog()
|
||||
|
||||
assert expected_path in window.selected_paths
|
||||
assert expected_path in window.bubble_widgets
|
||||
|
||||
window.files_are_dirty = False
|
||||
window.is_saved = True
|
||||
|
||||
|
||||
def test_open_folder_dialog(qtbot, tmp_path):
|
||||
"""Verify that open_folder_dialog recursively finds and loads all .snirf files."""
|
||||
sub_dir = tmp_path / "sub_folder"
|
||||
sub_dir.mkdir()
|
||||
|
||||
file1 = tmp_path / "root_file.snirf"
|
||||
file2 = sub_dir / "nested_file.snirf"
|
||||
ignored_file = tmp_path / "notes.txt"
|
||||
|
||||
file1.write_text("dummy snirf 1")
|
||||
file2.write_text("dummy snirf 2")
|
||||
ignored_file.write_text("text note")
|
||||
|
||||
folder_path = str(tmp_path)
|
||||
expected_paths = {
|
||||
os.path.normpath(str(file1)),
|
||||
os.path.normpath(str(file2)),
|
||||
}
|
||||
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch(
|
||||
"PySide6.QtWidgets.QFileDialog.getExistingDirectory",
|
||||
return_value=folder_path,
|
||||
):
|
||||
window.project_manager.open_folder_dialog()
|
||||
|
||||
loaded_paths = set(window.selected_paths)
|
||||
assert expected_paths.issubset(loaded_paths)
|
||||
assert os.path.normpath(str(ignored_file)) not in loaded_paths
|
||||
|
||||
window.files_are_dirty = False
|
||||
window.is_saved = True
|
||||
|
||||
|
||||
def test_load_project_dialog(qtbot, tmp_path):
|
||||
"""Verify loading a valid pickled .flare project restores application state."""
|
||||
project_file = tmp_path / "test_project.flare"
|
||||
|
||||
dummy_project_data = {
|
||||
"version": "1.1.7",
|
||||
"file_metadata": {"rel_sample.snirf": {"channels": 4}},
|
||||
"file_parameters": {"rel_sample.snirf": {"AGE": "25", "SEX": "M", "HAND": "R", "GROUP": "A"}},
|
||||
"roi_channel_map_dict": {},
|
||||
"file_list": ["rel_sample.snirf"],
|
||||
"progress_states": {"rel_sample.snirf": "completed"},
|
||||
"current_ui_params": {},
|
||||
}
|
||||
|
||||
with open(project_file, "wb") as f:
|
||||
pickle.dump(dummy_project_data, f)
|
||||
|
||||
file_path = str(project_file)
|
||||
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
if not hasattr(main, "DATA_SCHEMA"):
|
||||
main.DATA_SCHEMA = []
|
||||
|
||||
with patch(
|
||||
"PySide6.QtWidgets.QFileDialog.getOpenFileName",
|
||||
return_value=(file_path, "FLARE Project (*.flare)"),
|
||||
), patch("PySide6.QtWidgets.QMessageBox.information") as mock_info, patch.object(
|
||||
window, "show_files_as_bubbles_from_list"
|
||||
):
|
||||
|
||||
window.project_manager.load_project_dialog()
|
||||
|
||||
assert window.current_project_path == file_path
|
||||
mock_info.assert_called_once()
|
||||
|
||||
|
||||
def test_load_project_incompatible_version(qtbot, tmp_path):
|
||||
"""Verify that a missing required key triggers an incompatibility error."""
|
||||
invalid_file = tmp_path / "corrupt.flare"
|
||||
|
||||
incomplete_data = {
|
||||
"file_metadata": {},
|
||||
"file_parameters": {},
|
||||
"roi_channel_map_dict": {},
|
||||
}
|
||||
|
||||
with open(invalid_file, "wb") as f:
|
||||
pickle.dump(incomplete_data, f)
|
||||
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch("PySide6.QtWidgets.QMessageBox.critical") as mock_critical, \
|
||||
patch("PySide6.QtWidgets.QMessageBox.warning") as mock_warning:
|
||||
|
||||
window.project_manager.load_project(str(invalid_file))
|
||||
|
||||
assert mock_critical.called or mock_warning.called, "Expected a QMessageBox warning or critical popup."
|
||||
assert len(getattr(window, "selected_paths", [])) == 0
|
||||
|
||||
|
||||
def test_save_project_no_data_shows_warning(qtbot):
|
||||
"""Verify saving an empty project triggers a 'no data to save' warning."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch("PySide6.QtWidgets.QMessageBox.warning") as mock_warning:
|
||||
window.project_manager.save_project(ask=True)
|
||||
|
||||
mock_warning.assert_called_once()
|
||||
assert "no data" in mock_warning.call_args[0][2].lower()
|
||||
|
||||
|
||||
def test_save_project_success(qtbot, tmp_path):
|
||||
"""Verify saving a loaded project outputs a valid pickled .flare file."""
|
||||
save_file_path = tmp_path / "test_project.flare"
|
||||
dummy_snirf_path = str(tmp_path / "sample_subject.snirf")
|
||||
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
# 1. Satisfy 'has_files' check
|
||||
window.selected_paths = [dummy_snirf_path]
|
||||
|
||||
# 2. Add mock bubble widget so step 4 populates file_list
|
||||
mock_bubble = MagicMock()
|
||||
mock_bubble.file_path = dummy_snirf_path
|
||||
mock_bubble.current_step = 0
|
||||
window.bubble_widgets = {dummy_snirf_path: mock_bubble}
|
||||
|
||||
with patch("PySide6.QtWidgets.QFileDialog.getSaveFileName", return_value=(str(save_file_path), "FLARE Project (*.flare)")), \
|
||||
patch("PySide6.QtWidgets.QMessageBox.information") as mock_info:
|
||||
|
||||
window.project_manager.save_project(ask=True)
|
||||
|
||||
# Wait for SaveProjectThread to finish writing to disk
|
||||
qtbot.waitUntil(lambda: save_file_path.exists(), timeout=3000)
|
||||
mock_info.assert_called_once()
|
||||
|
||||
# 3. Verify the saved payload structure
|
||||
assert save_file_path.is_file()
|
||||
with open(save_file_path, "rb") as f:
|
||||
data = pickle.load(f)
|
||||
assert "version" in data
|
||||
# file_list contains relative paths normalized by sanitize()
|
||||
assert "sample_subject.snirf" in data["file_list"]
|
||||
|
||||
# Reset dirty state so teardown completes cleanly
|
||||
window.files_are_dirty = False
|
||||
window.is_saved = True
|
||||
|
||||
|
||||
def test_save_project_actions_pass_correct_ask_parameter(qtbot):
|
||||
"""
|
||||
Verify that the 'Save Project...' action calls save_project(ask=False)
|
||||
and 'Save Project As...' calls save_project(ask=True).
|
||||
"""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
file_menu = get_menu_by_title(window.menuBar(), "File")
|
||||
assert file_menu is not None, "File menu not found"
|
||||
|
||||
save_action = next(a for a in file_menu.actions() if a.text() == "Save Project...")
|
||||
save_as_action = next(a for a in file_menu.actions() if a.text() == "Save Project As...")
|
||||
|
||||
with patch.object(window.project_manager, 'save_project') as mock_save:
|
||||
save_action.trigger()
|
||||
mock_save.assert_called_once_with(ask=False)
|
||||
mock_save.reset_mock()
|
||||
|
||||
save_as_action.trigger()
|
||||
mock_save.assert_called_once_with(ask=True)
|
||||
|
||||
|
||||
def test_file_exit(qtbot):
|
||||
"""Verify that File → Exit calls QApplication.quit()."""
|
||||
with patch.object(QApplication, 'quit') as mock_quit:
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
file_menu = get_menu_by_title(window.menuBar(), "File")
|
||||
assert file_menu is not None, "File menu not found"
|
||||
|
||||
exit_action = next(a for a in file_menu.actions() if a.text() == "Exit")
|
||||
exit_action.trigger()
|
||||
mock_quit.assert_called_once()
|
||||
|
||||
|
||||
def test_file_recent_submenus_exist(qtbot):
|
||||
"""Verify that the 'Recent Files' and 'Recent Projects' submenus are created."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
file_menu = get_menu_by_title(window.menuBar(), "File")
|
||||
assert file_menu is not None, "File menu not found"
|
||||
|
||||
recent_files_action = next((a for a in file_menu.actions() if a.text() == "Recent Files"), None)
|
||||
assert recent_files_action is not None
|
||||
recent_files_menu = recent_files_action.menu()
|
||||
assert recent_files_menu is not None
|
||||
|
||||
recent_projects_action = next((a for a in file_menu.actions() if a.text() == "Recent Projects"), None)
|
||||
assert recent_projects_action is not None
|
||||
recent_projects_menu = recent_projects_action.menu()
|
||||
assert recent_projects_menu is not None
|
||||
|
||||
|
||||
# ===================== EDIT MENU =====================
|
||||
def test_edit_cut(qtbot):
|
||||
"""Verify Edit → Cut calls top_left_widget.cut()."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
edit_menu = get_menu_by_title(window.menuBar(), "Edit")
|
||||
assert edit_menu is not None, "Edit menu not found"
|
||||
|
||||
cut_action = next(a for a in edit_menu.actions() if a.text() == "Cut")
|
||||
with patch.object(window.top_left_widget, 'cut') as mock_cut:
|
||||
cut_action.trigger()
|
||||
mock_cut.assert_called_once()
|
||||
|
||||
|
||||
def test_edit_copy(qtbot):
|
||||
"""Verify Edit → Copy calls top_left_widget.copy()."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
edit_menu = get_menu_by_title(window.menuBar(), "Edit")
|
||||
assert edit_menu is not None, "Edit menu not found"
|
||||
|
||||
copy_action = next(a for a in edit_menu.actions() if a.text() == "Copy")
|
||||
with patch.object(window.top_left_widget, 'copy') as mock_copy:
|
||||
copy_action.trigger()
|
||||
mock_copy.assert_called_once()
|
||||
|
||||
|
||||
def test_edit_paste(qtbot):
|
||||
"""Verify Edit → Paste calls top_left_widget.paste()."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
edit_menu = get_menu_by_title(window.menuBar(), "Edit")
|
||||
assert edit_menu is not None, "Edit menu not found"
|
||||
|
||||
paste_action = next(a for a in edit_menu.actions() if a.text() == "Paste")
|
||||
with patch.object(window.top_left_widget, 'paste') as mock_paste:
|
||||
paste_action.trigger()
|
||||
mock_paste.assert_called_once()
|
||||
|
||||
|
||||
# ===================== VIEW MENU =====================
|
||||
def test_view_toggle_statusbar(qtbot):
|
||||
"""Verify View → Toggle Status Bar toggles visibility and calls _update_config_setting."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
view_menu = get_menu_by_title(window.menuBar(), "View")
|
||||
assert view_menu is not None, "View menu not found"
|
||||
|
||||
toggle_action = next(a for a in view_menu.actions() if a.text() == "Toggle Status Bar")
|
||||
assert toggle_action.isCheckable() is True
|
||||
|
||||
# Initially checked (True in create_menu_bar)
|
||||
assert toggle_action.isChecked() is True
|
||||
assert window.statusbar.isVisible() is True
|
||||
|
||||
# Trigger once to hide
|
||||
with patch.object(window, '_update_config_setting') as mock_update:
|
||||
toggle_action.trigger()
|
||||
assert not toggle_action.isChecked()
|
||||
assert not window.statusbar.isVisible()
|
||||
mock_update.assert_called_once_with("View", "status_bar", False)
|
||||
|
||||
# Trigger again to show
|
||||
with patch.object(window, '_update_config_setting') as mock_update:
|
||||
toggle_action.trigger()
|
||||
assert toggle_action.isChecked() is True
|
||||
assert window.statusbar.isVisible() is True
|
||||
mock_update.assert_called_once_with("View", "status_bar", True)
|
||||
|
||||
|
||||
def test_view_reset_layout(qtbot):
|
||||
"""Verify View → Reset Window Layout calls apply_splitter_ratios."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
view_menu = get_menu_by_title(window.menuBar(), "View")
|
||||
assert view_menu is not None, "View menu not found"
|
||||
|
||||
reset_action = next(a for a in view_menu.actions() if a.text() == "Reset Window Layout")
|
||||
with patch.object(window, 'apply_splitter_ratios') as mock_apply:
|
||||
reset_action.trigger()
|
||||
mock_apply.assert_called_once()
|
||||
|
||||
|
||||
# ===================== OPTIONS MENU =====================
|
||||
def test_about_window_opens(qtbot):
|
||||
"""Verify AboutWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "about", None) is None
|
||||
window.about_window()
|
||||
|
||||
assert window.about is not None
|
||||
assert window.about.isVisible() is True
|
||||
|
||||
first_instance = window.about
|
||||
window.about_window()
|
||||
assert window.about is first_instance
|
||||
|
||||
|
||||
def test_user_guide_window_opens(qtbot):
|
||||
"""Verify UserGuideWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "help", None) is None
|
||||
window.user_guide()
|
||||
|
||||
assert window.help is not None
|
||||
assert window.help.isVisible() is True
|
||||
|
||||
first_instance = window.help
|
||||
window.user_guide()
|
||||
assert window.help is first_instance
|
||||
|
||||
|
||||
def test_show_update_changelog(qtbot):
|
||||
"""Verify WelcomeDialog is instantiated and shown."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch.object(main, "WelcomeDialog") as mock_dialog_cls:
|
||||
mock_dialog_instance = MagicMock()
|
||||
mock_dialog_cls.return_value = mock_dialog_instance
|
||||
|
||||
window.show_update_changelog()
|
||||
|
||||
mock_dialog_cls.assert_called_once_with(window, direct=False)
|
||||
mock_dialog_instance.show.assert_called_once()
|
||||
|
||||
|
||||
def test_group_metadata_no_data_shows_msgbox(qtbot):
|
||||
"""Verify group_metadata triggers an information QMessageBox when file_metadata is empty."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
window.file_metadata = {}
|
||||
|
||||
with patch("PySide6.QtWidgets.QMessageBox.information") as mock_msgbox:
|
||||
window.group_metadata()
|
||||
|
||||
mock_msgbox.assert_called_once()
|
||||
assert "No Data" in mock_msgbox.call_args[0]
|
||||
|
||||
|
||||
def test_group_metadata_with_data_applies_mappings(qtbot):
|
||||
"""Verify group_metadata opens GroupAssignmentDialog and executes _apply_group_mappings on success."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
window.file_metadata = {"sub-01.snirf": {"age": "25"}}
|
||||
mock_result = ("Age", {"sub-01.snirf": "GroupA"})
|
||||
|
||||
with patch.object(main.GroupAssignmentDialog, "run", return_value=mock_result), \
|
||||
patch.object(window, "_apply_group_mappings") as mock_apply:
|
||||
|
||||
window.group_metadata()
|
||||
|
||||
mock_apply.assert_called_once_with({"sub-01.snirf": "GroupA"}, field_name="Age")
|
||||
|
||||
|
||||
def test_manual_check_for_updates(qtbot):
|
||||
"""Verify Options → Check for Updates triggers the updater method."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
options_menu = get_menu_by_title(window.menuBar(), "Options")
|
||||
assert options_menu is not None, "Options menu not found"
|
||||
|
||||
update_action = next(a for a in options_menu.actions() if a.text() == "Check for Updates")
|
||||
assert update_action is not None
|
||||
assert update_action.isEnabled() is True
|
||||
|
||||
# Patch the updater's manual_check_for_updates method
|
||||
with patch.object(window.updater, 'manual_check_for_updates') as mock_method:
|
||||
update_action.trigger()
|
||||
mock_method.assert_called_once()
|
||||
|
||||
|
||||
def test_update_optode_positions_opens(qtbot):
|
||||
"""Verify UpdateOptodesWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "optodes", None) is None
|
||||
window.update_optode_positions()
|
||||
|
||||
assert window.optodes is not None
|
||||
assert window.optodes.isVisible() is True
|
||||
|
||||
first_instance = window.optodes
|
||||
window.update_optode_positions()
|
||||
assert window.optodes is first_instance
|
||||
|
||||
|
||||
def test_update_event_markers_opens(qtbot):
|
||||
"""Verify UpdateEventsWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "events", None) is None
|
||||
window.update_event_markers()
|
||||
|
||||
assert window.events is not None
|
||||
assert window.events.isVisible() is True
|
||||
|
||||
first_instance = window.events
|
||||
window.update_event_markers()
|
||||
assert window.events is first_instance
|
||||
|
||||
|
||||
def test_update_event_markers_blazes_opens(qtbot):
|
||||
"""Verify UpdateEventsBlazesWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "events_blazes", None) is None
|
||||
window.update_event_markers_blazes()
|
||||
|
||||
assert window.events_blazes is not None
|
||||
assert window.events_blazes.isVisible() is True
|
||||
|
||||
first_instance = window.events_blazes
|
||||
window.update_event_markers_blazes()
|
||||
assert window.events_blazes is first_instance
|
||||
|
||||
|
||||
def test_reset_to_default_configuration_user_cancels(qtbot):
|
||||
"""Verify nothing is reset when the user clicks 'No' on the prompt."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch("main.QMessageBox.question", return_value=main.QMessageBox.StandardButton.No), \
|
||||
patch("main.open") as mock_open, \
|
||||
patch.object(window, "sync_app_with_config") as mock_sync:
|
||||
|
||||
window.reset_to_default_configuration()
|
||||
|
||||
mock_open.assert_not_called()
|
||||
mock_sync.assert_not_called()
|
||||
|
||||
|
||||
def test_reset_to_default_configuration_success(qtbot):
|
||||
"""Verify file write, widget resets, config sync, and singleShot timer call when confirmed."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
# Mock child ParamSection widgets
|
||||
mock_section1 = MagicMock()
|
||||
mock_section2 = MagicMock()
|
||||
|
||||
with patch("main.QMessageBox.question", return_value=main.QMessageBox.StandardButton.Yes), \
|
||||
patch("main.open") as mock_open, \
|
||||
patch("main.file_cfg") as mock_cfg, \
|
||||
patch.object(window, "findChildren", return_value=[mock_section1, mock_section2]), \
|
||||
patch.object(window, "sync_app_with_config") as mock_sync, \
|
||||
patch.object(window, "update_sections") as mock_update, \
|
||||
patch("main.QTimer.singleShot") as mock_timer:
|
||||
|
||||
window.reset_to_default_configuration()
|
||||
|
||||
# Check file overwrite and parser reload
|
||||
mock_open.assert_called_once()
|
||||
mock_cfg.read.assert_called_once_with(main.cfg_path)
|
||||
|
||||
# Check section UI resets and app syncing
|
||||
mock_section1.reset_to_defaults.assert_called_once()
|
||||
mock_section2.reset_to_defaults.assert_called_once()
|
||||
mock_sync.assert_called_once()
|
||||
mock_update.assert_called_once_with(0)
|
||||
|
||||
# Verify post-reset dialog singleShot queue
|
||||
mock_timer.assert_called_once_with(100, window._show_reset_success_dialog)
|
||||
|
||||
|
||||
def test_reset_to_default_configuration_file_error_fallback(qtbot):
|
||||
"""Verify fallback to in-memory read_string when file writing raises an Exception."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
with patch("main.QMessageBox.question", return_value=main.QMessageBox.StandardButton.Yes), \
|
||||
patch("main.open", side_effect=PermissionError("Access denied")), \
|
||||
patch("main.file_cfg") as mock_cfg, \
|
||||
patch.object(window, "sync_app_with_config"), \
|
||||
patch.object(window, "update_sections"), \
|
||||
patch("main.QTimer.singleShot"):
|
||||
|
||||
window.reset_to_default_configuration()
|
||||
|
||||
# Verify fallback read_string execution
|
||||
mock_cfg.read_string.assert_called_once_with(main.DEFAULT_CONFIG)
|
||||
|
||||
|
||||
def test_show_reset_success_dialog(qtbot):
|
||||
"""Verify success dialog pops up and statusbar updates."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
window.statusbar = MagicMock()
|
||||
|
||||
with patch("main.QMessageBox.information") as mock_info:
|
||||
window._show_reset_success_dialog()
|
||||
|
||||
mock_info.assert_called_once_with(
|
||||
window,
|
||||
"Reset Successful",
|
||||
"All application settings have been successfully restored to their default values."
|
||||
)
|
||||
window.statusbar.showMessage.assert_called_once_with(
|
||||
"All settings have been reset to their default values.", 5000
|
||||
)
|
||||
|
||||
|
||||
# ===================== PREFERENCES MENU =====================
|
||||
@pytest.mark.parametrize("action_text, config_key", [
|
||||
("2D Data Bypass", "2d_data_bypass"),
|
||||
("Incompatible Save Bypass", "incompatible_save_bypass"),
|
||||
("Missing Events Bypass", "missing_events_bypass"),
|
||||
("Analysis Clearing Bypass", "analysis_clearing_bypass"),
|
||||
("Folding Bypass", "folding_bypass"),
|
||||
("Show Advanced Parameters", "advanced_parameters"),
|
||||
])
|
||||
def test_preferences_actions(qtbot, action_text, config_key):
|
||||
"""
|
||||
Verify each Preferences action toggles checked state and updates config.
|
||||
Uses the current checked state as a starting point and verifies toggling.
|
||||
"""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
pref_menu = get_menu_by_title(window.menuBar(), "Preferences")
|
||||
assert pref_menu is not None, "Preferences menu not found"
|
||||
|
||||
action = next(a for a in pref_menu.actions() if a.text() == action_text)
|
||||
assert action.isCheckable() is True
|
||||
|
||||
# Record the initial state
|
||||
initial_checked = action.isChecked()
|
||||
|
||||
# Trigger once → state should toggle
|
||||
with patch.object(window, '_update_config_setting') as mock_update:
|
||||
action.trigger()
|
||||
assert action.isChecked() == (not initial_checked)
|
||||
mock_update.assert_called_once_with("Preferences", config_key, not initial_checked)
|
||||
|
||||
# Trigger again → should toggle back to initial
|
||||
with patch.object(window, '_update_config_setting') as mock_update:
|
||||
action.trigger()
|
||||
assert action.isChecked() == initial_checked
|
||||
mock_update.assert_called_once_with("Preferences", config_key, initial_checked)
|
||||
|
||||
|
||||
# ===================== TERMINAL MENU =====================
|
||||
def test_terminal_gui_opens(qtbot):
|
||||
"""Verify TerminalWindow opens and prevents duplicate instances."""
|
||||
window = main.MainApplication()
|
||||
qtbot.addWidget(window)
|
||||
window.show()
|
||||
|
||||
assert getattr(window, "terminal", None) is None
|
||||
window.terminal_gui()
|
||||
|
||||
assert window.terminal is not None
|
||||
assert window.terminal.isVisible() is True
|
||||
|
||||
first_instance = window.terminal
|
||||
window.terminal_gui()
|
||||
assert window.terminal is first_instance
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -87,6 +87,24 @@ PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
|
||||
}
|
||||
|
||||
|
||||
DESCRIPTION = """0. Spectral Coherence (functional_connectivity_spectral_epochs)
|
||||
\nTests for frequency-domain phase synchronization between channel pairs in a specific oscillatory band (0.04-0.2 Hz) across epoched data using multitaper spectral estimation. A significant connection means two brain regions share consistent, synchronized oscillatory phase dynamics across epochs, reflecting steady-state functional coupling. It does not tell you when during the epoch the interaction occurred (as time is integrated out), nor does it guarantee the interaction is neural, as shared systemic vascular oscillations or motion artifacts can drive spurious high coherence across distant sensors.
|
||||
\nIf connectivity appears lower or sparser than expected, common causes include: non-stationarity within the epoch (phase relationships that shift rapidly over time cancel out when averaged across the whole window); trial-to-trial timing jitter; or applying overly stringent edge thresholding or FDR correction across all unique channel pairs.
|
||||
|
||||
\n1. Envelope Correlation (functional_connectivity_envelope)
|
||||
\nExtracts the Hilbert amplitude envelope from bandpass-filtered signals (0.04-0.2 Hz) to measure slow amplitude power correlations across time within epoched data. A significant result indicates that the overall energy profiles or activation magnitudes of two regions co-vary over time, independent of sub-second phase locking. It says nothing about fast phase interactions or exact event-locked timing, and its power is heavily degraded if epoch lengths are too short (under ~10-15s) to capture multiple complete cycles of low-frequency hemodynamic fluctuations.
|
||||
\nIf this method underperforms compared to phase-based coherence, the most likely explanation is that your trial window is too brief for robust envelope extraction, or that the functional coupling between regions is purely phase-locked rather than power-coupled. Additionally, uncorrected global motion or systemic arterial pressure shifts can globally inflate envelope correlations across the whole head.
|
||||
|
||||
\n2. Time-Resolved Spectral Coherence (functional_connectivity_spectral_time)
|
||||
\nUses continuous Morlet wavelet time-frequency decomposition across multiple frequencies (0.04-0.2 Hz) to track how spectral coherence between channel pairs dynamically evolves over the duration of a trial. A significant result pinpoints the exact temporal window within a trial where functional coupling emerges or dissolves (e.g., during stimulus encoding vs. motor execution). It demands precise, jitter-free stimulus onset triggers and carries high computational complexity; it is also susceptible to wavelet edge artifacts at the start and end of epoch windows.
|
||||
\nIf expected temporal connectivity changes fail to emerge, check whether trial-to-trial onset latency variability across subjects is smearing the time-resolved average, or if the chosen wavelet cycle parameter (n_cycles) is oversmoothing short-lived, transient phase-coupling events.
|
||||
|
||||
\n3. Beta-Series Correlation (functional_connectivity_betas)
|
||||
\nFits a General Linear Model (GLM) using a flexible Finite Impulse Response (FIR) basis set to estimate trial-by-trial activation magnitudes (betas), optionally applies Global Signal Regression (GSR) to strip head-wide systemic noise, and correlates those beta series across events with FDR (q < alpha) and effect-size thresholding. A significant connection means that when Region A responds more strongly on a given trial, Region B also responds more strongly, isolating true task-evoked co-activation from background resting-state noise. It requires at least 4 (ideally 15+) repeated trials per condition to establish degrees of freedom for the correlation t-test, and relies heavily on a correctly specified trial annotation structure.
|
||||
\nIf this test returns no significant edges, the primary culprit is typically insufficient trial count (leading to severely underpowered degrees of freedom), poorly separated trials that induce severe multicollinearity in the FIR design matrix, or applying GSR when the underlying neural effect itself is diffuse, causing true network correlations to be over-regressed.
|
||||
"""
|
||||
|
||||
|
||||
class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidget):
|
||||
def __init__(
|
||||
self,
|
||||
@@ -102,7 +120,7 @@ class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidg
|
||||
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. "
|
||||
"By clicking OK, you accept that the images generated may not be factual.")
|
||||
|
||||
self.setup_participant_ui(["0 (Spectral Connectivity Epochs)", "1 (Envelope Correlation)", "2 (Betas)", "3 (Spectral Connectivity Epochs)",])
|
||||
self.setup_participant_ui(["0 (Spectral Connectivity Epochs)", "1 (Envelope Correlation)", "2 (Betas)", "3 (Spectral Connectivity Epochs)"], placeholder_text=DESCRIPTION)
|
||||
|
||||
|
||||
def process_request(self):
|
||||
|
||||
@@ -1877,7 +1877,8 @@ class InterGroupUIMixin:
|
||||
class ParticipantUIMixin:
|
||||
def setup_participant_ui(
|
||||
self,
|
||||
index_texts: Sequence[str]
|
||||
index_texts: Sequence[str],
|
||||
placeholder_text: str = ""
|
||||
) -> None:
|
||||
|
||||
# Create mappings: file_path -> participant label and dropdown display text
|
||||
@@ -1890,9 +1891,9 @@ class ParticipantUIMixin:
|
||||
self.participant_map[file_path] = short_label
|
||||
self.participant_dropdown_items.append(display_label)
|
||||
|
||||
self.layout = QVBoxLayout(self)
|
||||
self.main_layout = QVBoxLayout(self)
|
||||
self.top_bar = QHBoxLayout()
|
||||
self.layout.addLayout(self.top_bar)
|
||||
self.main_layout.addLayout(self.top_bar)
|
||||
|
||||
self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items)
|
||||
self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label)
|
||||
@@ -1917,12 +1918,16 @@ class ParticipantUIMixin:
|
||||
self.top_bar.addWidget(self.image_index_dropdown)
|
||||
self.top_bar.addWidget(self.submit_button)
|
||||
|
||||
self.scroll = QScrollArea()
|
||||
self.scroll.setWidgetResizable(True)
|
||||
self.scroll_area = QScrollArea()
|
||||
self.scroll_area.setWidgetResizable(True)
|
||||
self.scroll_content = QWidget()
|
||||
self.grid_layout = QGridLayout(self.scroll_content)
|
||||
self.scroll.setWidget(self.scroll_content)
|
||||
self.layout.addWidget(self.scroll)
|
||||
self.scroll_area.setWidget(self.scroll_content)
|
||||
self.placeholder_label = QLabel(placeholder_text)
|
||||
self.grid_layout.addWidget(self.placeholder_label, 0, 0)
|
||||
self.placeholder_label.setWordWrap(True)
|
||||
self.placeholder_label.setScaledContents(True)
|
||||
self.main_layout.addWidget(self.scroll_area)
|
||||
|
||||
self.thumb_size = QSize(280, 180)
|
||||
self.showMaximized()
|
||||
|
||||
+48
-2
@@ -8,11 +8,13 @@ License: GPL-3.0
|
||||
"""
|
||||
|
||||
# Built-in imports
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
# External library imports
|
||||
from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit
|
||||
from PySide6.QtCore import Qt, QThread, Signal
|
||||
from PySide6.QtCore import QProcess, Qt, QThread, Signal
|
||||
|
||||
from file_ext_registration import register_file_association, is_windows_admin
|
||||
from src.shared.shareddata import API_URL, API_URL_SECONDARY, APP_NAME, CURRENT_VERSION, PLATFORM_NAME
|
||||
@@ -55,6 +57,8 @@ class TerminalWindow(QWidget):
|
||||
layout.addWidget(self.input_line)
|
||||
self.setLayout(layout)
|
||||
|
||||
self._process: QProcess | None = None
|
||||
|
||||
self.commands: dict[str, Callable[..., Any]] = {
|
||||
"hello": self.cmd_hello,
|
||||
"help": self.cmd_help,
|
||||
@@ -62,6 +66,7 @@ class TerminalWindow(QWidget):
|
||||
"about": self.cmd_about,
|
||||
"assoc": self.cmd_assoc,
|
||||
"update": self.cmd_update,
|
||||
"utest": self.cmd_utest,
|
||||
}
|
||||
|
||||
self._pending_assoc_confirmation: bool = False
|
||||
@@ -107,7 +112,8 @@ class TerminalWindow(QWidget):
|
||||
return "Hello from the terminal!"
|
||||
|
||||
def cmd_help(self, *args: Any) -> str:
|
||||
return f"Available commands: {', '.join(self.commands.keys())}"
|
||||
available_cmds = [cmd for cmd in self.commands.keys() if cmd != "utest"]
|
||||
return f"Available commands: {', '.join(available_cmds)}"
|
||||
|
||||
def cmd_version(self, *args: Any) -> str:
|
||||
return f"{APP_NAME.upper()} is running version {CURRENT_VERSION}."
|
||||
@@ -164,3 +170,43 @@ class TerminalWindow(QWidget):
|
||||
def _on_assoc_result(self, ok: bool, msg: str) -> None:
|
||||
self.output_area.append(msg)
|
||||
self._assoc_worker = None
|
||||
|
||||
def cmd_utest(self, *args: Any) -> str | None:
|
||||
"""Executes a specific pre-defined python script non-blockingly."""
|
||||
if self._process and self._process.state() != QProcess.ProcessState.NotRunning:
|
||||
return "[Error] A process is already running."
|
||||
|
||||
target_script = Path("main_unit_tests.py")
|
||||
|
||||
if not target_script.exists():
|
||||
return f"[Error] Target script not found at: {target_script}"
|
||||
|
||||
self._process = QProcess(self)
|
||||
|
||||
# Stream stdout and stderr live to output_area
|
||||
self._process.readyReadStandardOutput.connect(self._handle_stdout)
|
||||
self._process.readyReadStandardError.connect(self._handle_stderr)
|
||||
self._process.finished.connect(self._handle_process_finished)
|
||||
|
||||
# Use current Python interpreter executable. Works when packaged?
|
||||
python_executable = sys.executable
|
||||
|
||||
self.output_area.append(f"Starting {target_script.name}...")
|
||||
self._process.start(python_executable, [str(target_script)])
|
||||
return None
|
||||
|
||||
def _handle_stdout(self) -> None:
|
||||
if self._process:
|
||||
data = self._process.readAllStandardOutput().data().decode("utf-8")
|
||||
if data.strip():
|
||||
self.output_area.append(data.strip())
|
||||
|
||||
def _handle_stderr(self) -> None:
|
||||
if self._process:
|
||||
data = self._process.readAllStandardError().data().decode("utf-8")
|
||||
if data.strip():
|
||||
self.output_area.append(f"[Error] {data.strip()}")
|
||||
|
||||
def _handle_process_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) -> None:
|
||||
self.output_area.append(f"Process finished with code {exit_code}.")
|
||||
self._process = None
|
||||
Reference in New Issue
Block a user