"""
Filename: participantfoldchannels.py
Description: Logic for the Participant fOLD Channels analysis window
Author: Tyler de Zeeuw
License: GPL-3.0
"""
import os
import time
import traceback
from multiprocessing import Process, current_process, Manager
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
from matplotlib.figure import Figure
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
from PySide6.QtWidgets import QFrame, QGridLayout, QHBoxLayout, QLabel, QProgressBar, QPushButton, QScrollArea, QSizePolicy, QWidget, QDialog, QVBoxLayout
from PySide6.QtCore import QThread, Qt, QSize, QTimer
from PySide6.QtGui import QPixmap, QImage
from src.shared.flaresbasewidget import FlaresBaseWidget
from src.shared.shareddata import APP_NAME, resource_path
class MultiProgressDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("fOLD Analysis Progress")
self.setFixedWidth(400)
self.setWindowModality(Qt.WindowModality.NonModal)
self.layout = QVBoxLayout(self)
self.bars = {}
self.allow_closing = False
def add_participant(self, label, total_steps):
clean_key = str(label).strip()
label_widget = QLabel(f"Analyzing {clean_key}...")
pbar = QProgressBar()
pbar.setMinimum(0)
pbar.setMaximum(int(total_steps)) # Ensure this is a strict integer
pbar.setValue(0)
self.layout.addWidget(label_widget)
self.layout.addWidget(pbar)
self.bars[label] = pbar
def update_bar(self, label, value):
if label in self.bars:
# Force integers to prevent QProgressBar from breaking or flickering
self.bars[label].setValue(int(value))
def closeEvent(self, event):
if self.allow_closing:
event.accept()
else:
event.ignore()
def force_close(self):
self.allow_closing = True
self.close()
def single_participant_worker(file_path, raw_data, result_queue, progress_queue):
""" Runs inside its own dedicated process """
p_name = os.path.basename(file_path)
try:
from flares import fold_channels
# Perform the heavy fold_channels logic
channel_results = fold_channels(raw_data, p_name, progress_queue)
# Hand back results and signal completion
result_queue.put({file_path: channel_results})
progress_queue.put(p_name)
except Exception as e:
progress_queue.put(f"ERROR: {p_name} - {str(e)}")
def get_landmark_color_map():
"""Generates the unified 40-color map for fOLD landmarks."""
landmarks = [
"1 - Primary Somatosensory Cortex", "2 - Primary Somatosensory Cortex",
"3 - Primary Somatosensory Cortex", "4 - Primary Motor Cortex",
"5 - Somatosensory Association Cortex", "6 - Pre-Motor and Supplementary Motor Cortex",
"7 - Somatosensory Association Cortex", "8 - Includes Frontal eye fields",
"9 - Dorsolateral prefrontal cortex", "10 - Frontopolar area",
"11 - Orbitofrontal area", "17 - Primary Visual Cortex (V1)",
"18 - Visual Association Cortex (V2)", "19 - V3", "20 - Inferior Temporal gyrus",
"21 - Middle Temporal gyrus", "22 - Superior Temporal Gyrus",
"23 - Ventral Posterior cingulate cortex", "24 - Ventral Anterior cingulate cortex",
"25 - Subgenual cortex", "32 - Dorsal anterior cingulate cortex",
"37 - Fusiform gyrus", "38 - Temporopolar area",
"39 - Angular gyrus, part of Wernicke's area", "40 - Supramarginal gyrus part of Wernicke's area",
"41 - Primary and Auditory Association Cortex", "42 - Primary and Auditory Association Cortex",
"43 - Subcentral area", "44 - pars opercularis, part of Broca's area",
"45 - pars triangularis Broca's area", "46 - Dorsolateral prefrontal cortex",
"47 - Inferior prefrontal gyrus", "48 - Retrosubicular area", "Brain_Outside"
]
# Sort logically
landmarks.sort(key=lambda x: (int(x.split(" - ")[0]) if x.split(" - ")[0].isdigit() else float('inf')))
cmap1 = plt.get_cmap('tab20')
cmap2 = plt.get_cmap('tab20b')
colors = [cmap1(i) for i in range(20)] + [cmap2(i) for i in range(20)]
return {landmark: colors[i % len(colors)] for i, landmark in enumerate(landmarks)}
class StaticChannelCanvas(FigureCanvas):
"""The Pop-up Window Canvas.
Renders the interactive pie chart on the left, and a matching PNG image on the right.
"""
def __init__(self, channel_name, data_list, color_map, image_path=None, parent=None):
self.fig = Figure(figsize=(11.0, 5.5))
self.ax = self.fig.subplots(1, 2)
super().__init__(self.fig)
self.setParent(parent)
self.setMouseTracking(True)
# --- 1. DATA PREPARATION ---
self.wedge_data_list = list(data_list)
total_specificity = sum(d['Specificity'] for d in self.wedge_data_list)
if total_specificity < 100.0:
remainder = 100.0 - total_specificity
if remainder > 0.01:
self.wedge_data_list.append({
'Landmark': 'Other / Unclassified Regions',
'Specificity': remainder
})
self.specificities = [d['Specificity'] for d in self.wedge_data_list]
self.landmarks = [d['Landmark'] for d in self.wedge_data_list]
self.colors = [color_map.get(lm, '#ccc') if 'Other' not in lm else '#d3d3d3' for lm in self.landmarks]
self.labels = [f"{lm.split(' - ')[0]}" if 'Other' not in lm and lm != 'Brain_Outside' else 'Other' if 'Other' in lm else 'B' for lm in self.landmarks]
# --- 2. LEFT SUBPLOT: PIE CHART ---
# Note we explicitly target self.ax[0] now
self.wedges, self.texts, self.autotexts = self.ax[0].pie(
self.specificities,
autopct='%1.1f%%',
startangle=90,
labels=self.labels,
colors=self.colors,
textprops={'fontsize': 10, 'fontweight': 'bold'},
labeldistance=1.1
)
self.ax[0].axis('equal')
# --- 3. RIGHT SUBPLOT: PNG IMAGE DISPLAY ---
# Note we explicitly target self.ax[1] now
if image_path:
try:
img = mpimg.imread(image_path)
self.ax[1].imshow(img)
except Exception as e:
self.ax[1].text(0.5, 0.5, f"Failed to load image:\n{e}",
ha='center', va='center', fontsize=10, color='red')
else:
# Fallback message if no image path is passed down
self.ax[1].text(0.5, 0.5, "No Reference Image\nProvided",
ha='center', va='center', fontsize=12, fontweight='bold', color='#777')
# Completely hide the background grid, spines, and axis lines for the image box
self.ax[1].axis('off')
# --- 4. CANVAS TEXT OVERLAY ---
# Main Title centered globally over both subplots
self.fig.suptitle(channel_name, fontsize=16, fontweight='bold', y=0.97)
# Shared info box text overlay centered horizontally across the whole window figure
self.info_text = self.ax[0].text(
0.5, 0.04, "",
transform=self.fig.transFigure,
ha="center", va="bottom",
fontsize=12, fontweight="bold",
bbox=dict(boxstyle="round,pad=0.5", facecolor="#fdfdfd", edgecolor="#bbb", alpha=0.95)
)
self.info_text.set_visible(False)
self.currently_exploded_idx = None
# Layout space optimization
self.fig.subplots_adjust(left=0.05, bottom=0.1, right=0.95, top=0.85, wspace=0.2)
self.draw()
self.mpl_connect('motion_notify_event', self._on_hover)
def _on_hover(self, event):
try:
# FIX: Only track mouse events when hovering over the LEFT axis frame containing the pie chart
if event.inaxes != self.ax[0]:
if self.currently_exploded_idx is not None:
self._reset_wedges()
self.info_text.set_visible(False)
self.currently_exploded_idx = None
self.draw_idle()
return
hovered_index = None
for idx, wedge in enumerate(self.wedges):
contained, _ = wedge.contains(event)
if contained:
hovered_index = idx
break
if hovered_index is not None:
if self.currently_exploded_idx != hovered_index:
self.currently_exploded_idx = hovered_index
self._explode_wedge(hovered_index)
displayed_pct = self.autotexts[hovered_index].get_text()
full_desc = self.landmarks[hovered_index]
self.info_text.set_text(f"{full_desc} | {displayed_pct}")
self.info_text.set_visible(True)
self.draw_idle()
else:
if self.currently_exploded_idx is not None:
self._reset_wedges()
self.info_text.set_visible(False)
self.currently_exploded_idx = None
self.draw_idle()
except Exception as err:
print("[ERROR] Internal failure inside _on_hover loop:")
traceback.print_exc()
def _explode_wedge(self, index_to_expand):
changed = False
for idx, wedge in enumerate(self.wedges):
if idx == index_to_expand:
theta = np.deg2rad((wedge.theta1 + wedge.theta2) / 2.0)
explode_distance = 0.08
new_x = explode_distance * np.cos(theta)
new_y = explode_distance * np.sin(theta)
if wedge.center != (new_x, new_y):
wedge.set_center((new_x, new_y))
changed = True
else:
if wedge.center != (0.0, 0.0):
wedge.set_center((0.0, 0.0))
changed = True
if changed:
self.draw_idle()
def _reset_wedges(self):
changed = False
for wedge in self.wedges:
if wedge.center != (0.0, 0.0):
wedge.set_center((0.0, 0.0))
changed = True
if changed:
self.draw_idle()
class StandaloneLegendDialog(QWidget):
def __init__(self, canvas_engine, title_prefix, parent=None):
super().__init__(None)
self.setWindowTitle("Full View - Brodmann Legend")
self.setMinimumSize(500, 600)
self.resize(500, 900)
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
# Reuse your exact card creation method to render inside the popup window
legend_card = canvas_engine.create_legend_card(title_prefix, self)
layout.addWidget(legend_card)
class InteractiveParticipantGridCanvas(FigureCanvas):
"""The Big Grid Canvas.
Dynamically scales row and column configurations to maintain a crisp 16:9 layout orientation.
"""
def __init__(self, channels_data, color_map, is_fullscreen_copy=False, parent=None):
self.channels_data = channels_data
self.color_map = color_map
self.is_fullscreen_copy = is_fullscreen_copy
num_channels = len(channels_data)
# --- FIX: DYNAMICALLY CALCULATE OPTIMAL 16:9 COLUMNS ---
target_ratio = 16 / 9
best_cols = 4
min_ratio_error = float('inf')
# Test configurations from 4 columns up to the total number of channels
for test_cols in range(4, num_channels + 1):
test_rows = (num_channels + test_cols - 1) // test_cols
# Approximate the visual aspect ratio based on cell dimensions
# Mini charts are slightly wider than tall, roughly 1.15 to 1.0 factor
current_ratio = (test_cols * 1.15) / (test_rows * 1.0)
error = abs(current_ratio - target_ratio)
if error < min_ratio_error:
min_ratio_error = error
best_cols = test_cols
cols = best_cols
rows = (num_channels + cols - 1) // cols
# Base figure sizing dynamically scales off the optimal matrix constraints
if is_fullscreen_copy:
# Maximized views stretch cleanly across standard display panels
figsize = (14.0, 14.0 / target_ratio)
else:
# Standard thumbnail views scaled down for participant cards
figsize = (7.5, 7.5 / target_ratio)
self.fig = Figure(figsize=figsize)
super().__init__(self.fig)
self.setParent(parent)
self.axes_data_registry = {}
for idx, (channel_name, data_list) in enumerate(channels_data.items()):
ax = self.fig.add_subplot(rows, cols, idx + 1)
padded_data_list = list(data_list)
total_specificity = sum(d['Specificity'] for d in padded_data_list)
if total_specificity < 100.0:
remainder = 100.0 - total_specificity
if remainder > 0.01:
padded_data_list.append({
'Landmark': 'Other / Unclassified Regions',
'Specificity': remainder
})
self.axes_data_registry[ax] = {
'channel_name': channel_name,
'data_list': padded_data_list
}
specificities = [d['Specificity'] for d in padded_data_list]
landmarks = [d['Landmark'] for d in padded_data_list]
colors = [color_map.get(lm, '#ccc') if 'Other' not in lm else '#d3d3d3' for lm in landmarks]
labels = [f"{lm.split(' - ')[0]}" if 'Other' not in lm and lm != 'Brain_Outside' else 'O' if 'Other' in lm else 'B' for lm in landmarks]
# Adjust label sizing dynamically based on how crowded the grid gets
font_sz = 5 if num_channels > 30 else (7 if is_fullscreen_copy else 6)
title_sz = 6 if num_channels > 30 else (9 if is_fullscreen_copy else 7)
ax.pie(
specificities,
startangle=90,
colors=colors,
labels=labels,
textprops={'fontsize': font_sz, 'fontweight': 'bold'},
labeldistance=1.05,
radius=0.75
)
ax.set_title(channel_name, fontsize=title_sz, fontweight='bold', pad=0, y=1.04)
ax.axis('equal')
# --- FIX: ADAPTIVE PADDING BOUNDS FOR EXTRA DENSE PLOTS ---
# Large multi-column plots require less spacing overhead to prevent clipping label masks
h_sp = 0.35 if num_channels > 30 else 0.18
w_sp = 0.25 if num_channels > 30 else 0.10
if is_fullscreen_copy:
self.fig.subplots_adjust(left=0.02, bottom=0.02, right=0.98, top=0.95, hspace=h_sp, wspace=w_sp)
else:
self.fig.set_layout_engine('constrained')
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
self.draw()
self.mpl_connect('button_press_event', self._on_canvas_click)
def create_matrix_card(self, title_prefix, layout_to_attach_to):
"""Wraps the channel matrix layout inside a responsive, matching hover-stylized card frame."""
# 1. Create matching styled container card frame
card_frame = QFrame()
card_frame.setFrameShape(QFrame.Shape.StyledPanel)
card_frame.setStyleSheet("""
QFrame {
background-color: #ffffff;
border: 2px solid #ced4da;
border-radius: 6px;
}
QFrame:hover {
border: 2px solid #4dabf7;
background-color: #f8f9fa;
}
""")
card_layout = QVBoxLayout(card_frame)
card_layout.setContentsMargins(6, 6, 6, 6)
card_layout.setSpacing(4)
# 2. Add header matching the summary card type architecture
header = QLabel(f"{title_prefix} - Channels Matrix")
header.setStyleSheet("font-weight: bold; font-size: 10pt; border: none; color: #212529; background: transparent;")
header.setAlignment(Qt.AlignmentFlag.AlignCenter)
card_layout.addWidget(header)
# 3. Nest this canvas instance cleanly inside the card frame layout
self.setParent(card_frame)
card_layout.addWidget(self)
card_layout.addStretch(0)
# 4. Make the remaining empty whitespace frame areas trigger the maximization loop
card_frame.mouseReleaseEvent = lambda event: self._open_fullscreen_grid() if event.button() == Qt.MouseButton.LeftButton else None
# Ensure underlying child mouse hits tunnel downstream properly to our parent container frame
header.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
layout_to_attach_to.addWidget(card_frame)
return card_frame
def _on_canvas_click(self, event):
# CASE 1: Whitespace Clicked -> Open full 25-matrix in fullscreen window
if event.inaxes is None:
self._open_fullscreen_grid()
return
# CASE 2: Specific Slice Clicked -> Open standard individual detailed channel popup
clicked_subplot_data = self.axes_data_registry.get(event.inaxes)
if clicked_subplot_data:
self._open_expanded_view(
clicked_subplot_data['channel_name'],
clicked_subplot_data['data_list']
)
def _open_fullscreen_grid(self):
"""Creates a maximized dialog window duplicating the full participant matrix view."""
if getattr(self, 'is_fullscreen_copy', False) or hasattr(self, '_is_fullscreen_flag_set'):
return
fullscreen_window = QWidget(None)
fullscreen_window.setWindowTitle("Participant Grid Monitor - Maximized View")
fullscreen_window.setWindowFlags(
Qt.WindowType.Window |
Qt.WindowType.WindowMinMaxButtonsHint |
Qt.WindowType.WindowCloseButtonHint
)
layout = QVBoxLayout(fullscreen_window)
layout.setContentsMargins(0, 0, 0, 0)
# Instantiate the copy
large_grid_canvas = InteractiveParticipantGridCanvas(
self.channels_data,
self.color_map,
is_fullscreen_copy=True,
parent=fullscreen_window
)
# Explicitly tag the new canvas object internally to block further clicks
large_grid_canvas._is_fullscreen_flag_set = True
layout.addWidget(large_grid_canvas)
# Open non-modally so it populates the taskbar and matches OS window behaviors
fullscreen_window.showMaximized()
# Keep a reference alive on the source canvas so Python doesn't garbage collect the window
if not hasattr(self, '_fullscreen_refs'):
self._fullscreen_refs = []
self._fullscreen_refs = [w for w in self._fullscreen_refs if w.isVisible()]
self._fullscreen_refs.append(fullscreen_window)
def _calculate_total_brodmann_profile(self, channels_data):
"""Sums and normalizes the specificity profile across all channels."""
totals = {}
num_channels = len(channels_data)
if num_channels == 0:
return []
# Sum up specificities across all channels
for channel_name, data_list in channels_data.items():
for entry in data_list:
landmark = entry['Landmark']
specificity = entry['Specificity']
totals[landmark] = totals.get(landmark, 0.0) + specificity
# Normalize back down to 100% total scale
normalized_data_list = []
for landmark, total_val in totals.items():
# If a landmark hit 20% in 10 channels, it's normalized relative to total channels
normalized_val = total_val / num_channels
if normalized_val > 0.01:
normalized_data_list.append({
'Landmark': landmark,
'Specificity': normalized_val
})
# Ensure "Other / Unclassified" fills any remaining precision gap
total_normalized = sum(d['Specificity'] for d in normalized_data_list)
if total_normalized < 100.0:
remainder = 100.0 - total_normalized
if remainder > 0.01:
normalized_data_list.append({
'Landmark': 'Other / Unclassified Regions',
'Specificity': remainder
})
return normalized_data_list
def _open_expanded_view(self, channel_name, data_list):
# 1. Create a plain QWidget with NO parent (None)
# This instantly makes it a top-level desktop window
popup = QWidget(None)
popup.setWindowTitle(f"Channel Specificity Detail - {channel_name}")
# 2. Add standard window control behaviors
popup.setWindowFlags(
Qt.WindowType.Window |
Qt.WindowType.WindowMinMaxButtonsHint |
Qt.WindowType.WindowCloseButtonHint
)
# 3. Build layout out exactly as before
layout = QVBoxLayout(popup)
layout.setContentsMargins(0, 0, 0, 0) # Strip extra outer layout spacing
target_png_path = resource_path("images/brain.png")
expanded_canvas = StaticChannelCanvas(
channel_name,
data_list,
self.color_map,
image_path=target_png_path,
parent=popup
)
layout.addWidget(expanded_canvas)
popup.resize(900, 520)
# 4. Display non-modally
popup.show()
# 5. Keep the reference alive so Python doesn't garbage collect it
if not hasattr(self, '_open_popups'):
self._open_popups = []
# Clean up closed windows from our tracking list to save memory
self._open_popups = [w for w in self._open_popups if w.isVisible()]
self._open_popups.append(popup)
def create_total_summary_card(self, title_prefix, layout_to_attach_to):
"""Generates a highly compact, clickable embedded card on the main window showing aggregated data."""
# 1. Calculate the normalized profile data payload using the instance's own data
summary_data = self._calculate_total_brodmann_profile(self.channels_data)
# 2. Create a styled container card frame
card_frame = QFrame()
card_frame.setFrameShape(QFrame.Shape.StyledPanel)
card_frame.setStyleSheet("""
QFrame {
background-color: #ffffff;
border: 2px solid #ced4da;
border-radius: 6px;
}
QFrame:hover {
border: 2px solid #4dabf7; /* Gives a subtle visual cue that it is clickable */
background-color: #f8f9fa; /* Slightly shifts background color on hover */
}
""")
card_layout = QVBoxLayout(card_frame)
card_layout.setContentsMargins(4, 4, 4, 4)
card_layout.setSpacing(2)
# Add a clear section header label containing the specific participant identity
header = QLabel(f"{title_prefix} - Total Profile")
header.setStyleSheet("font-weight: bold; font-size: 10pt; border: none; color: #212529;")
header.setAlignment(Qt.AlignmentFlag.AlignCenter)
card_layout.addWidget(header)
target_png_path = resource_path("images/brain.png")
# 3. Instantiate the canvas with a custom size flag or constraint
# Adjust your StaticChannelCanvas __init__ to check if it should render in 'compact' mode
summary_canvas = StaticChannelCanvas(
channel_name=f"{title_prefix} Combined",
data_list=summary_data,
color_map=self.color_map,
image_path=target_png_path,
parent=card_frame,
)
# --- CRITICAL: SHRINK MATPLOTLIB FIGURE ELEMENTS FOR THE EMBEDDED VIEWER ---
# Scale down the underlying canvas container so it doesn't balloon the layout grid
if hasattr(summary_canvas, 'fig'):
summary_canvas.fig.subplots_adjust(left=0.02, bottom=0.02, right=0.98, top=0.92, wspace=0.10)
for ax in summary_canvas.fig.axes:
for text in ax.texts:
text.set_fontsize(6)
summary_canvas.draw()
summary_canvas.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
card_layout.addWidget(summary_canvas)
card_layout.addStretch(0)
def handle_card_click(event):
# Only trigger expansion if it's a primary left-click action
if event.button() == Qt.MouseButton.LeftButton:
self._open_expanded_summary_window(title_prefix, summary_data)
card_frame.mouseReleaseEvent = handle_card_click
# Prevent clicks on the text/child elements from being swallowed up instead of passing to frame
header.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
summary_canvas.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents, True)
# 4. Inject completed card frame container assembly into target window layout position
layout_to_attach_to.addWidget(card_frame)
return card_frame
def create_legend_card(self, title_prefix, layout_to_attach_to):
card = QFrame()
card.setStyleSheet("QFrame { background-color: #ffffff; border-radius: 8px; border: 1px solid #e9ecef; }")
layout = QVBoxLayout(card)
layout.setContentsMargins(20, 20, 20, 20)
layout.setSpacing(10)
header_label = QLabel(f"{title_prefix}\nLandmarks")
header_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
header_label.setStyleSheet("font-size: 14px; font-weight: bold; color: #1a252f; border: none;")
layout.addWidget(header_label)
layout.addSpacing(10)
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setStyleSheet("QScrollArea { border: none; background: transparent; }")
scroll_content = QWidget()
scroll_content.setStyleSheet("background: transparent;")
scroll_layout = QVBoxLayout(scroll_content)
scroll_layout.setSpacing(6)
scroll_layout.setContentsMargins(0, 0, 0, 0)
true_color_map = get_landmark_color_map()
# Iterate over the sorted keys directly from your method
for landmark_text in true_color_map.keys():
item_row = QHBoxLayout()
item_row.setSpacing(12)
# Extract the RGBA tuple value assigned by matplotlib
rgba = true_color_map[landmark_text]
# Convert float tuple components (0.0 - 1.0) to standard CSS integer scales (0 - 255)
r, g, b = int(rgba[0] * 255), int(rgba[1] * 255), int(rgba[2] * 255)
color_hex = f"rgb({r}, {g}, {b})"
# Format display string nicely: "1 — Primary Somatosensory Cortex"
if " - " in landmark_text:
num, name = landmark_text.split(" - ", 1)
display_string = f"{num} — {name}"
else:
display_string = f"{landmark_text}"
dot = QLabel()
dot.setFixedSize(14, 14)
dot.setStyleSheet(f"background-color: {color_hex}; border-radius: 7px; border: none;")
label = QLabel(display_string)
label.setStyleSheet("font-size: 12px; color: #343a40; border: none;")
item_row.addWidget(dot)
item_row.addWidget(label, 1)
scroll_layout.addLayout(item_row)
scroll_area.setWidget(scroll_content)
layout.addWidget(scroll_area)
return card
def _open_expanded_summary_window(self, title_prefix, summary_data):
"""Pops open a beautifully scaled, independent large window when the card is clicked."""
popup = QWidget(None)
popup.setWindowTitle(f"Grand Total Profile Details - {title_prefix}")
popup.setWindowFlags(
Qt.WindowType.Window |
Qt.WindowType.WindowMinMaxButtonsHint |
Qt.WindowType.WindowCloseButtonHint
)
layout = QVBoxLayout(popup)
layout.setContentsMargins(10, 10, 10, 10)
target_png_path = resource_path("images/brain.png")
# This one renders full size (900x520) for analytical reading
expanded_canvas = StaticChannelCanvas(
f"{title_prefix} - All Channels Aggregated",
summary_data,
self.color_map,
image_path=target_png_path,
parent=popup,
)
layout.addWidget(expanded_canvas)
popup.resize(950, 550)
popup.show()
if not hasattr(self, '_summary_popups'):
self._summary_popups = []
self._summary_popups.append(popup)
from PySide6.QtCore import QObject, Signal
from multiprocessing import Manager, Process
class ProcessOrchestrator(QObject):
# Fires when Manager + Processes are completely ready
# Emits: (manager_instance, result_queue, progress_queue, active_processes_list)
setup_finished = Signal(object, object, object, list)
setup_failed = Signal(str)
def __init__(self, selected_files, haemo_dict, worker_func):
super().__init__()
self.selected_files = selected_files
self.haemo_dict = haemo_dict
self.worker_func = worker_func
def run(self):
try:
# Instantiate Manager completely off the main thread
manager = Manager()
result_queue = manager.Queue()
progress_queue = manager.Queue()
active_processes = []
# Perform heavy pickling loop safely in the background
for file_path in self.selected_files:
p = Process(
target=self.worker_func,
args=(file_path, self.haemo_dict[file_path], result_queue, progress_queue)
)
p.start()
active_processes.append(p)
# Deliver setup assets back to the GUI Main Thread
self.setup_finished.emit(manager, result_queue, progress_queue, active_processes)
except Exception as e:
self.setup_failed.emit(str(e))
class ParticipantFoldChannelsWidget(FlaresBaseWidget):
def __init__(self, haemo_dict, cha_dict):
super().__init__("ParticipantFoldChannels")
self.setWindowTitle(f"Participant Fold Channels Viewer - {APP_NAME.upper()}")
self.haemo_dict = haemo_dict
self.cha_dict = cha_dict
# Create mappings: file_path -> participant label and dropdown display text
self.participant_map = {} # file_path -> "Participant 1"
self.participant_dropdown_items = [] # "Participant 1 (filename)"
for i, file_path in enumerate(self.haemo_dict.keys(), start=1):
short_label = f"Participant {i}"
display_label = f"{short_label} ({os.path.basename(file_path)})"
self.participant_map[file_path] = short_label
self.participant_dropdown_items.append(display_label)
self.layout = QVBoxLayout(self)
self.top_bar = QHBoxLayout()
self.layout.addLayout(self.top_bar)
self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items)
self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label)
self.index_texts = [
"0 (Fold Channels)",
# "1 (second image)",
# "2 (third image)",
# "3 (fourth image)",
]
self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts)
self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label)
self.submit_button = QPushButton("Submit")
self.submit_button.clicked.connect(self.show_fold_images)
self.top_bar.addWidget(QLabel("Participants:"))
self.top_bar.addWidget(self.participant_dropdown)
self.top_bar.addWidget(QLabel("Fold Type:"))
self.top_bar.addWidget(self.image_index_dropdown)
self.top_bar.addWidget(self.submit_button)
self.scroll_area = QScrollArea(self)
self.scroll_area.setWidgetResizable(True)
self.scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
self.scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
self.scroll_area.setStyleSheet("QScrollArea { border: none; background-color: #f1f3f5; }")
# 2. Create the central canvas widget that inside the scroll block
self.scroll_content_widget = QWidget()
self.scroll_content_widget.setStyleSheet("background-color: #f1f3f5;")
# 3. Establish the strict 3-column layout grid engine
self.grid_layout = QGridLayout(self.scroll_content_widget)
self.grid_layout.setContentsMargins(12, 12, 12, 12)
self.grid_layout.setSpacing(15) # Controls breathing room gaps between cards
self.grid_layout.setColumnStretch(0, 1)
self.grid_layout.setColumnStretch(1, 1)
self.grid_layout.setColumnStretch(2, 1)
# 2. Force a uniform structural minimum width per column
# This blocks the dense matrices from hogging space and compressing the summary cards
self.grid_layout.setColumnMinimumWidth(0, 400)
self.grid_layout.setColumnMinimumWidth(1, 400)
self.grid_layout.setColumnMinimumWidth(2, 400)
# ----------------------------------------------------------
# Bind them together
self.scroll_area.setWidget(self.scroll_content_widget)
# Add the self.scroll_area widget to your root layout view frame panel
self.layout.addWidget(self.scroll_area)
self.thumb_size = QSize(280, 180)
self.showMaximized()
def show_fold_images(self):
selected_display_names = self._get_checked_items(self.participant_dropdown)
selected_indexes = [int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown)]
if not selected_display_names or 0 not in selected_indexes:
return
selected_files = [path for path, label in self.participant_map.items()
if f"{label} ({os.path.basename(path)})" in selected_display_names]
while self.grid_layout.count():
item = self.grid_layout.takeAt(0)
widget = item.widget()
if widget:
widget.deleteLater()
self.global_channels_data = {}
self.multi_progress = MultiProgressDialog(self)
for file_path in selected_files:
raw_data = self.haemo_dict[file_path]
# Dig out the exact channels list length matching your loop engine logic
hbo_channels = getattr(raw_data.copy().pick(picks='hbo'), "ch_names", [])
total_channels = len(hbo_channels) if hbo_channels else 1
self.multi_progress.add_participant(os.path.basename(file_path), total_channels)
from datetime import datetime
print(f"Before: {datetime.now()}")
self.multi_progress.show()
print(f"After 1: {datetime.now()}")
if current_process().name == 'MainProcess':
# Create a clean background thread worker execution channel
self.orchestrator_thread = QThread()
self.orchestrator = ProcessOrchestrator(selected_files, self.haemo_dict, single_participant_worker)
self.orchestrator.moveToThread(self.orchestrator_thread)
# Signal Routing
self.orchestrator_thread.started.connect(self.orchestrator.run)
self.orchestrator.setup_finished.connect(self.on_orchestration_success)
self.orchestrator.setup_failed.connect(self.on_orchestration_failed)
# Automatic lifecycle cleanup configuration
self.orchestrator.setup_finished.connect(self.orchestrator_thread.quit)
self.orchestrator.setup_failed.connect(self.orchestrator_thread.quit)
self.orchestrator_thread.finished.connect(self.orchestrator_thread.deleteLater)
self.orchestrator.setup_finished.connect(self.orchestrator.deleteLater)
self.orchestrator.setup_failed.connect(self.orchestrator.deleteLater)
self.orchestrator_thread.start()
print(f"After 4: {datetime.now()}")
def on_orchestration_success(self, manager, result_queue, progress_queue, active_processes):
""" Executed on the Main GUI Thread once background process setup finishes """
self.manager = manager
self.result_queue = result_queue
self.progress_queue = progress_queue
self.active_processes = active_processes
# Safely initialize and trigger the polling listener
self.completed_count = 0
self.result_timer = QTimer()
self.result_timer.timeout.connect(self.check_parallel_results)
self.result_timer.start()
def on_orchestration_failed(self, error_msg):
""" Fallback handler if Windows permissions or pickling fails in background """
if hasattr(self, 'multi_progress'):
self.multi_progress.close()
print(f"[CRITICAL FAILURE] Background Orchestration Failed:\n{error_msg}")
def check_parallel_results(self):
# Check for progress/completion signals
while not self.progress_queue.empty():
msg = self.progress_queue.get()
# CASE 1: Micro-step channel increment (Tuple tracking)
if isinstance(msg, tuple):
p_name, completed_channels = msg
clean_key = str(p_name).strip()
if hasattr(self, 'multi_progress') and clean_key in self.multi_progress.bars:
print(completed_channels)
self.multi_progress.update_bar(clean_key, completed_channels)
else:
# DEBUG LOG: This tells us exactly why a bar isn't moving
print(f"[DEBUG WARNING] Progress received for '{clean_key}' but no matching bar was found. Existing bars: {list(self.multi_progress.bars.keys())}")
continue
# CASE 2: Worker process crashed with an error string
if isinstance(msg, str) and msg.startswith("ERROR"):
print(f"Worker Error: {msg}")
#self.completed_count += 1 # Count as finished so the UI doesn't hang
# CASE 3: Final clean text string signal indicating complete file closure
elif isinstance(msg, str):
# Max out the progress bar visually on completion
if hasattr(self, 'multi_progress'):
if msg in self.multi_progress.bars:
max_val = self.multi_progress.bars[msg].maximum()
self.multi_progress.update_bar(msg, max_val)
self.completed_count += 1 # Increment the master task tracker
print(self.completed_count, time.time())
# Pull images as they become available
while not self.result_queue.empty():
result_dict = self.result_queue.get()
self.add_images_to_grid(result_dict)
# Clean up when all processes are done
if self.completed_count >= len(self.active_processes):
self.result_timer.stop()
# Close the custom multi-progress window
if hasattr(self, 'multi_progress'):
self.multi_progress.force_close()
# Clean up processes
for p in self.active_processes:
if p.is_alive():
p.join(timeout=1) # Give it a second to wrap up
p.close() # Explicitly close the process object
# Shut down the Manager process (the source of the 'rogue' process)
if hasattr(self, 'manager'):
self.manager.shutdown()
self.active_processes = []
print("Processing fully complete. All resources released.")
if hasattr(self, 'global_channels_data') and self.global_channels_data:
color_map = get_landmark_color_map()
# We feed the entire channel pool directly to your existing canvas engine class
global_canvas = InteractiveParticipantGridCanvas(self.global_channels_data, color_map)
# Create the summary card using your exact visual method
global_card = global_canvas.create_total_summary_card(
title_prefix="Grand Global Layout",
layout_to_attach_to=self.scroll_content_widget.layout()
)
# Match your exact layout positioning logic to place it next in the grid
count = self.grid_layout.count() - 1
row = count // 3
col = count % 3
self.grid_layout.addWidget(global_card, row, col)
legend_title = "Grand Total Brodmann Mapping Profile"
legend_card = global_canvas.create_legend_card(
title_prefix=legend_title,
layout_to_attach_to=self.scroll_content_widget.layout()
)
def handle_legend_click(event):
self.active_legend_window = StandaloneLegendDialog(global_canvas, legend_title, self)
self.active_legend_window.show()
legend_card.mousePressEvent = handle_legend_click
count = self.grid_layout.count()
row = count // 3
col = count % 3
self.grid_layout.addWidget(legend_card, row, col)
# def add_images_to_grid(self, result_dict):
# """
# result_dict format: { file_path: {"main": bytes, "legend": bytes} }
# """
# for file_path, images in result_dict.items():
# if self.grid_layout.count() == 0 and "legend" in images:
# self._add_legend_to_grid(images["legend"])
# # Create a container for this participant's results
# container = QFrame()
# container.setFrameShape(QFrame.StyledPanel)
# vbox = QVBoxLayout(container)
# participant_label = self.participant_map.get(file_path, os.path.basename(file_path))
# title = QLabel(f"{participant_label}")
# title.setAlignment(Qt.AlignCenter)
# vbox.addWidget(title)
# # We primarily want to show the 'main' plot in the grid
# if "main" in images:
# pixmap = self._bytes_to_pixmap(images["main"])
# img_label = QLabel()
# # Scale it to fit the thumbnail size defined in __init__
# img_label.setPixmap(pixmap.scaled(
# self.thumb_size,
# Qt.KeepAspectRatio,
# Qt.SmoothTransformation
# ))
# img_label.setAlignment(Qt.AlignCenter)
# # Optional: Click to open full size
# img_label.mousePressEvent = lambda e, p=pixmap, t=participant_label: self._open_full_size(p, t)
# vbox.addWidget(img_label)
# # Determine grid position (row-major order)
# count = self.grid_layout.count()
# row = count // 3 # 3 columns wide
# col = count % 3
# self.grid_layout.addWidget(container, row, col)
def add_images_to_grid(self, result_dict):
color_map = get_landmark_color_map()
for file_path, channels_data in result_dict.items():
participant_label = self.participant_map.get(file_path, os.path.basename(file_path))
if hasattr(self, 'global_channels_data'):
for ch_name, ch_data in channels_data.items():
unique_key = f"{participant_label}_{ch_name}"
self.global_channels_data[unique_key] = ch_data
# 1. Instantiate the background calculation engine matrix
participant_grid_canvas = InteractiveParticipantGridCanvas(channels_data, color_map)
# 2. Build Card A (Channels Matrix Frame Layout)
# The matrix automatically installs inside its layout box container slot
matrix_card = participant_grid_canvas.create_matrix_card(
title_prefix=participant_label,
layout_to_attach_to=self.scroll_content_widget.layout() # Maps directly to your grid layout
)
# Pin Card A to the sequential grid coordinate tracker layout
count = self.grid_layout.count() - 1 # Subtract 1 because widget registration steps index values forward
row = count // 3
col = count % 3
self.grid_layout.addWidget(matrix_card, row, col)
# 3. Build Card B (Total Summary Profile Frame Layout)
summary_card = participant_grid_canvas.create_total_summary_card(
title_prefix=participant_label,
layout_to_attach_to=self.scroll_content_widget.layout()
)
# Pin Card B directly next into the 3-column processing loop matrix layout tracker
count = self.grid_layout.count() - 1
row = count // 3
col = count % 3
self.grid_layout.addWidget(summary_card, row, col)
def _bytes_to_pixmap(self, png_bytes):
"""Converts raw bytes from the multiprocess queue to a QPixmap."""
image = QImage.fromData(png_bytes)
return QPixmap.fromImage(image)
def _open_full_size(self, pixmap, title):
"""Simple popup to view the image at a readable scale."""
view = QDialog(self)
view.setWindowTitle(f"Full View - {title}")
layout = QVBoxLayout(view)
label = QLabel()
label.setPixmap(pixmap)
layout.addWidget(label)
view.show()
def _add_legend_to_grid(self, legend_bytes):
"""Helper to put the legend in the first slot."""
container = QFrame()
container.setStyleSheet("background-color: #f9f9f9; border: 1px solid #ccc;")
vbox = QVBoxLayout(container)
title = QLabel("Brodmann Area Legend")
title.setAlignment(Qt.AlignCenter)
vbox.addWidget(title)
pixmap = self._bytes_to_pixmap(legend_bytes)
legend_label = QLabel()
# Legends are usually tall, so we scale it differently or keep it smaller
legend_label.setPixmap(pixmap.scaled(
self.thumb_size,
Qt.KeepAspectRatio,
Qt.SmoothTransformation
))
legend_label.setAlignment(Qt.AlignCenter)
legend_label.mousePressEvent = lambda e, p=pixmap: self._open_full_size(p, "Brodmann Legend")
vbox.addWidget(legend_label)
self.grid_layout.addWidget(container, 0, 0)