2624 lines
112 KiB
Python
2624 lines
112 KiB
Python
"""
|
|
Filename: main.py
|
|
Description: FLARES main executable
|
|
|
|
Author: Tyler de Zeeuw
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
# Built-in imports
|
|
import os
|
|
import sys
|
|
import time
|
|
import pickle
|
|
import shutil
|
|
import traceback
|
|
import subprocess
|
|
import configparser
|
|
import concurrent.futures
|
|
from queue import Empty
|
|
from pathlib import Path, PurePosixPath
|
|
from datetime import datetime
|
|
from multiprocessing import Process, current_process, freeze_support, Queue
|
|
|
|
# External library imports
|
|
import psutil
|
|
|
|
from mne.io import read_raw_snirf
|
|
from mne.preprocessing.nirs import source_detector_distances
|
|
from mne_nirs.channels import get_short_channels # type: ignore
|
|
|
|
from PySide6.QtWidgets import (
|
|
QApplication, QWidget, QMessageBox, QVBoxLayout, QHBoxLayout, QTextEdit, QScrollArea, QComboBox, QGridLayout, QSplitter,
|
|
QPushButton, QMainWindow, QFileDialog, QLabel, QLineEdit, QFrame, QSizePolicy, QGroupBox, QDialog, QMenu, QSpinBox
|
|
)
|
|
from PySide6.QtCore import QThread, Signal, Qt, QTimer, QPoint
|
|
from PySide6.QtGui import QAction, QKeySequence, QIcon
|
|
from PySide6.QtSvgWidgets import QSvgWidget # needed to show svgs when app is not frozen
|
|
|
|
from src.window.about import AboutWindow
|
|
from src.window.terminal import TerminalWindow
|
|
from src.window.updateevents import EventUpdateMode, UpdateEventsBlazesWindow, UpdateEventsWindow
|
|
from src.window.updateoptodes import UpdateOptodesWindow
|
|
from src.window.userguide import UserGuideWindow
|
|
from src.window.viewerlauncher import ViewerLauncherWidget
|
|
from src.window.welcome import WelcomeDialog
|
|
from src.shared.flaresbasewidget import ParamSection
|
|
from src.shared.shareddata import API_URL, API_URL_SECONDARY, APP_NAME, CURRENT_VERSION, PIPELINE_STAGES, PLATFORM_NAME
|
|
from updater import finish_update_if_needed, UpdateManager, LocalPendingUpdateCheckThread
|
|
|
|
|
|
DEFAULT_CONFIG = """
|
|
[File]
|
|
recent_files =
|
|
recent_projects =
|
|
|
|
[Edit]
|
|
|
|
[View]
|
|
status_bar = true
|
|
left_top = 0
|
|
left_bottom = 0
|
|
right = 0
|
|
|
|
[Options]
|
|
show_welcome_dialog = false
|
|
first_startup = true
|
|
|
|
[Preferences]
|
|
2d_data_bypass = false
|
|
incompatible_save_bypass = false
|
|
missing_events_bypass = false
|
|
analysis_clearing_bypass = false
|
|
folding_bypass = false
|
|
|
|
[Terminal]
|
|
|
|
[General]
|
|
|
|
"""
|
|
|
|
# Selectable parameters on the right side of the window
|
|
SECTIONS = [
|
|
{
|
|
"title": "Preprocessing",
|
|
"params": [
|
|
{"name": "DOWNSAMPLE", "default": True, "type": bool, "help": "Should the snirf files be downsampled? If this is set to True, DOWNSAMPLE_FREQUENCY will be used as the target frequency to downsample to."},
|
|
{"name": "DOWNSAMPLE_FREQUENCY", "default": 25, "type": int, "depends_on": "DOWNSAMPLE", "help": "Frequency (Hz) to downsample to. If this is set higher than the input data, new data will be interpolated."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Trimming",
|
|
"params": [
|
|
{"name": "TRIM", "default": True, "type": bool, "help": "Should the start of the files be trimmed?"},
|
|
{"name": "SECONDS_TO_KEEP", "default": 5, "type": float, "depends_on": "TRIM", "help": "Seconds to keep at the beginning of all loaded snirf files before the first annotation/event occurs. Calculation is done seperatly on all loaded snirf files. Setting this to 0 will have the first annotation/event be at time point 0. Only used if TRIM is set to True."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Verify Optode Placement",
|
|
"params": [
|
|
{"name": "OPTODE_PLACEMENT", "default": True, "type": bool, "help": "Should an image be generated for each participant outlining their optode placement on a head?"},
|
|
{"name": "SHOW_OPTODE_NAMES", "default": True, "type": bool, "depends_on": "OPTODE_PLACEMENT", "help": "Should the optode names be written next to their location in the image?"},
|
|
]
|
|
},
|
|
{
|
|
"title": "Short/Long Channels",
|
|
"params": [
|
|
{"name": "SHORT_CHANNEL", "default": True, "type": bool, "help": "This should be set to True if the data has a short channel present in the data. For more information about short channels, please visit the Wiki."},
|
|
{"name": "SHORT_CHANNEL_THRESH", "default": 0.015, "type": float, "depends_on": "SHORT_CHANNEL", "help": "The maximum distance the short channel can be in metres before it is no longer considered a short channel."},
|
|
{"name": "LONG_CHANNEL_THRESH", "default": 0.045, "type": float, "help": "The maximum distance channels can be in metres. Any channel longer than this distance will be discarded."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Heart Rate",
|
|
"params": [
|
|
{"name": "HEART_RATE", "default": True, "type": bool, "help": "Should an attempt be made to calculate the participants heart rate?"},
|
|
{"name": "SECONDS_TO_STRIP_HR", "default": 5, "type": int, "depends_on": "HEART_RATE", "help": "Will remove this many seconds from the start and end of the file. Useful if recording before cap is firmly placed, or participant removes cap while still recording."},
|
|
{"name": "MAX_LOW_HR", "default": 40, "type": int, "depends_on": "HEART_RATE", "help": "Any heart rate windows that average below this value will be rounded up to this value."},
|
|
{"name": "MAX_HIGH_HR", "default": 200, "type": int, "depends_on": "HEART_RATE", "help": "Any heart rate windows that average above this value will be rounded down to this value."},
|
|
{"name": "SMOOTHING_WINDOW_HR", "default": 100, "type": int, "depends_on": "HEART_RATE", "help": "How many individual data points to be used to create a single data point/window."},
|
|
{"name": "HEART_RATE_WINDOW", "default": 25, "type": int, "depends_on": "HEART_RATE", "help": "Only used for visualization. Shows the 'range' of the calculated heart rate, which is just the average +- this value."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Scalp Coupling Index",
|
|
"params": [
|
|
{"name": "SCI", "default": True, "type": bool, "help": "Calculate and mark channels bad based on their Scalp Coupling Index. This metric calculates the quality of the connection between the optode and the scalp."},
|
|
{"name": "SCI_TIME_WINDOW", "default": 3, "type": int, "depends_on": "SCI", "help": "Independent SCI calculations will be perfomed in a time window for the duration of the value provided, until the end of the file is reached."},
|
|
{"name": "SCI_THRESHOLD", "default": 0.6, "type": float, "depends_on": "SCI", "help": "SCI threshold on a scale of 0-1. A value of 0 is bad coupling while a value of 1 is perfect coupling. Any channels lower than this value will be marked as bad."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Signal to Noise Ratio",
|
|
"params": [
|
|
{"name": "SNR", "default": True, "type": bool, "help": "Calculate and mark channels bad based on their Signal to Noise Ratio. This metric calculates how much of the observed signal was noise versus how much of it was a useful signal."},
|
|
{"name": "SNR_THRESHOLD", "default": 5.0, "type": float, "depends_on": "SNR", "help": "SNR threshold (dB). A typical scale would be 0-25, but it is possible for values to be both above and below this range. Higher values correspond to a better signal. If SNR is True, any channels lower than this value will be marked as bad."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Peak Spectral Power",
|
|
"params": [
|
|
{"name": "PSP", "default": True, "type": bool, "help": "Calculate and mark channels bad based on their Peak Spectral Power. This metric calculates the amplitude or strength of a frequency component that is most prominent in a particular frequency range or spectrum."},
|
|
{"name": "PSP_TIME_WINDOW", "default": 3, "type": int, "depends_on": "PSP", "help": "Independent PSP calculations will be perfomed in a time window for the duration of the value provided, until the end of the file is reached."},
|
|
{"name": "PSP_THRESHOLD", "default": 0.1, "type": float, "depends_on": "PSP", "help": "PSP threshold. A typical scale would be 0-0.5, but it is possible for values to be above this range. Higher values correspond to a better signal. If PSP is True, any channels lower than this value will be marked as bad."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Cross Validation",
|
|
"params": [
|
|
{"name": "CV", "default": True, "type": bool, "help": "Identifies bad channels using the Coefficient of Variation."},
|
|
{"name": "CV_THRESHOLD", "default": 20, "type": int, "depends_on": "CV", "help": "Noise threshold (%)."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Median Absolute Deviation",
|
|
"params": [
|
|
{"name": "MAD", "default": True, "type": bool, "help": "Identifies bad channels using Mean Absolute Deviation."},
|
|
{"name": "MAD_THRESHOLD", "default": 4, "type": int, "depends_on": "MAD", "help": "Amount of deviations before the channel is flagged bad."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Power Spectral Density Noise",
|
|
"params": [
|
|
{"name": "PSD_NOISE", "default": True, "type": bool, "help": "Identifies bad channels based on their excessive power at high frequencies."},
|
|
{"name": "TARGET_FREQ_DIV", "default": 4, "type": int, "depends_on": "PSD_NOISE", "help": "What frequency to check for excessive power. Will take the recording frequency and divide by this number. Has to be greater than 2."},
|
|
{"name": "DB_LIMIT", "default": -60, "type": int, "depends_on": "PSD_NOISE", "help": "What db level the power level needs to be below at the target frequency."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Channel Variance",
|
|
"params": [
|
|
{"name": "CHANNEL_VAR", "default": True, "type": bool, "help": "Identifies bad channels based on comparing the variance of the first 25% of the data to the last 25%."},
|
|
{"name": "CHANNEL_THRESH", "default": 0.05, "type": float, "depends_on": "CHANNEL_VAR", "help": "If the end variance is less than this % of the start variance, the channel will be marked as bad."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Bad Channels Handling",
|
|
"params": [
|
|
{"name": "BAD_CHANNELS_HANDLING", "default": ["Interpolate"], "type": list, "options": ["Interpolate", "Remove", "None"], "exclusive": True, "help": "How should we deal with the bad channels that occurred? Note: Some analysis options will only work when this is set to 'Interpolate'."},
|
|
{"name": "MAX_DIST", "default": 0.03, "type": float, "depends_on": "BAD_CHANNELS_HANDLING", "depends_value": "Interpolate", "help": "The maximum distance to look for neighbours when interpolating. Used only when BAD_CHANNELS_HANDLING is set to 'Interpolate'."},
|
|
{"name": "MIN_NEIGHBORS", "default": 2, "type": int, "depends_on": "BAD_CHANNELS_HANDLING", "depends_value": "Interpolate", "help": "The minimumn amount of neighbours needed within the MAX_DIST parameter. Used only when BAD_CHANNELS_HANDLING is set to 'Interpolate'."},
|
|
{"name": "MAX_BAD_CHANNELS", "default": 12, "type": int, "depends_on": "BAD_CHANNELS_HANDLING", "depends_value": "Remove", "help": "Maximum amount of bad channels before the participant as a whole is marked as bad (exclusive). If this occurs, the participant will be prevented from processing any further. Used only when BAD_CHANNELS_HANDLING is set to 'Remove'."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Optical Density",
|
|
"params": [
|
|
# NOTE: Intentionally empty
|
|
]
|
|
},
|
|
{
|
|
"title": "Temporal Derivative Distribution Repair filtering",
|
|
"params": [
|
|
{"name": "TDDR", "default": True, "type": bool, "help": "Apply Temporal Derivitave Distribution Repair filtering - a method that removes baseline shift and spike artifacts from the data."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Wavelet filtering",
|
|
"params": [
|
|
{"name": "WAVELET", "default": True, "type": bool, "help": "Apply Wavelet filtering. It is a method to filter involving decomposition, threholding, and reconstruction."},
|
|
{"name": "IQR", "default": 1.5, "type": float, "depends_on": "WAVELET", "help": "Scaling factor for the Inter-Quartile Range."},
|
|
{"name": "WAVELET_TYPE", "default": "db4", "type": str, "depends_on": "WAVELET", "help": "Wavelet type. Valid values are ['bior1.1', 'bior1.3', 'bior1.5', 'bior2.2', 'bior2.4', 'bior2.6', 'bior2.8', 'bior3.1', 'bior3.3', 'bior3.5', 'bior3.7', 'bior3.9', 'bior4.4', 'bior5.5', 'bior6.8', 'coif1', 'coif2', 'coif3', 'coif4', 'coif5', 'coif6', 'coif7', 'coif8', 'coif9', 'coif10', 'coif11', 'coif12', 'coif13', 'coif14', 'coif15', 'coif16', 'coif17', 'db1', 'db2', 'db3', 'db4', 'db5', 'db6', 'db7', 'db8', 'db9', 'db10', 'db11', 'db12', 'db13', 'db14', 'db15', 'db16', 'db17', 'db18', 'db19', 'db20', 'db21', 'db22', 'db23', 'db24', 'db25', 'db26', 'db27', 'db28', 'db29', 'db30', 'db31', 'db32', 'db33', 'db34', 'db35', 'db36', 'db37', 'db38', 'dmey', 'haar', 'rbio1.1', 'rbio1.3', 'rbio1.5', 'rbio2.2', 'rbio2.4', 'rbio2.6', 'rbio2.8', 'rbio3.1', 'rbio3.3', 'rbio3.5', 'rbio3.7', 'rbio3.9', 'rbio4.4', 'rbio5.5', 'rbio6.8', 'sym2', 'sym3', 'sym4', 'sym5', 'sym6', 'sym7', 'sym8', 'sym9', 'sym10', 'sym11', 'sym12', 'sym13', 'sym14', 'sym15', 'sym16', 'sym17', 'sym18', 'sym19', 'sym20']"},
|
|
{"name": "WAVELET_LEVEL", "default": 3, "type": int, "depends_on": "WAVELET", "help": "Wavelet Decomposition level (must be >= 0)."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Haemoglobin Concentration",
|
|
"params": [
|
|
# NOTE: Intentionally empty
|
|
]
|
|
},
|
|
{
|
|
"title": "Enhance Negative Correlation",
|
|
"params": [
|
|
{"name": "ENHANCE_NEGATIVE_CORRELATION", "default": False, "type": bool, "help": "Apply Enhance Negative Correlation."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Filtering",
|
|
"params": [
|
|
{"name": "FILTER", "default": True, "type": bool, "help": "Should the data be bandpass filtered?"},
|
|
{"name": "L_FREQ", "default": 0.005, "type": float, "depends_on": "FILTER", "help": "Any frequencies lower than this value will be removed."},
|
|
{"name": "H_FREQ", "default": 0.3, "type": float, "depends_on": "FILTER", "help": "Any frequencies higher than this value will be removed."},
|
|
{"name": "L_TRANS_BANDWIDTH", "default": 0.002, "type": float, "depends_on": "FILTER", "help": "How wide the transitional period should be so the data doesn't just drop off on the lower bound."},
|
|
{"name": "H_TRANS_BANDWIDTH", "default": 0.002, "type": float, "depends_on": "FILTER", "help": "How wide the transitional period should be so the data doesn't just drop off on the upper bound."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Extracting Events*",
|
|
"params": [
|
|
#{"name": "EVENTS", "default": True, "type": bool, "help": "Calculate Peak Spectral Power."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Epoch Calculations",
|
|
"params": [
|
|
# TODO: implement drop
|
|
{"name": "EPOCH_HANDLING", "default": ["shift"], "type": list, "options": ["shift", "strict"], "help": "What to do if two unique events occur at the same time. Shift will automatically move one event to the first valid free index. Strict will raise an error processing the file. Drop will remove one of the events."},
|
|
{"name": "MAX_SHIFT", "default": 5, "type": int, "depends_on": "EPOCH_HANDLING", "depends_value": "shift", "help": "Amount of indexes to look ahead and see if there is a valid one to shift to. If none were found, will fall back to 'strict' behaviour."},
|
|
#{"name": "REJECT_BY_ANNOTATIONS", "default": True, "type": bool, "help": "Help."},
|
|
#{"name": "MAX_SHIFT", "default": 5, "type": int, "depends_on": "EPOCH_HANDLING", "depends_value": "shift", "help": "Amount of indexes to look ahead and see if there is a valid one to shift to. If none were found, will fall back to 'strict' behaviour."},
|
|
#{"name": "MAX_SHIFT", "default": 5, "type": int, "depends_on": "EPOCH_HANDLING", "depends_value": "shift", "help": "Amount of indexes to look ahead and see if there is a valid one to shift to. If none were found, will fall back to 'strict' behaviour."},
|
|
{"name": "T_MIN", "default": -5, "type": int, "help": "Seconds before the epoch to be used."},
|
|
{"name": "T_MAX", "default": 15, "type": int, "help": "Seconds after the epoch to be used."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Design Matrix",
|
|
"params": [
|
|
{"name": "RESAMPLE", "default": True, "type": bool, "help": "Should the data be resampled before calculating the design matrix? Downsampling is useful for speeding up calculations without losing overall data shape."},
|
|
{"name": "RESAMPLE_FREQ", "default": 1, "type": int, "help": "The frequency the data should be resampled to."},
|
|
{"name": "HRF_MODEL", "default": ["fir"], "type": list, "options": ["fir", "glover", "spm", "spm + derivative", "spm + derivative + dispersion", "glover + derivative", "glover + derivative + dispersion"], "exclusive": True, "help": "Specifies the haemodynamic response function."},
|
|
{"name": "STIM_DUR", "default": 0.5, "type": float, "help": "The length of your stimulus. If your HRF_MODEL is fir, this dictates how wide a bin should be."},
|
|
{"name": "DRIFT_MODEL", "default": ["cosine"], "type": list, "options": ["cosine", "polynomial"], "help": "Specifies the desired drift model."},
|
|
{"name": "HIGH_PASS", "default": 0.01, "type": float, "help": "High-pass frequency in case of a cosine model (in Hz)."},
|
|
{"name": "DRIFT_ORDER", "default": 1, "type": int, "help": "Order of the drift model (in case it is polynomial)"},
|
|
{"name": "FIR_DELAYS", "default": 15, "type": range, "depends_on": "HRF_MODEL", "depends_value": "fir", "help": "In case of FIR design, yields the array of delays used in the FIR model (in scans)."},
|
|
{"name": "MIN_ONSET", "default": -24, "type": int, "help": "Minimal onset relative to frame times (in seconds)"},
|
|
{"name": "OVERSAMPLING", "default": 50, "type": int, "help": "Oversampling factor used in temporal convolutions."},
|
|
{"name": "REMOVE_EVENTS", "default": "None", "type": list, "help": "Remove events matching the names provided before generating the Design Matrix"},
|
|
{"name": "SHORT_CHANNEL_REGRESSION", "default": True, "type": bool, "depends_on": "SHORT_CHANNEL", "help": "Should short channel regression be used to create the design matrix? This will use the 'signal' from the short channel and regress it out of all other channels."},
|
|
]
|
|
},
|
|
{
|
|
"title": "General Linear Model",
|
|
"params": [
|
|
{"name": "NOISE_MODEL", "default": "ar1", "type": str, "help": "The temporal variance model. Defaults to first order auto regressive model 'ar1'. The AR model can be set to any integer value by modifying the value of N. E.g. use ar5 for a fifth order model. If the string auto is provided a model with order 4 times the sample rate will be used."},
|
|
{"name": "BINS", "default": 0, "type": int, "help": "Maximum number of discrete bins for the AR coef histogram/clustering. By default the value is 0, which will set the number of bins to the number of channels, effectively estimating the AR model for each channel."},
|
|
{"name": "N_JOBS", "default": 1, "type": int, "help": "The number of CPUs to use to do the GLM computation. -1 means 'all CPUs'."},
|
|
]
|
|
},
|
|
{
|
|
"title": "Finishing Touches",
|
|
"params": [
|
|
# Intentionally empty (TODO)
|
|
]
|
|
},
|
|
{
|
|
"title": "Other",
|
|
"params": [
|
|
{"name": "TIME_WINDOW_START", "default": 0, "type": int, "help": "Where to start averaging the fir model bins. Only affects the significance and contrast images."},
|
|
{"name": "TIME_WINDOW_END", "default": 15, "type": int, "help": "Where to end averaging the fir model bins. Only affects the significance and contrast images."},
|
|
{"name": "MAX_WORKERS", "default": 6, "type": int, "help": "Number of files to be processed at once. Setting this to a small integer value may help on underpowered systems. Remove the value to use an automatic amount."},
|
|
{"name": "VERBOSITY", "default": False, "type": bool, "help": "Setting this to True will log lots of debugging information to the log file. Setting this to False will log minimal data."},
|
|
]
|
|
},
|
|
]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class SaveProjectThread(QThread):
|
|
finished_signal = Signal(str)
|
|
error_signal = Signal(str)
|
|
|
|
def __init__(self, filename, project_data):
|
|
super().__init__()
|
|
self.filename = filename
|
|
self.project_data = project_data
|
|
|
|
def run(self):
|
|
try:
|
|
with open(self.filename, "wb") as f:
|
|
pickle.dump(self.project_data, f)
|
|
self.finished_signal.emit(self.filename)
|
|
except Exception as e:
|
|
self.error_signal.emit(str(e))
|
|
|
|
|
|
|
|
class SavingOverlay(QDialog):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setWindowFlags(Qt.WindowType.Dialog | Qt.WindowType.FramelessWindowHint)
|
|
self.setModal(True)
|
|
self.setWindowModality(Qt.WindowModality.ApplicationModal)
|
|
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
|
|
|
|
layout = QVBoxLayout()
|
|
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
|
|
label = QLabel("Saving Project…")
|
|
label.setStyleSheet("font-size: 18px; color: white; background-color: rgba(0,0,0,150); padding: 20px; border-radius: 10px;")
|
|
layout.addWidget(label)
|
|
self.setLayout(layout)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ProgressBubble(QWidget):
|
|
"""
|
|
A clickable widget displaying a progress bar made of colored rectangles and a label.
|
|
|
|
Args:
|
|
display_name (str): Text to display above the progress bar.
|
|
file_path (str): Associated file path stored with the bubble.
|
|
|
|
"""
|
|
|
|
clicked = Signal(object)
|
|
rightClicked = Signal(object, QPoint)
|
|
|
|
def __init__(self, display_name, file_path):
|
|
super().__init__()
|
|
|
|
self.layout = QVBoxLayout()
|
|
self.label = QLabel(display_name)
|
|
self.loading_timer = QTimer(self)
|
|
self.loading_timer.timeout.connect(self._rotate_spinner)
|
|
self.spinner_frames = ["◐", "◓", "◑", "◒"] #cute
|
|
self.spinner_idx = 0
|
|
self.is_loading = False
|
|
self.base_text = display_name
|
|
self.label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
self.label.setStyleSheet("""
|
|
QLabel {
|
|
border: 1px solid #888;
|
|
border-radius: 10px;
|
|
padding: 8px 12px;
|
|
background-color: #e0f0ff;
|
|
}
|
|
""")
|
|
|
|
self.progress_layout = QHBoxLayout()
|
|
|
|
self.rects = []
|
|
for i in range(28):
|
|
rect = QFrame()
|
|
rect.setFixedSize(10, 18)
|
|
rect.setStyleSheet("background-color: white; border: 1px solid gray;")
|
|
stage_name = PIPELINE_STAGES[i]
|
|
rect.setToolTip(f"Stage {i + 1}: {stage_name}")
|
|
self.progress_layout.addWidget(rect)
|
|
self.rects.append(rect)
|
|
|
|
self.layout.addWidget(self.label)
|
|
self.layout.addLayout(self.progress_layout)
|
|
self.setLayout(self.layout)
|
|
|
|
# Store the file path
|
|
self.file_path = os.path.normpath(file_path)
|
|
|
|
self.current_step = 0
|
|
|
|
# Make the bubble appear to the user as clickable
|
|
self.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
|
|
# Resize policy to make bubbles responsive
|
|
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum)
|
|
|
|
def set_loading_state(self, loading=True):
|
|
self.is_loading = loading
|
|
if loading:
|
|
self.loading_timer.start(150) # Rotate every 150ms
|
|
else:
|
|
self.loading_timer.stop()
|
|
# Transition to a green checkmark
|
|
self.setSuffixText(" <span style='color: green;'>✔</span>")
|
|
|
|
def update_progress(self, step_index, active=True):
|
|
self.current_step = step_index
|
|
for i, rect in enumerate(self.rects):
|
|
if i < step_index:
|
|
rect.setStyleSheet("background-color: green; border: 1px solid gray;")
|
|
elif i == step_index:
|
|
color = "yellow" if active else "white"
|
|
rect.setStyleSheet(f"background-color: {color}; border: 1px solid gray;")
|
|
else:
|
|
rect.setStyleSheet("background-color: white; border: 1px solid gray;")
|
|
|
|
def mark_cancelled(self):
|
|
if 0 <= self.current_step < len(self.rects):
|
|
rect = self.rects[self.current_step]
|
|
rect.setStyleSheet("background-color: red; border: 1px solid gray;")
|
|
|
|
def mousePressEvent(self, event):
|
|
if event.button() == Qt.MouseButton.LeftButton:
|
|
self.clicked.emit(self)
|
|
elif event.button() == Qt.MouseButton.RightButton:
|
|
self.rightClicked.emit(self, event.globalPosition().toPoint())
|
|
super().mousePressEvent(event)
|
|
|
|
def setSuffixText(self, suffix):
|
|
if suffix:
|
|
self.label.setText(f"{self.base_text} {suffix}")
|
|
else:
|
|
self.label.setText(self.base_text)
|
|
|
|
def _rotate_spinner(self):
|
|
frame = self.spinner_frames[self.spinner_idx % len(self.spinner_frames)]
|
|
# Using HTML in setText allows us to style the spinner specifically
|
|
self.setSuffixText(f" <span style='color: #555;'>{frame}</span>")
|
|
self.spinner_idx += 1
|
|
|
|
|
|
|
|
|
|
|
|
class MainApplication(QMainWindow):
|
|
"""
|
|
Main application window that creates and sets up the UI.
|
|
"""
|
|
|
|
progress_update_signal = Signal(str, int)
|
|
metadata_processed = Signal(str, int)
|
|
metadata_ui_signal = Signal(dict, str, int)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setWindowTitle(f"{APP_NAME.upper()}")
|
|
self.setGeometry(100, 100, 1280, 720)
|
|
|
|
# Load the mne data in a seperate process
|
|
self.file_executor = concurrent.futures.ProcessPoolExecutor(max_workers=1)
|
|
|
|
self.about = None
|
|
self.help = None
|
|
self.optodes = None
|
|
self.events = None
|
|
self.terminal = None
|
|
self.bubble_widgets = {}
|
|
self.param_sections = []
|
|
self.folder_paths = []
|
|
self.section_widget = None
|
|
self.first_run = True
|
|
self.is_2d_bypass = False
|
|
self.incompatible_save_bypass = False
|
|
self.missing_events_bypass = False
|
|
self.analysis_clearing_bypass = False
|
|
self.folding_bypass = False
|
|
|
|
|
|
# Initialization to ensure that saving can occur
|
|
self.raw_haemo_dict = {} # Processed Hemodynamic data
|
|
self.config_dict = {} # Analysis parameters/settings
|
|
self.epochs_dict = {} # Timing/Event data
|
|
self.cha_dict = {} # Channel configurations
|
|
self.contrast_results_dict = {} # Statistical results
|
|
self.df_ind_dict = {} # Individual dataframes
|
|
self.design_matrix_dict = {} # GLM Design matrices
|
|
self.valid_dict = {} # Quality control/Validity flags
|
|
self.fig_bytes_dict = {} # Cached plot images (serialized)
|
|
self.file_metadata = {} # AGE, GENDER, GROUP
|
|
self.metadata_cache = {} # Internal file/path information metadata cache
|
|
self.bubble_widgets = {} # References to the UI "Bubble" objects
|
|
self.current_file = None # Tracks the currently selected absolute path
|
|
|
|
self.metadata_processed.connect(self._safe_ui_update)
|
|
self.metadata_ui_signal.connect(self._handle_metadata_ui_update)
|
|
|
|
self.files_total = 0 # total number of files to process
|
|
self.files_done = set() # set of file paths done (success or fail)
|
|
self.files_failed = set() # set of failed file paths
|
|
self.files_results = {} # dict for successful results (if needed)
|
|
|
|
self.platform_suffix = "-" + PLATFORM_NAME
|
|
|
|
self.updater = UpdateManager(
|
|
main_window=self,
|
|
api_url=API_URL,
|
|
api_url_sec=API_URL_SECONDARY,
|
|
current_version=CURRENT_VERSION,
|
|
platform_name=PLATFORM_NAME,
|
|
platform_suffix=self.platform_suffix,
|
|
app_name=APP_NAME
|
|
)
|
|
|
|
self.init_ui()
|
|
self.create_menu_bar()
|
|
|
|
self.pending_update_version = None
|
|
self.pending_update_path = None
|
|
self.last_clicked_bubble = None
|
|
self.installEventFilter(self)
|
|
|
|
# Start local pending update check thread
|
|
self.local_check_thread = LocalPendingUpdateCheckThread(CURRENT_VERSION, self.platform_suffix, PLATFORM_NAME, APP_NAME)
|
|
self.local_check_thread.pending_update_found.connect(self.updater.on_pending_update_found)
|
|
self.local_check_thread.no_pending_update.connect(self.updater.on_no_pending_update)
|
|
self.local_check_thread.start()
|
|
|
|
self.show()
|
|
|
|
# Check if we should pop up the welcome screen
|
|
should_show_welcome = file_cfg.getboolean("Options", "show_welcome_dialog", fallback=True)
|
|
first_startup = file_cfg.getboolean("Options", "first_startup", fallback=False)
|
|
|
|
if first_startup:
|
|
file_cfg.set("Options", "first_startup", "false")
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save preference: {e}")
|
|
|
|
welcome = WelcomeDialog(self, direct=True, first=first_startup)
|
|
welcome.show()
|
|
|
|
elif should_show_welcome:
|
|
file_cfg.set("Options", "show_welcome_dialog", "false")
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save preference: {e}")
|
|
|
|
welcome = WelcomeDialog(self, direct=True, first=False)
|
|
welcome.show()
|
|
|
|
|
|
|
|
|
|
def init_ui(self):
|
|
central = QWidget()
|
|
self.setCentralWidget(central)
|
|
main_layout = QHBoxLayout(central)
|
|
main_layout.setContentsMargins(5, 5, 5, 5)
|
|
|
|
self.main_h_splitter = QSplitter(Qt.Orientation.Horizontal)
|
|
self.main_h_splitter.setChildrenCollapsible(False)
|
|
main_layout.addWidget(self.main_h_splitter)
|
|
|
|
self.left_v_splitter = QSplitter(Qt.Orientation.Vertical)
|
|
self.left_v_splitter.setChildrenCollapsible(False)
|
|
self.left_v_splitter.setMinimumWidth(460)
|
|
|
|
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)
|
|
|
|
self.top_left_widget = QTextEdit()
|
|
self.top_left_widget.setReadOnly(True)
|
|
self.top_left_widget.setPlaceholderText("Click a file below to get started! No files below? Open one with File -> Open File!")
|
|
top_left_layout.addWidget(self.top_left_widget, stretch=4)
|
|
|
|
self.right_column_widget = QWidget()
|
|
right_column_layout = QVBoxLayout(self.right_column_widget)
|
|
self.meta_fields = {"AGE": QLineEdit(), "GENDER": QLineEdit(), "GROUP": QLineEdit()}
|
|
for key, field in self.meta_fields.items():
|
|
label = QLabel(key.capitalize())
|
|
right_column_layout.addWidget(label)
|
|
right_column_layout.addWidget(field)
|
|
field.textChanged.connect(self.sync_bubble_data)
|
|
|
|
label_desc = QLabel('<a href="#">Why are these useful?</a>')
|
|
label_desc.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
|
|
label_desc.linkActivated.connect(lambda: QMessageBox.information(None, "Info", "Parameter Info..."))
|
|
right_column_layout.addWidget(label_desc)
|
|
right_column_layout.addStretch()
|
|
self.right_column_widget.hide()
|
|
top_left_layout.addWidget(self.right_column_widget, stretch=1)
|
|
|
|
self.bubble_container = QWidget()
|
|
self.bubble_layout = QGridLayout(self.bubble_container)
|
|
self.bubble_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
|
|
|
|
self.scroll_area = QScrollArea()
|
|
self.scroll_area.setWidgetResizable(True)
|
|
self.scroll_area.setWidget(self.bubble_container)
|
|
self.scroll_area.setMinimumHeight(200)
|
|
|
|
self.left_v_splitter.addWidget(top_left_container)
|
|
self.left_v_splitter.addWidget(self.scroll_area)
|
|
|
|
self.right_container = QWidget()
|
|
self.right_container.setMinimumWidth(440)
|
|
right_container_layout = QVBoxLayout(self.right_container)
|
|
|
|
self.right_content_widget = QWidget()
|
|
right_content_layout = QVBoxLayout(self.right_content_widget)
|
|
self.rows_container = QWidget()
|
|
self.rows_layout = QVBoxLayout(self.rows_container)
|
|
right_content_layout.addWidget(self.rows_container)
|
|
right_content_layout.addStretch()
|
|
|
|
self.right_scroll_area = QScrollArea()
|
|
self.right_scroll_area.setWidgetResizable(True)
|
|
self.right_scroll_area.setWidget(self.right_content_widget)
|
|
|
|
buttons_widget = QWidget()
|
|
buttons_layout = QHBoxLayout(buttons_widget)
|
|
buttons_layout.addStretch()
|
|
|
|
self.button1, self.button2, self.button3 = QPushButton("Process"), QPushButton("Clear"), QPushButton("Analysis")
|
|
for btn in [self.button1, self.button2, self.button3]:
|
|
btn.setMinimumSize(100, 40)
|
|
buttons_layout.addWidget(btn)
|
|
|
|
self.button1.setVisible(False)
|
|
self.button3.setVisible(False)
|
|
self.button1.clicked.connect(self.on_run_task)
|
|
self.button2.clicked.connect(self.clear_all)
|
|
self.button3.clicked.connect(self.open_launcher_window)
|
|
|
|
right_container_layout.addWidget(self.right_scroll_area)
|
|
right_container_layout.addWidget(buttons_widget)
|
|
|
|
self.main_h_splitter.addWidget(self.left_v_splitter)
|
|
self.main_h_splitter.addWidget(self.right_container)
|
|
|
|
self.main_h_splitter.setSizes([600, 400])
|
|
self.left_v_splitter.setSizes([300, 700])
|
|
|
|
self.progress_update_signal.connect(self.update_file_progress)
|
|
self.update_sections(0)
|
|
|
|
#NOTE: leave this here for now
|
|
# def check_memory_leak(self):
|
|
# # 2. Take a snapshot
|
|
# snapshot = tracemalloc.take_snapshot()
|
|
|
|
# # 3. Filter to show the top 10 biggest "stayers"
|
|
# top_stats = snapshot.statistics('lineno')
|
|
|
|
# print("[ Top 10 Memory Consumers ]")
|
|
# for stat in top_stats[:10]:
|
|
# print(stat)
|
|
|
|
def create_menu_bar(self):
|
|
'''Menu Bar at the top of the screen'''
|
|
|
|
menu_bar = self.menuBar()
|
|
self.statusbar = self.statusBar()
|
|
|
|
def make_action(name, shortcut=None, slot=None, checkable=False, checked=False, icon=None):
|
|
action = QAction(name, self)
|
|
|
|
if shortcut:
|
|
action.setShortcut(QKeySequence(shortcut))
|
|
if slot:
|
|
action.triggered.connect(slot)
|
|
if checkable:
|
|
action.setCheckable(True)
|
|
action.setChecked(checked)
|
|
if icon:
|
|
action.setIcon(QIcon(icon))
|
|
return action
|
|
|
|
# File menu and actions
|
|
file_menu = menu_bar.addMenu("File")
|
|
file_actions = [
|
|
("Open File...", "Ctrl+O", self.open_file_dialog, resource_path("icons/file_open_24dp_1F1F1F.svg")),
|
|
("Open Folder...", "Ctrl+Alt+O", self.open_folder_dialog, resource_path("icons/folder_24dp_1F1F1F.svg")),
|
|
# ("Open Folders...", "Ctrl+Shift+O", self.open_folder_dialog, resource_path("icons/folder_copy_24dp_1F1F1F.svg")),
|
|
("Load Project...", "Ctrl+L", self.load_project, resource_path("icons/article_24dp_1F1F1F.svg")),
|
|
("Save Project...", "Ctrl+S", self.save_project, resource_path("icons/save_24dp_1F1F1F.svg")),
|
|
("Save Project As...", "Ctrl+Shift+S", self.save_project, resource_path("icons/save_as_24dp_1F1F1F.svg")),
|
|
]
|
|
|
|
for i, (name, shortcut, slot, icon) in enumerate(file_actions):
|
|
file_menu.addAction(make_action(name, shortcut, slot, icon=icon))
|
|
if i == 1:
|
|
self.recent_files_menu = file_menu.addMenu("Recent Files")
|
|
self.recent_files_menu.setIcon(QIcon(resource_path("icons/history_24dp_1F1F1F.svg"))) # optional icon
|
|
file_menu.addSeparator()
|
|
elif i == 2:
|
|
self.recent_projects_menu = file_menu.addMenu("Recent Projects")
|
|
self.recent_projects_menu.setIcon(QIcon(resource_path("icons/history_2_24dp_1F1F1F.svg")))
|
|
file_menu.addSeparator()
|
|
|
|
file_menu.addSeparator()
|
|
file_menu.addAction(make_action("Exit", "Ctrl+Q", QApplication.instance().quit, icon=resource_path("icons/exit_to_app_24dp_1F1F1F.svg")))
|
|
|
|
# Edit menu
|
|
edit_menu = menu_bar.addMenu("Edit")
|
|
edit_actions = [
|
|
("Cut", "Ctrl+X", self.cut_text, resource_path("icons/content_cut_24dp_1F1F1F.svg")),
|
|
("Copy", "Ctrl+C", self.copy_text, resource_path("icons/content_copy_24dp_1F1F1F.svg")),
|
|
("Paste", "Ctrl+V", self.paste_text, resource_path("icons/content_paste_24dp_1F1F1F.svg"))
|
|
]
|
|
for name, shortcut, slot, icon in edit_actions:
|
|
edit_menu.addAction(make_action(name, shortcut, slot, icon=icon))
|
|
|
|
# View menu
|
|
# TODO: Pretty this like the rest of the menus?
|
|
view_menu = menu_bar.addMenu("View")
|
|
toggle_statusbar_action = make_action("Toggle Status Bar", checkable=True, checked=True, slot=None)
|
|
view_menu.addAction(toggle_statusbar_action)
|
|
toggle_statusbar_action.toggled.connect(self.statusbar.setVisible)
|
|
|
|
# Reset Layout Action
|
|
view_menu.addSeparator()
|
|
reset_layout_action = make_action(
|
|
"Reset Window Layout",
|
|
"Ctrl+Shift+R",
|
|
self.reset_window_layout,
|
|
icon=resource_path("icons/grid_layout_side_24dp_1F1F1F.svg")
|
|
)
|
|
view_menu.addAction(reset_layout_action)
|
|
|
|
# Options menu (Help & About)
|
|
options_menu = menu_bar.addMenu("Options")
|
|
|
|
options_actions = [
|
|
("User Guide", "F1", self.user_guide, resource_path("icons/help_24dp_1F1F1F.svg")),
|
|
("Check for Updates", "F5", self.updater.manual_check_for_updates, resource_path("icons/update_24dp_1F1F1F.svg")),
|
|
("Show Update Changelog", "F6", self.show_update_changelog, resource_path("icons/article_shortcut_24dp_1F1F1.svg")),
|
|
("Update events in snirf file (BORIS)...", "F7", self.update_event_markers, resource_path("icons/upgrade_24dp_1F1F1F.svg")),
|
|
("Update events in snirf file (BLAZES)...", "F8", self.update_event_markers_blazes, resource_path("icons/upgrade_24dp_1F1F1F.svg")),
|
|
("Update optodes in snirf file...", "F9", self.update_optode_positions, resource_path("icons/upgrade_24dp_1F1F1F.svg")),
|
|
("Reset to Default Configuration", "F10", self.reset_to_default_configuration, resource_path("icons/reset_settings_24dp_1F1F1F.svg")),
|
|
("About", "F12", self.about_window, resource_path("icons/info_24dp_1F1F1F.svg"))
|
|
]
|
|
|
|
for i, (name, shortcut, slot, icon) in enumerate(options_actions):
|
|
options_menu.addAction(make_action(name, shortcut, slot, icon=icon))
|
|
if i == 2 or i == 5 or i == 6 or i == 7:
|
|
options_menu.addSeparator()
|
|
|
|
self.pref_actions = {}
|
|
|
|
preferences_menu = menu_bar.addMenu("Preferences")
|
|
preferences_actions = [
|
|
("2D Data Bypass", "", self.is_2d_bypass_func, resource_path("icons/warning_off_24dp_1F1F1F.svg"), "2d_data_bypass"),
|
|
("Incompatible Save Bypass", "", self.incompatable_save_bypass_func, resource_path("icons/warning_off_24dp_1F1F1F.svg"), "incompatible_save_bypass"),
|
|
("Missing Events Bypass", "", self.missing_events_bypass_func, resource_path("icons/warning_off_24dp_1F1F1F.svg"), "missing_events_bypass"),
|
|
("Analysis Clearing Bypass", "", self.analysis_clearing_bypass_func, resource_path("icons/warning_off_24dp_1F1F1F.svg"), "analysis_clearing_bypass"),
|
|
("Folding Bypass", "", self.folding_bypass_func, resource_path("icons/warning_off_24dp_1F1F1F.svg"), "folding_bypass"),
|
|
]
|
|
|
|
for name, shortcut, slot, icon, config_key in preferences_actions:
|
|
action = make_action(name, shortcut, slot, icon=icon, checkable=True)
|
|
preferences_menu.addAction(action)
|
|
self.pref_actions[config_key] = action
|
|
|
|
terminal_menu = menu_bar.addMenu("Terminal")
|
|
terminal_actions = [
|
|
("New Terminal", "Ctrl+Alt+T", self.terminal_gui, resource_path("icons/terminal_24dp_1F1F1F.svg")),
|
|
]
|
|
for name, shortcut, slot, icon in terminal_actions:
|
|
terminal_menu.addAction(make_action(name, shortcut, slot, icon=icon))
|
|
|
|
self.sync_app_with_config()
|
|
|
|
self.statusbar.showMessage("Ready")
|
|
|
|
|
|
def update_sections(self, index):
|
|
# Clear previous sections
|
|
for i in reversed(range(self.rows_layout.count())):
|
|
widget = self.rows_layout.itemAt(i).widget()
|
|
if widget is not None:
|
|
widget.deleteLater()
|
|
self.param_sections.clear()
|
|
|
|
self.global_param_widgets = {}
|
|
|
|
# Add ParamSection widgets from SECTIONS
|
|
for section in SECTIONS:
|
|
self.section_widget = ParamSection(section, self.global_param_widgets)
|
|
self.rows_layout.addWidget(self.section_widget)
|
|
|
|
self.param_sections.append(self.section_widget)
|
|
|
|
for sec in self.param_sections:
|
|
sec.update_dependencies()
|
|
|
|
|
|
def clear_all(self):
|
|
"""
|
|
Forcefully purges all data, kills background tasks,
|
|
and resets the memory heap.
|
|
"""
|
|
|
|
self.top_left_widget.clear()
|
|
|
|
if hasattr(self, "last_clicked_bubble"):
|
|
self.last_clicked_bubble = None
|
|
|
|
if hasattr(self, "result_timer") and self.result_timer:
|
|
self.result_timer.stop()
|
|
self.result_timer.deleteLater()
|
|
self.result_timer = None
|
|
|
|
if hasattr(self, "result_process") and self.result_process:
|
|
if self.result_process.is_alive():
|
|
self.result_process.terminate()
|
|
self.result_process.join(timeout=1)
|
|
self.result_process = None
|
|
|
|
if hasattr(self, "file_executor") and self.file_executor:
|
|
self.file_executor.shutdown(wait=False, cancel_futures=True)
|
|
self.file_executor = None
|
|
|
|
self.pending_files_count = 0
|
|
# Increment session so any 'in-flight' callbacks are ignored
|
|
if hasattr(self, "loading_session_id"):
|
|
self.loading_session_id += 1
|
|
|
|
# Disconnect the buttons to break potential closures
|
|
for btn in [self.button1, self.button3]:
|
|
try:
|
|
btn.clicked.disconnect()
|
|
except (TypeError, RuntimeError): #NOTE: Till raises RuntimeWarnings?
|
|
pass
|
|
|
|
# UI Cleanup
|
|
self.right_column_widget.hide()
|
|
while self.bubble_layout.count():
|
|
item = self.bubble_layout.takeAt(0)
|
|
widget = item.widget()
|
|
if widget:
|
|
# Forcefully disconnect signals to be safe
|
|
try:
|
|
widget.clicked.disconnect()
|
|
widget.rightClicked.disconnect()
|
|
except:
|
|
pass
|
|
widget.deleteLater()
|
|
|
|
self.bubble_layout.setSpacing(0)
|
|
self.bubble_layout.setContentsMargins(0, 0, 0, 0)
|
|
self.bubble_container.setMinimumSize(0, 0)
|
|
self.bubble_container.resize(0, 0)
|
|
self.scroll_area.updateGeometry()
|
|
|
|
# Data Purge
|
|
self.bubble_widgets = {}
|
|
self.files_results = {}
|
|
self.files_done = set()
|
|
self.files_failed = set()
|
|
|
|
self.raw_haemo_dict = {}
|
|
self.config_dict = {}
|
|
self.epochs_dict = {}
|
|
self.fig_bytes_dict = {}
|
|
self.cha_dict = {}
|
|
self.contrast_results_dict = {}
|
|
self.df_ind_dict = {}
|
|
self.design_matrix_dict = {}
|
|
self.valid_dict = {}
|
|
|
|
self.metadata_cache = {}
|
|
|
|
if hasattr(self, "selected_paths"): self.selected_paths = []
|
|
if hasattr(self, "selected_path"): self.selected_path = None
|
|
|
|
self.button1.setText("Process")
|
|
self.button1.clicked.connect(self.on_run_task)
|
|
self.button1.setVisible(False)
|
|
self.button3.setVisible(False)
|
|
|
|
self.statusBar().showMessage("All data has been cleared.")
|
|
|
|
#NOTE: leave this here for now. needs other parts uncommented to work
|
|
# self.check_memory_leak()
|
|
# self.find_referrers()
|
|
|
|
# snapshot2 = tracemalloc.take_snapshot()
|
|
|
|
# # 4. Show the "Compare" - This shows what REFUSED to die
|
|
# stats = snapshot2.compare_to(snapshot1, 'lineno')
|
|
# print("[ Memory that stayed after Clear ]")
|
|
# for stat in stats[:10]:
|
|
# print(stat)
|
|
# print("Top 10 growing object types in RAM:")
|
|
# objgraph.show_most_common_types(limit=10)
|
|
|
|
def update_recent_projects_menu(self):
|
|
"""Clears and rebuilds the Recent Projects submenu items."""
|
|
self.recent_projects_menu.clear()
|
|
|
|
raw_projects = file_cfg.get("File", "recent_projects", fallback="")
|
|
projects = [p.strip() for p in raw_projects.split(",") if p.strip()]
|
|
|
|
if not projects:
|
|
no_recent = self.recent_projects_menu.addAction("No Recent Projects")
|
|
no_recent.setEnabled(False)
|
|
return
|
|
|
|
for i, project_path in enumerate(projects):
|
|
action = QAction(f"{i+1}: {project_path}", self)
|
|
action.setToolTip(project_path)
|
|
action.triggered.connect(lambda checked, path=project_path: self.open_recent_project(path))
|
|
self.recent_projects_menu.addAction(action)
|
|
|
|
def add_to_recent_projects(self, project_path):
|
|
"""Adds a project path, moves it to the top, and hard caps at 10."""
|
|
raw_projects = file_cfg.get("File", "recent_projects", fallback="")
|
|
projects = [p.strip() for p in raw_projects.split(",") if p.strip()]
|
|
|
|
if project_path in projects:
|
|
projects.remove(project_path)
|
|
|
|
projects.insert(0, project_path)
|
|
projects = projects[:10] # Hard cap of 10 items
|
|
|
|
file_cfg.set("File", "recent_projects", ",".join(projects))
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save config history: {e}")
|
|
|
|
self.update_recent_projects_menu()
|
|
|
|
def open_recent_project(self, project_path):
|
|
"""The slot that executes when a recent project entry is clicked."""
|
|
if os.path.exists(project_path):
|
|
print(f"Opening recent project: {project_path}")
|
|
|
|
self.project_loader(project_path)
|
|
|
|
self.add_to_recent_projects(project_path)
|
|
else:
|
|
QMessageBox.warning(self, "Project Not Found", f"The project file could not be found:\n{project_path}")
|
|
# Clean out the broken path
|
|
raw_projects = file_cfg.get("File", "recent_projects", fallback="")
|
|
projects = [p.strip() for p in raw_projects.split(",") if p.strip() and p.strip() != project_path]
|
|
file_cfg.set("File", "recent_projects", ",".join(projects))
|
|
self.update_recent_projects_menu()
|
|
|
|
|
|
|
|
def update_recent_files_menu(self):
|
|
"""Clears and rebuilds the Recent Files submenu items."""
|
|
self.recent_files_menu.clear()
|
|
|
|
raw_files = file_cfg.get("File", "recent_files", fallback="")
|
|
files = [f.strip() for f in raw_files.split(",") if f.strip()]
|
|
|
|
if not files:
|
|
no_recent = self.recent_files_menu.addAction("No Recent Files")
|
|
no_recent.setEnabled(False)
|
|
return
|
|
|
|
for i, file_path in enumerate(files):
|
|
# Display just the file name (e.g. 'data.snirf'), but keep the full path as a tool tip
|
|
action = QAction(f"{i+1}: {file_path}", self)
|
|
# Connect it so it passes the specific path when clicked
|
|
action.triggered.connect(lambda checked, path=file_path: self.open_recent_file(path))
|
|
self.recent_files_menu.addAction(action)
|
|
|
|
|
|
def add_to_recent_files(self, file_path):
|
|
"""Adds a path, moves it to the top, and hard caps the list at 10."""
|
|
raw_files = file_cfg.get("File", "recent_files", fallback="")
|
|
files = [f.strip() for f in raw_files.split(",") if f.strip()]
|
|
|
|
if file_path in files:
|
|
files.remove(file_path)
|
|
|
|
files.insert(0, file_path)
|
|
files = files[:10]
|
|
|
|
file_cfg.set("File", "recent_files", ",".join(files))
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save config history: {e}")
|
|
|
|
self.update_recent_files_menu()
|
|
|
|
def open_recent_file(self, file_path):
|
|
"""The slot that executes when someone clicks a recent file entry."""
|
|
if os.path.exists(file_path):
|
|
print(f"Opening recent file: {file_path}")
|
|
self._load_files_into_pipeline([os.path.normpath(file_path)])
|
|
|
|
# Refresh position to top
|
|
self.add_to_recent_files(file_path)
|
|
else:
|
|
QMessageBox.warning(self, "File Not Found", f"The file could not be found:\n{file_path}")
|
|
# Clean up the broken link from history
|
|
raw_files = file_cfg.get("File", "recent_files", fallback="")
|
|
files = [f.strip() for f in raw_files.split(",") if f.strip() and f.strip() != file_path]
|
|
file_cfg.set("File", "recent_files", ",".join(files))
|
|
self.update_recent_files_menu()
|
|
|
|
|
|
def reset_window_layout(self):
|
|
"""
|
|
Snaps all draggable splitters back to their default proportional positions.
|
|
"""
|
|
total_width = self.main_h_splitter.width()
|
|
left_w = int(total_width * 26 / 45)
|
|
right_w = total_width - left_w
|
|
self.main_h_splitter.setSizes([left_w, right_w])
|
|
|
|
total_height = self.left_v_splitter.height()
|
|
top_h = int(total_height * 0.30)
|
|
bottom_h = total_height - top_h
|
|
self.left_v_splitter.setSizes([top_h, bottom_h])
|
|
|
|
self.statusBar().showMessage("Window layout reset to default.", 2000)
|
|
|
|
|
|
def open_launcher_window(self):
|
|
self.launcher_window = ViewerLauncherWidget(self.raw_haemo_dict, self.config_dict, self.fig_bytes_dict, self.cha_dict, self.contrast_results_dict, self.df_ind_dict, self.design_matrix_dict, self.epochs_dict, self.folding_bypass)
|
|
self.launcher_window.show()
|
|
|
|
def copy_text(self):
|
|
self.top_left_widget.copy() # Trigger copy
|
|
self.statusbar.showMessage("Copied to clipboard") # Show status message
|
|
|
|
def cut_text(self):
|
|
self.top_left_widget.cut() # Trigger cut
|
|
self.statusbar.showMessage("Cut to clipboard") # Show status message
|
|
|
|
def paste_text(self):
|
|
self.top_left_widget.paste() # Trigger paste
|
|
self.statusbar.showMessage("Pasted from clipboard") # Show status message
|
|
|
|
def _update_config_setting(self, key, value):
|
|
"""Helper to update memory configuration and save to disk."""
|
|
# configparser expects string values
|
|
file_cfg.set("Preferences", key, str(value).lower())
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save setting '{key}' to disk: {e}")
|
|
|
|
def is_2d_bypass_func(self, checked):
|
|
self.is_2d_bypass = checked
|
|
self._update_config_setting("2d_data_bypass", checked)
|
|
|
|
def incompatable_save_bypass_func(self, checked):
|
|
self.incompatible_save_bypass = checked
|
|
self._update_config_setting("incompatible_save_bypass", checked)
|
|
|
|
def missing_events_bypass_func(self, checked):
|
|
self.missing_events_bypass = checked
|
|
self._update_config_setting("missing_events_bypass", checked)
|
|
|
|
def analysis_clearing_bypass_func(self, checked):
|
|
self.analysis_clearing_bypass = checked
|
|
self._update_config_setting("analysis_clearing_bypass", checked)
|
|
|
|
def folding_bypass_func(self, checked):
|
|
self.folding_bypass = checked
|
|
self._update_config_setting("folding_bypass", checked)
|
|
|
|
def about_window(self):
|
|
if self.about is None or not self.about.isVisible():
|
|
self.about = AboutWindow(self)
|
|
self.about.show()
|
|
|
|
def user_guide(self):
|
|
if self.help is None or not self.help.isVisible():
|
|
self.help = UserGuideWindow(self)
|
|
self.help.show()
|
|
|
|
def terminal_gui(self):
|
|
if self.terminal is None or not self.terminal.isVisible():
|
|
self.terminal = TerminalWindow(self)
|
|
self.terminal.show()
|
|
|
|
def update_optode_positions(self):
|
|
if self.optodes is None or not self.optodes.isVisible():
|
|
self.optodes = UpdateOptodesWindow(self)
|
|
self.optodes.show()
|
|
|
|
def update_event_markers(self):
|
|
if self.events is None or not self.events.isVisible():
|
|
self.events = UpdateEventsWindow(self, EventUpdateMode.WRITE_SNIRF, "Manual SNIRF Edit")
|
|
self.events.show()
|
|
|
|
def update_event_markers_blazes(self):
|
|
if self.events is None or not self.events.isVisible():
|
|
self.events = UpdateEventsBlazesWindow(self, EventUpdateMode.WRITE_SNIRF, "Manual SNIRF Edit")
|
|
self.events.show()
|
|
|
|
def show_update_changelog(self):
|
|
welcome = WelcomeDialog(self, direct=False)
|
|
welcome.show()
|
|
|
|
def reset_to_default_configuration(self):
|
|
"""Asks user for confirmation, then resets all settings to defaults."""
|
|
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"Reset Configuration",
|
|
"Are you sure you want to reset the application and all settings to their default values? This cannot be undone.",
|
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
|
QMessageBox.StandardButton.No # Default focus on 'No'
|
|
)
|
|
|
|
# 2. If the user confirmed, perform the reset
|
|
if reply == QMessageBox.StandardButton.Yes:
|
|
try:
|
|
# Overwrite the file with the template string constant
|
|
with open(cfg_path, "w") as f:
|
|
f.write(DEFAULT_CONFIG.strip())
|
|
|
|
# Reload the config parser from the freshly written file
|
|
file_cfg.read(cfg_path)
|
|
print("Configuration reset to defaults successfully.")
|
|
|
|
except Exception as e:
|
|
print(f"Error resetting config file ({e}). Resetting in-memory only.")
|
|
# Fallback to loading the string into memory if file writing fails
|
|
file_cfg.read_string(DEFAULT_CONFIG)
|
|
|
|
self.sync_app_with_config()
|
|
|
|
self.statusbar.showMessage("All settings have been reset to their default values.", 5000)
|
|
|
|
|
|
def sync_app_with_config(self):
|
|
"""Reads values from file_cfg and updates both internal variables and UI checkmarks."""
|
|
# 1. Sync internal application state variables
|
|
self.is_2d_bypass = file_cfg.getboolean("Preferences", "2d_data_bypass", fallback=False)
|
|
self.incompatible_save_bypass = file_cfg.getboolean("Preferences", "incompatible_save_bypass", fallback=False)
|
|
self.missing_events_bypass = file_cfg.getboolean("Preferences", "missing_events_bypass", fallback=False)
|
|
self.analysis_clearing_bypass = file_cfg.getboolean("Preferences", "analysis_clearing_bypass", fallback=False)
|
|
self.folding_bypass = file_cfg.getboolean("Preferences", "folding_bypass", fallback=False)
|
|
|
|
|
|
# 2. Sync the UI Menu checkmarks visually
|
|
if hasattr(self, 'pref_actions'):
|
|
self.pref_actions["2d_data_bypass"].setChecked(self.is_2d_bypass)
|
|
self.pref_actions["incompatible_save_bypass"].setChecked(self.incompatible_save_bypass)
|
|
self.pref_actions["missing_events_bypass"].setChecked(self.missing_events_bypass)
|
|
self.pref_actions["analysis_clearing_bypass"].setChecked(self.analysis_clearing_bypass)
|
|
self.pref_actions["folding_bypass"].setChecked(self.folding_bypass)
|
|
|
|
if hasattr(self, 'recent_files_menu'):
|
|
self.update_recent_files_menu()
|
|
|
|
if hasattr(self, 'recent_projects_menu'):
|
|
self.update_recent_projects_menu()
|
|
|
|
def open_file_dialog(self):
|
|
file_path, _ = QFileDialog.getOpenFileName(self, "Open File", "", "SNIRF Files (*.snirf);;All Files (*)")
|
|
if file_path:
|
|
self._load_files_into_pipeline([os.path.normpath(file_path)])
|
|
|
|
def open_folder_dialog(self):
|
|
folder_path = QFileDialog.getExistingDirectory(self, "Select Folder", "")
|
|
if folder_path:
|
|
snirf_files = [os.path.normpath(str(f)) for f in Path(folder_path).glob("*.snirf")]
|
|
self._load_files_into_pipeline(snirf_files)
|
|
|
|
|
|
def _load_files_into_pipeline(self, file_paths):
|
|
if not file_paths:
|
|
return
|
|
|
|
# 1. Warm up the executor if needed
|
|
if not hasattr(self, 'file_executor') or self.file_executor is None:
|
|
self.file_executor = concurrent.futures.ProcessPoolExecutor(max_workers=1)
|
|
|
|
# 2. Track this session to prevent ghost updates
|
|
if not hasattr(self, 'loading_session_id'): self.loading_session_id = 0
|
|
self.loading_session_id += 1
|
|
current_session = self.loading_session_id
|
|
|
|
# 3. Setup internal tracking if not exists
|
|
if not hasattr(self, 'bubble_widgets'): self.bubble_widgets = {}
|
|
if not hasattr(self, 'selected_paths'): self.selected_paths = []
|
|
if not hasattr(self, 'metadata_cache'): self.metadata_cache = {}
|
|
|
|
# Filter out files already in the UI to avoid duplicates
|
|
new_files = [p for p in file_paths if p not in self.selected_paths]
|
|
if not new_files:
|
|
return
|
|
|
|
# Update the pending count for the current load batch
|
|
if not hasattr(self, 'pending_files_count'): self.pending_files_count = 0
|
|
self.pending_files_count += len(new_files)
|
|
|
|
for path in new_files:
|
|
self.selected_paths.append(path)
|
|
self.add_to_recent_files(path)
|
|
|
|
# Create the UI Bubble (Disconnected by default)
|
|
display_name = os.path.basename(path)
|
|
bubble = ProgressBubble(display_name, path)
|
|
bubble.setCursor(Qt.CursorShape.WaitCursor)
|
|
bubble.set_loading_state(True)
|
|
|
|
self.bubble_widgets[path] = bubble
|
|
self.bubble_layout.addWidget(bubble)
|
|
|
|
# 4. Queue the background work
|
|
future = self.file_executor.submit(_extract_metadata_worker, path)
|
|
# Use lambda with defaults to freeze the path and session at this moment
|
|
future.add_done_callback(
|
|
lambda f, p=path, s=current_session: self._on_metadata_ready(f, p, s)
|
|
)
|
|
|
|
self.button1.setVisible(True)
|
|
self.statusBar().showMessage(f"Loading {len(new_files)} new file(s)...")
|
|
|
|
|
|
# TODO: Is this needed?
|
|
# def open_multiple_folders_dialog(self):
|
|
# while True:
|
|
# folder_path = QFileDialog.getExistingDirectory(self, "Select Folder")
|
|
# if not folder_path:
|
|
# break
|
|
|
|
# snirf_files = [str(f) for f in Path(folder_path).glob("*.snirf")]
|
|
|
|
# if not hasattr(self, 'selected_paths'):
|
|
# self.selected_paths = []
|
|
|
|
# for file_path in snirf_files:
|
|
# if file_path not in self.selected_paths:
|
|
# self.selected_paths.append(file_path)
|
|
|
|
# self.show_files_as_bubbles(folder_path)
|
|
|
|
# # for section_widget in self.param_sections:
|
|
# # if hasattr(section_widget, 'update_annotation_dropdown_from_loaded_files'):
|
|
# # if "REMOVE_EVENTS" in section_widget.widgets:
|
|
# # section_widget.update_annotation_dropdown_from_loaded_files(self.bubble_widgets, self.button1)
|
|
# # break
|
|
# # else:
|
|
# # print("[MainWindow] Could not find ParamSection with 'REMOVE_EVENTS' widget")
|
|
|
|
|
|
# # Ask if the user wants to add another
|
|
# more = QMessageBox.question(
|
|
# self,
|
|
# "Add Another?",
|
|
# "Do you want to select another folder?",
|
|
# QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
|
# )
|
|
# if more == QMessageBox.StandardButton.No:
|
|
# break
|
|
|
|
# self.button1.setVisible(True)
|
|
|
|
|
|
def save_project(self, onCrash=False):
|
|
|
|
if hasattr(self, 'current_file') and self.current_file:
|
|
self.file_metadata[self.current_file] = {
|
|
key: field.text().strip() for key, field in self.meta_fields.items()
|
|
}
|
|
|
|
has_metadata = any(
|
|
any(val for val in meta.values())
|
|
for meta in self.file_metadata.values()
|
|
)
|
|
has_param_changes = any(section.has_any_changes() for section in self.param_sections)
|
|
|
|
# Check if there is processed data
|
|
has_processed_data = bool(getattr(self, 'raw_haemo_dict', None))
|
|
|
|
if not (has_processed_data or has_metadata or has_param_changes):
|
|
if not onCrash: # Don't show popups during a crash/autosave
|
|
QMessageBox.warning(
|
|
self,
|
|
"Save Project",
|
|
"There is no processed data to save. Please process some data before saving."
|
|
)
|
|
return
|
|
|
|
if hasattr(self, 'current_file') and self.current_file:
|
|
self.file_metadata[self.current_file] = {
|
|
key: field.text() for key, field in self.meta_fields.items()
|
|
}
|
|
|
|
if not onCrash:
|
|
filename, _ = QFileDialog.getSaveFileName(
|
|
self, "Save Project", "", "FLARE Project (*.flare)"
|
|
)
|
|
if not filename:
|
|
return
|
|
else:
|
|
if PLATFORM_NAME == "darwin":
|
|
filename = os.path.join(os.path.dirname(sys.executable), "../../../flares_autosave.flare")
|
|
else:
|
|
filename = os.path.join(os.getcwd(), "flares_autosave.flare")
|
|
|
|
try:
|
|
# Ensure the filename has the proper extension
|
|
if not filename.endswith(".flare"):
|
|
filename += ".flare"
|
|
|
|
project_path = Path(filename).resolve()
|
|
project_dir = project_path.parent
|
|
|
|
file_list = [
|
|
self._get_safe_path(bubble.file_path, project_dir)
|
|
for bubble in self.bubble_widgets.values()
|
|
]
|
|
|
|
progress_states = {
|
|
self._get_safe_path(bubble.file_path, project_dir): bubble.current_step
|
|
for bubble in self.bubble_widgets.values()
|
|
}
|
|
|
|
rel_metadata = {}
|
|
for full_path, meta in self.metadata_cache.items():
|
|
try:
|
|
# Resolve to absolute to be safe, then make relative to project_dir
|
|
safe_path = self._get_safe_path(full_path, project_dir)
|
|
rel_metadata[safe_path] = meta
|
|
except Exception as e:
|
|
print(f"Metadata conversion failed for {full_path}: {e}")
|
|
|
|
print(rel_metadata)
|
|
|
|
rel_file_params = {
|
|
self._get_safe_path(f_path, project_dir): meta
|
|
for f_path, meta in self.file_metadata.items()
|
|
}
|
|
|
|
|
|
current_params = self.get_all_current_ui_params()
|
|
|
|
# fallback - if UI reading fails, try the first processed file's config
|
|
if not current_params and self.config_dict:
|
|
first_file = next(iter(self.config_dict.keys()))
|
|
current_params = self.config_dict[first_file]
|
|
|
|
version = CURRENT_VERSION
|
|
project_data = {
|
|
"version": version,
|
|
"file_list": file_list,
|
|
"progress_states": progress_states,
|
|
"raw_haemo_dict": self.raw_haemo_dict,
|
|
"file_metadata": rel_metadata,
|
|
"file_parameters": rel_file_params,
|
|
"config_dict": self.config_dict,
|
|
"epochs_dict": self.epochs_dict,
|
|
"fig_bytes_dict": self.fig_bytes_dict,
|
|
"cha_dict": self.cha_dict,
|
|
"current_ui_params": current_params,
|
|
"contrast_results_dict": self.contrast_results_dict,
|
|
"df_ind_dict": self.df_ind_dict,
|
|
"design_matrix_dict": self.design_matrix_dict,
|
|
"valid_dict": self.valid_dict,
|
|
}
|
|
|
|
def sanitize(obj):
|
|
if isinstance(obj, Path):
|
|
return str(PurePosixPath(obj))
|
|
elif isinstance(obj, dict):
|
|
return {sanitize(k): sanitize(v) for k, v in obj.items()}
|
|
elif isinstance(obj, list):
|
|
return [sanitize(i) for i in obj]
|
|
return obj
|
|
|
|
project_data = sanitize(project_data)
|
|
|
|
self.saving_overlay = SavingOverlay(self)
|
|
self.saving_overlay.resize(self.size()) # Cover the main window
|
|
self.saving_overlay.show()
|
|
|
|
# Start the background save thread
|
|
self.save_thread = SaveProjectThread(filename, project_data)
|
|
|
|
# When finished, close overlay and show success
|
|
self.save_thread.finished_signal.connect(lambda f: (
|
|
self.saving_overlay.close(),
|
|
QMessageBox.information(self, "Success", f"Project saved to:\n{f}")
|
|
))
|
|
self.save_thread.error_signal.connect(lambda e: (
|
|
self.saving_overlay.close(),
|
|
QMessageBox.critical(self, "Error", f"Failed to save project:\n{e}")
|
|
))
|
|
|
|
self.save_thread.start()
|
|
|
|
except Exception as e:
|
|
if not onCrash:
|
|
QMessageBox.critical(self, "Error", f"Failed to save project:\n{e}")
|
|
|
|
|
|
def _get_safe_path(self, target_path, start_dir):
|
|
try:
|
|
# Convert both to absolute paths first
|
|
target = Path(target_path).resolve()
|
|
base = Path(start_dir).resolve()
|
|
|
|
rel = os.path.relpath(target, base)
|
|
return str(PurePosixPath(rel))
|
|
except ValueError:
|
|
return str(PurePosixPath(target))
|
|
|
|
|
|
def load_project(self):
|
|
filename, _ = QFileDialog.getOpenFileName(
|
|
self, "Load Project", "", "FLARE Project (*.flare)"
|
|
)
|
|
if not filename:
|
|
return
|
|
|
|
self.project_loader(filename=filename)
|
|
|
|
|
|
def project_loader(self, filename):
|
|
|
|
try:
|
|
with open(filename, "rb") as f:
|
|
data = pickle.load(f)
|
|
|
|
# Check for potentially broken saves
|
|
checks = [
|
|
("version", "<=1.1.7"),
|
|
("file_metadata", "<=1.2.2"),
|
|
("file_parameters", "<=1.3.0")
|
|
]
|
|
|
|
for key, ver_str in checks:
|
|
if key not in data:
|
|
msg = (f"This project was saved in an earlier version of FLARES ({ver_str}) "
|
|
"and is potentially not compatible with this version. ")
|
|
|
|
if self.incompatible_save_bypass:
|
|
QMessageBox.warning(self, f"Warning - {APP_NAME.upper()}", msg +
|
|
"You are receiving this warning because you have 'Incompatible Save Bypass' turned on. "
|
|
"FLARES will now attempt to load the project. It is strongly recommended to recreate the project file.")
|
|
break
|
|
else:
|
|
QMessageBox.critical(self, f"Error - {APP_NAME.upper()}", msg +
|
|
"The file can attempt to be loaded if 'Incompatible Save Bypass' is selected in the 'Preferences' menu.")
|
|
return
|
|
|
|
|
|
self.raw_haemo_dict = data.get("raw_haemo_dict", {})
|
|
self.config_dict = data.get("config_dict", {})
|
|
self.epochs_dict = data.get("epochs_dict", {})
|
|
self.fig_bytes_dict = data.get("fig_bytes_dict", {})
|
|
self.cha_dict = data.get("cha_dict", {})
|
|
self.contrast_results_dict = data.get("contrast_results_dict", {})
|
|
self.df_ind_dict = data.get("df_ind_dict", {})
|
|
self.design_matrix_dict = data.get("design_matrix_dict", {})
|
|
self.valid_dict = data.get("valid_dict", {})
|
|
|
|
project_dir = Path(filename).parent
|
|
|
|
saved_cache = data.get("file_metadata", {})
|
|
raw_params = data.get("file_parameters", {})
|
|
self.metadata_cache = {}
|
|
self.file_metadata = {}
|
|
|
|
for rel_path, meta_content in saved_cache.items():
|
|
abs_path = str((project_dir / Path(rel_path)).resolve())
|
|
self.metadata_cache[abs_path] = meta_content
|
|
|
|
# Convert saved relative paths to absolute paths
|
|
file_list = [str((project_dir / Path(rel_path)).resolve()) for rel_path in data["file_list"]]
|
|
|
|
# Also resolve progress_states with updated paths
|
|
raw_progress = data.get("progress_states", {})
|
|
progress_states = {
|
|
str((project_dir / Path(rel_path)).resolve()): step
|
|
for rel_path, step in raw_progress.items()
|
|
}
|
|
|
|
for rel_path in data["file_list"]:
|
|
abs_path = str((project_dir / Path(rel_path)).resolve())
|
|
|
|
if rel_path in raw_params:
|
|
# Scenario A: New format found
|
|
self.file_metadata[abs_path] = raw_params[rel_path]
|
|
elif abs_path in self.config_dict:
|
|
# Scenario B: Fallback to old config_dict
|
|
old_cfg = self.config_dict[abs_path]
|
|
self.file_metadata[abs_path] = {
|
|
"AGE": str(old_cfg.get("AGE", "")),
|
|
"GENDER": str(old_cfg.get("GENDER", "")),
|
|
"GROUP": str(old_cfg.get("GROUP", ""))
|
|
}
|
|
else:
|
|
# Scenario C: Empty default
|
|
self.file_metadata[abs_path] = {"AGE": "", "GENDER": "", "GROUP": ""}
|
|
|
|
self.show_files_as_bubbles_from_list(file_list, progress_states, filename)
|
|
|
|
if "current_ui_params" in data:
|
|
self.restore_sections_from_config(data["current_ui_params"])
|
|
|
|
elif self.config_dict:
|
|
first_file = next(iter(self.config_dict.keys()))
|
|
self.restore_sections_from_config(self.config_dict[first_file])
|
|
|
|
has_data = bool(self.raw_haemo_dict)
|
|
self.button1.setVisible(not has_data)
|
|
self.button3.setVisible(has_data)
|
|
|
|
self.add_to_recent_projects(os.path.normpath(filename))
|
|
|
|
QMessageBox.information(self, "Loaded", f"Project loaded from:\n{filename}")
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", f"Failed to load project:\n{e}")
|
|
|
|
|
|
def restore_sections_from_config(self, config):
|
|
"""
|
|
Fill all ParamSection widgets with values from a participant's config.
|
|
"""
|
|
for section_widget in self.param_sections:
|
|
widgets_dict = getattr(section_widget, 'widgets', None)
|
|
if widgets_dict is None:
|
|
continue
|
|
|
|
for name, widget_info in widgets_dict.items():
|
|
if name not in config:
|
|
continue
|
|
|
|
value = config[name]
|
|
|
|
widget = widget_info["widget"]
|
|
w_type = widget_info.get("type")
|
|
|
|
# QLineEdit (int, float, str)
|
|
if isinstance(widget, QLineEdit):
|
|
widget.blockSignals(True)
|
|
widget.setText(str(value))
|
|
widget.blockSignals(False)
|
|
widget.update()
|
|
|
|
# QComboBox (bool, list)
|
|
elif isinstance(widget, QComboBox):
|
|
widget.blockSignals(True)
|
|
widget.setCurrentText(str(value))
|
|
widget.blockSignals(False)
|
|
widget.update()
|
|
|
|
# QSpinBox (range)
|
|
elif isinstance(widget, QSpinBox):
|
|
widget.blockSignals(True)
|
|
try:
|
|
widget.setValue(int(value))
|
|
except Exception:
|
|
pass
|
|
widget.blockSignals(False)
|
|
widget.update()
|
|
|
|
# After restoring, make sure dependencies are updated
|
|
if hasattr(section_widget, 'update_dependencies'):
|
|
section_widget.update_dependencies()
|
|
|
|
|
|
# def show_files_as_bubbles(self, folder_paths):
|
|
|
|
# if isinstance(folder_paths, str):
|
|
# folder_paths = [folder_paths]
|
|
|
|
# # Clear previous bubbles
|
|
# # while self.bubble_layout.count():
|
|
# # item = self.bubble_layout.takeAt(0)
|
|
# # widget = item.widget()
|
|
# # if widget:
|
|
# # widget.deleteLater()
|
|
|
|
# temp_bubble = ProgressBubble("Test Bubble", "") # A dummy bubble for measurement
|
|
# temp_bubble.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy. Preferred)
|
|
# # temp_bubble.setAttribute(Qt.WA_OpaquePaintEvent) # Improve rendering?
|
|
# temp_bubble.adjustSize() # Adjust size after the widget is created
|
|
# bubble_width = temp_bubble.width() # Get the actual width of a bubble
|
|
# available_width = self.bubble_container.width()
|
|
|
|
# cols = max(1, available_width // bubble_width) # Ensure at least 1 column
|
|
|
|
# index = 0
|
|
# if not hasattr(self, 'selected_paths'):
|
|
# self.selected_paths = []
|
|
|
|
# for folder_path in folder_paths:
|
|
# if not os.path.isdir(folder_path):
|
|
# continue
|
|
|
|
# snirf_files = [str(f) for f in Path(folder_path).glob("*.snirf")]
|
|
|
|
# for full_path in snirf_files:
|
|
|
|
# display_name = f"{os.path.basename(folder_path)} / {os.path.basename(full_path)}"
|
|
# bubble = ProgressBubble(display_name, full_path)
|
|
# bubble.set_loading_state(True)
|
|
# bubble.setCursor(Qt.CursorShape.WaitCursor)
|
|
|
|
# self.bubble_widgets[full_path] = bubble
|
|
|
|
# if full_path not in self.selected_paths:
|
|
# self.selected_paths.append(full_path)
|
|
|
|
# row = index // cols
|
|
# col = index % cols
|
|
# self.bubble_layout.addWidget(bubble, row, col)
|
|
# index += 1
|
|
|
|
# self.statusBar().showMessage(f"{index} file(s) loaded from: {', '.join(folder_paths)}")
|
|
|
|
def get_all_current_ui_params(self):
|
|
"""Gathers current values from all UI widgets across all sections."""
|
|
current_ui_config = {}
|
|
try:
|
|
for section in self.param_sections:
|
|
# This calls the get_param_values() method you shared earlier
|
|
section_values = section.get_param_values()
|
|
current_ui_config.update(section_values)
|
|
return current_ui_config
|
|
except Exception as e:
|
|
print(f"Error reading UI parameters: {e}")
|
|
return None
|
|
|
|
def show_files_as_bubbles_from_list(self, file_list, progress_states=None, filenames=None):
|
|
if not hasattr(self, 'file_executor') or self.file_executor is None:
|
|
self.file_executor = concurrent.futures.ProcessPoolExecutor(max_workers=1)
|
|
progress_states = progress_states or {}
|
|
|
|
# Initialize trackers and clear layout
|
|
if not hasattr(self, 'selected_paths'):
|
|
self.selected_paths = []
|
|
|
|
self.bubble_widgets = {}
|
|
|
|
while self.bubble_layout.count():
|
|
item = self.bubble_layout.takeAt(0)
|
|
widget = item.widget()
|
|
if widget:
|
|
widget.deleteLater()
|
|
|
|
# Process the file list
|
|
for index, file_path in enumerate(file_list):
|
|
file_path = str(file_path)
|
|
|
|
display_name = f"{os.path.basename(os.path.dirname(file_path))} / {os.path.basename(file_path)}"
|
|
|
|
# Create bubble
|
|
bubble = ProgressBubble(display_name, file_path)
|
|
bubble.clicked.connect(self.on_bubble_clicked)
|
|
bubble.rightClicked.connect(self.on_bubble_right_clicked)
|
|
|
|
if hasattr(self, 'file_metadata') and file_path in self.file_metadata:
|
|
meta = self.file_metadata[file_path]
|
|
|
|
parts = []
|
|
for key in ["AGE", "GENDER", "GROUP"]:
|
|
value = meta.get(key, "").strip()
|
|
if value:
|
|
parts.append(f"{key}: {value}")
|
|
|
|
suffix = f"{', '.join(parts)}" if parts else ""
|
|
bubble.setSuffixText(suffix)
|
|
|
|
# Track it
|
|
self.bubble_widgets[file_path] = bubble
|
|
if file_path not in self.selected_paths:
|
|
self.selected_paths.append(file_path)
|
|
|
|
# Restore saved progress but keep loading state active
|
|
step = progress_states.get(file_path, 0)
|
|
bubble.update_progress(step, active=False)
|
|
|
|
# Add to layout
|
|
self.bubble_layout.addWidget(bubble, index, 1)
|
|
|
|
# 4. Status Bar
|
|
msg = f"Project loaded: {len(file_list)} files."
|
|
if filenames:
|
|
msg += f" Source: {os.path.basename(filenames)}"
|
|
self.statusBar().showMessage(msg)
|
|
|
|
|
|
def get_suffix_from_meta_fields(self):
|
|
parts = []
|
|
for key, line_edit in self.meta_fields.items():
|
|
val = line_edit.text().strip()
|
|
if val:
|
|
parts.append(f"{key}: {val}")
|
|
return ", ".join(parts)
|
|
|
|
def on_bubble_clicked(self, bubble):
|
|
|
|
if self.current_file:
|
|
self.save_metadata(self.current_file)
|
|
|
|
if self.last_clicked_bubble and self.last_clicked_bubble != bubble:
|
|
suffix = self.get_suffix_from_meta_fields()
|
|
self.last_clicked_bubble.setSuffixText(suffix)
|
|
|
|
self.last_clicked_bubble = bubble
|
|
|
|
# show age / gender / group
|
|
self.right_column_widget.show()
|
|
|
|
file_path = bubble.file_path
|
|
if not os.path.exists(file_path):
|
|
self.top_left_widget.setText("File not found.")
|
|
return
|
|
|
|
size = os.path.getsize(file_path)
|
|
created = time.ctime(os.path.getctime(file_path))
|
|
modified = time.ctime(os.path.getmtime(file_path))
|
|
|
|
snirf_info = self.get_snirf_metadata_mne(file_path)
|
|
|
|
lines = [
|
|
f"File: {os.path.basename(file_path)}",
|
|
f"Size: {size:,} bytes",
|
|
f"Created: {created}",
|
|
f"Modified: {modified}",
|
|
f"Full Path: {file_path}\n",
|
|
]
|
|
|
|
info = "\n".join(lines)
|
|
|
|
if snirf_info is None:
|
|
info += f"\nSNIRF Metadata could not be loaded!"
|
|
else:
|
|
info += "\nSNIRF Metadata:\n"
|
|
for k, v in snirf_info.items():
|
|
if isinstance(v, list):
|
|
info += f" {k}:\n"
|
|
for item in v:
|
|
info += f" - {item}\n"
|
|
else:
|
|
info += f" {k}: {v}\n"
|
|
|
|
self.top_left_widget.setText(info)
|
|
|
|
clicked_bubble = self.sender()
|
|
file_path = clicked_bubble.file_path
|
|
|
|
# Save current file's metadata
|
|
if self.current_file:
|
|
self.save_metadata(self.current_file)
|
|
|
|
# Update current file
|
|
self.current_file = file_path
|
|
|
|
if file_path not in self.file_metadata:
|
|
self.file_metadata[file_path] = {key: "" for key in self.meta_fields}
|
|
|
|
# Load new file's metadata into the fields
|
|
metadata = self.file_metadata.get(file_path, {})
|
|
for key, field in self.meta_fields.items():
|
|
field.blockSignals(True)
|
|
field.setText(metadata.get(key, ""))
|
|
field.blockSignals(False)
|
|
|
|
|
|
def on_bubble_right_clicked(self, bubble, global_pos):
|
|
menu = QMenu(self)
|
|
action1 = menu.addAction(QIcon(resource_path("icons/folder_eye_24dp_1F1F1F.svg")), "Reveal")
|
|
action2 = menu.addAction(QIcon(resource_path("icons/remove_24dp_1F1F1F.svg")), "Remove")
|
|
|
|
action = menu.exec(global_pos)
|
|
if action == action1:
|
|
path = bubble.file_path
|
|
if os.path.exists(path):
|
|
if PLATFORM_NAME == "windows":
|
|
subprocess.run(["explorer", "/select,", os.path.normpath(path)])
|
|
elif PLATFORM_NAME == "darwin": # macOS
|
|
subprocess.run(["open", "-R", path])
|
|
else: # Linux
|
|
folder = os.path.dirname(path)
|
|
subprocess.run(["xdg-open", folder])
|
|
else:
|
|
print("File not found:", path)
|
|
|
|
elif action == action2:
|
|
if self.button3.isVisible():
|
|
reply = QMessageBox.warning(
|
|
self,
|
|
"Confirm Remove",
|
|
"Are you sure you want to remove this file? This will remove the analysis option and the processing will have to be performed again.",
|
|
QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel
|
|
)
|
|
if reply != QMessageBox.StandardButton.Ok:
|
|
return
|
|
else:
|
|
self.button3.setVisible(False)
|
|
|
|
self.top_left_widget.clear()
|
|
self.right_column_widget.hide()
|
|
|
|
parent_layout = bubble.parent().layout()
|
|
if parent_layout is not None:
|
|
parent_layout.removeWidget(bubble)
|
|
|
|
key_to_delete = None
|
|
for path, b in self.bubble_widgets.items():
|
|
if b is bubble:
|
|
key_to_delete = path
|
|
break
|
|
|
|
if key_to_delete:
|
|
del self.bubble_widgets[key_to_delete]
|
|
|
|
# Remove from selected_paths
|
|
if hasattr(self, 'selected_paths'):
|
|
try:
|
|
self.selected_paths.remove(bubble.file_path)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Remove from selected_path (if used)
|
|
if hasattr(self, 'selected_path') and self.selected_path == bubble.file_path:
|
|
self.selected_path = None
|
|
|
|
# for section_widget in self.param_sections:
|
|
# if hasattr(section_widget, 'update_annotation_dropdown_from_loaded_files'):
|
|
# if "REMOVE_EVENTS" in section_widget.widgets:
|
|
# section_widget.update_annotation_dropdown_from_loaded_files(self.bubble_widgets, self.button1)
|
|
# break
|
|
|
|
bubble.setParent(None)
|
|
bubble.deleteLater()
|
|
|
|
if getattr(self, 'last_clicked_bubble', None) is bubble:
|
|
self.last_clicked_bubble = None
|
|
|
|
def sync_bubble_data(self):
|
|
"""Refreshes the bubble and saves data in real-time."""
|
|
if self.current_file and self.last_clicked_bubble:
|
|
# Save the current state of all fields
|
|
self.save_metadata(self.current_file)
|
|
|
|
# Grab the updated suffix and apply it immediately
|
|
suffix = self.get_suffix_from_meta_fields()
|
|
self.last_clicked_bubble.setSuffixText(suffix)
|
|
|
|
def placeholder(self):
|
|
QMessageBox.information(self, "Placeholder", "This feature is not implemented yet.")
|
|
|
|
def save_metadata(self, file_path):
|
|
if not file_path:
|
|
return
|
|
|
|
self.file_metadata[file_path] = {
|
|
key: field.text()
|
|
for key, field in self.meta_fields.items()
|
|
}
|
|
|
|
def get_all_metadata(self):
|
|
# First, make sure current file's edits are saved
|
|
|
|
for field in self.meta_fields.values():
|
|
field.clearFocus()
|
|
|
|
# Save current file's metadata
|
|
if self.current_file:
|
|
self.save_metadata(self.current_file)
|
|
|
|
return self.file_metadata
|
|
|
|
|
|
def cancel_task(self):
|
|
self.button1.clicked.disconnect(self.cancel_task)
|
|
self.button1.setText("Stopping...")
|
|
|
|
if hasattr(self, "result_process") and self.result_process.is_alive():
|
|
parent = psutil.Process(self.result_process.pid)
|
|
children = parent.children(recursive=True)
|
|
for child in children:
|
|
try:
|
|
child.kill()
|
|
except psutil.NoSuchProcess:
|
|
pass
|
|
self.result_process.terminate()
|
|
self.result_process.join()
|
|
|
|
if hasattr(self, "result_timer") and self.result_timer.isActive():
|
|
self.result_timer.stop()
|
|
|
|
# if hasattr(self, "result_process") and self.result_process.is_alive():
|
|
# self.result_process.terminate() # Forcefully terminate the process
|
|
# self.result_process.join() # Wait for it to properly close
|
|
|
|
# # Stop the QTimer if running
|
|
# if hasattr(self, "result_timer") and self.result_timer.isActive():
|
|
# self.result_timer.stop()
|
|
|
|
for bubble in self.bubble_widgets.values():
|
|
bubble.mark_cancelled()
|
|
|
|
self.statusbar.showMessage("Processing cancelled.")
|
|
self.button1.clicked.connect(self.on_run_task)
|
|
self.button1.setText("Process")
|
|
|
|
|
|
'''MODULE FILE'''
|
|
def on_run_task(self):
|
|
|
|
#do the check
|
|
if not self.analysis_clearing_bypass:
|
|
if self.button3.isVisible():
|
|
msg = QMessageBox(self)
|
|
msg.setWindowTitle("Confirm - FLARES")
|
|
msg.setText("Processing new data will clear the current analysis. Continue? (If you do not want this dialog box to appear, toggle 'Analysis Clearing Bypass' from the Preferences menu.)")
|
|
|
|
# Add the OK and Cancel buttons
|
|
msg.setStandardButtons(QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel)
|
|
|
|
# Set the default button (highlighted)
|
|
msg.setDefaultButton(QMessageBox.StandardButton.Cancel)
|
|
|
|
# Capture the result
|
|
response = msg.exec()
|
|
|
|
if response == QMessageBox.StandardButton.Ok:
|
|
print("User clicked OK")
|
|
else:
|
|
return
|
|
|
|
self.button3.setVisible(False)
|
|
|
|
self.raw_haemo_dict = {}
|
|
self.config_dict = {}
|
|
self.epochs_dict = {}
|
|
self.fig_bytes_dict = {}
|
|
self.cha_dict = {}
|
|
self.contrast_results_dict = {}
|
|
self.df_ind_dict = {}
|
|
self.design_matrix_dict = {}
|
|
self.valid_dict = {}
|
|
|
|
self.button1.clicked.disconnect(self.on_run_task)
|
|
self.button1.setText("Cancel")
|
|
self.button1.clicked.connect(self.cancel_task)
|
|
|
|
if not self.first_run:
|
|
for bubble in self.bubble_widgets.values():
|
|
pass
|
|
# bubble.mark_cancelled()
|
|
self.first_run = False
|
|
|
|
# Collect all selected snirf files in a flat list
|
|
snirf_files = []
|
|
|
|
if hasattr(self, "selected_paths") and self.selected_paths:
|
|
for path in self.selected_paths:
|
|
p = Path(path)
|
|
if p.is_dir():
|
|
snirf_files += [str(f) for f in p.glob("*.snirf")]
|
|
elif p.is_file() and p.suffix == ".snirf":
|
|
snirf_files.append(str(p))
|
|
|
|
elif hasattr(self, "selected_path") and self.selected_path:
|
|
p = Path(self.selected_path)
|
|
if p.is_dir():
|
|
snirf_files += [str(f) for f in p.glob("*.snirf")]
|
|
elif p.is_file() and p.suffix == ".snirf":
|
|
snirf_files.append(str(p))
|
|
|
|
else:
|
|
raise ValueError("No file(s) selected")
|
|
|
|
if not snirf_files:
|
|
raise ValueError("No .snirf files found in selection")
|
|
|
|
# TODO: Bad! read_raw_snirf doesnt release memory properly! Should be spawned in a seperate process and killed once completed
|
|
# # validate
|
|
# for i in snirf_files:
|
|
# x_coords = set()
|
|
# y_coords = set()
|
|
# z_coords = set()
|
|
# raw = read_raw_snirf(i)
|
|
# dig = raw.info.get('dig', None)
|
|
# if dig is not None:
|
|
# for point in dig:
|
|
# if point['kind'] == 3:
|
|
# coord = point['r']
|
|
# x_coords.add(coord[0])
|
|
# y_coords.add(coord[1])
|
|
# z_coords.add(coord[2])
|
|
# print(f"Coord: {coord}")
|
|
# is_2d = (
|
|
# all(abs(x) < 1e-6 for x in x_coords) or
|
|
# all(abs(y) < 1e-6 for y in y_coords) or
|
|
# all(abs(z) < 1e-6 for z in z_coords)
|
|
# )
|
|
|
|
# if is_2d:
|
|
# if self.is_2d_bypass == False:
|
|
# QMessageBox.critical(None, "Error - 2D Data Detected - FLARES", f"Error: 2 dimensional data was found in {i}. "
|
|
# "Please update the coordinates using the 'Update optodes in snirf file...' option from the Options menu or by pressing 'F6'. "
|
|
# "You may also select the '2D Data Bypass' option from the Preferences menu to ignore this warning and process anyway. ")
|
|
# self.button1.clicked.disconnect(self.cancel_task)
|
|
# self.button1.setText("Process")
|
|
# self.button1.clicked.connect(self.on_run_task)
|
|
# return
|
|
|
|
# raw.close()
|
|
# del raw
|
|
|
|
self.files_total = len(snirf_files)
|
|
self.files_done = set()
|
|
self.files_failed = set()
|
|
self.files_results = {}
|
|
|
|
all_params = {}
|
|
for section_widget in self.param_sections:
|
|
section_params = section_widget.get_param_values()
|
|
all_params.update(section_params)
|
|
|
|
if self.folding_bypass:
|
|
all_params['FOLDING_BYP'] = True
|
|
|
|
collected_data = {
|
|
"SNIRF_FILES": snirf_files,
|
|
"PARAMS": all_params, # add this line
|
|
"METADATA": self.get_all_metadata(), # optionally add metadata if needed
|
|
}
|
|
# Start processing
|
|
if current_process().name == 'MainProcess':
|
|
self.result_queue = Queue()
|
|
self.ack_queue = Queue()
|
|
self.progress_queue = Queue()
|
|
|
|
self.result_process = Process(
|
|
target=run_gui_entry_wrapper,
|
|
args=(collected_data, self.result_queue, self.progress_queue, self.ack_queue)
|
|
)
|
|
self.result_process.daemon = False
|
|
self.result_process.start()
|
|
|
|
self.statusbar.showMessage("Running processing in background...")
|
|
|
|
self.result_timer = QTimer()
|
|
self.result_timer.timeout.connect(self.check_for_pipeline_results)
|
|
self.result_timer.start()
|
|
|
|
self.statusbar.showMessage("Task started in separate process.")
|
|
|
|
|
|
|
|
def check_for_pipeline_results(self):
|
|
try:
|
|
while True:
|
|
try:
|
|
msg = self.result_queue.get_nowait()
|
|
except Empty:
|
|
break
|
|
|
|
if isinstance(msg, dict) and msg.get("type") == "file_done":
|
|
file_path = msg["file"]
|
|
self.files_done.add(file_path)
|
|
|
|
# print(f"[DEBUG] File Done: {os.path.basename(file_path)}")
|
|
# print(f"[DEBUG] Progress: {len(self.files_done)} / {self.files_total}")
|
|
|
|
if msg.get("success"):
|
|
# Unpack the massive tuple
|
|
raw_haemo, config, epochs, fig_bytes, cha, contrast, df_ind, design, valid = msg["result"]
|
|
|
|
# Initialize dictionaries once if needed
|
|
if not hasattr(self, 'raw_haemo_dict') or self.raw_haemo_dict is None:
|
|
attrs = ['raw_haemo_dict', 'config_dict', 'epochs_dict', 'fig_bytes_dict',
|
|
'cha_dict', 'contrast_results_dict', 'df_ind_dict',
|
|
'design_matrix_dict', 'valid_dict']
|
|
for attr in attrs:
|
|
setattr(self, attr, {})
|
|
|
|
self.files_results[file_path] = msg["result"]
|
|
self.raw_haemo_dict[file_path] = raw_haemo
|
|
self.config_dict[file_path] = config
|
|
self.epochs_dict[file_path] = epochs
|
|
self.fig_bytes_dict[file_path] = fig_bytes
|
|
self.cha_dict[file_path] = cha
|
|
self.contrast_results_dict[file_path] = contrast
|
|
self.df_ind_dict[file_path] = df_ind
|
|
self.design_matrix_dict[file_path] = design
|
|
self.valid_dict[file_path] = valid
|
|
|
|
self.statusbar.showMessage(f"Processed: {os.path.basename(file_path)}")
|
|
|
|
else:
|
|
self.files_failed.add(file_path)
|
|
error_msg = msg.get("error", "Unknown worker error")
|
|
print(f"[DEBUG] File Failed: {os.path.basename(file_path)} - {error_msg}")
|
|
self.show_error_popup(f"Error: {file_path}", error_msg, msg.get("traceback", ""))
|
|
self.statusbar.showMessage(f"Failed: {os.path.basename(file_path)}")
|
|
|
|
elif isinstance(msg, dict) and msg.get("type") == "FINISHED_SUCCESSFULLY":
|
|
# The child has finished its work AND its own cleanup.
|
|
# It is now safe for the GUI to stop the timer and clean up.
|
|
try:
|
|
self.ack_queue.put("ACK")
|
|
except: pass
|
|
self.result_timer.stop()
|
|
self.cleanup_after_process()
|
|
|
|
success_count = len(self.files_results)
|
|
fail_count = self.files_total - success_count
|
|
self.statusbar.showMessage(f"Complete: {success_count} succeeded, {fail_count} failed.")
|
|
|
|
if success_count > 0:
|
|
self.button3.setVisible(True)
|
|
|
|
# Reset the button
|
|
try: self.button1.clicked.disconnect()
|
|
except: pass
|
|
self.button1.setText("Process")
|
|
self.button1.clicked.connect(self.on_run_task)
|
|
return # Exit the method
|
|
|
|
elif isinstance(msg, dict) and msg.get("success") is True:
|
|
self.statusbar.showMessage("All files processed successfully!")
|
|
|
|
elif isinstance(msg, dict) and (msg.get("success") is False or msg.get("type") == "error"):
|
|
file_path = msg.get("file", "Process")
|
|
error_msg = msg.get("error", "Unknown error")
|
|
self.show_error_popup(f"Error: {file_path}", error_msg, msg.get("traceback", ""))
|
|
self.files_done.add(file_path)
|
|
if msg.get("success") is False: # Fatal crash
|
|
self.result_timer.stop()
|
|
self.cleanup_after_process()
|
|
return
|
|
|
|
elif isinstance(msg, tuple) and msg[0] == 'progress':
|
|
_, file_path, step_index = msg
|
|
self.progress_update_signal.emit(file_path, step_index)
|
|
|
|
|
|
except Exception as e:
|
|
print(f"Error in timer loop: {e}")
|
|
if not self.result_process.is_alive() and len(self.files_done) < self.files_total:
|
|
self.statusbar.showMessage("Background process died.")
|
|
self.result_timer.stop()
|
|
|
|
|
|
def show_error_popup(self, title, error_message, traceback_str=""):
|
|
msgbox = QMessageBox(self)
|
|
msgbox.setIcon(QMessageBox.Warning)
|
|
msgbox.setWindowTitle("Warning - FLARES")
|
|
|
|
message = (
|
|
f"FLARES has encountered an error processing the file {title}.<br><br>"
|
|
"This error was likely due to incorrect parameters on the right side of the screen and not an error with your data. "
|
|
"Processing of the remaining files continues in the background and this participant will be ignored in the analysis. "
|
|
"If you think the parameters on the right side are correct for your data, raise an issue <a href='https://git.research.dezeeuw.ca/tyler/flares/issues'>here</a>.<br><br>"
|
|
f"Error message: {error_message}"
|
|
)
|
|
|
|
msgbox.setTextFormat(Qt.TextFormat.RichText)
|
|
msgbox.setText(message)
|
|
msgbox.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
|
|
|
|
# Add traceback to detailed text
|
|
if traceback_str:
|
|
msgbox.setDetailedText(traceback_str)
|
|
|
|
msgbox.setStandardButtons(QMessageBox.Ok)
|
|
msgbox.show()
|
|
|
|
|
|
def cleanup_after_process(self):
|
|
|
|
if hasattr(self, 'result_process'):
|
|
self.result_process.join(timeout=0)
|
|
if self.result_process.is_alive():
|
|
self.result_process.terminate()
|
|
self.result_process.join()
|
|
|
|
if hasattr(self, 'result_queue'):
|
|
if 'AutoProxy' in repr(self.result_queue):
|
|
pass
|
|
else:
|
|
self.result_queue.close()
|
|
self.result_queue.join_thread()
|
|
|
|
if hasattr(self, 'progress_queue'):
|
|
if 'AutoProxy' in repr(self.progress_queue):
|
|
pass
|
|
else:
|
|
self.progress_queue.close()
|
|
self.progress_queue.join_thread()
|
|
|
|
|
|
def update_file_progress(self, file_path, step_index):
|
|
key = os.path.normpath(file_path)
|
|
bubble = self.bubble_widgets.get(key)
|
|
if bubble:
|
|
bubble.update_progress(step_index)
|
|
|
|
|
|
def get_snirf_metadata_mne(self, file_name):
|
|
# Check if we already have it (we should?)
|
|
if file_name in self.metadata_cache:
|
|
return self.metadata_cache[file_name]
|
|
|
|
print(self.metadata_cache)
|
|
|
|
# If the user clicked so fast it's not ready, do a one-off blocking call
|
|
print(f"Cache miss for {file_name}, fetching now...")
|
|
future = self.file_executor.submit(_extract_metadata_worker, file_name)
|
|
return future.result(timeout=5)
|
|
|
|
|
|
def closeEvent(self, event):
|
|
# Gracefully shut down multiprocessing children
|
|
print("Window is closing. Cleaning up...")
|
|
|
|
if hasattr(self, 'manager'):
|
|
self.manager.shutdown()
|
|
|
|
for child in self.findChildren(QWidget):
|
|
if child is not self and child.isVisible():
|
|
child.close()
|
|
|
|
kill_child_processes()
|
|
|
|
event.accept()
|
|
|
|
|
|
def _on_metadata_ready(self, future, file_path, session_id):
|
|
|
|
if session_id != self.loading_session_id:
|
|
return
|
|
|
|
try:
|
|
result = future.result()
|
|
|
|
if result is None:
|
|
result = {'status': 'error', 'reason': 'Worker returned no data.'}
|
|
# If it's a successful extraction, it won't have 'status' set yet
|
|
|
|
elif 'status' not in result:
|
|
# Wrap the raw extraction dictionary into our unified UI format
|
|
result = {'status': 'success', 'data': result}
|
|
|
|
except Exception as e:
|
|
result = {'status': 'error', 'reason': str(e)}
|
|
|
|
# Safely emit to the Main thread. No brittle QMetaObject needed!
|
|
self.metadata_ui_signal.emit(result, file_path, session_id)
|
|
|
|
|
|
def _handle_metadata_ui_update(self, result, file_path, session_id):
|
|
"""Executes safely on the MAIN GUI thread via Signal connection."""
|
|
if result.get('status') == 'error':
|
|
# 1. Pop up the warning safely on the main thread
|
|
QMessageBox.warning(
|
|
self,
|
|
"Invalid File",
|
|
f"Could not read metadata from: {os.path.basename(file_path)}\n\n"
|
|
f"Details: {result.get('reason', 'Unknown error')}"
|
|
)
|
|
# 2. Run your clean tracking removal
|
|
self._remove_file_from_pipeline(file_path)
|
|
return
|
|
|
|
# Success path
|
|
self.metadata_cache[file_path] = result.get('data', result)
|
|
self.metadata_processed.emit(file_path, session_id)
|
|
|
|
|
|
def _remove_file_from_pipeline(self, file_path):
|
|
"""Completely cleans up and removes all references to a file that failed to load."""
|
|
# 1. Decrement pending file count
|
|
if hasattr(self, 'pending_files_count') and self.pending_files_count > 0:
|
|
self.pending_files_count -= 1
|
|
|
|
# 2. Remove the UI widget cleanly
|
|
if hasattr(self, 'bubble_widgets') and file_path in self.bubble_widgets:
|
|
bubble = self.bubble_widgets.pop(file_path)
|
|
self.bubble_layout.removeWidget(bubble)
|
|
bubble.deleteLater() # Safely schedules the widget for deletion in Qt
|
|
|
|
# 3. Remove from tracking lists
|
|
if hasattr(self, 'selected_paths') and file_path in self.selected_paths:
|
|
self.selected_paths.remove(file_path)
|
|
|
|
# 4. Update Status Bar
|
|
if hasattr(self, 'pending_files_count') and self.pending_files_count == 0:
|
|
self.statusBar().showMessage("Ready.", 3000)
|
|
else:
|
|
self.statusBar().showMessage(f"Loading pending files... ({self.pending_files_count} left)")
|
|
|
|
|
|
def _safe_ui_update(self, file_path):
|
|
|
|
# 2. Update the Bubble safely
|
|
if file_path in self.bubble_widgets:
|
|
bubble = self.bubble_widgets[file_path]
|
|
# This is now thread-safe!
|
|
bubble.set_loading_state(False)
|
|
bubble.clicked.connect(self.on_bubble_clicked)
|
|
bubble.rightClicked.connect(self.on_bubble_right_clicked)
|
|
bubble.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
|
|
# 3. Handle the global counter/cleanup
|
|
self.pending_files_count -= 1
|
|
if self.pending_files_count <= 0:
|
|
self._cleanup_executor()
|
|
self.statusbar.showMessage("All files loaded sucessfully.")
|
|
|
|
def _cleanup_executor(self):
|
|
"""Safely shuts down the executor and clears the reference."""
|
|
if hasattr(self, 'file_executor') and self.file_executor is not None:
|
|
self.file_executor.shutdown(wait=False)
|
|
self.file_executor = None
|
|
print("[System] Background worker dismissed. RAM reclaimed.")
|
|
|
|
|
|
def _extract_metadata_worker(file_name):
|
|
"""Runs in the separate worker process. Returns a clean dict."""
|
|
|
|
# 1. Use preload=False! We only need metadata.
|
|
raw = None
|
|
|
|
try:
|
|
raw = read_raw_snirf(file_name, preload=False, verbose="ERROR")
|
|
snirf_info = {}
|
|
|
|
# 2. Measurement date
|
|
snirf_info['Measurement Date'] = str(raw.info.get('meas_date'))
|
|
|
|
# 3. Short Channels
|
|
try:
|
|
short_chans = get_short_channels(raw, max_dist=0.015)
|
|
names = list(short_chans.ch_names)
|
|
snirf_info['Short Channels'] = f"Likely - {names}"
|
|
if len(names) > 6:
|
|
snirf_info['Short Channels'] += "\n There are a lot of short channels. Optode distances are likely incorrect!"
|
|
except:
|
|
snirf_info['Short Channels'] = "Unlikely"
|
|
|
|
# 4. Distances
|
|
dist_vals = source_detector_distances(raw.info)
|
|
snirf_info['Source-Detector Distances'] = [
|
|
f"{name}: {d:.4f} m" for name, d in zip(raw.info['ch_names'], dist_vals)
|
|
]
|
|
|
|
# 5. Digitization
|
|
dig = raw.info.get('dig', None)
|
|
if dig is not None:
|
|
snirf_info['Digitization Points'] = [
|
|
f"Kind: {p['kind']}, ID: {p['ident']}, Coord: {p['r']}" for p in dig
|
|
]
|
|
else:
|
|
snirf_info['Digitization Points'] = "Not found"
|
|
|
|
# 6. Annotations (using our copy-to-string trick)
|
|
if raw.annotations is not None and len(raw.annotations) > 0:
|
|
snirf_info['Annotations'] = [
|
|
f"Onset: {o:.2f}s, Duration: {d:.2f}s, Description: {str(desc)}"
|
|
for o, d, desc in zip(raw.annotations.onset, raw.annotations.duration, raw.annotations.description)
|
|
]
|
|
else:
|
|
snirf_info['Annotations'] = "No annotations found"
|
|
|
|
return snirf_info
|
|
|
|
except Exception as e:
|
|
print(f"Worker safely caught failure on {file_name}: {str(e)}")
|
|
return {'status': 'error', 'reason': str(e)}
|
|
|
|
finally:
|
|
if raw is not None:
|
|
try:
|
|
raw.close()
|
|
except:
|
|
pass
|
|
|
|
|
|
|
|
def run_gui_entry_wrapper(config, gui_queue, progress_queue, ack_queue):
|
|
"""
|
|
Where the processing happens
|
|
"""
|
|
|
|
try:
|
|
import flares as flares
|
|
flares.gui_entry(config, gui_queue, progress_queue, ack_queue)
|
|
gui_queue.close()
|
|
# gui_queue.join_thread()
|
|
progress_queue.close()
|
|
# progress_queue.join_thread()
|
|
os._exit(0)
|
|
|
|
except Exception as e:
|
|
tb_str = traceback.format_exc()
|
|
gui_queue.put({
|
|
"success": False,
|
|
"error": f"Child process crashed: {str(e)}\nTraceback:\n{tb_str}"
|
|
})
|
|
os._exit(1)
|
|
|
|
|
|
def resource_path(relative_path):
|
|
"""
|
|
Get absolute path to resource regardless of running directly or packaged using PyInstaller
|
|
"""
|
|
|
|
if hasattr(sys, '_MEIPASS'):
|
|
# PyInstaller bundle path
|
|
base_path = sys._MEIPASS
|
|
else:
|
|
base_path = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
return os.path.join(base_path, relative_path)
|
|
|
|
|
|
def kill_child_processes():
|
|
"""
|
|
Goodbye children
|
|
"""
|
|
|
|
try:
|
|
parent = psutil.Process(os.getpid())
|
|
children = parent.children(recursive=True)
|
|
for child in children:
|
|
try:
|
|
child.kill()
|
|
except psutil.NoSuchProcess:
|
|
pass
|
|
psutil.wait_procs(children, timeout=5)
|
|
except Exception as e:
|
|
print(f"Error killing child processes: {e}")
|
|
|
|
|
|
def exception_hook(exc_type, exc_value, exc_traceback):
|
|
"""
|
|
Method that will display a popup when the program hard crashes containg what went wrong
|
|
"""
|
|
|
|
error_msg = "".join(traceback.format_exception(exc_type, exc_value, exc_traceback))
|
|
print(error_msg) # also print to console
|
|
|
|
kill_child_processes()
|
|
|
|
# Show error message box
|
|
# Make sure QApplication exists (or create a minimal one)
|
|
app = QApplication.instance()
|
|
if app is None:
|
|
app = QApplication(sys.argv)
|
|
|
|
show_critical_error(error_msg)
|
|
|
|
# Exit the app after user acknowledges
|
|
sys.exit(1)
|
|
|
|
def show_critical_error(error_msg):
|
|
msg_box = QMessageBox()
|
|
msg_box.setIcon(QMessageBox.Icon.Critical)
|
|
msg_box.setWindowTitle("Something went wrong!")
|
|
|
|
if PLATFORM_NAME == "darwin":
|
|
log_path = os.path.join(os.path.dirname(sys.executable), "../../../flares.log")
|
|
log_path2 = os.path.join(os.path.dirname(sys.executable), "../../../flares_error.log")
|
|
save_path = os.path.join(os.path.dirname(sys.executable), "../../../flares_autosave.flare")
|
|
|
|
else:
|
|
log_path = os.path.join(os.getcwd(), "flares.log")
|
|
log_path2 = os.path.join(os.getcwd(), "flares_error.log")
|
|
save_path = os.path.join(os.getcwd(), "flares_autosave.flare")
|
|
|
|
|
|
shutil.copy(log_path, log_path2)
|
|
log_path2 = Path(log_path2).absolute().as_posix()
|
|
autosave_path = Path(save_path).absolute().as_posix()
|
|
log_link = f"file:///{log_path2}"
|
|
autosave_link = f"file:///{autosave_path}"
|
|
|
|
window.save_project(True) #TODO: If the window is the one to crash, the file can't get saved. Could be fine as the window is what was storing the data to begin with?
|
|
|
|
message = (
|
|
f"{APP_NAME.upper()} has encountered an unrecoverable error and needs to close.<br><br>"
|
|
f"We are sorry for the inconvenience. An autosave was attempted to be saved to <a href='{autosave_link}'>{autosave_path}</a>, but it may not have been saved. "
|
|
"If the file was saved, it still may not be intact, openable, or contain the correct data. Use the autosave at your discretion.<br><br>"
|
|
f"This unrecoverable error was likely due to an error with {APP_NAME.upper()} and not your data.<br>"
|
|
f"Please raise an issue <a href='https://git.research.dezeeuw.ca/tyler/{APP_NAME}/issues'>here</a> and attach the error file located at <a href='{log_link}'>{log_path2}</a><br><br>"
|
|
f"<pre>{error_msg}</pre>"
|
|
)
|
|
|
|
msg_box.setTextFormat(Qt.TextFormat.RichText)
|
|
msg_box.setText(message)
|
|
msg_box.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
|
|
msg_box.setStandardButtons(QMessageBox.StandardButton.Ok)
|
|
|
|
msg_box.exec()
|
|
|
|
|
|
def config_init():
|
|
|
|
ref_cfg.read_string(DEFAULT_CONFIG)
|
|
|
|
if not os.path.exists(cfg_path):
|
|
try:
|
|
with open(cfg_path, "w") as f:
|
|
f.write(DEFAULT_CONFIG.strip())
|
|
print(f"Created default configuration file at {cfg_path}")
|
|
file_cfg.read_string(DEFAULT_CONFIG)
|
|
except Exception as e:
|
|
print(f"Warning: Could not create config file ({e}). Using in-memory defaults.")
|
|
file_cfg.read_string(DEFAULT_CONFIG)
|
|
|
|
else:
|
|
try:
|
|
# Load the user's actual file first
|
|
file_cfg.read(cfg_path)
|
|
has_changes = False
|
|
|
|
for section in file_cfg.sections():
|
|
if not ref_cfg.has_section(section):
|
|
file_cfg.remove_section(section)
|
|
has_changes = True
|
|
continue
|
|
|
|
for option in list(file_cfg.options(section)):
|
|
if not ref_cfg.has_option(section, option):
|
|
file_cfg.remove_option(section, option)
|
|
has_changes = True
|
|
|
|
for section in ref_cfg.sections():
|
|
if not file_cfg.has_section(section):
|
|
file_cfg.add_section(section)
|
|
has_changes = True
|
|
|
|
for option in list(ref_cfg.options(section)):
|
|
if not file_cfg.has_option(section, option):
|
|
default_val = ref_cfg.get(section, option)
|
|
file_cfg.set(section, option, default_val)
|
|
has_changes = True
|
|
|
|
# 4. If we added or removed anything, save the sanitized file back to disk
|
|
if has_changes:
|
|
with open(cfg_path, "w") as f:
|
|
file_cfg.write(f)
|
|
print("Configuration file synchronized: removed old keys and appended new ones.")
|
|
else:
|
|
print("Configuration loaded successfully. Schema is up to date.")
|
|
|
|
except Exception as e:
|
|
print(f"Error validating config file ({e}). Falling back completely to defaults.")
|
|
file_cfg.read_string(DEFAULT_CONFIG)
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Redirect exceptions to the popup window
|
|
sys.excepthook = exception_hook
|
|
|
|
# Set up application logging and configuration
|
|
if PLATFORM_NAME == "darwin":
|
|
log_path = os.path.join(os.path.dirname(sys.executable), f"../../../{APP_NAME}.log")
|
|
cfg_path = os.path.join(os.path.dirname(sys.executable), f"../../../{APP_NAME}.cfg")
|
|
else:
|
|
log_path = os.path.join(os.getcwd(), f"{APP_NAME}.log")
|
|
cfg_path = os.path.join(os.getcwd(), f"{APP_NAME}.cfg")
|
|
try:
|
|
os.remove(log_path)
|
|
except:
|
|
pass
|
|
|
|
sys.stdout = open(log_path, "a", buffering=1)
|
|
sys.stderr = sys.stdout
|
|
print(f"\n=== App started at {datetime.now()} ===\n")
|
|
|
|
file_cfg = configparser.ConfigParser()
|
|
ref_cfg = configparser.ConfigParser()
|
|
config_init()
|
|
|
|
freeze_support() # Required for PyInstaller + multiprocessing
|
|
|
|
# Only run GUI in the main process
|
|
if current_process().name == 'MainProcess':
|
|
app = QApplication(sys.argv)
|
|
finish_update_if_needed(PLATFORM_NAME, APP_NAME, cfg_path)
|
|
window = MainApplication()
|
|
|
|
if PLATFORM_NAME == "darwin":
|
|
app.setWindowIcon(QIcon(resource_path("icons/main.icns")))
|
|
window.setWindowIcon(QIcon(resource_path("icons/main.icns")))
|
|
else:
|
|
app.setWindowIcon(QIcon(resource_path("icons/main.ico")))
|
|
window.setWindowIcon(QIcon(resource_path("icons/main.ico")))
|
|
window.show()
|
|
sys.exit(app.exec())
|
|
|
|
# Not 6000 lines yay! |