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

start of plugin support?

Browse Source
This commit is contained in:
tyler
2026-09-01 02:14:47 -07:00
parent 815f342ead
commit ca203fcb56
9 changed files with 476 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/shared/shareddata.py
+1 -1
View File
@@ -13,7 +13,7 @@ import sys
import platform
CURRENT_VERSION = "1.7.0"
CURRENT_VERSION = "1.7.1"
APP_NAME = "flares"
APP_NAME_EXPANDED = "fNIRS Lightweight Analysis, Research, & Evaluation Suite"
API_URL = f"https://git.research.dezeeuw.ca/api/v1/repos/tyler/{APP_NAME}/releases"
src/window/plugins.py
+224
View File
@@ -0,0 +1,224 @@
"""
Filename: plugins.py
Description: Plugins window
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in imports
import json
import sys
import urllib.request
from pathlib import Path
from typing import Any, cast
# External library imports
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem, QPushButton, QTabWidget, QVBoxLayout, QWidget
from src.shared.shareddata import APP_NAME, PLATFORM_NAME, PLUGINS_URL
class PluginsWindow(QWidget):
"""
Plugins window containing two tabs: Installed Plugins and Plugin Browser.
Args:
parent (QWidget | None, optional): Parent widget of this window. Defaults to None.
"""
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent, Qt.WindowType.Window)
self.setWindowTitle(f"{APP_NAME.upper()} - Plugins")
self.resize(650, 500)
self.plugins_dir: Path = self._resolve_plugins_dir()
self.repository_urls: list[str] = [PLUGINS_URL]
self.remote_plugins_data: list[dict[str, Any]] = []
self._has_fetched_remote: bool = False
main_layout = QVBoxLayout(self)
self.tab_widget = QTabWidget(self)
self.installed_tab = self._create_installed_tab()
self.browser_tab = self._create_browser_tab()
self.tab_widget.addTab(self.installed_tab, "Installed Plugins")
self.tab_widget.addTab(self.browser_tab, "Plugin Browser")
self.tab_widget.setCurrentIndex(0)
self.tab_widget.currentChanged.connect(self._on_tab_changed)
main_layout.addWidget(self.tab_widget)
self.setLayout(main_layout)
self.refresh_installed_plugins()
def _resolve_plugins_dir(self) -> Path:
"""Determines the local plugins directory based on execution context."""
if PLATFORM_NAME == "darwin":
base_dir = Path(sys.executable).parent / "../../.."
else:
base_dir = Path.cwd()
return (base_dir / "plugins").resolve()
def _on_tab_changed(self, index: int) -> None:
"""Triggers remote fetch only when switching specifically to the Plugin Browser tab."""
if index == 1 and not self._has_fetched_remote:
self.fetch_remote_plugins()
def refresh_installed_plugins(self) -> None:
"""Scans local plugins directory silently without UI popups."""
self.installed_list.clear()
self.plugins_dir.mkdir(parents=True, exist_ok=True)
installed_plugins: list[str] = []
for entry in self.plugins_dir.iterdir():
if entry.is_dir() and not entry.name.startswith((".", "__")):
installed_plugins.append(entry.name)
elif entry.is_file() and entry.suffix == ".py" and entry.stem != "__init__":
installed_plugins.append(entry.stem)
if installed_plugins:
self.installed_list.addItems(sorted(installed_plugins))
else:
self.installed_list.addItem("No plugins installed.")
def add_custom_repository(self) -> None:
"""Adds a custom repository URL from the input field and refreshes the browser list."""
url = self.repo_input.text().strip()
if not url:
return
if not (url.startswith("http://") or url.startswith("https://")):
self._show_browser_error("Invalid URL format. Must start with http:// or https://")
return
if url not in self.repository_urls:
self.repository_urls.append(url)
self.repo_input.clear()
self.fetch_remote_plugins()
def fetch_remote_plugins(self) -> None:
"""
Fetches plugins.json from all configured repository URLs directly into memory.
Aggregates results and handles network failures gracefully in-line.
"""
self.browser_list.clear()
self.browser_list.addItem("Fetching remote repositories...")
aggregated_plugins: list[dict[str, Any]] = []
failed_count: int = 0
for url in self.repository_urls:
try:
req = urllib.request.Request(
url,
headers={"User-Agent": f"{APP_NAME}-PluginManager"},
)
with urllib.request.urlopen(req, timeout=5) as response:
if response.status == 200:
raw_data = response.read().decode("utf-8")
plugins = json.loads(raw_data)
if isinstance(plugins, list):
aggregated_plugins.extend(cast(list[Any], plugins))
else:
failed_count += 1
except Exception:
failed_count += 1
self.remote_plugins_data = aggregated_plugins
self._has_fetched_remote = True
if aggregated_plugins:
self._populate_browser_list()
elif failed_count == len(self.repository_urls):
self._show_browser_error(
"Unable to load plugins from any repository. Check your connection or URLs."
)
else:
self._show_browser_error("No plugins found across configured repositories.")
def _show_browser_error(self, message: str) -> None:
"""Renders error text directly in the list widget."""
self.browser_list.clear()
self.browser_list.addItem(message)
def _populate_browser_list(self) -> None:
"""Populates list widget with parsed remote plugin data."""
self.browser_list.clear()
for plugin in self.remote_plugins_data:
name = plugin.get("name", "Unknown Plugin")
version = plugin.get("version", "v0.0")
desc = plugin.get("description", "No description provided.")
display_text = f"{name} ({version}) - {desc}"
item = QListWidgetItem(display_text, self.browser_list)
item.setData(Qt.ItemDataRole.UserRole, plugin)
def _create_installed_tab(self) -> QWidget:
"""Creates 'Installed Plugins' tab UI."""
tab = QWidget()
layout = QVBoxLayout(tab)
path_label = QLabel(f"Directory: {self.plugins_dir}", tab)
self.installed_list = QListWidget(tab)
button_layout = QHBoxLayout()
self.btn_refresh_local = QPushButton("Refresh List", tab)
self.btn_enable_disable = QPushButton("Enable/Disable", tab)
self.btn_uninstall = QPushButton("Uninstall", tab)
self.btn_refresh_local.clicked.connect(self.refresh_installed_plugins)
button_layout.addWidget(self.btn_refresh_local)
button_layout.addStretch()
button_layout.addWidget(self.btn_enable_disable)
button_layout.addWidget(self.btn_uninstall)
layout.addWidget(path_label)
layout.addWidget(self.installed_list)
layout.addLayout(button_layout)
return tab
def _create_browser_tab(self) -> QWidget:
"""Creates 'Plugin Browser' tab UI."""
tab = QWidget()
layout = QVBoxLayout(tab)
# Custom Repository Input Controls
repo_layout = QHBoxLayout()
self.repo_input = QLineEdit(tab)
self.repo_input.setPlaceholderText("Enter custom plugins.json URL...")
self.btn_add_repo = QPushButton("Add Repo", tab)
self.btn_add_repo.clicked.connect(self.add_custom_repository)
repo_layout.addWidget(self.repo_input)
repo_layout.addWidget(self.btn_add_repo)
label = QLabel("Available Plugins from Repositories:", tab)
self.browser_list = QListWidget(tab)
button_layout = QHBoxLayout()
self.btn_fetch_remote = QPushButton("Fetch Remote Lists", tab)
self.btn_install = QPushButton("Install Plugin", tab)
self.btn_fetch_remote.clicked.connect(self.fetch_remote_plugins)
button_layout.addWidget(self.btn_fetch_remote)
button_layout.addStretch()
button_layout.addWidget(self.btn_install)
layout.addLayout(repo_layout)
layout.addWidget(label)
layout.addWidget(self.browser_list)
layout.addLayout(button_layout)
return tab
src/window/updateevents.py
+201 -170
View File
@@ -1,6 +1,7 @@
"""
Filename: updateevents.py
Description: Methods to update snirf events for FLARES
Description: Methods to update snirf events
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
License: GPL-3.0
@@ -9,27 +10,184 @@ License: GPL-3.0
# Built-in imports
import os
import json
import concurrent.futures
from enum import Enum, auto
from typing import Any, List, Optional, cast
from typing import Any, List, Optional, Sequence, Union, cast
# External library imports
from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QLineEdit, QPushButton, QComboBox, QHBoxLayout, QMessageBox, QFileDialog
from PySide6.QtCore import Qt
from mne import Annotations
from mne.io import read_raw_snirf #type: ignore
from mne_nirs.io import write_raw_snirf #type: ignore
from mne.io import read_raw_snirf #type: ignore
from mne_nirs.io import write_raw_snirf #type: ignore
from mne.io.base import BaseRaw
from src.shared.shareddata import APP_NAME
def _load_annotations_worker(file_path: str) -> Annotations:
"""Isolated worker process to load SNIRF annotations without leaving HDF5 memory leaks in main process."""
raw: BaseRaw = read_raw_snirf(file_path, preload=False, verbose=False)
return cast(Annotations, getattr(raw, "annotations"))
def load_snirf_annotations_isolated(file_path: str) -> Annotations:
"""Spawns a short-lived process to extract annotations and cleanly releases C-memory allocations."""
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
future = executor.submit(_load_annotations_worker, file_path)
return future.result()
def _write_snirf_worker(
file_path: str,
save_path: str,
onsets: list[float],
durations: list[float],
descriptions: list[str]
) -> None:
"""Isolated process worker to read, annotate, write, and immediately free memory/file handles."""
raw = read_raw_snirf(file_path, preload=True, verbose=False)
new_annotations = Annotations(onset=onsets, duration=durations, description=descriptions)
raw.set_annotations(new_annotations) #type: ignore
write_raw_snirf(raw, save_path)
def write_snirf_with_annotations_isolated(
file_path: str,
save_path: str,
onsets: list[float],
durations: list[float],
descriptions: list[str]
) -> None:
"""Executes SNIRF writing in a dedicated worker process to ensure 100% memory/handle cleanup."""
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
future = executor.submit(
_write_snirf_worker,
file_path,
save_path,
onsets,
durations,
descriptions
)
future.result()
def _align_boris_and_write_snirf_worker(
file_path: str,
save_path: str,
boris_events: list[Any],
time_shift: float
) -> int:
"""
Isolated process worker to read SNIRF, align BORIS events, write updated file,
and cleanly exit releasing all HDF5 C-memory allocations and file locks.
"""
raw = read_raw_snirf(file_path, preload=True, verbose=False)
# Type-safe extraction of sampling frequency from raw.info
info_dict = cast(dict[str, Any], raw.info)
sfreq = float(info_dict.get("sfreq", 10.0))
min_shift = 1.0 / sfreq if sfreq > 0 else 0.1
max_attempts = 10
onsets: list[float] = []
durations: list[float] = []
descriptions: list[str] = []
open_events: dict[str, list[float]] = {}
label_counts: dict[str, int] = {}
used_times: set[float] = set()
for raw_event in boris_events:
if not isinstance(raw_event, (list, tuple)):
continue
event = cast(Sequence[Any], raw_event)
if len(event) < 3:
continue
event_time = float(event[0])
label = str(event[2])
count = label_counts.get(label, 0) + 1
label_counts[label] = count
if label not in open_events:
open_events[label] = []
if count % 2 == 1:
open_events[label].append(event_time)
else:
if open_events[label]:
start_time = open_events[label].pop(0)
duration = event_time - start_time
if duration <= 0:
continue
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
continue
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
onsets.append(adjusted_time)
durations.append(round(duration, 6))
descriptions.append(label)
# Handle unmatched start markers
for label, starts in open_events.items():
for start_time in starts:
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
continue
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
onsets.append(adjusted_time)
durations.append(0.0)
descriptions.append(label)
new_annotations = Annotations(onset=onsets, duration=durations, description=descriptions)
raw.set_annotations(new_annotations) #type: ignore
write_raw_snirf(raw, save_path)
return len(onsets)
def align_boris_and_write_snirf_isolated(
file_path: str,
save_path: str,
boris_events: list[Any],
time_shift: float
) -> int:
"""Executes BORIS alignment and SNIRF writing in a dedicated worker process."""
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
future = executor.submit(
_align_boris_and_write_snirf_worker,
file_path,
save_path,
boris_events,
time_shift
)
return future.result()
class EventUpdateMode(Enum):
WRITE_SNIRF = auto() # destructive
WRITE_JSON = auto() # non-destructive
class UpdateEventsWindow(QWidget):
def __init__(self, parent: Optional[QWidget]=None, mode=EventUpdateMode.WRITE_SNIRF, caller=None):
def __init__(self, parent: Optional[QWidget]=None, mode: EventUpdateMode=EventUpdateMode.WRITE_SNIRF, caller: Optional[Union[str, object]] = None,):
super().__init__(parent, Qt.WindowType.Window)
self.mode = mode
@@ -210,12 +368,13 @@ class UpdateEventsWindow(QWidget):
if file_path:
self.line_edit_file_a.setText(file_path)
try:
# 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_path, preload=False)
annotations = raw.annotations
# Memory leak safe: Extracted in a separate process that terminates immediately
annotations: Annotations = load_snirf_annotations_isolated(file_path)
print(f"Loaded {len(annotations)} annotations from {file_path}")
# Build individual event entries
event_entries = []
event_entries: list[str] = []
for onset, description in zip(annotations.onset, annotations.description):
event_str = f"{description} @ {onset:.3f}s"
event_entries.append(event_str)
@@ -251,6 +410,7 @@ 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: dict[str, Any]) -> List[str]:
if "observations" not in data:
@@ -274,11 +434,16 @@ class UpdateEventsWindow(QWidget):
self.combo_events.setEnabled(False)
return
event_entries = []
for event in events:
if isinstance(event, list) and len(event) >= 3:
timestamp = event[0]
label = event[2]
event_entries: list[str] = []
for raw_event in events:
if not isinstance(raw_event, (list, tuple)):
continue
event = cast(Sequence[Any], raw_event)
if len(event) >= 3:
timestamp = float(event[0])
label = str(event[2])
display = f"{label} @ {timestamp:.3f}"
event_entries.append(display)
@@ -363,71 +528,15 @@ class UpdateEventsWindow(QWidget):
save_path += ".snirf"
try:
raw = read_raw_snirf(file_a, preload=True)
# Memory leak safe: Worker handles alignment calculation, read, write, and process termination
event_count = align_boris_and_write_snirf_isolated(
file_a,
save_path,
boris_events,
time_shift
)
# --- Align BORIS events to SNIRF ---
boris_events = boris_obs.get("events", [])
onsets, durations, descriptions = [], [], []
open_events = {} # label -> list of start times
label_counts = {}
used_times = set()
sfreq = raw.info['sfreq']
min_shift = 1.0 / sfreq
max_attempts = 10
for event in boris_events:
if not isinstance(event, list) or len(event) < 3:
continue
event_time = event[0]
label = event[2]
count = label_counts.get(label, 0) + 1
label_counts[label] = count
if label not in open_events:
open_events[label] = []
if count % 2 == 1:
open_events[label].append(event_time)
else:
if open_events[label]:
start_time = open_events[label].pop(0)
duration = event_time - start_time
if duration <= 0:
continue
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
continue
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
onsets.append(adjusted_time)
durations.append(duration)
descriptions.append(label)
# Handle unmatched starts
for label, starts in open_events.items():
for start_time in starts:
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
continue
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
onsets.append(adjusted_time)
durations.append(0.0)
descriptions.append(label)
new_annotations = Annotations(onset=onsets, duration=durations, description=descriptions)
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
QMessageBox.information(self, "Success", "SNIRF file updated with aligned BORIS events.")
QMessageBox.information(self, "Success", f"SNIRF file updated with {event_count} aligned BORIS events.")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to update SNIRF file:\n{e}")
@@ -466,79 +575,9 @@ 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):
# 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
# # 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
# # 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)
# 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
# ):
# 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)
# return save_path
class UpdateEventsBlazesWindow(QWidget):
def __init__(self, parent: Optional[QWidget]=None, mode=EventUpdateMode.WRITE_SNIRF, caller=None):
def __init__(self, parent: Optional[QWidget]=None, mode: EventUpdateMode=EventUpdateMode.WRITE_SNIRF, caller: Optional[Union[str, object]] = None,):
super().__init__(parent, Qt.WindowType.Window)
self.mode = mode
@@ -692,12 +731,13 @@ class UpdateEventsBlazesWindow(QWidget):
if file_path:
self.line_edit_file_a.setText(file_path)
try:
# 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_path, preload=False)
annotations = raw.annotations
# Memory leak safe: Extracted in a separate process that terminates immediately
annotations: Annotations = load_snirf_annotations_isolated(file_path)
print(f"Loaded {len(annotations)} annotations from {file_path}")
# Build individual event entries
event_entries = []
event_entries: list[str] = []
for onset, description in zip(annotations.onset, annotations.description):
event_str = f"{description} @ {onset:.3f}s"
event_entries.append(event_str)
@@ -748,12 +788,11 @@ class UpdateEventsBlazesWindow(QWidget):
event_strings: List[str] = []
# The new format is a flat list chronologically ordered
# Flat list chronologically ordered
for event in data["events"]:
track_name = event.get("track_name", "Unknown")
onset = event.get("start_sec", 0.0)
# Formatting to match your SNIRF style: "Event Name @ 0.000s"
display_str = f"{track_name} @ {onset:.3f}s"
event_strings.append(display_str)
@@ -792,7 +831,9 @@ class UpdateEventsBlazesWindow(QWidget):
time_shift = snirf_anchor_time - json_anchor_time
onsets, durations, descriptions = [], [], []
onsets: list[float] = []
durations: list[float] = []
descriptions: list[str] = []
skipped_count = 0
try:
@@ -806,7 +847,7 @@ class UpdateEventsBlazesWindow(QWidget):
original_end = event.get("end_sec", original_start)
duration = original_end - original_start
# FILTER: Minimum 0.1s duration
# Minimum 0.1s duration
if duration < 0.1:
skipped_count += 1
continue
@@ -834,19 +875,9 @@ class UpdateEventsBlazesWindow(QWidget):
if not save_path.lower().endswith(".snirf"): save_path += ".snirf"
try:
raw = read_raw_snirf(file_a, preload=True)
# Create annotations
new_annotations = Annotations(
onset=onsets,
duration=durations,
description=descriptions
)
# Replace existing annotations with the new aligned JSON tracks
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
# Memory leak safe: Worker handles read, write, and complete memory reclamation upon exit
write_snirf_with_annotations_isolated(file_a, save_path, onsets, durations, descriptions)
QMessageBox.information(self, "Success",
f"Aligned {len(onsets)} events.\n(Filtered out {skipped_count} short events)")
except Exception as e:
src/window/updateoptodes.py
+1 -2
View File
@@ -1,6 +1,6 @@
"""
Filename: updateoptodes.py
Description: Methods to update optode locations for FLARES
Description: Methods to update optode locations
Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw
@@ -8,7 +8,6 @@ License: GPL-3.0
"""
# Built-in imports
import os
from pathlib import Path
from typing import Dict, Optional, Union