""" Filename: project_manager.py Description: Manager file for anything project related Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in imports import os import sys import copy import pickle import concurrent from pathlib import Path, PurePosixPath # External library imports import pandas as pd from PySide6.QtWidgets import QMessageBox, QVBoxLayout, QFileDialog, QLabel, QDialog from PySide6.QtCore import QThread, Signal, Qt, QTimer from PySide6.QtGui import QAction from mne.io import read_raw_snirf from mne.preprocessing.nirs import source_detector_distances from mne_nirs.channels import get_short_channels # type: ignore from src.shared.flaresbasewidget import ProgressBubble from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME, DATA_SCHEMA 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 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, file_cfg, cfg_path): self.app = app self.file_cfg = file_cfg self.cfg_path = cfg_path # ========================================================================= # Path Utilities # ========================================================================= def get_safe_path(self, target_path, project_dir): """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 open_file_dialog(self): """Opens dialog to pick a single .snirf file.""" file_path, _ = QFileDialog.getOpenFileName( self.app, "Open File", "", "SNIRF Files (*.snirf);;All Files (*)" ) if file_path: self._load_files_into_pipeline([os.path.normpath(file_path)]) def open_folder_dialog(self): """Recursively finds all .snirf files in a selected directory.""" folder_path = QFileDialog.getExistingDirectory(self.app, "Select Folder", "") if folder_path: snirf_files = [os.path.normpath(str(f)) for f in Path(folder_path).rglob("*.snirf")] self._load_files_into_pipeline(snirf_files) def _load_files_into_pipeline(self, file_paths): """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 app.loading_session_id += 1 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 files already in the UI to avoid duplicates new_files = [p for p in file_paths if p not in app.selected_paths] 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): 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): """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): """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): """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(not 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=False, ask=False): """ 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 = { 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): 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): 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): 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): """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): """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): """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): """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): """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): """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): """Captures current file_metadata state as baseline.""" self.app.saved_file_metadata = copy.deepcopy(getattr(self.app, "file_metadata", {})) def is_metadata_dirty(self): """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): """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) -> dict[str, str]: result = {} 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-/sub-_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): """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