tyler/flares
2
0
Fork 0
Code Issues 24 Pull Requests Actions Packages Projects Releases 25 Wiki Activity

improved heart rate calculations

Browse Source
This commit is contained in:
tyler
2026-08-24 16:40:02 -07:00
parent d9d5b6d940
commit 2fa3188296
4 changed files with 139 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
project_manager.py
+40 -30
View File
@@ -6,18 +6,22 @@ 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
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
from PySide6.QtWidgets import QMessageBox, QVBoxLayout, QFileDialog, QLabel, QDialog, QWidget
from PySide6.QtCore import QThread, Signal, Qt, QTimer
from PySide6.QtGui import QAction
@@ -29,17 +33,20 @@ 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, project_data):
def __init__(self, filename: str, project_data: dict[str, Any]) -> None:
super().__init__()
self.filename = filename
self.project_data = project_data
def run(self):
def run(self) -> None:
try:
with open(self.filename, "wb") as f:
pickle.dump(self.project_data, f)
@@ -50,7 +57,7 @@ class SaveProjectThread(QThread):
class SavingOverlay(QDialog):
def __init__(self, parent=None):
def __init__(self, parent: Optional[QWidget] = None) -> None:
super().__init__(parent)
self.setWindowFlags(Qt.WindowType.Dialog | Qt.WindowType.FramelessWindowHint)
self.setModal(True)
@@ -75,7 +82,11 @@ class ProjectManager:
- State baseline synchronization (dirty tracking)
"""
def __init__(self, app, file_cfg, cfg_path):
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
@@ -83,7 +94,7 @@ class ProjectManager:
# =========================================================================
# Path Utilities
# =========================================================================
def get_safe_path(self, target_path, project_dir):
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()
@@ -96,7 +107,7 @@ class ProjectManager:
# =========================================================================
# File & Folder Opening Dialogs
# =========================================================================
def open_file_dialog(self):
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 (*)"
@@ -104,14 +115,14 @@ class ProjectManager:
if file_path:
self._load_files_into_pipeline([os.path.normpath(file_path)])
def open_folder_dialog(self):
def open_folder_dialog(self)-> None:
"""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):
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:
@@ -151,7 +162,7 @@ class ProjectManager:
# Queue chunked widget creation
CHUNK_SIZE = 10
def process_chunk(file_queue):
def process_chunk(file_queue: List[str]) -> None:
chunk = file_queue[:CHUNK_SIZE]
remaining = file_queue[CHUNK_SIZE:]
@@ -187,7 +198,7 @@ class ProjectManager:
process_chunk(new_files)
def add_files_to_project(self, file_paths):
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]
@@ -220,7 +231,7 @@ class ProjectManager:
# =========================================================================
# Project Loading
# =========================================================================
def load_project_dialog(self):
def load_project_dialog(self) -> None:
"""Prompts for a project file and loads it."""
app = self.app
filename, _ = QFileDialog.getOpenFileName(
@@ -229,7 +240,7 @@ class ProjectManager:
if filename:
self.load_project(filename)
def load_project(self, filename):
def load_project(self, filename: str) -> None:
"""Loads a .flare project file into the application."""
app = self.app
try:
@@ -343,7 +354,7 @@ class ProjectManager:
# =========================================================================
# Project Saving (Save / Save As)
# =========================================================================
def save_project(self, onCrash=False, ask=False):
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).
@@ -436,7 +447,7 @@ class ProjectManager:
current_params = app.config_dict[first_file]
# 5. Build Serialized Payload
project_data = {
project_data: dict[str, Any] = {
item["key"]: getattr(app, item["key"], {}) for item in DATA_SCHEMA
}
project_data.update({
@@ -448,7 +459,7 @@ class ProjectManager:
"current_ui_params": current_params,
})
def sanitize(obj):
def sanitize(obj: Any) -> Any:
if isinstance(obj, Path):
return str(PurePosixPath(obj))
elif isinstance(obj, dict):
@@ -469,7 +480,7 @@ class ProjectManager:
app.save_thread = SaveProjectThread(filename, project_data)
def _on_save_success(saved_file):
def _on_save_success(saved_file: str) -> None:
if hasattr(app, "saving_overlay"):
app.saving_overlay.close()
@@ -483,7 +494,7 @@ class ProjectManager:
app, "Success", f"Project saved to:\n{saved_file}"
)
def _on_save_error(error_msg):
def _on_save_error(error_msg: str) -> None:
if hasattr(app, "saving_overlay"):
app.saving_overlay.close()
if not onCrash:
@@ -500,7 +511,7 @@ class ProjectManager:
QMessageBox.critical(app, "Error", f"Failed to save project:\n{e}")
def update_recent_projects_menu(self):
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"):
@@ -526,7 +537,7 @@ class ProjectManager:
)
app.recent_projects_menu.addAction(action)
def add_to_recent_projects(self, project_path):
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()]
@@ -546,7 +557,7 @@ class ProjectManager:
self.update_recent_projects_menu()
def open_recent_project(self, project_path):
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}")
@@ -577,7 +588,7 @@ class ProjectManager:
# =========================================================================
# Recent Files Operations
# =========================================================================
def update_recent_files_menu(self):
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"):
@@ -600,7 +611,7 @@ class ProjectManager:
)
app.recent_files_menu.addAction(action)
def add_to_recent_files(self, file_path):
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()]
@@ -620,7 +631,7 @@ class ProjectManager:
self.update_recent_files_menu()
def open_recent_file(self, file_path):
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}")
@@ -647,11 +658,11 @@ class ProjectManager:
# =========================================================================
# Baseline Synchronization & Dirty Checks
# =========================================================================
def sync_metadata_baseline(self):
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):
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", {})
@@ -671,7 +682,7 @@ class ProjectManager:
return False
def reset_all_dirty_states(self):
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", []))
@@ -692,7 +703,6 @@ class ProjectManager:
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) -
@@ -708,7 +718,7 @@ def _get_bids_demographics(snirf_path: str) -> dict[str, str]:
if not sub_id:
return {}
def _row_to_dict(row) -> dict[str, str]:
def _row_to_dict(row: pd.Series) -> dict[str, str]:
result = {}
for field in fields:
if field not in row:
@@ -756,7 +766,7 @@ def _get_bids_demographics(snirf_path: str) -> dict[str, str]:
def extract_metadata_worker(file_name):
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.