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

typos and pylance

Browse Source
This commit is contained in:
tyler
2026-08-23 00:12:22 -07:00
parent e37275a1bb
commit d9d5b6d940
18 changed files with 115 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
file_ext_registration.py
+1 -1
View File
@@ -16,6 +16,7 @@ from typing import Optional, Tuple
# External library imports
from src.shared.shareddata import APP_NAME, PLATFORM_NAME
ELEVATION_FLAG = "--register_file_association_elevated"
@@ -25,7 +26,6 @@ def register_file_association(ext: Optional[str] = None,
bundle_id: Optional[str] = None,
force_admin: bool = False,
) -> Tuple[bool, str]:
"""
Registers a custom file extension across Windows, Linux, and macOS.
Handles non-admin Windows users by falling back to local user registry.
flares.py
+16 -5
View File
@@ -25,7 +25,7 @@ from copy import deepcopy
import multiprocessing as mp
from itertools import compress
from queue import Empty, Queue
from typing import Any, Optional, Sequence, cast, Literal, Union
from typing import Any, Optional, Sequence, cast, Literal, Union, List
# External library imports
import matplotlib.pyplot as plt
@@ -6274,10 +6274,21 @@ def _single_subject_epoch_coherence(
# ============================================================================
def run_group_functional_connectivity_betas(
haemo_dict, selected_paths, event_name, n_lines, vmin,
*, drift_model="cosine", drift_order=1, hrf_model="glover",
apply_gsr=True, alpha=0.05, min_participants=3, resample_freq=4.0,
haemo_dict: dict[str, BaseRaw],
selected_paths: List[str],
event_name: Optional[str],
n_lines: int,
vmin: float,
*,
drift_model: str = "cosine",
drift_order: int = 1,
hrf_model: str = "glover",
apply_gsr: bool = True,
alpha: float = 0.05,
min_participants: int = 3,
resample_freq: float = 4.0,
) -> None:
subject_results = []
for path in selected_paths:
raw = haemo_dict.get(path)
@@ -6311,7 +6322,7 @@ def run_group_functional_connectivity_betas(
def run_group_functional_connectivity_epochs(
epochs_dict: dict[str | Path, Epochs],
epochs_dict: dict[str, Epochs],
selected_paths: list[str],
event_name: str | None,
n_lines: int,
flares_updater.py
+2 -4
View File
@@ -14,15 +14,14 @@ import time
import shlex
import psutil
import shutil
import platform
import subprocess
from typing import Union
from pathlib import Path
from datetime import datetime
# External library imports
from src.shared.shareddata import APP_NAME, PLATFORM_NAME
PLATFORM_NAME = platform.system().lower()
APP_NAME = "flares"
if PLATFORM_NAME == 'darwin':
_log_path = os.path.join(os.path.dirname(sys.executable), f"../../../{APP_NAME}_updater.log")
@@ -189,7 +188,6 @@ def main():
update_folder = sys.argv[1]
main_exe = sys.argv[2]
# Interesting naming convention
main_exe_path = Path(main_exe).resolve()
app_dir = main_exe_path.parent
bundle_dir = main_exe_path.parents[2]
project_manager.py
+3 -3
View File
@@ -21,9 +21,9 @@ from PySide6.QtWidgets import QMessageBox, QVBoxLayout, QFileDialog, QLabel, QDi
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 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
pylance_progress
+4 -4
View File
@@ -1,7 +1,7 @@
src\analysis\participantfoldchannels.py 157
src\analysis\participantfoldchannels.py 158
src\shared\flaresbasewidget.py 1001+
src\window\updateevents.py 151
src\window\updateevents.py 83
flares.py 1001+
main_unit_tests.py 153
main.py 709
project_manager.py 407
main.py 705
project_manager.py 405
src/analysis/exporttocsv.py
+1 -2
View File
@@ -9,7 +9,6 @@ License: GPL-3.0
# Built-in imports
import os
from pathlib import Path
from typing import Any
# External library imports
@@ -27,7 +26,7 @@ from src.shared.shareddata import APP_NAME
class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
cha_dict: dict[str, DataFrame],
df_ind_dict: dict[str, DataFrame],
design_matrix_dict: dict[str, DataFrame],
src/analysis/intergroupbrainimage.py
+2 -3
View File
@@ -7,8 +7,7 @@ Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in Imports
from pathlib import Path
# Built-in imports
from typing import Any, cast
# External library imports
@@ -69,7 +68,7 @@ DESCRIPTION = """\n1. Group Contrast 2D/3D (plot_2d_3d_contrasts_between_groups)
class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
df_ind_dict: dict[str, DataFrame],
design_matrix_dict: dict[str, DataFrame],
contrast_results_dict: dict[str, dict[str, Any]],
src/analysis/intergroupstats.py
+1 -2
View File
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
from pathlib import Path
from typing import Any, cast
# External library imports
@@ -147,7 +146,7 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
cha_dict: dict[str, DataFrame],
df_ind_dict: dict[str, DataFrame],
design_matrix_dict: dict[str, DataFrame],
src/analysis/intragroupbrainimage.py
+2 -3
View File
@@ -7,8 +7,7 @@ Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in Imports
from pathlib import Path
# Built-in imports
from typing import Any, cast
# External library imports
@@ -79,7 +78,7 @@ DESCRIPTION = """0. FIR Model Results (plot_fir_model_results)
class IntraGroupBrainImageWidget(IntraGroupUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
cha_dict: dict[str, DataFrame],
df_ind_dict: dict[str, DataFrame],
design_matrix_dict: dict[str, DataFrame],
src/analysis/intragroupfunctionalconnectivity.py
+6 -3
View File
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
from pathlib import Path
from typing import Any, cast
# External library imports
@@ -61,7 +60,7 @@ DESCRIPTION = """0. Beta-Series Correlation (run_group_functional_connectivity_b
class IntraGroupFunctionalConnectivityWidget(IntraGroupUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
epochs_dict: dict[str, Epochs],
group_dict: dict[str, str],
) -> None:
@@ -99,7 +98,11 @@ class IntraGroupFunctionalConnectivityWidget(IntraGroupUIMixin, FlaresBaseWidget
min_participants = params.get("min_participants", 3)
run_group_functional_connectivity_betas(
self.haemo_dict, selected_file_paths, selected_event, n_lines, vmin,
self.haemo_dict,
selected_file_paths,
selected_event,
n_lines,
vmin,
drift_model=drift_model,
drift_order=drift_order,
hrf_model=hrf_model,
src/analysis/intragroupstats.py
+5 -4
View File
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
from pathlib import Path
from typing import Any, cast
# External library imports
@@ -159,7 +158,7 @@ DESCRIPTION = """0. ROI vs. Zero (run_roi_second_level_analysis)
class IntraGroupStatsWidget(IntraGroupUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
cha_dict: dict[str, DataFrame],
df_ind_dict: dict[str, DataFrame],
design_matrix_dict: dict[str, DataFrame],
@@ -240,6 +239,10 @@ class IntraGroupStatsWidget(IntraGroupUIMixin, FlaresBaseWidget):
if correction_method == "None":
correction_method = None
if not selected_event:
print("Warning: No event condition selected for ROI analysis.")
continue
if df_group.empty:
print("No ROI data (df_ind) found for selected participants.")
continue
@@ -265,7 +268,6 @@ class IntraGroupStatsWidget(IntraGroupUIMixin, FlaresBaseWidget):
else:
all_cha_filtered = all_cha
run_roi_second_level_analysis(
df_roi_all=df_filtered,
condition=selected_event,
@@ -303,7 +305,6 @@ class IntraGroupStatsWidget(IntraGroupUIMixin, FlaresBaseWidget):
print("Both ROI A and ROI B must be specified.")
continue
print(min_subjects)
run_roi_paired_contrast_analysis(
df_roi_all=df_group,
roi_pairs=(roi_a, roi_b),
src/analysis/participantbrain.py
+1 -2
View File
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
from pathlib import Path
from typing import Any, cast
# External library imports
@@ -69,7 +68,7 @@ PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = {
class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
cha_dict: dict[str, DataFrame],
) -> None:
src/analysis/participantfoldchannels.py
+1 -1
View File
@@ -748,7 +748,7 @@ class ProcessOrchestrator(QObject):
def __init__(self,
selected_files,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
worker_func
):
src/analysis/participantfunctionalconnectivity.py
+2 -3
View File
@@ -7,8 +7,7 @@ Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in Imports
from pathlib import Path
# Built-in imports
from typing import Any, cast
# External library imports
@@ -81,7 +80,7 @@ DESCRIPTION = """0. Spectral Coherence (functional_connectivity_spectral_epochs)
class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
epochs_dict: dict[str, Epochs],
) -> None:
src/analysis/participantimage.py
+1 -1
View File
@@ -7,7 +7,7 @@ Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in Imports
# Built-in imports
import os.path as op
from pathlib import Path
from datetime import datetime
src/window/terminal.py
+2 -2
View File
@@ -13,7 +13,7 @@ from pathlib import Path
from typing import Any, Callable
# External library imports
from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit
from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit, QMainWindow
from PySide6.QtCore import QProcess, Qt, QThread, Signal
from file_ext_registration import register_file_association, is_windows_admin
@@ -124,7 +124,7 @@ class TerminalWindow(QWidget):
def cmd_update(self, *args: Any) -> str:
main_win = self.parent()
if not isinstance(main_win, QWidget):
if not isinstance(main_win, QMainWindow):
return "[Error] Main window context not found."
self.updater = UpdateManager(
src/window/updateevents.py
+63 -68
View File
@@ -10,12 +10,9 @@ License: GPL-3.0
import os
import json
from enum import Enum, auto
from datetime import datetime
from typing import Optional
from typing import Any, List, Optional, cast
# External library imports
import numpy as np
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QLineEdit, QPushButton, QComboBox, QHBoxLayout, QMessageBox, QFileDialog
from PySide6.QtCore import Qt
@@ -255,16 +252,14 @@ class UpdateEventsWindow(QWidget):
except (json.JSONDecodeError, FileNotFoundError, KeyError) as e:
QMessageBox.warning(self, "Error", f"Failed to parse BORIS file:\n{e}")
def extract_boris_observation_keys(self, data):
def extract_boris_observation_keys(self, data: dict[str, Any]) -> List[str]:
if "observations" not in data:
raise KeyError("Missing 'observations' key in BORIS file.")
observations = data["observations"]
if not isinstance(observations, dict):
raise TypeError("'observations' must be a dictionary.")
observations = cast(dict[str, Any], data["observations"])
return list(observations.keys())
def on_observation_selected(self):
selected_obs = self.combo_suffix.currentText()
if not selected_obs or not hasattr(self, 'boris_data'):
@@ -454,7 +449,7 @@ class UpdateEventsWindow(QWidget):
save_path += ".json"
# Build JSON dict
json_data = {
json_data: dict[str, Any] = {
"observation": selected_obs,
"snirf_anchor": {"label": snirf_label, "time": snirf_anchor_time},
"boris_anchor": {"label": boris_label, "time": boris_anchor_time},
@@ -471,73 +466,73 @@ class UpdateEventsWindow(QWidget):
QMessageBox.critical(self, "Error", f"Failed to write JSON:\n{e}")
def update_optode_positions(self, file_a, file_b, save_path):
# def update_optode_positions(self, file_a, file_b, save_path):
fiducials = {}
ch_positions = {}
# fiducials = {}
# ch_positions = {}
# Read the lines from the optode file
with open(file_b, 'r') as f:
for line in f:
if line.strip():
# Split by the semicolon and convert to meters
ch_name, coords_str = line.split(":")
coords = np.array(list(map(float, coords_str.strip().split()))) * 0.001
# # Read the lines from the optode file
# with open(file_b, 'r') as f:
# for line in f:
# if line.strip():
# # Split by the semicolon and convert to meters
# ch_name, coords_str = line.split(":")
# coords = np.array(list(map(float, coords_str.strip().split()))) * 0.001
# The key we have is a fiducial
if ch_name.lower() in ['lpa', 'nz', 'rpa']:
fiducials[ch_name.lower()] = coords
# # The key we have is a fiducial
# if ch_name.lower() in ['lpa', 'nz', 'rpa']:
# fiducials[ch_name.lower()] = coords
# The key we have is a source or detector
else:
ch_positions[ch_name.upper()] = coords
# # The key we have is a source or detector
# else:
# ch_positions[ch_name.upper()] = coords
# Create montage with updated coords in head space
initial_montage = make_dig_montage(ch_pos=ch_positions, nasion=fiducials.get('nz'), lpa=fiducials.get('lpa'), rpa=fiducials.get('rpa'), coord_frame='head') # type: ignore
# # Create montage with updated coords in head space
# initial_montage = make_dig_montage(ch_pos=ch_positions, nasion=fiducials.get('nz'), lpa=fiducials.get('lpa'), rpa=fiducials.get('rpa'), coord_frame='head') # type: ignore
# Read the SNIRF file, set the montage, and write it back
# TODO: Bad! read_raw_snirf doesnt release memory properly! Should be spawned in a seperate process and killed once completed
raw = read_raw_snirf(file_a, preload=True)
raw.set_montage(initial_montage)
write_raw_snirf(raw, save_path)
# # Read the SNIRF file, set the montage, and write it back
# # TODO: Bad! read_raw_snirf doesnt release memory properly! Should be spawned in a seperate process and killed once completed
# raw = read_raw_snirf(file_a, preload=True)
# raw.set_montage(initial_montage)
# write_raw_snirf(raw, save_path)
def _apply_events_to_snirf(self, raw, new_annotations, save_path):
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
# def _apply_events_to_snirf(self, raw, new_annotations, save_path):
# raw.set_annotations(new_annotations)
# write_raw_snirf(raw, save_path)
def _write_event_mapping_json(
self,
file_a,
file_b,
selected_obs,
snirf_anchor,
boris_anchor,
time_shift,
mapped_events,
save_path
):
# def _write_event_mapping_json(
# self,
# file_a,
# file_b,
# selected_obs,
# snirf_anchor,
# boris_anchor,
# time_shift,
# mapped_events,
# save_path
# ):
payload = {
"source": {
"called_from": self.caller,
"snirf_file": os.path.basename(file_a),
"boris_file": os.path.basename(file_b),
"observation": selected_obs
},
"alignment": {
"snirf_anchor": snirf_anchor,
"boris_anchor": boris_anchor,
"time_shift_seconds": time_shift
},
"events": mapped_events,
"created_at": datetime.utcnow().isoformat() + "Z"
}
# payload = {
# "source": {
# "called_from": self.caller,
# "snirf_file": os.path.basename(file_a),
# "boris_file": os.path.basename(file_b),
# "observation": selected_obs
# },
# "alignment": {
# "snirf_anchor": snirf_anchor,
# "boris_anchor": boris_anchor,
# "time_shift_seconds": time_shift
# },
# "events": mapped_events,
# "created_at": datetime.utcnow().isoformat() + "Z"
# }
with open(save_path, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
# with open(save_path, "w", encoding="utf-8") as f:
# json.dump(payload, f, indent=2)
return save_path
# return save_path
@@ -747,11 +742,11 @@ class UpdateEventsBlazesWindow(QWidget):
self.combo_events.setEnabled(False)
def extract_json_observation_strings(self, data):
def extract_json_observation_strings(self, data: dict[str, Any]) -> List[str]:
if "events" not in data:
raise KeyError("Missing 'events' key in JSON file.")
event_strings = []
event_strings: List[str] = []
# The new format is a flat list chronologically ordered
for event in data["events"]:
@@ -772,7 +767,7 @@ class UpdateEventsBlazesWindow(QWidget):
def go_action(self) -> None:
file_a = self.line_edit_file_a.text()
file_b = self.line_edit_file_b.text()
_ = self.line_edit_file_b.text()
suffix = APP_NAME
if not hasattr(self, "json_data") or self.combo_events.count() == 0 or self.combo_snirf_events.count() == 0:
src/window/viewerlauncher.py
+1 -2
View File
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
from pathlib import Path
from typing import Any, Callable, Type
# External library imports
@@ -36,7 +35,7 @@ from src.shared.shareddata import APP_NAME
class ViewerLauncherWidget(QWidget):
def __init__(
self,
haemo_dict: dict[str | Path, BaseRaw],
haemo_dict: dict[str, BaseRaw],
epochs_dict: dict[str, Epochs],
cha_dict: dict[str, DataFrame],
df_ind_dict: dict[str, DataFrame],