2175 lines
85 KiB
Python
2175 lines
85 KiB
Python
"""
|
|
Filename: main.py
|
|
Description: LIGHTS main executable
|
|
|
|
Author: Tyler de Zeeuw
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
|
|
|
|
# Built-in imports
|
|
import os
|
|
import re
|
|
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 h5py
|
|
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
|
|
from matplotlib.figure import Figure
|
|
import psutil
|
|
import pyqtgraph.opengl as gl
|
|
from pyqtgraph.opengl import MeshData
|
|
import numpy as np
|
|
|
|
from mne.io import read_raw_snirf
|
|
from mne.preprocessing.nirs import source_detector_distances
|
|
|
|
from PySide6.QtWidgets import (
|
|
QApplication, QPlainTextEdit, 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, QVector3D
|
|
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 CameraQueryThread, CameraRowWidget
|
|
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]
|
|
|
|
[Edit]
|
|
|
|
[View]
|
|
status_bar = true
|
|
left_top = 0
|
|
left_bottom = 0
|
|
right = 0
|
|
|
|
[Options]
|
|
show_welcome_dialog = false
|
|
first_startup = true
|
|
|
|
[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."},
|
|
]
|
|
},
|
|
]
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).parent
|
|
|
|
# Venv folder (NOTE THE LEADING DOT)
|
|
VENV_ROOT = PROJECT_ROOT / ".gphoto2_venv"
|
|
|
|
# Python executable - try Scripts first (Windows), fallback to bin (Unix/MSYS2 style)
|
|
PYTHON_EXE = VENV_ROOT / "bin" / "python.exe"
|
|
|
|
# Bridge script location
|
|
BRIDGE_SCRIPT = PROJECT_ROOT / "part_A.py"
|
|
|
|
|
|
|
|
|
|
class LogConsoleDialog(QDialog):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setWindowTitle("Subprocess Execution Console")
|
|
self.resize(600, 400)
|
|
|
|
layout = QVBoxLayout(self)
|
|
|
|
# The scrolling log field
|
|
self.log_viewer = QPlainTextEdit(self)
|
|
self.log_viewer.setReadOnly(True)
|
|
layout.addWidget(self.log_viewer)
|
|
|
|
# Trigger button inside the popup
|
|
self.trigger_btn = QPushButton("Execute Synced Burst Capture", self)
|
|
self.trigger_btn.setEnabled(False) # Disabled until setup finishes
|
|
layout.addWidget(self.trigger_btn)
|
|
|
|
def append_log(self, text):
|
|
self.log_viewer.appendPlainText(text)
|
|
# Auto-scroll to the bottom
|
|
self.log_viewer.ensureCursorVisible()
|
|
|
|
|
|
|
|
class SubprocessWorker(QThread):
|
|
# Emits every line of output to append to our log window
|
|
line_received = Signal(str)
|
|
finished = Signal(int)
|
|
|
|
def __init__(self, snirf_file_path, is_2d_bypass, parent=None):
|
|
super().__init__(parent)
|
|
self.snirf_file_path = snirf_file_path
|
|
self.is_2d_bypass = is_2d_bypass
|
|
|
|
def run(self):
|
|
try:
|
|
cmd_args = [str(PYTHON_EXE), "-u", str(BRIDGE_SCRIPT), "--cmd", "gui"]
|
|
|
|
if self.is_2d_bypass:
|
|
cmd_args.append("--bypass-2d")
|
|
|
|
if self.snirf_file_path:
|
|
cmd_args.append("--snirf")
|
|
cmd_args.append(str(self.snirf_file_path))
|
|
|
|
self.process = subprocess.Popen(
|
|
cmd_args,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1
|
|
)
|
|
|
|
for line in iter(self.process.stdout.readline, ""):
|
|
self.line_received.emit(line.strip())
|
|
|
|
# Loop drops here automatically when the process dies (sys.exit)
|
|
return_code = self.process.wait()
|
|
self.finished.emit(return_code)
|
|
|
|
except Exception as e:
|
|
self.line_received.emit(f"[Thread Error] {str(e)}")
|
|
self.finished.emit(-1)
|
|
|
|
def send_input(self, text):
|
|
"""Safely writes text directly to the running process's stdin"""
|
|
if hasattr(self, 'process') and self.process.poll() is None:
|
|
try:
|
|
self.process.stdin.write(text + "\n")
|
|
self.process.stdin.flush() # Force Windows to send it immediately
|
|
except Exception as e:
|
|
print(f"Failed to write to stdin: {e}")
|
|
|
|
|
|
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 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.snirf_file_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(400)
|
|
|
|
top_left_container = QGroupBox("3D Viewer")
|
|
top_left_container.setStyleSheet("QGroupBox { font-weight: bold; }")
|
|
top_left_container.setMinimumHeight(340)
|
|
top_left_layout = QHBoxLayout(top_left_container)
|
|
|
|
# --- NEW 3D VIEWER CODE START ---
|
|
# Create the interactive 3D view widget
|
|
self.top_left_widget = gl.GLViewWidget()
|
|
self.top_left_widget.setBackgroundColor('k')
|
|
self.top_left_widget.setCameraPosition(distance=28, elevation=30, azimuth=45)
|
|
|
|
# 1. Base grid layout
|
|
grid = gl.GLGridItem()
|
|
grid.scale(1, 1, 1)
|
|
self.top_left_widget.addItem(grid)
|
|
|
|
# 2. Central Reference Sphere (Blue)
|
|
sphere_md = MeshData.sphere(rows=20, cols=20, radius=2.0)
|
|
self.sphere_item = gl.GLMeshItem(meshdata=sphere_md, smooth=True, color=(0.2, 0.6, 1.0, 1.0), shader='shaded')
|
|
self.top_left_widget.addItem(self.sphere_item)
|
|
|
|
# 3. Base Forward Direction Indicator (Arrow/Ray going down the -Y or +X axis)
|
|
# Let's define Forward as along the +Y horizontal axis (Green line)
|
|
arrow_points = np.array([
|
|
[0, 0, 0], [0, 3, 0], # Main shaft
|
|
[0, 3, 0], [-0.5, 2, 0], # Left arrowhead barb
|
|
[0, 3, 0], [0.5, 2, 0] # Right arrowhead barb
|
|
])
|
|
|
|
# Using mode='lines' treats every pairs of coordinates as independent lines
|
|
self.forward_arrow = gl.GLLinePlotItem(
|
|
pos=arrow_points,
|
|
color=(0.0, 1.0, 0.0, 1.0),
|
|
width=4,
|
|
mode='lines'
|
|
)
|
|
self.top_left_widget.addItem(self.forward_arrow)
|
|
|
|
# 4. Storage registry to track rendering handles for active cameras
|
|
self.rendered_camera_items = []
|
|
|
|
top_left_layout.addWidget(self.top_left_widget, stretch=4)
|
|
|
|
self.camera_pool_wrapper = QWidget()
|
|
wrapper_layout = QVBoxLayout(self.camera_pool_wrapper)
|
|
wrapper_layout.setContentsMargins(0, 0, 0, 0)
|
|
|
|
# 1. Add the Refresh Button to the wrapper layout
|
|
self.refresh_cameras_btn = QPushButton("🔄 Refresh Connected Cameras")
|
|
self.refresh_cameras_btn.setStyleSheet("""
|
|
QPushButton {
|
|
padding: 6px;
|
|
font-weight: bold;
|
|
background-color: #2c3e50;
|
|
color: white;
|
|
border-radius: 4px;
|
|
}
|
|
QPushButton:hover { background-color: #34495e; }
|
|
QPushButton:disabled { background-color: #7f8c8d; color: #bdc3c7; }
|
|
""")
|
|
self.refresh_cameras_btn.clicked.connect(self.refresh_hardware_cameras)
|
|
wrapper_layout.addWidget(self.refresh_cameras_btn)
|
|
|
|
# 2. Setup your existing scroll list setup
|
|
self.camera_pool_container = QWidget()
|
|
self.camera_pool_layout = QGridLayout(self.camera_pool_container)
|
|
self.camera_pool_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
|
|
|
|
self.camera_scroll_area = QScrollArea()
|
|
self.camera_scroll_area.setWidgetResizable(True)
|
|
self.camera_scroll_area.setWidget(self.camera_pool_container)
|
|
self.camera_scroll_area.setMinimumHeight(160)
|
|
|
|
# Add the scroll area right under the button inside the wrapper
|
|
wrapper_layout.addWidget(self.camera_scroll_area)
|
|
|
|
# --- REPLACED SPLITTER LAYOUT ATTACHMENT ---
|
|
self.left_v_splitter.addWidget(top_left_container)
|
|
# Instead of adding the scroll area alone, add the wrapper containing both the button and scroll area
|
|
self.left_v_splitter.addWidget(self.camera_pool_wrapper)
|
|
|
|
# --- REST OF YOUR RIGHT CONTAINER SETUP UNCHANGED ---
|
|
self.right_container = QWidget()
|
|
self.right_container.setMinimumWidth(380)
|
|
right_container_layout = QVBoxLayout(self.right_container)
|
|
|
|
# Scroll area setup
|
|
self.right_scroll_area = QScrollArea()
|
|
self.right_scroll_area.setWidgetResizable(True)
|
|
|
|
self.right_content_widget = QWidget()
|
|
self.right_content_layout = QVBoxLayout(self.right_content_widget)
|
|
self.right_content_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
|
|
|
|
# Add Configuration Management Row Right below Scroll area
|
|
control_actions_widget = QWidget()
|
|
control_actions_layout = QHBoxLayout(control_actions_widget)
|
|
control_actions_layout.setContentsMargins(5, 5, 5, 5)
|
|
|
|
self.add_camera_btn = QPushButton("+ Add Camera")
|
|
self.add_camera_btn.setMinimumHeight(35)
|
|
self.add_camera_btn.setStyleSheet("font-weight: bold; background-color: #2ecc71; color: white; border-radius: 4px;")
|
|
self.add_camera_btn.clicked.connect(self.add_new_camera_row)
|
|
control_actions_layout.addWidget(self.add_camera_btn)
|
|
|
|
# Inject the '+' button panel above your bottom process buttons
|
|
right_container_layout.addWidget(control_actions_widget)
|
|
|
|
# Dedicated layout container where rows will be dynamically injected
|
|
self.rows_container = QWidget()
|
|
self.rows_layout = QVBoxLayout(self.rows_container)
|
|
self.rows_layout.setContentsMargins(0, 0, 0, 0)
|
|
self.right_content_layout.addWidget(self.rows_container)
|
|
self.right_scroll_area.setWidget(self.right_content_widget)
|
|
right_container_layout.addWidget(self.right_scroll_area)
|
|
|
|
|
|
self.right_scroll_area2 = QScrollArea()
|
|
self.right_scroll_area2.setWidgetResizable(True)
|
|
|
|
plot_container_widget = QWidget()
|
|
plot_container_layout = QVBoxLayout(plot_container_widget)
|
|
plot_container_layout.setContentsMargins(5, 5, 5, 5)
|
|
|
|
# 1. Create the File Actions Row (Pick / Remove)
|
|
file_actions_widget = QWidget()
|
|
file_actions_layout = QHBoxLayout(file_actions_widget)
|
|
file_actions_layout.setContentsMargins(0, 0, 0, 0)
|
|
|
|
self.pick_file_btn = QPushButton("Pick File")
|
|
self.pick_file_btn.setMinimumHeight(30)
|
|
self.pick_file_btn.clicked.connect(self.select_file) # Connect to file picker
|
|
|
|
self.remove_file_btn = QPushButton("Remove File")
|
|
self.remove_file_btn.setMinimumHeight(30)
|
|
self.remove_file_btn.clicked.connect(self.clear_file) # Connect to clearer
|
|
|
|
file_actions_layout.addWidget(self.pick_file_btn)
|
|
file_actions_layout.addWidget(self.remove_file_btn)
|
|
plot_container_layout.addWidget(file_actions_widget)
|
|
|
|
# NEW: Text label to display the active file path, located right beneath the buttons
|
|
self.filepath_label = QLabel("No SNIRF file selected")
|
|
self.filepath_label.setStyleSheet("color: #7f8c8d; font-style: italic; padding: 2px;")
|
|
self.filepath_label.setWordWrap(True) # Keeps ultra-long file paths from clipping horizontally
|
|
plot_container_layout.addWidget(self.filepath_label)
|
|
|
|
file_actions_layout.addWidget(self.pick_file_btn)
|
|
file_actions_layout.addWidget(self.remove_file_btn)
|
|
plot_container_layout.addWidget(file_actions_widget)
|
|
|
|
# 2. Setup the Matplotlib Figure and Canvas
|
|
# tight_layout keeps the plot from clipping at smaller container sizes
|
|
self.fig = Figure(figsize=(3, 2), dpi=100, tight_layout=True)
|
|
self.canvas = FigureCanvas(self.fig)
|
|
self.ax = self.fig.add_subplot(111)
|
|
self.ax.axis('off') # Hides the X/Y axes ticks for a cleaner image look
|
|
|
|
plot_container_layout.addWidget(self.canvas)
|
|
|
|
# 3. Bind the container widget to the second scroll area
|
|
self.right_scroll_area2.setWidget(plot_container_widget)
|
|
right_container_layout.addWidget(self.right_scroll_area2)
|
|
|
|
|
|
|
|
|
|
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.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(buttons_widget)
|
|
|
|
self.main_h_splitter.addWidget(self.left_v_splitter)
|
|
self.main_h_splitter.addWidget(self.right_container)
|
|
|
|
self.main_h_splitter.setSizes([760, 240])
|
|
self.left_v_splitter.setSizes([800, 200])
|
|
|
|
self.progress_update_signal.connect(self.update_file_progress)
|
|
self.update_sections(0)
|
|
self.refresh_hardware_cameras()
|
|
|
|
def select_file(self):
|
|
file_path, _ = QFileDialog.getOpenFileName(
|
|
self,
|
|
"Select SNIRF File",
|
|
"",
|
|
"SNIRF Files (*.snirf)"
|
|
)
|
|
|
|
if not file_path:
|
|
return
|
|
|
|
self.snirf_file_path = file_path
|
|
|
|
# Update the UI text label with the path
|
|
self.filepath_label.setText(file_path)
|
|
self.filepath_label.setStyleSheet("color: #2c3e50; font-weight: bold; padding: 2px;")
|
|
|
|
try:
|
|
# Clear previous elements from the plot area
|
|
self.ax.clear()
|
|
|
|
# 1. H5PY Parse Routine
|
|
with h5py.File(file_path, "r") as f:
|
|
nirs_key = [k for k in f.keys() if "nirs" in k][0]
|
|
nirs = f[nirs_key]
|
|
probe = nirs["probe"]
|
|
|
|
src_pos = probe["sourcePos2D"][:, :2]
|
|
det_pos = probe["detectorPos2D"][:, :2]
|
|
wavelengths = probe["wavelengths"][:]
|
|
|
|
oxy_wavelength = max(wavelengths)
|
|
channels = []
|
|
|
|
def collect_channels(name, obj):
|
|
if "measurementList" in name and isinstance(obj, h5py.Group):
|
|
short_name = name.split('/')[-1]
|
|
num = int(re.search(r'\d+', short_name).group())
|
|
|
|
source_idx = obj["sourceIndex"][()]
|
|
detector_idx = obj["detectorIndex"][()]
|
|
wavelength_idx = obj["wavelengthIndex"][()]
|
|
wl_value = wavelengths[wavelength_idx - 1]
|
|
|
|
channels.append({
|
|
"num": num,
|
|
"source": source_idx,
|
|
"detector": detector_idx,
|
|
"wavelength": wl_value
|
|
})
|
|
|
|
nirs.visititems(collect_channels)
|
|
|
|
# 2. Extract specific oxygenated layout linkages
|
|
oxy_channels_for_plot = []
|
|
for ch in channels:
|
|
if ch["wavelength"] == oxy_wavelength:
|
|
oxy_channels_for_plot.append({
|
|
"source": ch["source"] - 1,
|
|
"detector": ch["detector"] - 1
|
|
})
|
|
|
|
# 3. Build & Render plot layout onto the active widget figure
|
|
self.ax.axis('on') # Re-enable the plot framing system
|
|
|
|
# Draw measurement paths
|
|
for ch in oxy_channels_for_plot:
|
|
s_idx = ch["source"]
|
|
d_idx = ch["detector"]
|
|
self.ax.plot([src_pos[s_idx, 0], det_pos[d_idx, 0]],
|
|
[src_pos[s_idx, 1], det_pos[d_idx, 1]],
|
|
color='gray', linestyle='--', alpha=0.5, zorder=1)
|
|
|
|
# Draw Sources (Red Circles)
|
|
self.ax.scatter(src_pos[:, 0], src_pos[:, 1], color='#e74c3c', s=180, label='Sources (S)', zorder=2)
|
|
for i, pos in enumerate(src_pos):
|
|
self.ax.text(pos[0], pos[1], f"S{i+1}", color='white', ha='center', va='center', fontweight='bold', fontsize=9)
|
|
|
|
# Draw Detectors (Blue Squares)
|
|
self.ax.scatter(det_pos[:, 0], det_pos[:, 1], color='#3498db', marker='s', s=180, label='Detectors (D)', zorder=2)
|
|
for i, pos in enumerate(det_pos):
|
|
self.ax.text(pos[0], pos[1], f"D{i+1}", color='white', ha='center', va='center', fontweight='bold', fontsize=9)
|
|
|
|
# Apply Titles and Spacing Styles
|
|
self.ax.legend(loc='upper right', frameon=True, shadow=True, fontsize=6)
|
|
self.ax.grid(True, linestyle=':', alpha=0.5)
|
|
self.ax.axis('equal')
|
|
|
|
# Force redraw canvas window frame
|
|
self.canvas.draw()
|
|
|
|
except Exception as e:
|
|
self.filepath_label.setText(f"Error loading SNIRF file: {str(e)}")
|
|
self.filepath_label.setStyleSheet("color: #c0392b; font-weight: bold;")
|
|
self.clear_file()
|
|
|
|
def clear_file(self):
|
|
# Reset text to original placeholder state
|
|
self.filepath_label.setText("No SNIRF file selected")
|
|
self.filepath_label.setStyleSheet("color: #7f8c8d; font-style: italic; padding: 2px;")
|
|
|
|
# Reset plot to an empty frame or default state
|
|
self.ax.clear()
|
|
self.ax.axis('off')
|
|
self.canvas.draw()
|
|
|
|
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 = [
|
|
("Photo Taking 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.statusbar.showMessage("Ready")
|
|
|
|
|
|
def update_sections(self, index=0):
|
|
"""Replaced the old method to act as our initial setup cleaner."""
|
|
self.clear_all_camera_rows()
|
|
# Seed with one default camera row to begin with
|
|
self.add_new_camera_row()
|
|
|
|
def refresh_hardware_cameras(self):
|
|
"""Spins up the background hardware bridge subprocess."""
|
|
# 1. Lock the button UI feedback
|
|
self.refresh_cameras_btn.setEnabled(False)
|
|
self.refresh_cameras_btn.setText("⏳ Scanning USB Ports...")
|
|
|
|
# 2. Clear old list items while scanning
|
|
while self.camera_pool_layout.count() > 0:
|
|
item = self.camera_pool_layout.takeAt(0)
|
|
if item.widget():
|
|
item.widget().deleteLater()
|
|
|
|
# 3. Inform the user visually in the grid area
|
|
from PySide6.QtWidgets import QLabel
|
|
self.camera_pool_layout.addWidget(QLabel("Querying subprocess bridge..."), 0, 0)
|
|
|
|
# 4. Fire the thread
|
|
self.query_thread = CameraQueryThread()
|
|
self.query_thread.cameras_found.connect(self.on_cameras_detected)
|
|
self.query_thread.query_failed.connect(self.on_camera_query_error)
|
|
self.query_thread.start()
|
|
|
|
def on_cameras_detected(self, camera_list):
|
|
"""Callback when hardware signals back successfully via ast.literal_eval."""
|
|
# Unlock refresh UI
|
|
self.refresh_cameras_btn.setEnabled(True)
|
|
self.refresh_cameras_btn.setText("🔄 Refresh Connected Cameras")
|
|
|
|
# Clear the scanning placeholder
|
|
while self.camera_pool_layout.count() > 0:
|
|
item = self.camera_pool_layout.takeAt(0)
|
|
if item.widget():
|
|
item.widget().deleteLater()
|
|
|
|
if not camera_list:
|
|
from PySide6.QtWidgets import QLabel
|
|
self.camera_pool_layout.addWidget(QLabel("No hardware cameras detected over USB."), 0, 0)
|
|
return
|
|
|
|
# Populate your grid layout with your new tags
|
|
from PySide6.QtWidgets import QLabel
|
|
for idx, (name, port) in enumerate(camera_list):
|
|
lbl = QLabel(f"📷 {name} [{port}]")
|
|
lbl.setStyleSheet("padding: 5px; background: #27ae60; color: white; border-radius: 4px; font-weight: bold;")
|
|
self.camera_pool_layout.addWidget(lbl, idx, 0)
|
|
|
|
def on_camera_query_error(self, error_msg):
|
|
"""Callback handles process crash or script paths missing."""
|
|
# Unlock refresh UI
|
|
self.refresh_cameras_btn.setEnabled(True)
|
|
self.refresh_cameras_btn.setText("🔄 Refresh Connected Cameras")
|
|
|
|
# Clear the scanning placeholder
|
|
while self.camera_pool_layout.count() > 0:
|
|
item = self.camera_pool_layout.takeAt(0)
|
|
if item.widget():
|
|
item.widget().deleteLater()
|
|
|
|
from PySide6.QtWidgets import QLabel
|
|
err_lbl = QLabel(f"Scan Failure:\n{error_msg}")
|
|
err_lbl.setWordWrap(True)
|
|
err_lbl.setStyleSheet("color: #e74c3c; font-weight: bold; padding: 4px;")
|
|
self.camera_pool_layout.addWidget(err_lbl, 0, 0)
|
|
|
|
|
|
def add_new_camera_row(self):
|
|
"""Creates a configuration row and binds changes to 3D rendering updates."""
|
|
next_index = self.rows_layout.count() + 1
|
|
new_row = CameraRowWidget(row_number=next_index)
|
|
|
|
# Connect signals
|
|
new_row.remove_requested.connect(self.remove_camera_row)
|
|
new_row.color_changed.connect(self.sync_3d_view)
|
|
new_row.config_changed.connect(self.sync_3d_view) # Trigger on dropdown change
|
|
|
|
self.rows_layout.addWidget(new_row)
|
|
self.sync_3d_view()
|
|
|
|
def remove_camera_row(self, row_widget):
|
|
if self.rows_layout.count() <= 1:
|
|
return
|
|
self.rows_layout.removeWidget(row_widget)
|
|
row_widget.deleteLater()
|
|
|
|
from PySide6.QtCore import QTimer
|
|
QTimer.singleShot(0, self.reindex_and_sync)
|
|
|
|
def reindex_and_sync(self):
|
|
self.reindex_camera_labels()
|
|
self.sync_3d_view()
|
|
|
|
def sync_3d_view(self):
|
|
"""Clears old 3D meshes and draws distinct cameras with non-touching indicators."""
|
|
# 1. Clean out the previous camera widgets
|
|
for item in self.rendered_camera_items:
|
|
self.top_left_widget.removeItem(item)
|
|
self.rendered_camera_items.clear()
|
|
|
|
# 2. Iterate through your configurations
|
|
for i in range(self.rows_layout.count()):
|
|
row = self.rows_layout.itemAt(i).widget()
|
|
if not isinstance(row, CameraRowWidget):
|
|
continue
|
|
|
|
config = row.get_values()
|
|
pos_preset = config["position"]
|
|
rgb = row.current_color.getRgbF()
|
|
|
|
# 3. Establish precise camera coordinates (assuming sphere radius is 2.0)
|
|
distance = 6.0 # Orbit radius away from the center sphere
|
|
cam_x, cam_y, cam_z = 0, 0, 0
|
|
|
|
# Target offset components for the arrowhead projection
|
|
dx, dy, dz = 0, 0, 0
|
|
|
|
# Diagonal components for 45-degree angle layouts (4.24 ≈ 6.0 * cos(45°))
|
|
diag = 4.24
|
|
|
|
# Assuming standard coordinate space:
|
|
# +Y is Forward (Front), -Y is Backward (Back)
|
|
# +X is Right, -X is Left
|
|
# +Z is Up (Top)
|
|
|
|
if pos_preset == "Front":
|
|
cam_x, cam_y, cam_z = 0.0, distance, 0.0
|
|
dy = -1.5
|
|
elif pos_preset == "Back":
|
|
cam_x, cam_y, cam_z = 0.0, -distance, 0.0
|
|
dy = 1.5
|
|
elif pos_preset == "Left":
|
|
cam_x, cam_y, cam_z = -distance, 0.0, 0.0
|
|
dx = 1.5
|
|
elif pos_preset == "Right":
|
|
cam_x, cam_y, cam_z = distance, 0.0, 0.0
|
|
dx = -1.5
|
|
|
|
# Diagonals / Corner Angles
|
|
elif pos_preset == "Front-Left": # Front-Left
|
|
cam_x, cam_y, cam_z = -diag, diag, 0.0
|
|
dx, dy = 1.0, -1.0
|
|
elif pos_preset == "Front-Right": # Right-Front
|
|
cam_x, cam_y, cam_z = diag, diag, 0.0
|
|
dx, dy = -1.0, -1.0
|
|
elif pos_preset == "Back-Left": # Left-Rear / Left-Back
|
|
cam_x, cam_y, cam_z = -diag, -diag, 0.0
|
|
dx, dy = 1.0, 1.0
|
|
elif pos_preset == "Back-Right": # Right-Rear / Right-Back
|
|
cam_x, cam_y, cam_z = diag, -diag, 0.0
|
|
dx, dy = -1.0, 1.0
|
|
|
|
# Overhead / Zenith Angle
|
|
elif pos_preset == "Top":
|
|
cam_x, cam_y, cam_z = 0.0, 0.0, distance
|
|
dz = -1.5
|
|
else:
|
|
# Custom/Fallback configuration if unexpected string input
|
|
cam_x, cam_y, cam_z = 0.0, distance, 2.0
|
|
dy, dz = -1.5, -0.5
|
|
|
|
# 4. Draw the Camera Cube cleanly
|
|
# We explicitly define dimensions and use GLBoxItem
|
|
cam_mesh = gl.GLBoxItem()
|
|
|
|
# Pass sizes as a standard Python tuple/list (X, Y, Z) to setSize
|
|
cam_mesh.setSize(1.0, 1.0, 1.0)
|
|
|
|
# Set the color using its dedicated method
|
|
cam_mesh.setColor(rgb)
|
|
|
|
# Center the box cleanly over our calculated coordinate point
|
|
cam_mesh.translate(cam_x - 0.5, cam_y - 0.5, cam_z - 0.5)
|
|
self.top_left_widget.addItem(cam_mesh)
|
|
self.rendered_camera_items.append(cam_mesh)
|
|
|
|
# 5. Draw a SHORT directional vector pointing out of the camera box
|
|
# It starts at the camera and extends a short distance toward the sphere, never touching it.
|
|
arrow_start = np.array([cam_x, cam_y, cam_z])
|
|
arrow_end = arrow_start + np.array([dx, dy, dz])
|
|
|
|
# Calculate standard orthogonal barbs to form a visible head tip
|
|
# Normalize direction vector to scale the arrowhead barbs correctly
|
|
vec = arrow_end - arrow_start
|
|
length = np.linalg.norm(vec)
|
|
if length > 0:
|
|
u = vec / length # Unit vector pointing toward sphere
|
|
|
|
# Create a simple perpendicular vector for the arrowhead wings
|
|
# if looking along Z, use X; otherwise shift cross product
|
|
perp = np.array([-u[1], u[0], 0]) if abs(u[2]) < 0.9 else np.array([0, -u[2], u[1]])
|
|
perp = (perp / np.linalg.norm(perp)) * 0.3 # barb width scale
|
|
|
|
# Back-step along the arrow line for barb length
|
|
barb_base = arrow_end - (u * 0.4)
|
|
|
|
# Combine shaft and the two arrow wings into an independent lines matrix
|
|
cam_arrow_points = np.array([
|
|
arrow_start, arrow_end, # Shaft
|
|
arrow_end, barb_base + perp, # Barb Wing 1
|
|
arrow_end, barb_base - perp # Barb Wing 2
|
|
])
|
|
else:
|
|
cam_arrow_points = np.array([arrow_start, arrow_end])
|
|
|
|
camera_arrow = gl.GLLinePlotItem(
|
|
pos=cam_arrow_points,
|
|
color=rgb,
|
|
width=3,
|
|
mode='lines' # Renders pairs of coordinates cleanly as broken segments
|
|
)
|
|
self.top_left_widget.addItem(camera_arrow)
|
|
self.rendered_camera_items.append(camera_arrow)
|
|
|
|
def clear_all_camera_rows(self):
|
|
"""Cleans out everything inside the list wrapper layout."""
|
|
while self.rows_layout.count() > 0:
|
|
item = self.rows_layout.takeAt(0)
|
|
widget = item.widget()
|
|
if widget is not None:
|
|
widget.deleteLater()
|
|
|
|
def get_all_camera_configs(self):
|
|
"""Call this inside on_run_task to export data as an array of dicts."""
|
|
configs = []
|
|
for i in range(self.rows_layout.count()):
|
|
widget = self.rows_layout.itemAt(i).widget()
|
|
if isinstance(widget, CameraRowWidget):
|
|
configs.append(widget.get_values())
|
|
return configs
|
|
|
|
|
|
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.button3.setVisible(False)
|
|
|
|
self.statusBar().showMessage("All data has been cleared.")
|
|
|
|
|
|
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 * 30 / 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.90)
|
|
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):
|
|
return
|
|
|
|
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 is_2d_bypass_func(self, checked):
|
|
self.is_2d_bypass = checked
|
|
# self._update_config_setting("2d_data_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 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.statusbar.showMessage("All settings have been reset to their default values.", 5000)
|
|
|
|
|
|
|
|
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 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):
|
|
|
|
self.button1.clicked.disconnect(self.on_run_task)
|
|
self.button1.setText("Cancel")
|
|
self.button1.clicked.connect(self.cancel_task)
|
|
|
|
|
|
if current_process().name == 'MainProcess':
|
|
# self.ack_queue = Queue()
|
|
# self.progress_queue = Queue()
|
|
self.console = LogConsoleDialog(self)
|
|
self.console.trigger_btn.clicked.connect(self.send_burst_trigger_signal)
|
|
self.console.show()
|
|
|
|
self.worker = SubprocessWorker(self.snirf_file_path, self.is_2d_bypass)
|
|
self.worker.line_received.connect(self.handle_live_logs)
|
|
self.worker.finished.connect(self.on_subprocess_complete)
|
|
self.worker.start()
|
|
|
|
# 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 handle_live_logs(self, text):
|
|
# Append raw log to the popup window text field
|
|
self.console.append_log(text)
|
|
|
|
# If the script says it is waiting, enable the action button
|
|
if "waiting for gui trigger" in text.strip().lower():
|
|
self.console.trigger_btn.setEnabled(True)
|
|
self.statusbar.showMessage("Hardware initialized. Awaiting user verification.")
|
|
|
|
def send_burst_trigger_signal(self):
|
|
# Disable the button so it can't be multi-clicked
|
|
self.console.trigger_btn.setEnabled(False)
|
|
self.statusbar.showMessage("Burst signal sent!")
|
|
|
|
# Write the file token that part_A.py is waiting for
|
|
self.worker.send_input("")
|
|
|
|
def on_subprocess_complete(self, returncode):
|
|
if returncode != 0:
|
|
self.console.append_log(f"\n[GUI SYSTEM ERROR]\n{returncode}")
|
|
self.statusbar.showMessage("Process failed.")
|
|
else:
|
|
self.statusbar.showMessage("Sequence completed successfully.")
|
|
|
|
|
|
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():
|
|
"""
|
|
Where the processing happens
|
|
"""
|
|
|
|
try:
|
|
print("here")
|
|
import part_A as light_a
|
|
light_a.gui_entry()
|
|
# gui_queue.join_thread()
|
|
# progress_queue.join_thread()
|
|
print("done")
|
|
os._exit(0)
|
|
|
|
except Exception as e:
|
|
print(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! |