Coverage for main/ui/pipeline_review_page.py: 99%
152 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-03 15:51 +0000
1import json
2import os
3import glob
5from PyQt6.QtGui import QFont
6from PyQt6.QtWidgets import QWidget, QScrollArea, QVBoxLayout, QLabel
7from PyQt6.QtCore import Qt, QCoreApplication
9from components.collapsible_info_frame import CollapsibleInfoFrame
10from components.collapsible_patient_frame import CollapsiblePatientFrame
11from ui.pipeline_execution_page import PipelineExecutionPage
12from page import Page
13from logger import get_logger
15log = get_logger()
18class PipelineReviewPage(Page):
19 """
20 GUI page that allows users to review and adjust the configuration
21 before running the medical imaging pipeline.
23 This page:
24 - Loads the latest configuration JSON file.
25 - Displays collapsible panels for each patient and data type.
26 - Allows editing, saving, and validating configurations.
27 - Transitions to the execution page once all data is ready.
28 """
30 def __init__(self, context=None, previous_page=None):
31 """
32 Initialize the review page and build its UI.
34 Args:
35 context (dict): Shared application context (workspace path, logger, signals, etc.).
36 previous_page (Page): The previous page in the workflow.
37 """
38 super().__init__()
39 self.context = context
40 self.workspace_path = context["workspace_path"]
41 self.previous_page = previous_page
42 self.next_page = None
44 # Locate and load the latest configuration file
45 self.config_path = self._find_latest_config()
46 self.pipeline_config = self._load_config()
48 # Main layout container
49 self.main_layout = QVBoxLayout(self)
51 # Build interface
52 self._setup_ui()
54 # Apply translation and connect to language change signal (if available)
55 self._translate_ui()
56 if context and "language_changed" in context:
57 context["language_changed"].connect(self._translate_ui)
59 # ─────────────────────────────────────────────
60 # CONFIGURATION MANAGEMENT
61 # ─────────────────────────────────────────────
62 def _find_latest_config(self):
63 """
64 Find the most recent pipeline configuration file in the workspace.
66 Returns:
67 str: Absolute path to the newest configuration JSON file.
68 """
69 pipeline_dir = os.path.join(self.workspace_path, "pipeline")
71 # Return a default path if the directory does not exist
72 if not os.path.exists(pipeline_dir):
73 return os.path.join(pipeline_dir, "pipeline_config.json")
75 # Locate all files matching the *_config.json pattern
76 config_files = glob.glob(os.path.join(pipeline_dir, "*_config.json"))
78 if not config_files:
79 return os.path.join(pipeline_dir, "pipeline_config.json")
81 # Identify the latest config file by its numeric ID prefix
82 max_id = 0
83 latest_config = None
84 for config_file in config_files:
85 filename = os.path.basename(config_file)
86 try:
87 config_id = int(filename.split('_')[0]) # Extract numeric prefix
88 if config_id > max_id:
89 max_id = config_id
90 latest_config = config_file
91 except (ValueError, IndexError):
92 continue # Skip non-conforming filenames
94 if latest_config:
95 return latest_config
97 return os.path.join(pipeline_dir, "pipeline_config.json")
99 def _load_config_from_path(self, config_path):
100 """
101 Load configuration data from a specific JSON file.
103 Args:
104 config_path (str): Path to the configuration file.
106 Returns:
107 dict: Parsed configuration data, or an empty dict if load fails.
108 """
109 if not os.path.exists(config_path):
110 log.warning(f"Config file not found: {config_path}")
111 return {}
113 try:
114 with open(config_path, "r", encoding="utf-8") as f:
115 return json.load(f)
116 except (json.JSONDecodeError, IOError) as e:
117 log.error(f"Error loading config {config_path}: {e}")
118 return {}
120 def _load_config(self):
121 """Load the current configuration using the latest detected file."""
122 return self._load_config_from_path(self.config_path)
124 # ─────────────────────────────────────────────
125 # UI SETUP AND LAYOUT
126 # ─────────────────────────────────────────────
127 def _setup_ui(self):
128 """Construct the entire user interface for the review page."""
129 layout = self.main_layout
131 # Clear any existing widgets
132 while layout.count():
133 item = layout.takeAt(0)
134 widget = item.widget()
135 if widget:
136 widget.deleteLater()
138 # ── Header ──────────────────────────────
139 self.header = QLabel("Pipeline Configuration Review")
140 self.header.setAlignment(Qt.AlignmentFlag.AlignCenter)
141 self.header.setFont(QFont("Arial", 18, QFont.Weight.Bold))
142 layout.addWidget(self.header)
144 # ── Config file info ─────────────────────
145 self.config_info = QLabel(f"Reviewing: {os.path.basename(self.config_path)}")
146 self.config_info.setAlignment(Qt.AlignmentFlag.AlignCenter)
147 self.config_info.setWordWrap(True)
148 self.config_info.setStyleSheet("""
149 font-size: 11px;
150 color: #666;
151 font-style: italic;
152 margin-bottom: 6px;
153 """)
154 layout.addWidget(self.config_info)
156 # ── Collapsible instructions ─────────────
157 info_frame = CollapsibleInfoFrame(self.context)
158 layout.addWidget(info_frame)
160 # ── Info label ──────────────────────────
161 self.info_label = QLabel(
162 "<strong>Click</strong> a frame to review file selections. "
163 "<strong>Save</strong> yellow frames after selection."
164 )
165 self.info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
166 self.info_label.setWordWrap(True)
167 self.info_label.setStyleSheet("""
168 color: #666;
169 font-size: 12px;
170 margin: 6px 0;
171 """)
172 layout.addWidget(self.info_label)
174 # ── Scroll area for patient panels ───────
175 scroll = QScrollArea()
176 scroll.setWidgetResizable(True)
177 scroll.setStyleSheet("QScrollArea { border: none; }")
178 layout.addWidget(scroll)
180 content = QWidget()
181 content_layout = QVBoxLayout(content)
182 content_layout.setContentsMargins(6, 6, 6, 6)
183 content_layout.setSpacing(8)
184 scroll.setWidget(content)
186 # ── Patient frames ───────────────────────
187 self.patient_widgets = {}
189 # Define expected data categories and file patterns
190 categories = {
191 "mri": [os.path.join(self.workspace_path, "{pid}", "anat", "*_flair.nii*")],
192 "mri_str": [os.path.join(self.workspace_path, "derivatives", "skullstrips", "{pid}", "anat", "*_brain.nii*")],
193 "pet": [os.path.join(self.workspace_path, "{pid}", "ses-01", "pet", "*_pet.nii*")],
194 "pet4d": [os.path.join(self.workspace_path, "{pid}", "ses-02", "pet", "*_pet.nii*")],
195 "tumor_mri": [
196 os.path.join(self.workspace_path, "derivatives", "manual_masks", "{pid}", "anat", "*_mask.nii*"),
197 os.path.join(self.workspace_path, "derivatives", "deep_learning_seg", "{pid}", "anat", "*_seg.nii*"),
198 ],
199 }
201 # Create collapsible frames for each patient
202 for patient_id, files in self.pipeline_config.items():
203 patient_patterns = {
204 cat: [pat.format(pid=patient_id) for pat in pats]
205 for cat, pats in categories.items()
206 }
207 multiple_choice = bool(files.get("need_revision", False))
209 frame = CollapsiblePatientFrame(
210 self.context,
211 patient_id,
212 files,
213 patient_patterns,
214 multiple_choice,
215 save_callback=self._save_single_patient,
216 )
218 self.patient_widgets[patient_id] = frame
219 content_layout.addWidget(frame)
221 content_layout.addStretch()
223 # ─────────────────────────────────────────────
224 # SAVE & UPDATE LOGIC
225 # ─────────────────────────────────────────────
226 def _save_single_patient(self, patient_id, files):
227 """
228 Save the configuration for a single patient.
230 Args:
231 patient_id (str): Patient identifier.
232 files (dict): Updated file paths and metadata.
233 """
234 self.pipeline_config[patient_id] = files
235 with open(self.config_path, "w", encoding="utf-8") as f:
236 json.dump(self.pipeline_config, f, indent=4)
238 def on_enter(self):
239 """
240 Hook called when the page becomes visible.
242 Checks for updated configuration files and refreshes the UI
243 if any changes are detected.
244 """
245 new_config_path = self._find_latest_config()
246 new_pipeline_config = self._load_config_from_path(new_config_path)
248 config_changed = (
249 new_config_path != self.config_path or
250 new_pipeline_config != self.pipeline_config
251 )
253 if config_changed:
254 log.info(f"Config changed, reloading UI: {os.path.basename(new_config_path)}")
255 self.config_path = new_config_path
256 self.pipeline_config = new_pipeline_config
257 self._setup_ui()
258 else:
259 log.debug("Config unchanged, keeping existing UI")
261 # ─────────────────────────────────────────────
262 # NAVIGATION HANDLERS
263 # ─────────────────────────────────────────────
264 def next(self, context):
265 """
266 Proceed to the next page, executing the pipeline if all configurations are valid.
268 Args:
269 context (dict): Shared context.
271 Returns:
272 Page: The next page instance, or self if configuration is incomplete.
273 """
274 # Check for patients requiring revision
275 unsaved_patients = [
276 pid for pid, files in self.pipeline_config.items()
277 if files.get("need_revision", False)
278 ]
280 if unsaved_patients:
281 # Warn the user about unsaved patients
282 from PyQt6.QtWidgets import QMessageBox
283 msg = QMessageBox()
284 msg.setIcon(QMessageBox.Icon.Warning)
285 msg.setWindowTitle(QCoreApplication.translate("PipelineReviewPage", "Configuration Incomplete"))
286 msg.setText(QCoreApplication.translate("PipelineReviewPage", "Some patients still require configuration review."))
287 msg.setInformativeText(
288 QCoreApplication.translate(
289 "PipelineReviewPage",
290 "Please review and save configuration for: {patients}"
291 ).format(patients=", ".join(unsaved_patients))
292 )
293 msg.exec()
294 return self # Stay on current page
296 # Proceed to pipeline execution page
297 if not self.next_page:
298 self.next_page = PipelineExecutionPage(context, self)
299 self.context["history"].append(self.next_page)
300 self.next_page.on_enter()
301 return self.next_page
303 def back(self):
304 """
305 Navigate back to the previous page.
307 If the output folder for the current config does not exist,
308 the config file is deleted to keep the workspace clean.
309 """
310 if os.path.exists(self.config_path):
311 try:
312 config_filename = os.path.basename(self.config_path)
313 config_id = config_filename.split('_config.json')[0]
314 pipeline_dir = os.path.dirname(self.config_path)
315 output_folder_path = os.path.join(pipeline_dir, f"{config_id}_output")
317 if not os.path.exists(output_folder_path):
318 os.remove(self.config_path)
319 log.info(f"Removed unused config file: {self.config_path}")
320 else:
321 log.info(f"Keeping config file: {self.config_path}")
322 except (OSError, IndexError, ValueError) as e:
323 log.error(f"Error processing config file {self.config_path}: {e}")
324 else:
325 log.error(f"Invalid config path: {self.config_path}")
327 if self.previous_page:
328 self.previous_page.on_enter()
329 return self.previous_page
330 return None
332 # ─────────────────────────────────────────────
333 # TRANSLATION
334 # ─────────────────────────────────────────────
335 def _translate_ui(self):
336 """Translate all static text for internationalization."""
337 self.header.setText(QCoreApplication.translate("PipelineReviewPage", "Pipeline Configuration Review"))
338 self.config_info.setText(
339 QCoreApplication.translate("PipelineReviewPage", "Reviewing: {0}").format(os.path.basename(self.config_path))
340 )
341 self.info_label.setText(
342 QCoreApplication.translate(
343 "PipelineReviewPage",
344 "<strong>Click</strong> a frame to review file selections. <strong>Save</strong> yellow frames after selection."
345 )
346 )