""" 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-/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: 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