Source code for mafw.steering_gui.views.globals_editor

#  Copyright 2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""Expose the globals editor fields required by the steering GUI.

:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Provide input fields for the analysis metadata and flags managed by the controller.
"""

from __future__ import annotations

from PySide6.QtCore import Qt, QTimer, Signal
from PySide6.QtGui import QColor, QDrag, QDropEvent
from PySide6.QtWidgets import (
    QAbstractItemView,
    QCheckBox,
    QFormLayout,
    QHBoxLayout,
    QHeaderView,
    QLineEdit,
    QPushButton,
    QTableWidget,
    QTableWidgetItem,
    QToolButton,
    QVBoxLayout,
    QWidget,
)

from mafw.processor.path_types import PathRole
from mafw.steering_gui.utils import block_signals
from mafw.steering_gui.views.path_editor import PathEditor


[docs] class _EnvironmentFilesTable(QTableWidget): """Editable ordered list of dotenv paths and their load status.""" paths_changed = Signal(list) rows_moved = Signal() def __init__(self, parent: QWidget | None = None) -> None: """Create the three-column dotenv table with internal row reordering.""" super().__init__(0, 3, parent) self._dragged_rows: list[int] = [] self._internal_drag_active = False self._path_change_timer = QTimer(self) self._path_change_timer.setSingleShot(True) self._path_change_timer.setInterval(250) self._path_change_timer.timeout.connect(self._emit_paths_changed) self.setHorizontalHeaderLabels(['Dotenv file', 'Valid', 'Loaded']) self.horizontalHeader().setStretchLastSection(False) self.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch) self.setColumnWidth(1, 72) self.setColumnWidth(2, 72) self.verticalHeader().setVisible(False) self.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) self.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection) self.setDragEnabled(True) self.setAcceptDrops(True) self.setDropIndicatorShown(True) self.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove) self.setDragDropOverwriteMode(False) self.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
[docs] def paths(self) -> list[str]: """Return configured path strings in their current table order.""" paths: list[str] = [] for row in range(self.rowCount()): editor = self.cellWidget(row, 0) paths.append(editor.value() if isinstance(editor, PathEditor) else '') return paths
[docs] def set_files(self, paths: list[str], statuses: list[dict[str, object]]) -> None: """Replace table rows from the steering path list and current validation state.""" self.blockSignals(True) try: self.setRowCount(0) for index, path in enumerate(paths): status = statuses[index] if index < len(statuses) else {} self._append_row( path, valid=bool(status.get('valid', False)), loaded=bool(status.get('loaded', False)), ) finally: self.blockSignals(False)
[docs] def set_statuses(self, statuses: list[dict[str, object]]) -> None: """Refresh the validity and loaded flags without replacing path editors.""" for row in range(self.rowCount()): status = statuses[row] if row < len(statuses) else {} self._set_flag(row, 1, bool(status.get('valid', False))) self._set_flag(row, 2, bool(status.get('loaded', False))) self._set_row_error_style(row, not bool(status.get('valid', False)))
[docs] def add_row(self) -> PathEditor: """Append an empty path and return its focused editor.""" editor = self._append_row('', valid=False, loaded=False) self.setCurrentCell(self.rowCount() - 1, 0) editor.setFocus() self.paths_changed.emit(self.paths()) return editor
[docs] def remove_selected_rows(self) -> None: """Remove all selected rows and report the new ordered path list.""" for row in sorted({index.row() for index in self.selectionModel().selectedRows()}, reverse=True): self.removeRow(row) self.paths_changed.emit(self.paths())
[docs] def startDrag(self, supported_actions: Qt.DropAction) -> None: # noqa: N802 """Start an internal drag while retaining selected row identities. ``supported_actions`` uses ``Qt.DropAction`` from PySide6. """ self._dragged_rows = sorted({index.row() for index in self.selectionModel().selectedRows()}) if not self._dragged_rows: return model = self.model() if model is None: return drag = QDrag(self) drag.setMimeData(model.mimeData(self.selectedIndexes())) self._internal_drag_active = True try: drag.exec(supported_actions, Qt.DropAction.MoveAction) finally: self._internal_drag_active = False self._dragged_rows.clear()
[docs] def dropEvent(self, event: QDropEvent) -> None: # noqa: N802 """Move selected rows as a block and preserve their path and status cells.""" if not self._internal_drag_active: event.ignore() return rows = self._dragged_rows position = event.position().toPoint() target_index = self.indexAt(position) if not target_index.isValid(): target = self.rowCount() else: target = target_index.row() if position.y() >= self.visualRect(target_index).center().y(): target += 1 self._move_rows(rows, target) event.setDropAction(Qt.DropAction.MoveAction) event.accept()
[docs] def _move_rows(self, selected_rows: list[int], target: int) -> None: """Rebuild the table in the requested order without losing cell widgets.""" records = self._records() moving = [records[row] for row in selected_rows] remaining = [record for index, record in enumerate(records) if index not in selected_rows] insertion = target - sum(row < target for row in selected_rows) insertion = max(0, min(insertion, len(remaining))) reordered = remaining[:insertion] + moving + remaining[insertion:] self.set_files( [record[0] for record in reordered], [{'valid': record[1], 'loaded': record[2]} for record in reordered], ) self.rows_moved.emit()
[docs] def _records(self) -> list[tuple[str, bool, bool]]: """Return path and status values for each current row.""" records = [] for row in range(self.rowCount()): editor = self.cellWidget(row, 0) valid_item = self.item(row, 1) loaded_item = self.item(row, 2) valid = valid_item is not None and valid_item.checkState() == Qt.CheckState.Checked loaded = loaded_item is not None and loaded_item.checkState() == Qt.CheckState.Checked records.append((editor.value() if isinstance(editor, PathEditor) else '', valid, loaded)) return records
[docs] def _append_row(self, path: str, *, valid: bool, loaded: bool) -> PathEditor: """Create a path widget and read-only bool cells for one row.""" row = self.rowCount() self.insertRow(row) editor = PathEditor(self, role=PathRole.INPUT_FILE) editor.set_value(path) editor.value_changed.connect(lambda _value: self._path_change_timer.start()) editor.editing_finished.connect(self._emit_paths_changed) editor.focused.connect(lambda: self.selectRow(row)) self.setCellWidget(row, 0, editor) self._set_flag(row, 1, valid) self._set_flag(row, 2, loaded) self._set_row_error_style(row, not valid) return editor
[docs] def _emit_paths_changed(self) -> None: """Validate edited paths after a short pause in typing.""" self._path_change_timer.stop() self.paths_changed.emit(self.paths())
[docs] def _set_flag(self, row: int, column: int, value: bool) -> None: """Set a check indicator that users cannot toggle manually.""" item = self.item(row, column) if item is None: item = QTableWidgetItem() self.setItem(row, column, item) item.setCheckState(Qt.CheckState.Checked if value else Qt.CheckState.Unchecked) item.setFlags(Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable)
[docs] def _set_row_error_style(self, row: int, invalid: bool) -> None: """Paint invalid rows red while leaving valid rows with the default palette.""" color = QColor('#ffd6d6') if invalid else QColor('white') for column in (1, 2): item = self.item(row, column) if item is not None: item.setBackground(color) editor = self.cellWidget(row, 0) if isinstance(editor, PathEditor): editor.setStyleSheet('QLineEdit { background-color: #ffd6d6; }' if invalid else '')
[docs] class GlobalsEditor(QWidget): """Editor that exposes analysis metadata and global flags without touching the builder.""" analysis_name_changed = Signal(str) description_changed = Signal(str) new_only_changed = Signal(bool) create_standard_tables_changed = Signal(bool) env_files_changed = Signal(list) reload_environment_requested = Signal() def __init__(self, parent: QWidget | None = None) -> None: super().__init__(parent) self.analysis_name_edit = QLineEdit() self.description_edit = QLineEdit() self.new_only_checkbox = QCheckBox('Run only new data') self.new_only_checkbox.setChecked(False) self.create_standard_tables_checkbox = QCheckBox('Create standard tables') self.create_standard_tables_checkbox.setChecked(False) self.env_files_table = _EnvironmentFilesTable(self) self.add_env_file_button = QToolButton() self.add_env_file_button.setText('+') self.add_env_file_button.setToolTip('Add a dotenv file') self.remove_env_file_button = QToolButton() self.remove_env_file_button.setText('−') self.remove_env_file_button.setToolTip('Remove selected dotenv files') self.remove_env_file_button.setEnabled(False) self.reload_environment_button = QPushButton('Apply / Reload environment') self.analysis_name_edit.textChanged.connect(self.analysis_name_changed) self.description_edit.textChanged.connect(self.description_changed) self.new_only_checkbox.toggled.connect(self.new_only_changed) self.create_standard_tables_checkbox.toggled.connect(self.create_standard_tables_changed) self.env_files_table.paths_changed.connect(self.env_files_changed) self.env_files_table.rows_moved.connect(self._emit_env_files) self.env_files_table.itemSelectionChanged.connect(self._update_remove_button) self.add_env_file_button.clicked.connect(self._add_env_file) self.remove_env_file_button.clicked.connect(self.env_files_table.remove_selected_rows) self.reload_environment_button.clicked.connect(self.reload_environment_requested) layout = QFormLayout() layout.addRow('Analysis Name', self.analysis_name_edit) layout.addRow('Description', self.description_edit) layout.addRow('', self.new_only_checkbox) layout.addRow('', self.create_standard_tables_checkbox) env_files_row = QHBoxLayout() env_files_row.addWidget(self.env_files_table, 1) env_file_buttons = QVBoxLayout() env_file_buttons.addWidget(self.add_env_file_button) env_file_buttons.addWidget(self.remove_env_file_button) env_file_buttons.addStretch(1) env_files_row.addLayout(env_file_buttons) layout.addRow('Environment files', env_files_row) layout.addRow('', self.reload_environment_button) self.setLayout(layout)
[docs] def set_data( self, analysis_name: str | None, description: str | None, new_only: bool | None, create_standard_tables: bool | None, env_files: list[str] | None = None, env_file_statuses: list[dict[str, object]] | None = None, ) -> None: """Populate the widgets and avoid re-triggering their change signals.""" with block_signals( self.analysis_name_edit, self.description_edit, self.new_only_checkbox, self.create_standard_tables_checkbox, ): self.analysis_name_edit.setText(analysis_name or '') self.description_edit.setText(description or '') self.new_only_checkbox.setChecked(bool(new_only)) self.create_standard_tables_checkbox.setChecked(bool(create_standard_tables)) self.env_files_table.set_files(env_files or [], env_file_statuses or []) self._update_remove_button()
[docs] def set_environment_statuses(self, statuses: list[dict[str, object]]) -> None: """Refresh table flags after validation or environment reload.""" self.env_files_table.set_statuses(statuses)
[docs] def current_values(self) -> tuple[str, str, bool, bool]: """Return the current UI values without normalizing them.""" return ( self.analysis_name_edit.text(), self.description_edit.text(), self.new_only_checkbox.isChecked(), self.create_standard_tables_checkbox.isChecked(), )
[docs] def _emit_env_files(self) -> None: """Emit the current ordered paths after a drag operation.""" self.env_files_changed.emit(self.env_files_table.paths())
[docs] def _add_env_file(self) -> None: """Add an empty path row and focus its line editor.""" self.env_files_table.add_row()
[docs] def _update_remove_button(self) -> None: """Enable removal only while at least one complete row is selected.""" self.remove_env_file_button.setEnabled(bool(self.env_files_table.selectionModel().selectedRows()))