# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Processor parameter editor for steering GUI pipelines.
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Display and edit processor parameters, replica metadata, and overrides in the GUI.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any, get_args, get_origin
from PySide6.QtCore import (
QAbstractItemModel,
QAbstractTableModel,
QModelIndex,
QPersistentModelIndex,
QPoint,
Qt,
QTimer,
Signal,
)
from PySide6.QtGui import QColor, QDrag, QDropEvent, QFont, QIcon, QKeyEvent
from PySide6.QtWidgets import (
QAbstractItemDelegate,
QAbstractItemView,
QApplication,
QCheckBox,
QComboBox,
QDialog,
QDoubleSpinBox,
QFormLayout,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QMenu,
QMessageBox,
QPushButton,
QSizePolicy,
QSpinBox,
QStackedWidget,
QStyledItemDelegate,
QStyleOptionViewItem,
QTableView,
QTableWidget,
QTableWidgetItem,
QToolButton,
QVBoxLayout,
QWidget,
)
from mafw.processor.path_types import (
_list_element_type,
is_path_type,
path_role_of,
)
from mafw.steering.models import ParameterConfig, ParameterSchemaStatus, ParameterSource, ProcessorConfig
from mafw.steering_gui.utils import FLOAT_DECIMALS, FLOAT_RANGE, INT_RANGE, block_signals
from mafw.steering_gui.views.path_editor import PathEditor
from mafw.storage import LogicalFileReference
from mafw.tools.regexp import parse_processor_name
_ACTIVE_COLUMN_LABEL = 'Active'
"""Header label for the parameter active toggle column."""
_COLUMNS = (
_ACTIVE_COLUMN_LABEL,
'Name',
'Value',
'Type',
'Source',
'Status',
)
"""Column labels for the processor parameter table."""
_RESOURCE_DIR = Path(__file__).resolve().parent.parent / 'resources'
_STATUS_ICON_FILES: dict[ParameterSchemaStatus, str] = {
ParameterSchemaStatus.DEPRECATED: 'mynaui--danger-triangle.svg',
ParameterSchemaStatus.OK: 'ix--namur-ok.svg',
ParameterSchemaStatus.UNKNOWN: 'circum--square-question.svg',
ParameterSchemaStatus.NEW: 'fluent-emoji-high-contrast--new-button.svg',
}
[docs]
class _ParameterRoles:
"""Custom roles used to expose parameter metadata to delegates."""
PARAMETER = Qt.ItemDataRole.UserRole + 1
TYPE = Qt.ItemDataRole.UserRole + 2
VALUE = Qt.ItemDataRole.UserRole + 3
PENDING = Qt.ItemDataRole.UserRole + 4
BROWSING = Qt.ItemDataRole.UserRole + 5
[docs]
def _set_list_item_value(
model: QAbstractItemModel,
index: QModelIndex | QPersistentModelIndex,
value: Path | LogicalFileReference | str,
) -> None:
"""Keep a list row readable while retaining its selected path type."""
display_value = ListEditDialog._display_value(value)
model.setData(index, display_value, Qt.ItemDataRole.EditRole)
model.setData(index, value, _ParameterRoles.VALUE)
[docs]
def _remove_pending_item_later(table: QTableWidget, index: QModelIndex | QPersistentModelIndex) -> None:
"""Remove an empty new row after Qt has finished closing its active editor."""
persistent_index = QPersistentModelIndex(index)
def remove_pending_item() -> None:
if persistent_index.isValid() and persistent_index.data(_ParameterRoles.PENDING):
table.removeRow(persistent_index.row())
QTimer.singleShot(0, remove_pending_item)
[docs]
def _is_empty_list_editor(editor: QWidget) -> bool:
"""Check whether a newly added list row still has no selected value."""
if isinstance(editor, PathLogicalUnionEditor):
if editor._selector.currentIndex() == 0:
return not editor._path.value().strip()
return not str(editor._logical.value()).strip()
if isinstance(editor, PathEditor):
return not editor.value().strip()
if isinstance(editor, LogicalReferenceEditor):
return not str(editor.value()).strip()
if isinstance(editor, QLineEdit):
return not editor.text().strip()
return False
[docs]
def _commit_list_editor_value(
delegate: ParameterDelegate,
editor: QWidget,
model: QAbstractItemModel,
index: QModelIndex | QPersistentModelIndex,
value: Any,
) -> None:
"""Commit a list row, dropping a newly added row dismissed while empty."""
if index.data(_ParameterRoles.PARAMETER) is not None:
model.setData(index, value, Qt.ItemDataRole.EditRole)
return
if index.data(_ParameterRoles.BROWSING):
return
if (
not index.data(_ParameterRoles.PENDING)
and _is_empty_list_editor(editor)
and isinstance(index.data(_ParameterRoles.VALUE), (Path, LogicalFileReference))
):
# Browse commits the selected value before the modal dialog returns. Do
# not let the now-empty, possibly closing editor overwrite that choice.
return
if index.data(_ParameterRoles.PENDING) and _is_empty_list_editor(editor):
table = delegate.parent()
if isinstance(table, QTableWidget):
_remove_pending_item_later(table, index)
return
model.setData(index, None, _ParameterRoles.PENDING)
if isinstance(value, (Path, LogicalFileReference)):
_set_list_item_value(model, index, value)
else:
model.setData(index, value, Qt.ItemDataRole.EditRole)
[docs]
@dataclass
class _ParameterRow:
"""Container describing a single parameter table row."""
config: ParameterConfig
order: int
[docs]
class ParameterTableModel(QAbstractTableModel):
"""Qt table model exposing processor parameters for editing."""
parameter_value_changed = Signal(str, object)
parameter_active_changed = Signal(str, bool, object)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._rows: list[_ParameterRow] = []
self._inherited_font = QFont()
self._inherited_font.setItalic(True)
self._warning_color = QColor('red')
self._status_icons = {
status: QIcon(str(_RESOURCE_DIR / filename)) for status, filename in _STATUS_ICON_FILES.items()
}
self._all_row_roles = [
Qt.ItemDataRole.DisplayRole,
Qt.ItemDataRole.EditRole,
Qt.ItemDataRole.CheckStateRole,
Qt.ItemDataRole.DecorationRole,
Qt.ItemDataRole.FontRole,
Qt.ItemDataRole.ForegroundRole,
]
[docs]
def set_parameters(self, parameters: Iterable[ParameterConfig]) -> None:
"""Replace the rows with the provided parameters."""
self.beginResetModel()
self._rows = [_ParameterRow(config=param, order=index) for index, param in enumerate(parameters)]
self.endResetModel()
def rowCount(
self,
parent: QModelIndex | QPersistentModelIndex = QModelIndex(),
) -> int: # noqa: N802
if parent.isValid():
return 0
return len(self._rows)
def columnCount(
self,
parent: QModelIndex | QPersistentModelIndex = QModelIndex(),
) -> int: # noqa: N802
if parent.isValid():
return 0
return len(_COLUMNS)
def headerData(
self,
section: int,
orientation: Qt.Orientation,
role: int = Qt.ItemDataRole.DisplayRole,
) -> object:
if orientation == Qt.Orientation.Horizontal and role == Qt.ItemDataRole.DisplayRole:
if 0 <= section < len(_COLUMNS):
return _COLUMNS[section]
return None
def data(
self,
index: QModelIndex | QPersistentModelIndex,
role: int = Qt.ItemDataRole.DisplayRole,
) -> object:
if not index.isValid():
return None
row = index.row()
col = index.column()
if row < 0 or row >= len(self._rows):
return None
config = self._rows[row].config
if role == Qt.ItemDataRole.ToolTipRole:
return config.help or None
if role == _ParameterRoles.PARAMETER:
return config
if role == _ParameterRoles.TYPE:
return self._effective_type(config)
if role == Qt.ItemDataRole.FontRole and config.source == ParameterSource.INHERITED:
return self._inherited_font
if (
role == Qt.ItemDataRole.ForegroundRole
and isinstance(config.status, ParameterSchemaStatus)
and config.status in (ParameterSchemaStatus.UNKNOWN, ParameterSchemaStatus.DEPRECATED)
):
return self._warning_color
if role == Qt.ItemDataRole.CheckStateRole and col == 0:
return Qt.CheckState.Checked if config.active_override else Qt.CheckState.Unchecked
if role == Qt.ItemDataRole.CheckStateRole and col == 2 and self._effective_type(config) is bool:
return Qt.CheckState.Checked if bool(config.value) else Qt.CheckState.Unchecked
if role == Qt.ItemDataRole.DecorationRole and col == 5:
icon = self._status_icons.get(config.status)
if icon and not icon.isNull():
return icon
if role in (Qt.ItemDataRole.DisplayRole, Qt.ItemDataRole.EditRole):
if col == 2 and self._effective_type(config) is bool:
if role == Qt.ItemDataRole.DisplayRole:
return ''
return bool(config.value)
return self._data_for_column(config, col, role)
return None
def flags(self, index: QModelIndex | QPersistentModelIndex) -> Qt.ItemFlag:
if not index.isValid():
return Qt.ItemFlag.ItemIsEnabled
flags = Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable
if index.column() == 0:
return flags | Qt.ItemFlag.ItemIsUserCheckable
if index.column() == 2:
config = self._rows[index.row()].config
if self._effective_type(config) is bool:
return flags | Qt.ItemFlag.ItemIsUserCheckable
return flags | Qt.ItemFlag.ItemIsEditable
return flags
def setData(
self,
index: QModelIndex | QPersistentModelIndex,
value: object,
role: int = Qt.ItemDataRole.EditRole,
) -> bool:
if not index.isValid():
return False
row = index.row()
col = index.column()
if row < 0 or row >= len(self._rows):
return False
config = self._rows[row].config
if col == 0 and role in (Qt.ItemDataRole.CheckStateRole, Qt.ItemDataRole.EditRole):
# probably the CheckStateRole is enough.
if role == Qt.ItemDataRole.EditRole and isinstance(value, bool):
active = value
else:
active = Qt.CheckState(value) == Qt.CheckState.Checked
config.active_override = active
config.source = ParameterSource.CONFIG if active else ParameterSource.DEFAULT
self._refresh_row(row)
self.parameter_active_changed.emit(config.name, active, config.value)
return True
if col == 2 and role == Qt.ItemDataRole.CheckStateRole and self._effective_type(config) is bool:
config.value = Qt.CheckState(value) == Qt.CheckState.Checked
config.source = ParameterSource.CONFIG
config.active_override = True
self._refresh_row(row)
self.parameter_value_changed.emit(config.name, config.value)
return True
if col == 2 and role == Qt.ItemDataRole.EditRole:
config.value = value
config.source = ParameterSource.CONFIG
config.active_override = True
self._refresh_row(row)
self.parameter_value_changed.emit(config.name, value)
return True
return False
def sort(self, column: int, order: Qt.SortOrder = Qt.SortOrder.AscendingOrder) -> None:
if column not in (0, 1):
return
reverse = order == Qt.SortOrder.DescendingOrder
self.layoutAboutToBeChanged.emit()
if column == 1:
self._rows.sort(key=lambda row: row.config.name, reverse=reverse)
else:
self._rows.sort(key=lambda row: row.order, reverse=reverse)
self.layoutChanged.emit()
@staticmethod
def _effective_type(config: ParameterConfig) -> Any:
if config.status in (ParameterSchemaStatus.UNKNOWN, ParameterSchemaStatus.DEPRECATED):
return Any
return config.type
@staticmethod
def _format_value(value: Any) -> str:
if value is None:
return ''
return str(value)
def _data_for_column(self, config: ParameterConfig, col: int, role: int) -> object:
if col == 0:
return None
if col == 1:
return config.name
if col == 2:
value = config.value
if isinstance(value, LogicalFileReference):
if role == Qt.ItemDataRole.EditRole and _is_path_logical_union(config.type):
return value
return value.to_string()
if isinstance(value, dict) and value.get('__param_type__') in {'Path', 'LogicalFileReference'}:
if role == Qt.ItemDataRole.EditRole:
return value
return str(value.get('value', ''))
if isinstance(value, list):
rendered = [item.to_string() if isinstance(item, LogicalFileReference) else item for item in value]
return rendered if role == Qt.ItemDataRole.EditRole else str(rendered)
return value if role == Qt.ItemDataRole.EditRole else self._format_value(value)
if col == 3:
return self._format_type(self._effective_type(config))
if col == 4:
return config.source.value if isinstance(config.source, ParameterSource) else str(config.source)
if col == 5:
return config.status.value if isinstance(config.status, ParameterSchemaStatus) else str(config.status)
return None
def _refresh_row(self, row: int) -> None:
if not (0 <= row < len(self._rows)):
return
first = self.index(row, 0)
last = self.index(row, len(_COLUMNS) - 1)
self.dataChanged.emit(first, last, self._all_row_roles)
[docs]
def _is_list_type(annotation: Any) -> bool:
"""Return whether *annotation* denotes a ``list`` (plain or parameterised).
Any :class:`~typing.Annotated` wrapper is stripped first, so
``Annotated[list[int], marker]`` is recognised as a list as well.
"""
if annotation is None or annotation is Any:
return False
if hasattr(annotation, '__metadata__'):
annotation = get_args(annotation)[0]
if annotation is list:
return True
return get_origin(annotation) is list
[docs]
def _contains_annotation(annotation: Any, target: Any) -> bool:
"""Return whether an annotation contains *target*, including nested wrappers."""
return annotation is target or any(_contains_annotation(argument, target) for argument in get_args(annotation))
[docs]
class _ListItemsTable(QTableWidget):
"""One-column list table with add/remove keyboard commands."""
add_requested = Signal()
remove_requested = Signal()
def __init__(self, rows: int, columns: int, parent: QWidget | None = None) -> None:
"""Initialize a table that can move its selected rows internally."""
super().__init__(rows, columns, parent)
self._internal_drag_active = False
self._dragged_rows: list[int] = []
[docs]
def startDrag(self, supported_actions: Any) -> None: # noqa: N802
"""Run an internal drag without Qt removing rows after the custom drop."""
model = self.model()
if model is None:
return
self._dragged_rows = sorted({index.row() for index in self.selectionModel().selectedRows()})
if not self._dragged_rows:
return
mime_data = model.mimeData(self.selectedIndexes())
drag = QDrag(self)
drag.setMimeData(mime_data)
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 keyPressEvent(self, event: QKeyEvent) -> None: # noqa: N802
"""Route list editing keys through the dialog's shared actions."""
key = event.key()
if key in (Qt.Key.Key_Plus, Qt.Key.Key_Equal) and (
key == Qt.Key.Key_Plus or event.modifiers() & Qt.KeyboardModifier.ShiftModifier
):
self.add_requested.emit()
event.accept()
return
if key in (Qt.Key.Key_Minus, Qt.Key.Key_Delete, Qt.Key.Key_Backspace, Qt.Key.Key_Cancel):
self.remove_requested.emit()
event.accept()
return
super().keyPressEvent(event)
[docs]
def dropEvent(self, event: QDropEvent) -> None: # noqa: N802
"""Move selected rows as a block without letting the table overwrite items."""
if not self._internal_drag_active:
event.ignore()
return
rows = self._dragged_rows or sorted({index.row() for index in self.selectionModel().selectedRows()})
if not rows:
event.ignore()
return
target_item = self.itemAt(event.position().toPoint())
if target_item is None:
target_row = self.rowCount()
else:
target_row = target_item.row()
item_rect = self.visualItemRect(target_item)
if event.position().toPoint().y() >= item_rect.center().y():
target_row += 1
moved_items = [(row, self.takeItem(row, 0)) for row in rows]
insert_row = target_row - sum(row < target_row for row in rows)
for row in reversed(rows):
self.removeRow(row)
insert_row = max(0, min(insert_row, self.rowCount()))
for offset, (_, item) in enumerate(moved_items):
if item is not None:
row = insert_row + offset
self.insertRow(row)
self.setItem(row, 0, item)
self.clearSelection()
for row in range(insert_row, insert_row + len(moved_items)):
self.selectRow(row)
event.setDropAction(Qt.DropAction.MoveAction)
event.accept()
[docs]
class ListEditDialog(QDialog):
"""Dialog for editing the items in one list-valued processor parameter."""
def __init__(
self,
parameter_name: str,
values: Iterable[Any],
element_type: Any = None,
default_values: Iterable[Any] = (),
volume_names: Iterable[str] = (),
parent: QWidget | None = None,
) -> None:
"""Build the item table and controls for one list parameter."""
super().__init__(parent)
self._element_type = element_type
self._default_values = list(default_values)
self.setWindowTitle(f'Edit {parameter_name}')
self.resize(440, 320)
outer = QVBoxLayout(self)
outer.addWidget(QLabel(parameter_name))
content = QHBoxLayout()
self._table = _ListItemsTable(0, 1)
self._table.setHorizontalHeaderLabels(['Items'])
self._table.horizontalHeader().setStretchLastSection(True)
self._table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self._table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
self._table.setDragEnabled(True)
self._table.setAcceptDrops(True)
self._table.setDropIndicatorShown(True)
self._table.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove)
self._table.setDragDropOverwriteMode(False)
self._table.setEditTriggers(
QAbstractItemView.EditTrigger.DoubleClicked | QAbstractItemView.EditTrigger.EditKeyPressed
)
delegate = ParameterDelegate(self._table)
delegate.volume_names = list(volume_names)
self._table.setItemDelegateForColumn(0, delegate)
content.addWidget(self._table, 1)
controls = QVBoxLayout()
self._add_button = QToolButton()
self._add_button.setText('+')
self._add_button.setToolTip('Add an item')
self._remove_button = QToolButton()
self._remove_button.setText('−')
self._remove_button.setToolTip('Remove the selected item')
self._remove_button.setFocusPolicy(Qt.FocusPolicy.NoFocus)
self._remove_button.setEnabled(False)
controls.addWidget(self._add_button)
controls.addWidget(self._remove_button)
controls.addStretch(1)
content.addLayout(controls)
outer.addLayout(content, 1)
self._close_button = QPushButton('Close')
self._close_button.clicked.connect(self._accept_if_valid)
outer.addWidget(self._close_button, alignment=Qt.AlignmentFlag.AlignRight)
self._context_menu = QMenu(self._table)
self._add_action = self._context_menu.addAction('Add')
self._remove_action = self._context_menu.addAction('Remove')
self._remove_all_action = self._context_menu.addAction('Remove all')
self._reset_action = self._context_menu.addAction('Reset to default')
self._remove_all_action.setEnabled(True)
self._reset_action.setEnabled(True)
self._add_button.clicked.connect(self._add_item)
self._remove_button.clicked.connect(self._remove_item)
self._remove_button.pressed.connect(self._close_active_row_editor)
self._table.add_requested.connect(self._add_item)
self._table.remove_requested.connect(self._remove_item)
self._add_action.triggered.connect(self._add_item)
self._remove_action.triggered.connect(self._remove_item)
self._remove_all_action.triggered.connect(self._remove_all)
self._reset_action.triggered.connect(self._reset_to_default)
self._table.itemSelectionChanged.connect(self._update_remove_button)
self._table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._table.customContextMenuRequested.connect(self._show_context_menu)
for value in values:
self._insert_item(value)
self._update_remove_button()
[docs]
def _insert_item(self, value: Any) -> QTableWidgetItem:
"""Append one editable row and return its table item."""
row = self._table.rowCount()
self._table.insertRow(row)
item = QTableWidgetItem(self._display_value(value))
item.setData(_ParameterRoles.TYPE, self._element_type)
if isinstance(value, (Path, LogicalFileReference)):
item.setData(_ParameterRoles.VALUE, value)
self._table.setItem(row, 0, item)
return item
[docs]
@staticmethod
def _display_value(value: Any) -> str:
"""Render typed logical references using their portable serialized form."""
return value.to_string() if isinstance(value, LogicalFileReference) else str(value)
[docs]
def _add_item(self) -> None:
"""Add an empty row and begin editing it immediately."""
item = self._insert_item('')
item.setData(_ParameterRoles.PENDING, True)
self._table.setCurrentItem(item)
self._table.setFocus()
self._table.editItem(item)
[docs]
def _remove_item(self) -> None:
"""Remove every selected row."""
self._remove_rows_later(self._selected_rows())
[docs]
def _close_active_row_editor(self) -> None:
"""Close the active cell editor before the selected row is removed."""
delegate = self._table.itemDelegateForColumn(0)
if isinstance(delegate, ParameterDelegate):
delegate.cancel_active_editor()
[docs]
def _remove_all(self) -> None:
"""Remove every row using the same deletion path as selected removal."""
self._remove_rows_later(range(self._table.rowCount()))
[docs]
def _remove_rows_later(self, rows: Iterable[int]) -> None:
"""Delete selected rows after Qt has completed any editor focus-out commit."""
indexes = [QPersistentModelIndex(self._table.model().index(row, 0)) for row in set(rows)]
def remove_rows() -> None:
current_rows = sorted({index.row() for index in indexes if index.isValid()}, reverse=True)
self._remove_rows(current_rows)
QTimer.singleShot(0, remove_rows)
[docs]
def _remove_rows(self, rows: Iterable[int]) -> None:
"""Delete unique rows in reverse order and refresh selection actions."""
for row in sorted(set(rows), reverse=True):
if 0 <= row < self._table.rowCount():
self._table.removeRow(row)
self._update_remove_button()
[docs]
def _selected_rows(self) -> list[int]:
"""Return selected table row numbers without duplicates."""
return sorted({index.row() for index in self._table.selectionModel().selectedRows()})
[docs]
def _reset_to_default(self) -> None:
"""Replace all rows with the parameter's ActiveParameter default items."""
self._remove_all()
for value in self._default_values:
self._insert_item(value)
self._update_remove_button()
[docs]
def _accept_if_valid(self) -> None:
"""Close only when all non-empty items can be converted to their declared type."""
try:
self.value()
except ValueError:
QMessageBox.warning(self, 'Invalid list item', 'Enter a valid value for every list item.')
return
self.accept()
[docs]
def value(self) -> list[Any]:
"""Return non-empty rows converted to the parameter's element type."""
values: list[Any] = []
for row in range(self._table.rowCount()):
item = self._table.item(row, 0)
if item is None:
continue
typed_value = item.data(_ParameterRoles.VALUE)
if isinstance(typed_value, LogicalFileReference):
values.append(typed_value)
continue
if isinstance(typed_value, Path):
values.append(str(typed_value))
continue
raw_value = item.data(Qt.ItemDataRole.EditRole)
if isinstance(raw_value, LogicalFileReference):
values.append(raw_value)
continue
text = item.text() if raw_value is None else str(raw_value)
if not text.strip():
continue
if self._element_type is int:
values.append(int(text))
elif self._element_type is float:
values.append(float(text))
elif _contains_annotation(self._element_type, LogicalFileReference):
try:
values.append(LogicalFileReference.from_string(text))
except ValueError:
values.append(str(Path(text)) if is_path_type(self._element_type) else text)
else:
# Paths are serialized in TOML as strings, like other path parameters.
values.append(str(Path(text)) if is_path_type(self._element_type) else text)
return values
[docs]
class ListEditor(QWidget):
"""Compact list cell editor that opens a dedicated :class:`ListEditDialog`."""
def __init__(
self,
parent: QWidget | None = None,
element_type: Any = None,
parameter_name: str = '',
on_selected: Callable[[list[Any]], None] | None = None,
default_values: Iterable[Any] = (),
volume_names: Iterable[str] = (),
) -> None:
"""Build the cell summary and its right-aligned Edit button."""
super().__init__(parent)
self._element_type = element_type
self._parameter_name = parameter_name
self._on_selected = on_selected
self._default_values = list(default_values)
self._volume_names = list(volume_names)
self._value: list[Any] = []
self._summary = QLabel('[]')
self._edit_button = QPushButton('Edit')
self._edit_button.clicked.connect(self._edit)
layout = QHBoxLayout(self)
layout.setContentsMargins(4, 0, 4, 0)
layout.addWidget(self._summary, 1)
layout.addWidget(self._edit_button, 0, Qt.AlignmentFlag.AlignRight)
[docs]
def value(self) -> list[Any]:
"""Return the current list value."""
return list(self._value)
[docs]
def set_value(self, value: Any) -> None:
"""Update the displayed list summary."""
self._value = list(value) if isinstance(value, (list, tuple)) else []
if _contains_annotation(self._element_type, LogicalFileReference):
converted: list[Any] = []
for item in self._value:
try:
converted.append(LogicalFileReference.from_parameter_value(item))
except (TypeError, ValueError):
converted.append(str(item))
self._value = converted
invalid = any(not isinstance(item, LogicalFileReference) for item in self._value)
unavailable = any(
isinstance(item, LogicalFileReference) and item.volume not in self._volume_names for item in self._value
)
self._summary.setStyleSheet('color: red' if invalid else 'color: #a66b00' if unavailable else '')
if invalid:
self._summary.setToolTip('One or more values are not valid logical file references.')
elif unavailable:
self._summary.setToolTip('One or more references use a volume that is not configured.')
else:
self._summary.setToolTip('')
self._summary.setText(str([ListEditDialog._display_value(item) for item in self._value]))
[docs]
def _edit(self) -> None:
"""Open the independent dialog and report its result through the model callback."""
# The table may destroy this inline editor on focus change, so use captured
# values and the callback after the parentless modal dialog returns.
values = self.value()
name = self._parameter_name
element_type = self._element_type
on_selected = self._on_selected
default_values = self._default_values
volume_names = self._volume_names
dialog = ListEditDialog(name, values, element_type, default_values, volume_names)
if dialog.exec() == QDialog.DialogCode.Accepted:
selected = dialog.value()
if on_selected is not None:
on_selected(selected)
else:
self.set_value(selected)
[docs]
class LogicalReferenceEditor(QWidget):
"""Keep the serialized value visible and offer a structured edit dialog."""
commit_requested = Signal()
structured_commit_requested = Signal()
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._volumes: list[str] = []
self._line = QLineEdit()
self._button = QPushButton('Edit')
self._error = QLabel()
self._last_committed: LogicalFileReference | str | None = None
self._error.setStyleSheet('color: red')
self._error.setMaximumWidth(140)
self._error.setWordWrap(True)
self._error.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred)
self._button.setMinimumWidth(self._button.sizeHint().width())
self._button.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
self._line.setMinimumWidth(60)
layout = QHBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(self._line, 1)
layout.addWidget(self._button)
layout.addWidget(self._error)
self._button.clicked.connect(self._edit)
self._line.textChanged.connect(self._show_validation_state)
self._line.editingFinished.connect(self._commit_if_changed)
[docs]
def set_volumes(self, volumes: Iterable[str]) -> None:
"""Set configured volume choices used by the structured editor."""
self._volumes = list(volumes)
[docs]
def set_text(self, value: str) -> None:
"""Display existing serialized text without discarding malformed input."""
self._line.setText(value)
self._last_committed = self.value()
[docs]
def _show_validation_state(self, value: str) -> None:
"""Show syntax feedback beside the active parameter editor."""
try:
reference = LogicalFileReference.from_string(value)
except (TypeError, ValueError) as exc:
self._error.setText(str(exc))
self._error.setToolTip(str(exc))
self._error.setStyleSheet('color: red')
return
if reference.volume not in self._volumes:
message = f"Volume '{reference.volume}' is not configured"
self._error.setText(message)
self._error.setToolTip(message)
self._error.setStyleSheet('color: #a66b00')
else:
self._error.setText('')
self._error.setToolTip('')
self._error.setStyleSheet('')
[docs]
def value(self) -> LogicalFileReference | str:
"""Return a typed reference when valid, retaining malformed text otherwise."""
try:
return LogicalFileReference.from_string(self._line.text())
except (TypeError, ValueError):
return self._line.text()
[docs]
def _commit_if_changed(self) -> None:
"""Ask the delegate to store user-edited text once it leaves the field."""
if QApplication.focusWidget() is self._button:
return
value = self.value()
if value != self._last_committed:
self._last_committed = value
self.commit_requested.emit()
[docs]
def _edit(self) -> None:
"""Open a small volume/path form and commit only valid references."""
from PySide6.QtWidgets import QComboBox, QDialog, QDialogButtonBox, QFormLayout, QLineEdit
dialog = QDialog(self)
dialog.setWindowTitle('Edit logical file reference')
layout = QFormLayout(dialog)
volume = QComboBox(dialog)
volume.setEditable(True)
volume.addItems(self._volumes)
path = QLineEdit(dialog)
try:
current = LogicalFileReference.from_string(self._line.text())
if current.volume not in self._volumes:
volume.addItem(f'{current.volume} (unavailable)', current.volume)
volume.setCurrentIndex(volume.count() - 1)
else:
volume.setCurrentText(current.volume)
path.setText(current.relative_path.as_posix())
except (TypeError, ValueError):
if ':' in self._line.text():
volume.setCurrentText(self._line.text().split(':', 1)[0])
path.setText(self._line.text().split(':', 1)[1])
layout.addRow('Volume', volume)
layout.addRow('Relative path', path)
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
buttons.accepted.connect(dialog.accept)
buttons.rejected.connect(dialog.reject)
layout.addRow(buttons)
if dialog.exec() == QDialog.DialogCode.Accepted:
try:
selected_volume = volume.currentData() or volume.currentText()
self._line.setText(LogicalFileReference(selected_volume, PurePosixPath(path.text())).to_string())
value = self.value()
self._last_committed = value
self.commit_requested.emit()
self.structured_commit_requested.emit()
except (TypeError, ValueError) as exc:
QMessageBox.warning(self, 'Invalid logical reference', str(exc))
[docs]
def _is_path_logical_union(annotation: Any) -> bool:
"""Return whether an annotation is one of the supported path/reference unions."""
import types
from typing import Union
from mafw.processor.path_types import is_path_type
origin = get_origin(annotation)
args = get_args(annotation)
return (
origin in (Union, types.UnionType)
and len(args) == 2
and any(is_path_type(arg) for arg in args)
and LogicalFileReference in args
)
[docs]
class PathLogicalUnionEditor(QWidget):
"""Select and edit one physical or logical path alternative."""
commit_requested = Signal()
def __init__(
self,
parent: QWidget,
annotation: Any,
volumes: Iterable[str],
on_path_selected: Callable[[str], None] | None = None,
on_browse_started: Callable[[], None] | None = None,
on_browse_cancelled: Callable[[], None] | None = None,
) -> None:
super().__init__(parent)
from mafw.processor.path_types import is_path_type, path_role_of
self._selector = QComboBox(self)
self._selector.addItems(['Physical path', 'Logical reference'])
self._path = PathEditor(
self,
role=path_role_of(next(arg for arg in get_args(annotation) if is_path_type(arg))),
on_selected=on_path_selected,
on_browse_started=on_browse_started,
on_browse_cancelled=on_browse_cancelled,
)
self._logical = LogicalReferenceEditor(self)
self._logical.set_volumes(volumes)
self._stack = QStackedWidget(self)
self._stack.addWidget(self._path)
self._stack.addWidget(self._logical)
layout = QHBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(self._selector)
layout.addWidget(self._stack, 1)
self._selector.currentIndexChanged.connect(self._stack.setCurrentIndex)
self._path.editing_finished.connect(self._commit_if_focus_leaves)
self._logical.commit_requested.connect(self._commit_if_focus_leaves)
self._logical.structured_commit_requested.connect(self.commit_requested)
[docs]
def _commit_if_focus_leaves(self) -> None:
"""Ignore focus changes between child controls; commit when the editor loses focus."""
focused = QApplication.focusWidget()
if focused is not None and (focused is self or self.isAncestorOf(focused)):
return
self.commit_requested.emit()
[docs]
def value(self) -> Path | LogicalFileReference | str:
"""Return the selected typed alternative."""
if not self._selector.currentIndex():
return Path(self._path.value())
return self._logical.value()
[docs]
def set_value(self, value: Any) -> None:
"""Display the currently selected path alternative."""
if isinstance(value, dict) and set(value) == {'__param_type__', 'value'}:
if value['__param_type__'] == 'LogicalFileReference':
try:
value = LogicalFileReference.from_parameter_value(value['value'])
except (TypeError, ValueError):
self._selector.setCurrentIndex(1)
self._logical.set_text(str(value['value']))
return
elif value['__param_type__'] == 'Path':
value = Path(value['value'])
if isinstance(value, LogicalFileReference):
self._selector.setCurrentIndex(1)
self._logical.set_text(value.to_string())
else:
self._selector.setCurrentIndex(0)
self._path.set_value(value)
[docs]
class ParameterDelegate(QStyledItemDelegate):
"""Delegate that picks editors based on parameter type."""
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.volume_names: list[str] = []
self._active_editor: QWidget | None = None
[docs]
def _track_editor(self, editor: QWidget) -> QWidget:
"""Remember the live editor so list actions can close it before row mutation."""
self._active_editor = editor
def clear_active_editor(*_: object) -> None:
if self._active_editor is editor:
self._active_editor = None
editor.destroyed.connect(clear_active_editor)
return editor
[docs]
def cancel_active_editor(self) -> None:
"""Close the current cell editor without committing its in-progress value."""
editor = self._active_editor
if editor is not None:
self._active_editor = None
self.closeEditor.emit(editor, QAbstractItemDelegate.EndEditHint.RevertModelCache)
def createEditor(
self,
parent: QWidget,
option: QStyleOptionViewItem,
index: QModelIndex | QPersistentModelIndex,
) -> QWidget:
param_type = index.data(_ParameterRoles.TYPE)
if _is_path_logical_union(param_type):
model = index.model()
persistent_index = QPersistentModelIndex(index)
is_list_item = index.data(_ParameterRoles.PARAMETER) is None
def browse_started() -> None:
"""Protect a new row while the modal chooser owns focus."""
if is_list_item and persistent_index.isValid():
model.setData(persistent_index, True, _ParameterRoles.BROWSING)
def browse_cancelled() -> None:
"""Discard an empty pending row after the chooser is cancelled."""
if is_list_item and persistent_index.isValid():
model.setData(persistent_index, None, _ParameterRoles.BROWSING)
if persistent_index.data(_ParameterRoles.PENDING):
table = self.parent()
if isinstance(table, QTableWidget):
_remove_pending_item_later(table, persistent_index)
def set_selected_path(path: str) -> None:
"""Commit a browse result after the modal dialog closes the editor."""
if persistent_index.isValid():
selected: Any = Path(path)
if is_list_item:
model.setData(persistent_index, None, _ParameterRoles.BROWSING)
model.setData(persistent_index, None, _ParameterRoles.PENDING)
_set_list_item_value(model, persistent_index, selected)
else:
model.setData(persistent_index, selected, Qt.ItemDataRole.EditRole)
editor = PathLogicalUnionEditor(
parent,
param_type,
self.volume_names,
set_selected_path,
browse_started,
browse_cancelled,
)
def commit_union_editor() -> None:
"""Store nested editor changes without emitting Qt's invalid child commit."""
if persistent_index.isValid():
self.setModelData(editor, model, persistent_index)
self.closeEditor.emit(editor, QAbstractItemDelegate.EndEditHint.NoHint)
editor.commit_requested.connect(commit_union_editor)
return self._track_editor(editor)
if param_type is LogicalFileReference:
logical_editor = LogicalReferenceEditor(parent)
logical_editor.set_volumes(self.volume_names)
logical_editor.commit_requested.connect(lambda: self.commitData.emit(logical_editor))
logical_editor.commit_requested.connect(
lambda: self.closeEditor.emit(logical_editor, QAbstractItemDelegate.EndEditHint.NoHint)
)
return self._track_editor(logical_editor)
if _is_list_type(param_type):
model = index.model()
persistent_index = QPersistentModelIndex(index)
parameter = index.data(_ParameterRoles.PARAMETER)
def set_selected_values(values: list[Any]) -> None:
"""Commit dialog results even if the inline editor has closed."""
if persistent_index.isValid():
model.setData(persistent_index, values, Qt.ItemDataRole.EditRole)
return self._track_editor(
ListEditor(
parent,
element_type=_list_element_type(param_type),
parameter_name=parameter.name,
on_selected=set_selected_values,
default_values=parameter.default or (),
volume_names=self.volume_names,
)
)
if param_type is int:
spin_box = QSpinBox(parent)
spin_box.setRange(*INT_RANGE)
return self._track_editor(spin_box)
if param_type is float:
double_box = QDoubleSpinBox(parent)
double_box.setRange(*FLOAT_RANGE)
double_box.setDecimals(FLOAT_DECIMALS)
return self._track_editor(double_box)
if param_type is bool:
# this is not used any more. it is left just for legacy reason and can be removed, since boolean
# values are now directly rendered with a checkbox without going through the delegate.
checkbox = QCheckBox(parent)
checkbox.toggled.connect(self._commit_checkbox)
return self._track_editor(checkbox)
if is_path_type(param_type):
# Any Path/Path subclass (optionally Annotated with a PathRole) routes here.
model = index.model()
persistent_index = QPersistentModelIndex(index)
def set_selected_path(path: str) -> None:
"""Write the selection through the model after the editor closes."""
if persistent_index.isValid():
model.setData(persistent_index, path, Qt.ItemDataRole.EditRole)
return self._track_editor(PathEditor(parent, role=path_role_of(param_type), on_selected=set_selected_path))
return self._track_editor(QLineEdit(parent))
def setEditorData(
self,
editor: QWidget,
index: QModelIndex | QPersistentModelIndex,
) -> None:
value = index.data(Qt.ItemDataRole.EditRole)
if isinstance(editor, PathLogicalUnionEditor):
typed_value = index.data(_ParameterRoles.VALUE)
editor.set_value(typed_value if isinstance(typed_value, (Path, LogicalFileReference)) else value)
return
if isinstance(editor, QSpinBox):
editor.setValue(int(value) if value not in (None, '') else 0)
return
if isinstance(editor, LogicalReferenceEditor):
editor.set_text(
value
if isinstance(value, str)
else value.to_string()
if isinstance(value, LogicalFileReference)
else ''
)
return
if isinstance(editor, QDoubleSpinBox):
editor.setValue(float(value) if value not in (None, '') else 0.0)
return
if isinstance(editor, QCheckBox):
editor.setChecked(bool(value))
return
if isinstance(editor, ListEditor):
editor.set_value(value)
return
if isinstance(editor, PathEditor):
editor.set_value(value)
return
if isinstance(editor, QLineEdit):
editor.setText('' if value is None else str(value))
return
super().setEditorData(editor, index)
def setModelData(
self,
editor: QWidget,
model: QAbstractItemModel,
index: QModelIndex | QPersistentModelIndex,
) -> None:
if isinstance(editor, QSpinBox):
model.setData(index, editor.value(), Qt.ItemDataRole.EditRole)
return
if isinstance(editor, PathLogicalUnionEditor):
value = editor.value()
_commit_list_editor_value(self, editor, model, index, value)
return
if isinstance(editor, QDoubleSpinBox):
model.setData(index, editor.value(), Qt.ItemDataRole.EditRole)
return
if isinstance(editor, QCheckBox):
model.setData(index, editor.isChecked(), Qt.ItemDataRole.EditRole)
return
if isinstance(editor, ListEditor):
model.setData(index, editor.value(), Qt.ItemDataRole.EditRole)
return
if isinstance(editor, PathEditor):
_commit_list_editor_value(self, editor, model, index, editor.value())
return
if isinstance(editor, QLineEdit):
_commit_list_editor_value(self, editor, model, index, editor.text())
return
if isinstance(editor, LogicalReferenceEditor):
value = editor.value()
if index.data(_ParameterRoles.PARAMETER) is None:
_commit_list_editor_value(self, editor, model, index, value)
return
model.setData(index, value, Qt.ItemDataRole.EditRole)
return
super().setModelData(editor, model, index)
def _commit_checkbox(self, _: bool) -> None:
editor = self.sender()
if isinstance(editor, QWidget):
self.commitData.emit(editor)
self.closeEditor.emit(editor)
[docs]
class ProcessorParameterEditor(QWidget):
"""Widget used to edit parameters and metadata of a processor entry."""
replica_name_changed = Signal(str)
new_only_changed = Signal(object)
inheritance_changed = Signal(object)
parameter_value_changed = Signal(str, object)
parameter_active_changed = Signal(str, bool, object)
reset_defaults_requested = Signal()
remove_requested = Signal()
edit_filters_requested = Signal()
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._current_name: str | None = None
self._current_replica: str | None = None
self._processor_label = QLabel('Processor: -')
self._replica_edit = QLineEdit()
self._new_only_checkbox = QCheckBox('only new items')
self._inherit_checkbox = QCheckBox('inherit from base processor')
self._new_only_checkbox.setTristate(True)
self._inherit_checkbox.setTristate(True)
self._table_model = ParameterTableModel(self)
self._table = QTableView()
self._table.setModel(self._table_model)
self._parameter_delegate = ParameterDelegate(self._table)
self._table.setItemDelegateForColumn(2, self._parameter_delegate)
self._table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self._table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self._table.setEditTriggers(QAbstractItemView.EditTrigger.DoubleClicked)
self._table.clicked.connect(self._activate_parameter_editor_on_click)
self._table.setSortingEnabled(True)
header = self._table.horizontalHeader()
header.setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
header.setStretchLastSection(False)
header.setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch)
header.setSectionsClickable(True)
self._table.verticalHeader().setVisible(False)
self._reset_button = QPushButton('Reset defaults')
self._remove_button = QPushButton('Remove processor')
self._edit_filters_button = QPushButton('Edit filters')
self._setup_layout()
self._wire_signals()
[docs]
def set_processor_data(self, config: ProcessorConfig) -> None:
"""Populate the editor with processor configuration data."""
self._current_name = config.name
base_name, replica = parse_processor_name(config.name)
self._current_replica = replica
with block_signals(self._replica_edit, self._new_only_checkbox, self._inherit_checkbox):
self._processor_label.setText(f'Processor: {base_name}')
self._replica_edit.setText(replica or '')
self._new_only_checkbox.setCheckState(self._encode_tristate(config.new_only))
self._inherit_checkbox.setEnabled(replica is not None)
self._inherit_checkbox.setCheckState(self._encode_tristate(config.inheritance))
parameters = list(config.parameters.values())
self._table_model.set_parameters(parameters)
[docs]
def set_storage_volumes(self, volumes: Iterable[str]) -> None:
"""Update the volume choices offered by logical-reference editors."""
self._parameter_delegate.volume_names = list(volumes)
[docs]
def current_name(self) -> str | None:
"""Return the currently displayed processor full name."""
return self._current_name
[docs]
def _activate_parameter_editor_on_click(self, index: QModelIndex) -> None:
"""Show the list or path editor after one click on its value cell."""
param_type = index.data(_ParameterRoles.TYPE)
if index.column() == 2 and (
param_type is LogicalFileReference or _is_list_type(param_type) or is_path_type(param_type)
):
self._table.edit(index)
def _setup_layout(self) -> None:
form_layout = QFormLayout()
form_layout.addRow('Replica:', self._replica_edit)
toggle_layout = QHBoxLayout()
toggle_layout.addWidget(self._new_only_checkbox)
toggle_layout.addWidget(self._inherit_checkbox)
toggle_layout.addStretch(1)
button_layout = QHBoxLayout()
button_layout.addWidget(self._reset_button)
button_layout.addWidget(self._remove_button)
button_layout.addWidget(self._edit_filters_button)
button_layout.addStretch(1)
layout = QVBoxLayout(self)
layout.addWidget(self._processor_label)
layout.addLayout(form_layout)
layout.addLayout(toggle_layout)
layout.addWidget(self._table)
layout.addLayout(button_layout)
def _wire_signals(self) -> None:
self._replica_edit.editingFinished.connect(self._on_replica_changed)
self._new_only_checkbox.stateChanged.connect(self._on_new_only_state_changed)
self._inherit_checkbox.stateChanged.connect(self._on_inheritance_state_changed)
self._table_model.parameter_value_changed.connect(self.parameter_value_changed)
self._table_model.parameter_active_changed.connect(self.parameter_active_changed)
self._reset_button.clicked.connect(self.reset_defaults_requested)
self._remove_button.clicked.connect(self.remove_requested)
self._edit_filters_button.clicked.connect(self.edit_filters_requested)
def _on_replica_changed(self) -> None:
new_replica = self._replica_edit.text().strip() or None
if new_replica == self._current_replica:
return
self._current_replica = new_replica
self.replica_name_changed.emit(new_replica or '')
@staticmethod
def _resolve_tristate(state: int) -> bool | None:
if Qt.CheckState(state) == Qt.CheckState.Checked:
value = True
elif Qt.CheckState(state) == Qt.CheckState.Unchecked:
value = False
else:
value = None
return value
def _on_new_only_state_changed(self, state: int) -> None:
self.new_only_changed.emit(self._resolve_tristate(state))
def _on_inheritance_state_changed(self, state: int) -> None:
self.inheritance_changed.emit(self._resolve_tristate(state))
@staticmethod
def _encode_tristate(value: bool | None) -> Qt.CheckState:
if value:
return Qt.CheckState.Checked
if value is False:
return Qt.CheckState.Unchecked
return Qt.CheckState.PartiallyChecked