# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Pipeline editor widget exposing processors and groups in execution order.
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Provide a table-based editor to reorder, edit, and add pipeline entries without
direct controller access.
"""
from __future__ import annotations
from PySide6.QtCore import QItemSelectionModel, QModelIndex, Qt, Signal
from PySide6.QtGui import QKeyEvent
from PySide6.QtWidgets import (
QAbstractItemView,
QHBoxLayout,
QHeaderView,
QPushButton,
QStyle,
QTableView,
QToolButton,
QVBoxLayout,
QWidget,
)
from mafw.steering_gui.models import PipelineItem, SteeringTreeModel, SteeringTreeRoles
[docs]
class _PipelineTableView(QTableView):
"""Internal table view that exposes delete key handling."""
delete_pressed = Signal()
def keyPressEvent(self, event: QKeyEvent) -> None:
if event.key() in (Qt.Key.Key_Delete, Qt.Key.Key_Backspace):
self.delete_pressed.emit()
event.accept()
return
super().keyPressEvent(event)
[docs]
class PipelineEditor(QWidget):
"""Widget exposing the processor pipeline as a flat table plus toolbox actions."""
add_processor_requested = Signal()
add_group_requested = Signal()
edit_requested = Signal(object)
remove_requested = Signal(list)
move_requested = Signal(str, list)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._model: SteeringTreeModel | None = None
self._selection_model: QItemSelectionModel | None = None
self.table = _PipelineTableView()
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self.table.setDragEnabled(True)
self.table.setAcceptDrops(True)
self.table.setDropIndicatorShown(True)
self.table.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove)
self.table.verticalHeader().setVisible(False)
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
self.table.horizontalHeader().setStretchLastSection(True)
self.table.horizontalHeader().setVisible(False)
self.move_up_button = QToolButton()
self.move_down_button = QToolButton()
self.edit_button = QToolButton()
self.remove_button = QToolButton()
self._apply_tool_icons()
self.add_processor_button = QPushButton('Add processor')
self.add_processor_button.setToolTip('Add a new processor to the pipeline')
self.add_group_button = QPushButton('Add group')
self.add_group_button.setToolTip('Add a new group to the pipeline')
self._setup_layout()
self._wire_signals()
self._update_action_state()
[docs]
def set_model(self, model: SteeringTreeModel) -> None:
"""Attach the steering tree model and focus on the processors section."""
if self._model is not None:
self._disconnect_model_signals(self._model)
if self._selection_model is not None:
try:
self._selection_model.selectionChanged.disconnect(self._update_action_state)
except (RuntimeError, TypeError):
pass
self._model = model
self.table.setModel(model)
self._set_root_index(model)
self._connect_model_signals(model)
selection_model = self.table.selectionModel()
if selection_model is not None:
selection_model.selectionChanged.connect(self._update_action_state)
self._selection_model = selection_model
self._update_action_state()
[docs]
def refresh(self) -> None:
"""Refresh the processors root index after model resets."""
if self._model is None:
return
self._set_root_index(self._model)
self._update_action_state()
def _apply_tool_icons(self) -> None:
style = self.style()
self.move_up_button.setIcon(style.standardIcon(QStyle.StandardPixmap.SP_ArrowUp))
self.move_down_button.setIcon(style.standardIcon(QStyle.StandardPixmap.SP_ArrowDown))
self.edit_button.setIcon(style.standardIcon(QStyle.StandardPixmap.SP_FileDialogDetailedView))
self.remove_button.setIcon(style.standardIcon(QStyle.StandardPixmap.SP_TrashIcon))
self.move_up_button.setToolTip('Move up')
self.move_down_button.setToolTip('Move down')
self.edit_button.setToolTip('Edit')
self.remove_button.setToolTip('Delete')
def _setup_layout(self) -> None:
toolbar = QHBoxLayout()
toolbar.addWidget(self.move_up_button)
toolbar.addWidget(self.move_down_button)
toolbar.addWidget(self.edit_button)
toolbar.addWidget(self.remove_button)
toolbar.addStretch(1)
add_row = QHBoxLayout()
add_row.addWidget(self.add_processor_button)
add_row.addWidget(self.add_group_button)
add_row.addStretch(1)
layout = QVBoxLayout()
layout.addWidget(self.table)
layout.addLayout(toolbar)
layout.addLayout(add_row)
self.setLayout(layout)
def _wire_signals(self) -> None:
self.table.doubleClicked.connect(self._on_edit_selected)
self.table.delete_pressed.connect(self._on_remove_selected)
self.add_processor_button.clicked.connect(self.add_processor_requested.emit)
self.add_group_button.clicked.connect(self.add_group_requested.emit)
self.move_up_button.clicked.connect(self._on_move_up)
self.move_down_button.clicked.connect(self._on_move_down)
self.edit_button.clicked.connect(self._on_edit_selected)
self.remove_button.clicked.connect(self._on_remove_selected)
def _connect_model_signals(self, model: SteeringTreeModel) -> None:
model.modelReset.connect(self.refresh)
model.layoutChanged.connect(self.refresh)
def _disconnect_model_signals(self, model: SteeringTreeModel) -> None:
for signal_name in ('modelReset', 'layoutChanged'):
signal = getattr(model, signal_name)
try:
signal.disconnect(self.refresh)
except (RuntimeError, TypeError):
pass
def _set_root_index(self, model: SteeringTreeModel) -> None:
root_index = model.index(0, 0, QModelIndex())
processors_index = QModelIndex()
for row in range(model.rowCount(root_index)):
candidate = model.index(row, 0, root_index)
if model.data(candidate, SteeringTreeRoles.SECTION_KEY) == 'processors':
processors_index = candidate
break
if processors_index.isValid():
self.table.setRootIndex(processors_index)
else:
self.table.setRootIndex(QModelIndex())
def _selected_items(self) -> list[PipelineItem]:
if self._model is None or self._selection_model is None:
return []
indexes = self._selection_model.selectedRows()
items: list[PipelineItem] = []
seen_ids: set[int] = set()
for index in sorted(indexes, key=lambda idx: idx.row()):
value = self._model.data(index, SteeringTreeRoles.PIPELINE_ITEM)
if isinstance(value, PipelineItem):
item_id = id(value)
if item_id in seen_ids:
continue
seen_ids.add(item_id)
items.append(value)
return items
def _update_action_state(self, *_: object) -> None:
has_selection = bool(self._selected_items())
for button in (self.move_up_button, self.move_down_button, self.edit_button, self.remove_button):
button.setEnabled(has_selection)
def _on_edit_selected(self, *_: object) -> None:
items = self._selected_items()
if not items:
return
self.edit_requested.emit(items[0])
def _on_remove_selected(self) -> None:
items = self._selected_items()
if not items:
return
self.remove_requested.emit(items)
def _on_move_up(self) -> None:
self._emit_move('up')
def _on_move_down(self) -> None:
self._emit_move('down')
def _emit_move(self, direction: str) -> None:
items = self._selected_items()
if not items:
return
self.move_requested.emit(direction, items)