""" Main application window for the MRI sequence interpreter GUI. Layout ────── [Button bar] ← QPushButton row, never clips text ┌─────────────┬────────────────────────────┬──────────────────────┐ │ Left │ Centre │ Right │ │ ─────────── │ [Sequence scheme] │ Block Details │ │ Metadata │ ───────────────────────── │ Warnings │ │ HW delays │ Waveform plots │ Sync XML │ │ Warnings │ │ POST JSON │ │ │ [Block table — hidden] │ Log │ └─────────────┴────────────────────────────┴──────────────────────┘ Design rules ──────────── • All heavy work runs in QThread workers — UI thread stays responsive. • Block table is hidden by default; toggled with the "Blocks" button. • Log output lives in the right panel (PreviewPanel Log tab). • Sequence load state is shown as a coloured status indicator in the button bar. • hw_config.json is never silently overwritten. """ from __future__ import annotations import logging import os from datetime import datetime from PySide6.QtCore import Qt, QSize from PySide6.QtGui import QFont, QColor from PySide6.QtWidgets import ( QApplication, QMainWindow, QWidget, QSplitter, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout, QLabel, QListWidget, QListWidgetItem, QFrame, QPushButton, QProgressBar, QStatusBar, QMessageBox, QScrollArea, QSizePolicy, QFileDialog, ) from seq_interp.src.gui.adapters import ( build_block_rows, seq_metadata, validate_timing, find_block_at_time, ) from seq_interp.src.gui.block_table import BlockTable from seq_interp.src.gui.controls_panel import DelayControlsPanel from seq_interp.src.gui.plot_panel import PlotPanel from seq_interp.src.gui.preview_panel import PreviewPanel from seq_interp.src.gui.scheme_panel import SchemePanel, system_is_dark from seq_interp.src.gui.workers import ( LoadInterpWorker, SyncOnlyWorker, ExportWorker, XmlPreviewWorker, ) # ── loading-state style maps (light / dark) ─────────────────────────────────── _STATE_LIGHT: dict[str, tuple[str, str]] = { "idle": ("#757575", "●"), "selected": ("#1565c0", "●"), "loading": ("#e65100", "⟳"), "loaded": ("#2e7d32", "✓"), "failed": ("#c62828", "✗"), } _STATE_DARK: dict[str, tuple[str, str]] = { "idle": ("#9e9e9e", "●"), "selected": ("#64b5f6", "●"), "loading": ("#ff9800", "⟳"), "loaded": ("#81c784", "✓"), "failed": ("#ef9a9a", "✗"), } class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("MRI Sequence Interpreter") self.setMinimumSize(900, 600) # Size relative to available screen so we never start off-screen on # small monitors (e.g. 1366×768 laptops). Cap at 1600×940. screen = QApplication.primaryScreen() if screen is not None: ag = screen.availableGeometry() w = min(1600, max(900, int(ag.width() * 0.92))) h = min(940, max(600, int(ag.height() * 0.90))) self.resize(w, h) else: self.resize(1600, 940) # ── application state ────────────────────────────────────────── self._seq_path: str | None = None self._hw_config_path: str | None = None self._output_dir: str | None = None self._seq_data: dict | None = None self._sync_data: dict | None = None self._hw = None self._block_rows: list = [] self._worker = None self._xml_preview_worker = None self._pending_table_select: int | None = None # remembered when table hidden # ── build UI ─────────────────────────────────────────────────── central = QWidget() self.setCentralWidget(central) root_layout = QVBoxLayout(central) root_layout.setContentsMargins(0, 0, 0, 0) root_layout.setSpacing(0) root_layout.addWidget(self._build_button_bar()) root_layout.addWidget(self._build_seq_status_bar()) root_layout.addWidget(self._build_main_splitter(), stretch=1) self._build_statusbar() self._setup_logging() # ================================================================== # # Button bar (replaces QToolBar — never clips text) # # ================================================================== # def _build_button_bar(self) -> QWidget: bar = QWidget() bar.setObjectName("ButtonBar") bar.setStyleSheet( "#ButtonBar { background: palette(window); border-bottom: 1px solid palette(mid); }" ) lay = QHBoxLayout(bar) lay.setContentsMargins(6, 4, 6, 4) lay.setSpacing(4) def sep() -> QFrame: f = QFrame() f.setFrameShape(QFrame.VLine) f.setFrameShadow(QFrame.Sunken) f.setFixedWidth(2) return f def btn(label: str, tip: str, slot=None, enabled: bool = True) -> QPushButton: b = QPushButton(label) b.setToolTip(tip) b.setEnabled(enabled) b.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) if slot: b.clicked.connect(slot) lay.addWidget(b) return b self._btn_load_seq = btn("📂 Load .seq", "Open Pulseq .seq file", self._load_seq) self._btn_load_hw = btn("⚙ HW Config", "Load hardware constraints JSON", self._load_hw_config) self._btn_out_dir = btn("📁 Output Dir", "Choose output directory", self._choose_output_dir) lay.addWidget(sep()) self._btn_run = btn("▶ Run", "Run interpretation pipeline", self._run, enabled=False) self._btn_export = btn("💾 Export", "Export all artifacts to output dir", self._export, enabled=False) lay.addWidget(sep()) self._btn_fit = btn("🔍 Fit All", "Fit all plots to data", lambda: self._plots.fit_all()) self._btn_blocks = btn("📋 Blocks ▼", "Toggle block table open/closed", self._toggle_table) lay.addStretch() self._progress = QProgressBar() self._progress.setRange(0, 0) self._progress.setFixedWidth(120) self._progress.setVisible(False) lay.addWidget(self._progress) return bar def _build_seq_status_bar(self) -> QWidget: bar = QWidget() bar.setObjectName("SeqStatus") bar.setStyleSheet( "#SeqStatus { background: palette(window); border-bottom: 1px solid palette(mid); }" ) lay = QHBoxLayout(bar) lay.setContentsMargins(8, 2, 8, 2) self._seq_status = QLabel(" ● No file selected") self._seq_status.setFont(QFont("Arial", 9)) self._seq_status.setStyleSheet("color: #9e9e9e;") lay.addWidget(self._seq_status) lay.addStretch() return bar # ================================================================== # # Main three-panel splitter # # ================================================================== # def _build_main_splitter(self) -> QSplitter: root = QSplitter(Qt.Horizontal) root.addWidget(self._build_left_panel()) root.addWidget(self._build_centre_panel()) self._preview = PreviewPanel() self._preview.setMinimumWidth(260) self._preview.setMaximumWidth(480) root.addWidget(self._preview) root.setSizes([280, 1040, 280]) # Wire signals self._plots.blockClicked.connect(self._on_block_from_plot) self._plots.timeHovered.connect(self._on_hover) self._table.blockSelected.connect(self._on_block_from_table) self._scheme.blockClicked.connect(self._on_block_from_scheme) self._controls.rerun.connect(self._rerun) self._controls.reloadConfig.connect(self._reload_hw_config) return root # ── left panel ──────────────────────────────────────────────────────────── def _build_left_panel(self) -> QScrollArea: left = QWidget() left.setMinimumWidth(220) left.setMaximumWidth(360) lay = QVBoxLayout(left) lay.setContentsMargins(4, 4, 4, 4) lay.setSpacing(6) # Metadata meta_grp = QGroupBox("Sequence Metadata") meta_form = QFormLayout(meta_grp) meta_form.setFieldGrowthPolicy(QFormLayout.AllNonFixedFieldsGrow) self._meta_labels: dict[str, QLabel] = {} for key in [ "Total blocks (orig)", "RF blocks", "ADC blocks", "Grad blocks", "RF raster (µs)", "Grad raster (µs)", "ADC raster (ns)", "Block raster (µs)", "RF delay (ns)", "TR delay (ns)", "Start delay (µs)", "Min block dur (ns)", "Gamma (MHz/T)", "RF scale", ]: lbl = QLabel("—") lbl.setFont(QFont("Courier New", 9)) meta_form.addRow(QLabel(f"{key}:"), lbl) self._meta_labels[key] = lbl lay.addWidget(meta_grp) # Delay controls self._controls = DelayControlsPanel() lay.addWidget(self._controls) # Warnings (compact, left-panel copy) warn_grp = QGroupBox("Warnings") warn_lay = QVBoxLayout(warn_grp) self._warn_list = QListWidget() self._warn_list.setFont(QFont("Arial", 9)) self._warn_list.setMaximumHeight(110) self._warn_list.setStyleSheet( "QListWidget { color: palette(text); background: palette(alternateBase); } " "QListWidget::item { padding: 2px; }" ) warn_lay.addWidget(self._warn_list) lay.addWidget(warn_grp) lay.addStretch() scroll = QScrollArea() scroll.setWidget(left) scroll.setWidgetResizable(True) scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) return scroll # ── centre panel ────────────────────────────────────────────────────────── def _build_centre_panel(self) -> QSplitter: vsplit = QSplitter(Qt.Vertical) self._centre_vsplit = vsplit # kept for table-toggle height adjustment # Scheme (always visible, compact) self._scheme = SchemePanel() vsplit.addWidget(self._scheme) # Plots self._plots = PlotPanel() vsplit.addWidget(self._plots) # Block table (hidden by default) self._table_container = QWidget() tc_lay = QVBoxLayout(self._table_container) tc_lay.setContentsMargins(0, 0, 0, 0) self._table = BlockTable() tc_lay.addWidget(self._table) self._table_container.setVisible(False) vsplit.addWidget(self._table_container) vsplit.setSizes([64, 700, 0]) vsplit.setCollapsible(2, True) return vsplit # ================================================================== # # Status bar # # ================================================================== # def _build_statusbar(self) -> None: sb = QStatusBar() self.setStatusBar(sb) self._status_lbl = QLabel("Ready") sb.addWidget(self._status_lbl) def _setup_logging(self) -> None: log_dir = os.path.join( os.path.dirname(os.path.dirname( os.path.dirname(os.path.abspath(__file__)) )), "log", ) os.makedirs(log_dir, exist_ok=True) ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") handler = logging.FileHandler( os.path.join(log_dir, f"gui_{ts}.log"), encoding="utf-8" ) handler.setFormatter( logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") ) logging.getLogger().addHandler(handler) logging.getLogger().setLevel(logging.INFO) # ================================================================== # # File / directory actions # # ================================================================== # def _load_seq(self) -> None: path, _ = QFileDialog.getOpenFileName( self, "Open Pulseq sequence", "", "Pulseq files (*.seq);;All files (*)" ) if not path: return self._seq_path = path name = os.path.basename(path) self._set_seq_state("selected", name) self._btn_run.setEnabled(True) self._log(f"Sequence selected: {path}") def _load_hw_config(self) -> None: path, _ = QFileDialog.getOpenFileName( self, "Open HW config", "", "JSON files (*.json);;All files (*)" ) if not path: return self._hw_config_path = path self._log(f"HW config: {path}") def _choose_output_dir(self) -> None: path = QFileDialog.getExistingDirectory(self, "Choose output directory") if path: self._output_dir = path self._log(f"Output dir: {path}") def _reload_hw_config(self) -> None: if self._hw and self._hw_config_path: self._hw.load_from_json(self._hw_config_path) self._controls.load_from_hw(self._hw) self._log("HW config reloaded from file") elif self._hw: from seq_interp.src.hardware.constraints import HardwareConstraints self._hw = HardwareConstraints() self._controls.load_from_hw(self._hw) self._log("HW config reset to class defaults") # ================================================================== # # Table toggle # # ================================================================== # def _toggle_table(self) -> None: visible = not self._table_container.isVisible() self._table_container.setVisible(visible) self._btn_blocks.setText("📋 Blocks ▲" if visible else "📋 Blocks ▼") if visible: # Splitter stores size 0 for the hidden widget; explicitly give it # 200 px when revealed by taking from the plots section. sizes = self._centre_vsplit.sizes() if sizes[2] < 100: target = 200 plots_h = max(100, sizes[1] - target) self._centre_vsplit.setSizes([sizes[0], plots_h, target]) if self._pending_table_select is not None: self._table.select_by_sync_index(self._pending_table_select) self._pending_table_select = None # ================================================================== # # Run / rerun / export # # ================================================================== # def _run(self) -> None: if not self._seq_path: return name = os.path.basename(self._seq_path) self._set_seq_state("loading", name) self._start_busy("Loading and interpreting…") overrides = self._controls.get_overrides() if self._hw else {} self._worker = LoadInterpWorker( self._seq_path, hw_config_path=self._hw_config_path, hw_overrides=overrides, ) self._worker.log_msg.connect(self._log) self._worker.finished.connect(self._on_interp_finished) self._worker.error.connect(self._on_worker_error) self._worker.start() def _rerun(self) -> None: if self._seq_data is None: self._run() return self._start_busy("Re-synchronizing…") self._worker = SyncOnlyWorker( self._seq_data, hw_config_path=self._hw_config_path, hw_overrides=self._controls.get_overrides(), ) self._worker.log_msg.connect(self._log) self._worker.finished.connect(self._on_sync_finished) self._worker.error.connect(self._on_worker_error) self._worker.start() def _export(self) -> None: if self._seq_data is None or self._sync_data is None: return out = self._output_dir if not out: out = QFileDialog.getExistingDirectory(self, "Choose output directory") if not out: return self._output_dir = out self._start_busy("Exporting artifacts…") self._worker = ExportWorker( self._seq_data, self._sync_data, self._hw, out ) self._worker.log_msg.connect(self._log) self._worker.finished.connect(self._on_export_finished) self._worker.error.connect(self._on_worker_error) self._worker.start() # ================================================================== # # Worker callbacks # # ================================================================== # def _on_interp_finished(self, seq_data: dict, sync_data: dict, hw) -> None: self._seq_data = seq_data self._sync_data = sync_data self._hw = hw self._stop_busy() self._apply_results(seq_data, sync_data, hw) def _on_sync_finished(self, sync_data: dict, hw) -> None: self._sync_data = sync_data self._hw = hw self._stop_busy() self._apply_results(self._seq_data, sync_data, hw) def _on_export_finished(self, output_dir: str, xml_text: str, post_text: str) -> None: self._stop_busy() self._preview.set_xml_text(xml_text) self._preview.set_post_json_text(post_text) self._log(f"Export complete → {output_dir}") QMessageBox.information( self, "Export complete", f"Artifacts written to:\n{output_dir}" ) def _on_worker_error(self, msg: str) -> None: name = os.path.basename(self._seq_path or "") self._set_seq_state("failed", name, msg[:80]) self._stop_busy() self._log(f"ERROR: {msg}", error=True) self._preview.add_error(msg) QMessageBox.critical(self, "Error", msg) # ================================================================== # # Results display # # ================================================================== # def _apply_results(self, seq_data: dict, sync_data: dict, hw) -> None: # Metadata labels meta = seq_metadata(seq_data, hw) for key, lbl in self._meta_labels.items(): lbl.setText(str(meta.get(key, "—"))) # Controls — populate once on first load, don't overwrite user edits if not self._controls._defaults: self._controls.load_from_hw(hw) # Warnings warnings = validate_timing(hw, seq_data, sync_data) self._refresh_warnings(warnings) self._preview.set_warnings(warnings) # Block table and scheme self._block_rows = build_block_rows(seq_data, sync_data) self._table.load_rows(self._block_rows) self._scheme.load_rows(self._block_rows) # Plots self._plots.plot_all(seq_data, sync_data) # XML / POST preview (async, non-blocking) pw = XmlPreviewWorker(sync_data, hw) pw.finished.connect( lambda xml, post: ( self._preview.set_xml_text(xml), self._preview.set_post_json_text(post), ) ) pw.error.connect(lambda e: self._log(f"XML preview: {e}", error=True)) pw.start() self._xml_preview_worker = pw # keep reference alive # Seq status blocks = seq_data.get("blocks", []) total_s = sum(sync_data.get("blocks_duration", [])) parts = [f"{len(blocks)} blocks"] if any("RF" in b.get("type", []) for b in blocks): parts.append("RF") if any(b.get("has_adc") for b in blocks): parts.append("ADC") if any("GRAD" in b.get("type", []) for b in blocks): parts.append("Grad") if total_s >= 1e-3: parts.append(f"{total_s * 1e3:.2f} ms") else: parts.append(f"{total_s * 1e6:.1f} µs") name = os.path.basename(self._seq_path or "") self._set_seq_state("loaded", name, " · ".join(parts)) self._act_enabled(run=True, export=True) self._status_lbl.setText( f"{len(blocks)} input blocks → {sync_data['number_of_blocks']} sync blocks" ) def _refresh_warnings(self, warnings: list[str]) -> None: self._warn_list.clear() warn_color = QColor("#ff9800" if system_is_dark() else "#e65100") for w in warnings: item = QListWidgetItem(f"⚠ {w}") item.setForeground(warn_color) self._warn_list.addItem(item) # ================================================================== # # Block selection sync # # ================================================================== # def _on_block_from_plot(self, sync_index: int) -> None: self._select_block(sync_index, source="plot") def _on_block_from_table(self, sync_index: int) -> None: self._select_block(sync_index, source="table") def _on_block_from_scheme(self, sync_index: int) -> None: self._select_block(sync_index, source="scheme") def _select_block(self, sync_index: int, source: str) -> None: row = self._table.row_for_sync_index(sync_index) self._preview.show_block_details(row) self._scheme.select_block(sync_index) if source != "plot": self._plots.highlight_block(sync_index) if self._table_container.isVisible(): if source != "table": self._table.select_by_sync_index(sync_index) else: # Remember for when the table is opened self._pending_table_select = sync_index # ================================================================== # # Status / log helpers # # ================================================================== # def _set_seq_state(self, state: str, name: str = "", detail: str = "") -> None: _state_map = _STATE_DARK if system_is_dark() else _STATE_LIGHT color, icon = _state_map.get(state, ("#9e9e9e", "●")) text = f" {icon} {name}" if name else f" {icon} No file selected" if detail: text += f" — {detail}" self._seq_status.setStyleSheet(f"color: {color}; font-weight: bold;") self._seq_status.setText(text) def _on_hover(self, t_s: float, channel: str, value: float) -> None: block = find_block_at_time(t_s, self._block_rows) if self._block_rows else None blk = f" block #{block.sync_index} [{block.block_type}]" if block else "" self._status_lbl.setText( f"t = {t_s * 1e6:.4f} µs {channel} = {value:.4g}{blk}" ) def _log(self, msg: str, error: bool = False) -> None: self._preview.append_log(msg, error=error) if error: logging.error(msg) else: logging.info(msg) def _start_busy(self, tip: str) -> None: self._progress.setVisible(True) self._status_lbl.setText(tip) self._act_enabled(run=False, export=False) def _stop_busy(self) -> None: self._progress.setVisible(False) def _act_enabled(self, run: bool, export: bool) -> None: self._btn_run.setEnabled(run and bool(self._seq_path)) self._btn_export.setEnabled(export and self._seq_data is not None)