934 lines
35 KiB
Python
934 lines
35 KiB
Python
"""
|
|
Filename: project_manager.py
|
|
Description: Manager file for anything project related
|
|
|
|
Author: Tyler de Zeeuw
|
|
License: GPL-3.0
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
# Built-in imports
|
|
import os
|
|
import sys
|
|
import copy
|
|
import pickle
|
|
import concurrent
|
|
import configparser
|
|
import concurrent.futures
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import TYPE_CHECKING, Any, List, Optional, Union
|
|
|
|
# External library imports
|
|
import pandas as pd
|
|
|
|
from PySide6.QtWidgets import QMessageBox, QVBoxLayout, QFileDialog, QLabel, QDialog, QWidget
|
|
from PySide6.QtCore import QThread, Signal, Qt, QTimer
|
|
from PySide6.QtGui import QAction
|
|
|
|
from mne.io import read_raw_snirf # type: ignore
|
|
from mne.preprocessing.nirs import source_detector_distances # type: ignore
|
|
from mne_nirs.channels import get_short_channels # type: ignore
|
|
|
|
from src.shared.flaresbasewidget import ProgressBubble
|
|
from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME, DATA_SCHEMA
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from main import MainApplication
|
|
|
|
|
|
class SaveProjectThread(QThread):
|
|
finished_signal = Signal(str)
|
|
error_signal = Signal(str)
|
|
|
|
def __init__(self, filename: str, project_data: dict[str, Any]) -> None:
|
|
super().__init__()
|
|
self.filename = filename
|
|
self.project_data = project_data
|
|
|
|
def run(self) -> None:
|
|
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: Optional[QWidget] = None) -> 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 ProjectManager:
|
|
"""
|
|
Central manager for all I/O operations:
|
|
- File loading (individual files & folders)
|
|
- Project loading & saving (Save vs Save As)
|
|
- Relative / absolute path utilities
|
|
- State baseline synchronization (dirty tracking)
|
|
"""
|
|
|
|
def __init__(self,
|
|
app: MainApplication,
|
|
file_cfg: configparser.ConfigParser,
|
|
cfg_path: str,
|
|
) -> None:
|
|
self.app = app
|
|
self.file_cfg = file_cfg
|
|
self.cfg_path = cfg_path
|
|
|
|
# =========================================================================
|
|
# Path Utilities
|
|
# =========================================================================
|
|
def get_safe_path(self, target_path: Union[str, Path], project_dir: Union[str, Path]) -> str:
|
|
"""Converts an absolute file path to a relative path relative to project_dir."""
|
|
try:
|
|
target = Path(target_path).resolve()
|
|
proj = Path(project_dir).resolve()
|
|
return str(PurePosixPath(target.relative_to(proj)))
|
|
except ValueError:
|
|
# Fall back to absolute path string if on a different drive/volume
|
|
return str(PurePosixPath(Path(target_path).resolve()))
|
|
|
|
# =========================================================================
|
|
# File & Folder Opening Dialogs
|
|
# =========================================================================
|
|
|
|
def is_valid_snirf(self, path: str) -> bool:
|
|
"""Fast header check to verify HDF5/SNIRF signature and ignore corrupt/shadow files."""
|
|
try:
|
|
if os.path.getsize(path) < 8:
|
|
return False
|
|
with open(path, "rb") as f:
|
|
return f.read(8) == b"\x89HDF\r\n\x1a\n"
|
|
except Exception:
|
|
return False
|
|
|
|
def open_file_dialog(self) -> None:
|
|
"""Opens dialog to pick a single .snirf file."""
|
|
file_path, _ = QFileDialog.getOpenFileName(
|
|
self.app, "Open File", "", "SNIRF Files (*.snirf);;All Files (*)"
|
|
)
|
|
if not file_path:
|
|
return
|
|
|
|
norm_path = os.path.normpath(file_path)
|
|
filename = os.path.basename(norm_path)
|
|
|
|
# 1. Check specifically for macOS shadow files
|
|
if filename.startswith("._"):
|
|
QMessageBox.warning(
|
|
self.app,
|
|
"Invalid SNIRF File",
|
|
f"'{filename}' is a macOS system shadow file (Apple Double resource fork), not a actual data file."
|
|
)
|
|
return
|
|
|
|
# 2. Check for general header validity / corruption
|
|
if not self.is_valid_snirf(norm_path):
|
|
QMessageBox.warning(
|
|
self.app,
|
|
"Invalid SNIRF File",
|
|
f"'{filename}' is not a valid SNIRF file or has a corrupted header."
|
|
)
|
|
return
|
|
|
|
self._load_files_into_pipeline([norm_path])
|
|
|
|
def open_folder_dialog(self)-> None:
|
|
"""Recursively finds all .snirf files in a selected directory."""
|
|
folder_path = QFileDialog.getExistingDirectory(self.app, "Select Folder", "")
|
|
if not folder_path:
|
|
return
|
|
|
|
# Automatically filter out shadow files and invalid headers in batch mode
|
|
snirf_files = [
|
|
os.path.normpath(str(f))
|
|
for f in Path(folder_path).rglob("*.snirf")
|
|
if self.is_valid_snirf(str(f))
|
|
]
|
|
|
|
if not snirf_files:
|
|
QMessageBox.information(
|
|
self.app,
|
|
"No Valid Files",
|
|
"No valid .snirf files were found in the selected directory."
|
|
)
|
|
return
|
|
|
|
def load_dropped_files(self, file_paths: List[str]) -> None:
|
|
"""Public entry point for handling files or folders dropped onto the UI."""
|
|
if not file_paths:
|
|
return
|
|
|
|
# 1. Single file drop: Give explicit feedback like open_file_dialog
|
|
if len(file_paths) == 1:
|
|
norm_path = os.path.normpath(file_paths[0])
|
|
filename = os.path.basename(norm_path)
|
|
|
|
if filename.startswith("._"):
|
|
QMessageBox.warning(
|
|
self.app,
|
|
"Invalid SNIRF File",
|
|
f"'{filename}' is a macOS system shadow file (Apple Double resource fork), not an actual data file."
|
|
)
|
|
return
|
|
|
|
if not self.is_valid_snirf(norm_path):
|
|
QMessageBox.warning(
|
|
self.app,
|
|
"Invalid SNIRF File",
|
|
f"'{filename}' is not a valid SNIRF file or has a corrupted header."
|
|
)
|
|
return
|
|
|
|
valid_files = [norm_path]
|
|
|
|
# 2. Batch drop: Silently filter out shadow/corrupt files like open_folder_dialog
|
|
else:
|
|
valid_files = [
|
|
os.path.normpath(p)
|
|
for p in file_paths
|
|
if self.is_valid_snirf(os.path.normpath(p))
|
|
]
|
|
|
|
if not valid_files:
|
|
QMessageBox.information(
|
|
self.app,
|
|
"No Valid Files",
|
|
"None of the dropped items were valid .snirf files."
|
|
)
|
|
return
|
|
|
|
# Delegate to internal pipeline
|
|
self._load_files_into_pipeline(valid_files)
|
|
|
|
def _load_files_into_pipeline(self, file_paths: List[str]) -> None:
|
|
"""Loads .snirf files into UI using chunked batches and background workers."""
|
|
app = self.app
|
|
if not file_paths:
|
|
return
|
|
|
|
# 1. Warm up the executor if needed
|
|
if not hasattr(app, "file_executor") or app.file_executor is None:
|
|
app.file_executor = concurrent.futures.ProcessPoolExecutor(max_workers=1)
|
|
|
|
# 2. Track this session to prevent ghost updates
|
|
if not hasattr(app, "loading_session_id"):
|
|
app.loading_session_id = 0
|
|
current_session = app.loading_session_id
|
|
|
|
# 3. Setup internal tracking if not exists
|
|
if not hasattr(app, "bubble_widgets"):
|
|
app.bubble_widgets = {}
|
|
if not hasattr(app, "selected_paths"):
|
|
app.selected_paths = []
|
|
if not hasattr(app, "metadata_cache"):
|
|
app.metadata_cache = {}
|
|
|
|
# Filter out duplicates AND non-SNIRF files (including macOS ._ shadow files)
|
|
new_files = [
|
|
p for p in file_paths
|
|
if p not in app.selected_paths and self.is_valid_snirf(p)
|
|
]
|
|
if not new_files:
|
|
return
|
|
|
|
# Update the pending count for the current load batch
|
|
if not hasattr(app, "pending_files_count"):
|
|
app.pending_files_count = 0
|
|
app.pending_files_count += len(new_files)
|
|
|
|
app.button1.setVisible(True)
|
|
app.statusBar().showMessage(f"Loading {len(new_files)} new file(s)...")
|
|
|
|
# Queue chunked widget creation
|
|
CHUNK_SIZE = 10
|
|
|
|
def process_chunk(file_queue: List[str]) -> None:
|
|
chunk = file_queue[:CHUNK_SIZE]
|
|
remaining = file_queue[CHUNK_SIZE:]
|
|
|
|
for path in chunk:
|
|
app.selected_paths.append(path)
|
|
self.add_to_recent_files(path)
|
|
|
|
display_name = os.path.basename(path)
|
|
bubble = ProgressBubble(display_name, path)
|
|
bubble.setCursor(Qt.CursorShape.WaitCursor)
|
|
bubble.set_loading_state(True)
|
|
|
|
app.bubble_widgets[path] = bubble
|
|
app.bubble_layout.addWidget(bubble)
|
|
|
|
# Submit background task as each bubble is constructed
|
|
future = app.file_executor.submit(extract_metadata_worker, path)
|
|
future.add_done_callback(
|
|
lambda f, p=path, s=current_session: app._on_metadata_ready(f, p, s)
|
|
)
|
|
|
|
app.files_are_dirty = True
|
|
app.is_saved = False
|
|
|
|
if hasattr(app, "check_if_app_is_dirty"):
|
|
app.check_if_app_is_dirty()
|
|
elif hasattr(app, "update_window_title"):
|
|
app.update_window_title()
|
|
|
|
# Schedule remaining files
|
|
if remaining:
|
|
QTimer.singleShot(0, lambda: process_chunk(remaining))
|
|
|
|
process_chunk(new_files)
|
|
|
|
def add_files_to_project(self, file_paths: List[str]) -> None:
|
|
"""Adds file paths to the application state, creating bubble UI items."""
|
|
app = self.app
|
|
normalized_paths = [os.path.normpath(p) for p in file_paths]
|
|
|
|
# Merge with existing selected paths avoiding duplicates
|
|
existing_paths = getattr(app, "selected_paths", [])
|
|
new_paths = [p for p in normalized_paths if p not in existing_paths]
|
|
|
|
if not new_paths:
|
|
return
|
|
|
|
app.selected_paths = existing_paths + new_paths
|
|
|
|
# Render file bubbles in UI if method exists
|
|
if hasattr(app, "show_files_as_bubbles_from_list"):
|
|
progress_states = getattr(app, "progress_states", {})
|
|
current_project = getattr(app, "current_project_path", "")
|
|
app.show_files_as_bubbles_from_list(
|
|
app.selected_paths, progress_states, current_project
|
|
)
|
|
|
|
# Record in recent files menu
|
|
for path in new_paths:
|
|
self.add_to_recent_files(path)
|
|
|
|
# Trigger dirty check
|
|
if hasattr(app, "check_if_app_is_dirty"):
|
|
app.check_if_app_is_dirty()
|
|
|
|
# =========================================================================
|
|
# Project Loading
|
|
# =========================================================================
|
|
def load_project_dialog(self) -> None:
|
|
"""Prompts for a project file and loads it."""
|
|
app = self.app
|
|
filename, _ = QFileDialog.getOpenFileName(
|
|
app, "Load Project", "", "FLARE Project (*.flare)"
|
|
)
|
|
if filename:
|
|
self.load_project(filename)
|
|
|
|
def load_project(self, filename: str) -> None:
|
|
"""Loads a .flare project file into the application."""
|
|
app = self.app
|
|
try:
|
|
with open(filename, "rb") as f:
|
|
data = pickle.load(f)
|
|
|
|
checks = [
|
|
("version", "<=1.1.7"),
|
|
("file_metadata", "<=1.2.2"),
|
|
("file_parameters", "<=1.3.0"),
|
|
("roi_channel_map_dict", "<=1.5.2"),
|
|
]
|
|
|
|
for key, ver_str in checks:
|
|
if key not in data:
|
|
msg = (
|
|
f"This project was saved in an earlier version of {APP_NAME.upper()} ({ver_str}) "
|
|
"and is potentially not compatible with this version. "
|
|
)
|
|
if getattr(app, "incompatible_save_bypass", False):
|
|
QMessageBox.warning(
|
|
app, f"Warning - {APP_NAME.upper()}", msg + "Attempting load."
|
|
)
|
|
break
|
|
else:
|
|
QMessageBox.critical(
|
|
app,
|
|
f"Error - {APP_NAME.upper()}",
|
|
msg + "Enable bypass in Preferences to load.",
|
|
)
|
|
return
|
|
|
|
# Clear existing UI bubbles
|
|
if hasattr(app, "bubble_widgets"):
|
|
for bubble in list(app.bubble_widgets.values()):
|
|
bubble.setParent(None)
|
|
bubble.deleteLater()
|
|
app.bubble_widgets.clear()
|
|
|
|
app.selected_paths = []
|
|
app.current_project_path = filename
|
|
|
|
# Restore Data Schema
|
|
for item in DATA_SCHEMA:
|
|
setattr(app, item["key"], data.get(item["key"], {}))
|
|
|
|
project_dir = Path(filename).parent
|
|
saved_cache = data.get("file_metadata", {})
|
|
raw_params = data.get("file_parameters", {})
|
|
app.metadata_cache = {}
|
|
app.file_metadata = {}
|
|
|
|
for rel_path, meta_content in saved_cache.items():
|
|
abs_path = str((project_dir / Path(rel_path)).resolve())
|
|
app.metadata_cache[abs_path] = meta_content
|
|
|
|
file_list = [
|
|
str((project_dir / Path(rel_path)).resolve()) for rel_path in data["file_list"]
|
|
]
|
|
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:
|
|
app.file_metadata[abs_path] = raw_params[rel_path]
|
|
elif hasattr(app, "config_dict") and abs_path in app.config_dict:
|
|
old_cfg = app.config_dict[abs_path]
|
|
app.file_metadata[abs_path] = {
|
|
"AGE": str(old_cfg.get("AGE", "")),
|
|
"SEX": str(old_cfg.get("SEX", "")),
|
|
"HAND": str(old_cfg.get("HAND", "")),
|
|
"GROUP": str(old_cfg.get("GROUP", "")),
|
|
}
|
|
else:
|
|
app.file_metadata[abs_path] = {
|
|
"AGE": "",
|
|
"SEX": "",
|
|
"HAND": "",
|
|
"GROUP": "",
|
|
}
|
|
|
|
app.show_files_as_bubbles_from_list(file_list, progress_states, filename)
|
|
|
|
if "current_ui_params" in data:
|
|
app.restore_sections_from_config(data["current_ui_params"])
|
|
elif getattr(app, "config_dict", None):
|
|
first_file = next(iter(app.config_dict.keys()))
|
|
app.restore_sections_from_config(app.config_dict[first_file])
|
|
|
|
has_data = any(len(getattr(app, item["key"], {})) > 0 for item in DATA_SCHEMA)
|
|
if hasattr(app, "button1"):
|
|
app.button1.setVisible(has_data)
|
|
if hasattr(app, "button3"):
|
|
app.button3.setVisible(has_data)
|
|
|
|
|
|
self.add_to_recent_projects(os.path.normpath(filename))
|
|
|
|
# Reset baselines cleanly
|
|
self.reset_all_dirty_states()
|
|
|
|
QMessageBox.information(app, "Loaded", f"Project loaded from:\n{filename}")
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(app, "Error", f"Failed to load project:\n{e}")
|
|
|
|
# =========================================================================
|
|
# Project Saving (Save / Save As)
|
|
# =========================================================================
|
|
def save_project(self, onCrash: bool = False, ask: bool = False) -> None:
|
|
"""
|
|
Saves the project to disk.
|
|
- ask=False: Quick Save to self.app.current_project_path (prompts if unsaved).
|
|
- ask=True: Save As (always prompts for location).
|
|
"""
|
|
app = self.app
|
|
|
|
# 1. Sync active text fields into active metadata dict
|
|
if hasattr(app, "current_file") and app.current_file and hasattr(app, "meta_fields"):
|
|
if not hasattr(app, "file_metadata"):
|
|
app.file_metadata = {}
|
|
app.file_metadata[app.current_file] = {
|
|
key: field.text().strip() for key, field in app.meta_fields.items()
|
|
}
|
|
|
|
# 2. Check if saveable state exists
|
|
has_files = len(getattr(app, "selected_paths", [])) > 0
|
|
has_metadata = any(
|
|
any(val for val in m.values()) for m in getattr(app, "file_metadata", {}).values()
|
|
)
|
|
has_param_changes = any(
|
|
s.has_any_changes() for s in getattr(app, "param_sections", [])
|
|
)
|
|
has_processed_data = any(
|
|
len(getattr(app, item["key"], {})) > 0 for item in DATA_SCHEMA
|
|
)
|
|
|
|
if not (has_files or has_processed_data or has_metadata or has_param_changes):
|
|
if not onCrash:
|
|
QMessageBox.warning(
|
|
app,
|
|
"Save Project",
|
|
"There is no data or configuration to save.",
|
|
)
|
|
return
|
|
|
|
# 3. Path Resolution
|
|
filename = None
|
|
if not onCrash:
|
|
existing_path = getattr(app, "current_project_path", None)
|
|
if ask or not existing_path:
|
|
start_dir = existing_path if existing_path else ""
|
|
filename, _ = QFileDialog.getSaveFileName(
|
|
app, "Save Project", start_dir, "FLARE Project (*.flare)"
|
|
)
|
|
if not filename:
|
|
return
|
|
else:
|
|
filename = existing_path
|
|
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:
|
|
if not filename.endswith(".flare"):
|
|
filename += ".flare"
|
|
|
|
project_path = Path(filename).resolve()
|
|
project_dir = project_path.parent
|
|
|
|
# 4. Convert Paths to Relative
|
|
bubble_widgets = getattr(app, "bubble_widgets", {})
|
|
file_list = [
|
|
self.get_safe_path(b.file_path, project_dir) for b in bubble_widgets.values()
|
|
]
|
|
progress_states = {
|
|
self.get_safe_path(b.file_path, project_dir): getattr(b, "current_step", 0)
|
|
for b in bubble_widgets.values()
|
|
}
|
|
|
|
rel_metadata = {}
|
|
for full_path, meta in getattr(app, "metadata_cache", {}).items():
|
|
try:
|
|
rel_metadata[self.get_safe_path(full_path, project_dir)] = meta
|
|
except Exception as e:
|
|
print(f"Metadata conversion failed for {full_path}: {e}")
|
|
|
|
rel_file_params = {
|
|
self.get_safe_path(f_path, project_dir): meta
|
|
for f_path, meta in getattr(app, "file_metadata", {}).items()
|
|
}
|
|
|
|
current_params = app.get_all_current_ui_params()
|
|
if not current_params and getattr(app, "config_dict", None):
|
|
first_file = next(iter(app.config_dict.keys()))
|
|
current_params = app.config_dict[first_file]
|
|
|
|
# 5. Build Serialized Payload
|
|
project_data: dict[str, Any] = {
|
|
item["key"]: getattr(app, item["key"], {}) for item in DATA_SCHEMA
|
|
}
|
|
project_data.update({
|
|
"version": CURRENT_VERSION,
|
|
"file_list": file_list,
|
|
"progress_states": progress_states,
|
|
"file_metadata": rel_metadata,
|
|
"file_parameters": rel_file_params,
|
|
"current_ui_params": current_params,
|
|
})
|
|
|
|
def sanitize(obj: Any) -> Any:
|
|
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.add_to_recent_projects(os.path.normpath(filename))
|
|
|
|
# 6. Background Saving Execution
|
|
if not onCrash:
|
|
app.saving_overlay = SavingOverlay(app)
|
|
app.saving_overlay.resize(app.size())
|
|
app.saving_overlay.show()
|
|
|
|
app.save_thread = SaveProjectThread(filename, project_data)
|
|
|
|
def _on_save_success(saved_file: str) -> None:
|
|
if hasattr(app, "saving_overlay"):
|
|
app.saving_overlay.close()
|
|
|
|
self.add_to_recent_projects(os.path.normpath(saved_file))
|
|
|
|
app.current_project_path = saved_file
|
|
self.reset_all_dirty_states()
|
|
|
|
if not onCrash:
|
|
QMessageBox.information(
|
|
app, "Success", f"Project saved to:\n{saved_file}"
|
|
)
|
|
|
|
def _on_save_error(error_msg: str) -> None:
|
|
if hasattr(app, "saving_overlay"):
|
|
app.saving_overlay.close()
|
|
if not onCrash:
|
|
QMessageBox.critical(
|
|
app, "Error", f"Failed to save project:\n{error_msg}"
|
|
)
|
|
|
|
app.save_thread.finished_signal.connect(_on_save_success)
|
|
app.save_thread.error_signal.connect(_on_save_error)
|
|
app.save_thread.start()
|
|
|
|
except Exception as e:
|
|
if not onCrash:
|
|
QMessageBox.critical(app, "Error", f"Failed to save project:\n{e}")
|
|
|
|
|
|
def update_recent_projects_menu(self) -> None:
|
|
"""Clears and rebuilds the Recent Projects submenu items."""
|
|
app = self.app
|
|
if not hasattr(app, "recent_projects_menu"):
|
|
return
|
|
|
|
app.recent_projects_menu.clear()
|
|
|
|
raw_projects = self.file_cfg.get("File", "recent_projects", fallback="")
|
|
projects = [p.strip() for p in raw_projects.split(",") if p.strip()]
|
|
|
|
if not projects:
|
|
no_recent = app.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}", app)
|
|
action.setToolTip(project_path)
|
|
action.triggered.connect(
|
|
lambda checked, path=project_path: self.open_recent_project(path)
|
|
)
|
|
app.recent_projects_menu.addAction(action)
|
|
|
|
def add_to_recent_projects(self, project_path: str) -> None:
|
|
"""Adds a project path, moves it to the top, and hard caps at 10."""
|
|
raw_projects = self.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
|
|
|
|
self.file_cfg.set("File", "recent_projects", ",".join(projects))
|
|
try:
|
|
with open(self.cfg_path, "w") as f:
|
|
self.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: str) -> None:
|
|
"""The slot that executes when a recent project entry is clicked."""
|
|
if os.path.exists(project_path):
|
|
print(f"Opening recent project: {project_path}")
|
|
|
|
# Route project loading through ProjectManager or app's loader
|
|
if hasattr(self.app, "project_loader"):
|
|
self.app.project_loader(project_path)
|
|
else:
|
|
self.load_project(project_path)
|
|
|
|
self.add_to_recent_projects(project_path)
|
|
else:
|
|
QMessageBox.warning(
|
|
self.app,
|
|
"Project Not Found",
|
|
f"The project file could not be found:\n{project_path}",
|
|
)
|
|
# Clean out the broken path
|
|
raw_projects = self.file_cfg.get("File", "recent_projects", fallback="")
|
|
projects = [
|
|
p.strip()
|
|
for p in raw_projects.split(",")
|
|
if p.strip() and p.strip() != project_path
|
|
]
|
|
self.file_cfg.set("File", "recent_projects", ",".join(projects))
|
|
self.update_recent_projects_menu()
|
|
|
|
# =========================================================================
|
|
# Recent Files Operations
|
|
# =========================================================================
|
|
def update_recent_files_menu(self) -> None:
|
|
"""Clears and rebuilds the Recent Files submenu items."""
|
|
app = self.app
|
|
if not hasattr(app, "recent_files_menu"):
|
|
return
|
|
|
|
app.recent_files_menu.clear()
|
|
|
|
raw_files = self.file_cfg.get("File", "recent_files", fallback="")
|
|
files = [f.strip() for f in raw_files.split(",") if f.strip()]
|
|
|
|
if not files:
|
|
no_recent = app.recent_files_menu.addAction("No Recent Files")
|
|
no_recent.setEnabled(False)
|
|
return
|
|
|
|
for i, file_path in enumerate(files):
|
|
action = QAction(f"{i+1}: {file_path}", app)
|
|
action.triggered.connect(
|
|
lambda checked, path=file_path: self.open_recent_file(path)
|
|
)
|
|
app.recent_files_menu.addAction(action)
|
|
|
|
def add_to_recent_files(self, file_path: str) -> None:
|
|
"""Adds a path, moves it to the top, and hard caps the list at 10."""
|
|
raw_files = self.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]
|
|
|
|
self.file_cfg.set("File", "recent_files", ",".join(files))
|
|
try:
|
|
with open(self.cfg_path, "w") as f:
|
|
self.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: str) -> None:
|
|
"""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.app,
|
|
"File Not Found",
|
|
f"The file could not be found:\n{file_path}",
|
|
)
|
|
# Clean up the broken link from history
|
|
raw_files = self.file_cfg.get("File", "recent_files", fallback="")
|
|
files = [
|
|
f.strip()
|
|
for f in raw_files.split(",")
|
|
if f.strip() and f.strip() != file_path
|
|
]
|
|
self.file_cfg.set("File", "recent_files", ",".join(files))
|
|
self.update_recent_files_menu()
|
|
|
|
# =========================================================================
|
|
# Baseline Synchronization & Dirty Checks
|
|
# =========================================================================
|
|
def sync_metadata_baseline(self) -> None:
|
|
"""Captures current file_metadata state as baseline."""
|
|
self.app.saved_file_metadata = copy.deepcopy(getattr(self.app, "file_metadata", {}))
|
|
|
|
def is_metadata_dirty(self) -> bool:
|
|
"""Returns True if file metadata has been modified relative to saved baseline."""
|
|
current_meta = getattr(self.app, "file_metadata", {})
|
|
saved_meta = getattr(self.app, "saved_file_metadata", {})
|
|
|
|
if current_meta != saved_meta:
|
|
return True
|
|
|
|
if (
|
|
hasattr(self.app, "current_file")
|
|
and self.app.current_file
|
|
and hasattr(self.app, "meta_fields")
|
|
):
|
|
active_saved = saved_meta.get(self.app.current_file, {})
|
|
for key, field in getattr(self.app, "meta_fields", {}).items():
|
|
if field.text().strip() != active_saved.get(key, "").strip():
|
|
return True
|
|
|
|
return False
|
|
|
|
def reset_all_dirty_states(self) -> None:
|
|
"""Resets parameter, file, and metadata baselines after load/save."""
|
|
# 1. Sync file list baseline
|
|
self.app.saved_selected_paths = copy.deepcopy(getattr(self.app, "selected_paths", []))
|
|
self.app.files_are_dirty = False
|
|
|
|
# 2. Sync metadata baseline
|
|
self.sync_metadata_baseline()
|
|
|
|
# 3. Sync parameter baselines
|
|
for section_widget in getattr(self.app, "param_sections", []):
|
|
if hasattr(section_widget, "save_current_as_baseline"):
|
|
section_widget.save_current_as_baseline()
|
|
|
|
# 4. Clear dirty status & update UI title
|
|
self.app.is_saved = True
|
|
if hasattr(self.app, "update_window_title"):
|
|
self.app.update_window_title()
|
|
|
|
|
|
|
|
def _get_bids_demographics(snirf_path: str) -> dict[str, str]:
|
|
"""Traverses the path of a SNIRF file to extract age/sex/hand from BIDS TSV files.
|
|
'hand' is only included if a value is present and isn't 'n/a' (case-insensitive) -
|
|
many datasets leave it unset/inapplicable, so surfacing 'n/a' explicitly just adds noise.
|
|
"""
|
|
path = Path(snirf_path)
|
|
fields = ["age", "sex", "hand"]
|
|
|
|
# Extract sub-XX and ses-YY labels from the path
|
|
sub_id = next((part for part in path.parts if part.startswith("sub-")), None)
|
|
ses_id = next((part for part in path.parts if part.startswith("ses-")), None)
|
|
|
|
if not sub_id:
|
|
return {}
|
|
|
|
def _row_to_dict(row: pd.Series) -> dict[str, str]:
|
|
result: dict[str, str] = {}
|
|
for field in fields:
|
|
if field not in row:
|
|
continue
|
|
val = row[field]
|
|
if pd.isna(val):
|
|
continue
|
|
val_str = str(val).strip()
|
|
if field == "hand" and val_str.lower() in ("n/a", "na", ""):
|
|
continue
|
|
result[field] = val_str
|
|
return result
|
|
|
|
# 1. Look for sub-<id>/sub-<id>_sessions.tsv
|
|
sub_dir = next((p for p in path.parents if p.name == sub_id), None)
|
|
if sub_dir and ses_id:
|
|
sessions_tsv = sub_dir / f"{sub_id}_sessions.tsv"
|
|
if sessions_tsv.exists():
|
|
try:
|
|
df = pd.read_csv(sessions_tsv, sep="\t")
|
|
matching = df[df["session_id"].astype(str).str.replace("ses-", "") == ses_id.replace("ses-", "")]
|
|
if not matching.empty:
|
|
result = _row_to_dict(matching.iloc[0])
|
|
if result:
|
|
return result
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Fallback: Check dataset root participants.tsv
|
|
bids_root = sub_dir.parent if sub_dir else None
|
|
if bids_root:
|
|
participants_tsv = bids_root / "participants.tsv"
|
|
if participants_tsv.exists():
|
|
try:
|
|
df = pd.read_csv(participants_tsv, sep="\t")
|
|
matching = df[df["participant_id"].astype(str).str.replace("sub-", "") == sub_id.replace("sub-", "")]
|
|
if not matching.empty:
|
|
result = _row_to_dict(matching.iloc[0])
|
|
if result:
|
|
return result
|
|
except Exception:
|
|
pass
|
|
|
|
return {}
|
|
|
|
|
|
|
|
def extract_metadata_worker(file_name: str) -> dict[str, Any]:
|
|
"""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}"
|
|
total_chans = len(raw.ch_names)
|
|
pct_short = (len(names) / total_chans * 100) if total_chans else 0
|
|
if pct_short > 35:
|
|
snirf_info['Short Channels'] += "\n There are a lot of short channels. Perhaps the optode distances are 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"
|
|
|
|
demographics = _get_bids_demographics(file_name)
|
|
if "age" in demographics:
|
|
snirf_info["BIDS - Age"] = demographics["age"]
|
|
if "sex" in demographics:
|
|
snirf_info["BIDS - Sex"] = demographics["sex"]
|
|
if "hand" in demographics:
|
|
snirf_info["BIDS - Handedness"] = demographics["hand"]
|
|
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 |