"""
Filename: plugins.py
Description: roi-builder-pack plugin
Author: Tyler de Zeeuw
License: GPL-3.0
"""
# Built-in imports
import os
import json
from typing import Optional, Any
# External library imports
import h5py
from PySide6.QtCore import Qt
from PySide6.QtGui import QAction
from PySide6.QtWidgets import (QAbstractItemView, QComboBox, QFileDialog, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget,
QMenu, QMessageBox, QPushButton, QSplitter, QTreeWidget, QTreeWidgetItem, QVBoxLayout, QWidget,)
from mne.io import read_raw_snirf
from mne.preprocessing.nirs import optical_density, beer_lambert_law
from flares import fold_channels
class RoiBuilderWidget(QWidget):
"""UI Widget for opening SNIRF files, grouping channels into ROIs, and exporting JSON files."""
NEW_GROUP_OPTION = "+ Create New Group"
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setWindowTitle("ROI Channel Builder")
self.resize(850, 620)
self.extracted_channels: list[str] = []
self._setup_ui()
def _setup_ui(self) -> None:
main_layout = QVBoxLayout(self)
file_box = QGroupBox("SNIRF File Selection", self)
file_layout = QHBoxLayout(file_box)
self.txt_file_path = QLineEdit(self)
self.txt_file_path.setReadOnly(True)
self.txt_file_path.setPlaceholderText("Select a .snirf file to load channels...")
btn_browse = QPushButton("Browse...", self)
btn_browse.clicked.connect(self._on_browse_snirf)
file_layout.addWidget(self.txt_file_path)
file_layout.addWidget(btn_browse)
main_layout.addWidget(file_box)
splitter = QSplitter(Qt.Orientation.Horizontal, self)
# Left Column: Extracted Channels
left_widget = QWidget(self)
left_layout = QVBoxLayout(left_widget)
left_layout.setContentsMargins(0, 0, 0, 0)
left_layout.addWidget(QLabel("Available Channels:", self))
self.list_channels = QListWidget(self)
self.list_channels.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
left_layout.addWidget(self.list_channels)
# Draft Auto-Grouping Button
self.btn_auto_group = QPushButton("Auto ROI Grouping", self)
self.btn_auto_group.setToolTip("Automatically splits available channels via their physical locations.")
self.btn_auto_group.clicked.connect(self._on_auto_group)
left_layout.addWidget(self.btn_auto_group)
splitter.addWidget(left_widget)
# Right Column: Defined ROIs & Inputs
right_widget = QWidget(self)
right_layout = QVBoxLayout(right_widget)
right_layout.setContentsMargins(0, 0, 0, 0)
# Form for selecting or adding ROI
form_box = QGroupBox("Assign Channels to Group", self)
form_layout = QVBoxLayout(form_box)
self.combo_group_select = QComboBox(self)
self.combo_group_select.addItem(self.NEW_GROUP_OPTION)
self.combo_group_select.currentIndexChanged.connect(self._on_group_selection_changed)
self.txt_roi_name = QLineEdit(self)
self.txt_roi_name.setPlaceholderText("ROI Name (e.g., Left_PFC)")
self.txt_roi_desc = QLineEdit(self)
self.txt_roi_desc.setPlaceholderText("Description (e.g., Left prefrontal cortex)")
btn_add_roi = QPushButton("Assign Selected Channels to Group", self)
btn_add_roi.setStyleSheet("font-weight: bold;")
btn_add_roi.clicked.connect(self._on_assign_channels)
form_layout.addWidget(QLabel("Target Group:"))
form_layout.addWidget(self.combo_group_select)
form_layout.addWidget(self.txt_roi_name)
form_layout.addWidget(self.txt_roi_desc)
form_layout.addWidget(btn_add_roi)
right_layout.addWidget(form_box)
# ROI Tree View
right_layout.addWidget(QLabel("Defined Regions of Interest:", self))
self.tree_rois = QTreeWidget(self)
self.tree_rois.setHeaderLabels(["ROI / Channel", "Description"])
self.tree_rois.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
self.tree_rois.header().setSectionResizeMode(0, QHeaderView.ResizeMode.ResizeToContents)
right_layout.addWidget(self.tree_rois)
# Unassign & Remove Buttons
btn_layout = QHBoxLayout()
btn_unassign_channel = QPushButton("Unassign Selected Channel(s)", self)
btn_unassign_channel.clicked.connect(self._on_unassign_channels)
btn_remove_roi = QPushButton("Remove Selected Group", self)
btn_remove_roi.clicked.connect(self._on_remove_roi)
btn_layout.addWidget(btn_unassign_channel)
btn_layout.addWidget(btn_remove_roi)
right_layout.addLayout(btn_layout)
splitter.addWidget(right_widget)
splitter.setSizes([280, 570])
main_layout.addWidget(splitter)
bottom_layout = QHBoxLayout()
self.lbl_status = QLabel("Ready.", self)
btn_export = QPushButton("Export ROI JSON...", self)
btn_export.setStyleSheet("font-weight: bold;")
btn_export.clicked.connect(self._on_export_json)
bottom_layout.addWidget(self.lbl_status)
bottom_layout.addStretch()
bottom_layout.addWidget(btn_export)
main_layout.addLayout(bottom_layout)
def _add_channels_to_available_list(self, channels: list[str]) -> None:
"""Restores channels to the available list and keeps them sorted."""
existing = [self.list_channels.item(i).text() for i in range(self.list_channels.count())]
combined = set(existing).union(channels)
sorted_channels = sorted(list(combined), key=parse_sd_key)
self.list_channels.clear()
for ch in sorted_channels:
self.list_channels.addItem(ch)
def _find_roi_item_by_name(self, name: str) -> QTreeWidgetItem | None:
"""Finds an existing top-level tree item matching the group name."""
for i in range(self.tree_rois.topLevelItemCount()):
item = self.tree_rois.topLevelItem(i)
if item:
data = item.data(0, Qt.ItemDataRole.UserRole)
if data and data.get("name", "").strip().lower() == name.strip().lower():
return item
return None
def _on_group_selection_changed(self, index: int) -> None:
selected_text = self.combo_group_select.currentText()
if selected_text == self.NEW_GROUP_OPTION:
self.txt_roi_name.clear()
self.txt_roi_desc.clear()
self.txt_roi_name.setReadOnly(False)
else:
roi_item = self._find_roi_item_by_name(selected_text)
if roi_item:
data = roi_item.data(0, Qt.ItemDataRole.UserRole)
self.txt_roi_name.setText(data.get("name", ""))
self.txt_roi_desc.setText(data.get("description", ""))
self.txt_roi_name.setReadOnly(True)
def _on_browse_snirf(self) -> None:
file_path, _ = QFileDialog.getOpenFileName(
self, "Select SNIRF File", "", "SNIRF Files (*.snirf *.h5);;All Files (*)"
)
if not file_path:
return
self.txt_file_path.setText(file_path)
try:
self.extracted_channels = extract_snirf_channels(file_path)
self.tree_rois.clear()
self.list_channels.clear()
# Reset combo box
self.combo_group_select.blockSignals(True)
self.combo_group_select.clear()
self.combo_group_select.addItem(self.NEW_GROUP_OPTION)
self.combo_group_select.blockSignals(False)
for ch in self.extracted_channels:
self.list_channels.addItem(ch)
self.lbl_status.setText(f"Loaded {len(self.extracted_channels)} channels from file.")
except Exception as e:
QMessageBox.critical(self, "Error Reading SNIRF", f"Failed to load channels:\n{e}")
self.lbl_status.setText("Failed to load SNIRF file.")
def _on_assign_channels(self) -> None:
name = self.txt_roi_name.text().strip()
desc = self.txt_roi_desc.text().strip()
selected_items = self.list_channels.selectedItems()
if not name:
QMessageBox.warning(self, "Input Required", "Please specify a Group / ROI Name.")
return
if not selected_items:
QMessageBox.warning(self, "Selection Required", "Please select at least one channel from the left panel.")
return
new_channels = [item.text() for item in selected_items]
existing_item = self._find_roi_item_by_name(name)
if existing_item:
# Merge channels into existing group
data = existing_item.data(0, Qt.ItemDataRole.UserRole)
combined_channels = set(data.get("channels", [])).union(new_channels)
sorted_channels = sorted(list(combined_channels), key=parse_sd_key)
data["description"] = desc
data["channels"] = sorted_channels
existing_item.setText(1, desc)
existing_item.setData(0, Qt.ItemDataRole.UserRole, data)
# Rebuild child channel items
existing_item.takeChildren()
for ch in sorted_channels:
child = QTreeWidgetItem(existing_item)
child.setText(0, ch)
self.lbl_status.setText(f"Updated group '{name}' with {len(new_channels)} additional channels.")
else:
# Create new group tree item
roi_item = QTreeWidgetItem(self.tree_rois)
roi_item.setText(0, name)
roi_item.setText(1, desc)
sorted_channels = sorted(new_channels, key=parse_sd_key)
roi_data = {"name": name, "description": desc, "channels": sorted_channels}
roi_item.setData(0, Qt.ItemDataRole.UserRole, roi_data)
for ch in sorted_channels:
child = QTreeWidgetItem(roi_item)
child.setText(0, ch)
self.tree_rois.expandItem(roi_item)
self.combo_group_select.addItem(name)
self.combo_group_select.setCurrentText(name)
self.lbl_status.setText(f"Created group '{name}' with {len(new_channels)} channels.")
# REMOVE assigned channels from available channels list
for item in selected_items:
self.list_channels.takeItem(self.list_channels.row(item))
def _on_unassign_channels(self) -> None:
"""Unassigns individually selected channels from tree items and returns them to available list."""
selected_tree_items = self.tree_rois.selectedItems()
if not selected_tree_items:
QMessageBox.information(
self, "Selection Required", "Please select channel item(s) inside an ROI group to unassign."
)
return
channels_to_restore: list[str] = []
for item in selected_tree_items:
parent = item.parent()
# Check if this item is a child channel node
if parent is not None:
ch_name = item.text(0)
channels_to_restore.append(ch_name)
# Remove child from tree
parent.removeChild(item)
# Update parent's channel dataset
data = parent.data(0, Qt.ItemDataRole.UserRole)
if data and "channels" in data and ch_name in data["channels"]:
data["channels"].remove(ch_name)
parent.setData(0, Qt.ItemDataRole.UserRole, data)
if channels_to_restore:
self._add_channels_to_available_list(channels_to_restore)
self.lbl_status.setText(f"Unassigned {len(channels_to_restore)} channels back to available list.")
else:
QMessageBox.information(
self, "Select Channels", "Please select specific channel items under a group (not the group header)."
)
def _on_remove_roi(self) -> None:
"""Removes an entire ROI group and returns all its channels back to available list."""
selected_item = self.tree_rois.currentItem()
if not selected_item:
return
root_item = selected_item.parent() if selected_item.parent() else selected_item
index = self.tree_rois.indexOfTopLevelItem(root_item)
if index != -1:
data = root_item.data(0, Qt.ItemDataRole.UserRole)
name = data.get("name", "") if data else ""
channels = data.get("channels", []) if data else []
# Return channels to available list
if channels:
self._add_channels_to_available_list(channels)
self.tree_rois.takeTopLevelItem(index)
# Remove from combo box
cb_idx = self.combo_group_select.findText(name)
if cb_idx != -1:
self.combo_group_select.removeItem(cb_idx)
self.combo_group_select.setCurrentIndex(0)
self.lbl_status.setText(f"Removed group '{name}' and restored {len(channels)} channels.")
def _on_export_json(self) -> None:
roi_list: list[dict[str, Any]] = []
for i in range(self.tree_rois.topLevelItemCount()):
item = self.tree_rois.topLevelItem(i)
if item:
data = item.data(0, Qt.ItemDataRole.UserRole)
if data and data.get("channels"):
roi_list.append(data)
if not roi_list:
QMessageBox.warning(self, "No ROIs Defined", "Please create at least one non-empty group before exporting.")
return
save_path, _ = QFileDialog.getSaveFileName(
self, "Save Regions of Interest JSON", "regions_of_interest.json", "JSON Files (*.json)"
)
if not save_path:
return
output_data = {"regions_of_interest": roi_list}
try:
with open(save_path, "w", encoding="utf-8") as f:
json.dump(output_data, f, indent=2)
QMessageBox.information(self, "Export Successful", f"Saved ROI definitions to:\n{save_path}")
self.lbl_status.setText(f"Successfully exported {len(roi_list)} groups.")
except Exception as e:
QMessageBox.critical(self, "Export Failed", f"Could not save JSON file:\n{e}")
def _on_auto_group(self) -> None:
"""Automatically parses fOLD results and populates the tree and combo UI components."""
file_path = self.txt_file_path.text().strip()
if not file_path:
QMessageBox.warning(self, "No File Selected", "Please select a .snirf file first.")
return
self.btn_auto_group.setEnabled(False)
try:
# 1. Fetch fold specificity data
channel_results = get_channel_fold_results(
file_path=file_path,
atlas="Brodmann"
)
# 2. Determine groups automatically
auto_groups = _auto_group_channels(channel_results)
if not auto_groups:
QMessageBox.information(self, "Auto Grouping", "No valid ROI groups could be determined.")
return
# Map available channel strings to QListWidgetItems
available_items = {
self.list_channels.item(i).text(): self.list_channels.item(i)
for i in range(self.list_channels.count())
}
assigned_count = 0
groups_created = 0
for group_name, ch_list in auto_groups.items():
clean_channels: list[str] = []
items_to_remove: list = []
# Clean channel names (e.g., 'S1_D1 hbo' -> 'S1_D1') and match against available list
for raw_ch in ch_list:
ch_id = raw_ch.split()[0]
if ch_id in available_items:
clean_channels.append(ch_id)
items_to_remove.append(available_items[ch_id])
if not clean_channels:
continue
sorted_channels = sorted(list(set(clean_channels)), key=parse_sd_key)
# Check if group tree item exists or create new
existing_item = self._find_roi_item_by_name(group_name)
if existing_item:
data = existing_item.data(0, Qt.ItemDataRole.UserRole) or {}
combined = set(data.get("channels", [])).union(sorted_channels)
final_channels = sorted(list(combined), key=parse_sd_key)
data["name"] = group_name
data["description"] = group_name
data["channels"] = final_channels
existing_item.setData(0, Qt.ItemDataRole.UserRole, data)
existing_item.takeChildren()
for ch in final_channels:
child = QTreeWidgetItem(existing_item)
child.setText(0, ch)
else:
roi_item = QTreeWidgetItem(self.tree_rois)
roi_item.setText(0, group_name)
roi_item.setText(1, group_name)
roi_data = {
"name": group_name,
"description": group_name,
"channels": sorted_channels
}
roi_item.setData(0, Qt.ItemDataRole.UserRole, roi_data)
for ch in sorted_channels:
child = QTreeWidgetItem(roi_item)
child.setText(0, ch)
self.tree_rois.expandItem(roi_item)
if self.combo_group_select.findText(group_name) == -1:
self.combo_group_select.addItem(group_name)
groups_created += 1
# Remove assigned channels from available QListWidget
for item in items_to_remove:
row = self.list_channels.row(item)
if row != -1:
self.list_channels.takeItem(row)
ch_key = item.text()
if ch_key in available_items:
del available_items[ch_key]
assigned_count += len(clean_channels)
self.lbl_status.setText(f"Auto-grouped {assigned_count} channels into {groups_created} ROI regions.")
QMessageBox.information(
self,
"Auto Grouping Complete",
f"Successfully created {groups_created} ROI groups and assigned {assigned_count} channels."
)
except Exception as e:
QMessageBox.critical(self, "Fold Error", f"Failed to calculate fold channels:\n{e}")
finally:
self.btn_auto_group.setEnabled(True)
class Plugin:
"""Plugin entry point contract loaded by PluginManager."""
def __init__(self, main_window: QWidget) -> None:
self.main_window = main_window
self.name = "ROI Channel Builder"
self.widget_instance: RoiBuilderWidget | None = None
def register_menu(self, plugin_menu: QMenu) -> None:
"""Registers plugin options into the application's top menubar."""
open_action = QAction("Open ROI Builder Tool", self.main_window)
open_action.triggered.connect(self.show_widget)
about_action = QAction("About ROI Builder", self.main_window)
about_action.triggered.connect(self.show_about)
plugin_menu.addAction(open_action)
plugin_menu.addAction(about_action)
def show_widget(self) -> None:
"""Instantiates or focuses the ROI Builder window."""
if self.widget_instance is None or not self.widget_instance.isVisible():
self.widget_instance = RoiBuilderWidget()
self.widget_instance.show()
else:
self.widget_instance.raise_()
self.widget_instance.activateWindow()
def show_about(self) -> None:
"""Displays plugin information."""
QMessageBox.about(
self.main_window,
"About ROI Channel Builder",
"This plugin loads SNIRF binary files using h5py, extracts source-detector channel pairs, "
"and exports custom ROI channel group JSON files.",
)
def parse_sd_key(ch: str) -> tuple[int, int]:
"""Helper for natural sorting of S{source}_D{detector} strings."""
parts = ch.split("_")
s_num = int(parts[0].lstrip("S")) if len(parts) > 0 and parts[0].lstrip("S").isdigit() else 0
d_num = int(parts[1].lstrip("D")) if len(parts) > 1 and parts[1].lstrip("D").isdigit() else 0
return s_num, d_num
def extract_snirf_channels(file_path: str) -> list[str]:
"""Reads a .snirf file using h5py and extracts unique channel names (S{source}_D{detector})."""
channels: set[str] = set()
with h5py.File(file_path, "r") as h5_file:
if "nirs" not in h5_file:
raise ValueError("Invalid SNIRF file: missing top-level '/nirs' HDF5 group.")
nirs_group = h5_file["nirs"]
data_keys = [k for k in nirs_group.keys() if k.startswith("data")]
for d_key in data_keys:
data_group = nirs_group[d_key]
ml_keys = [k for k in data_group.keys() if k.startswith("measurementList")]
for ml_key in ml_keys:
ml = data_group[ml_key]
if "sourceIndex" in ml and "detectorIndex" in ml:
src = ml["sourceIndex"][()]
det = ml["detectorIndex"][()]
if hasattr(src, "item"):
src = src.item()
if hasattr(det, "item"):
det = det.item()
channels.add(f"S{src}_D{det}")
return sorted(list(channels), key=parse_sd_key)
def get_channel_fold_results(
file_path: str,
atlas: str = 'Brodmann',
progress_queue: Optional[Any] = None
) -> dict[str, list[dict[str, Any]]]:
"""
Directly loads a .snirf file, performs a lightweight HbO conversion,
and runs fold_channels without executing any GLM, QC, or pipeline steps.
"""
p_name = os.path.basename(file_path)
raw = read_raw_snirf(file_path, preload=True, verbose=False)
raw_od = optical_density(raw)
raw_haemo = beer_lambert_law(raw_od)
channel_results = fold_channels(
raw=raw_haemo,
p_name=p_name,
atlas=atlas,
progress_queue=progress_queue
)
return channel_results
def _auto_group_channels(channel_results: dict) -> dict[str, list[str]]:
"""Clusters channels by their highest specificity landmark (excluding Brain_Outside)."""
roi_groups: dict[str, list[str]] = {}
for ch_name, landmarks in channel_results.items():
if not landmarks:
continue
# Filter out Brain_Outside unless it's the only option
valid_landmarks = [
lm for lm in landmarks
if lm.get("Landmark") != "Brain_Outside"
]
if not valid_landmarks:
valid_landmarks = landmarks
# Pick highest specificity landmark
top_landmark = max(valid_landmarks, key=lambda x: x.get("Specificity", 0))
raw_name = top_landmark.get("Landmark", "Unassigned")
# Format label (e.g. "7 - Somatosensory..." -> "BA 7: Somatosensory Association Cortex")
parts = raw_name.split(" - ", 1)
if len(parts) == 2 and parts[0].strip().isdigit():
group_name = f"BA {parts[0].strip()}: {parts[1].strip()}"
else:
group_name = raw_name
roi_groups.setdefault(group_name, []).append(ch_name)
return roi_groups