# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Present a hierarchical representation of the steering file sections.
:Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
:Description: Show globals, database, UI, and processors in a tree so the user can select editors.
"""
from __future__ import annotations
from PySide6.QtCore import QItemSelectionModel, QModelIndex, QPoint, Qt, Signal
from PySide6.QtWidgets import (
QAbstractItemView,
QHBoxLayout,
QMenu,
QTreeView,
QWidget,
)
from mafw.steering_gui.models import PipelineItem, SteeringTreeModel
[docs]
class SteeringTreeView(QWidget):
"""Widget exposing the steering-file sections as a :class:`QTreeView`."""
section_selected = Signal(str)
pipeline_selection_changed = Signal(list)
add_processor_requested = Signal()
add_group_requested = Signal()
pipeline_items_delete_requested = Signal(list)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._tree = QTreeView()
self._model: SteeringTreeModel | None = None
self._tree.header().hide()
self._tree.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
self._tree.setDragEnabled(True)
self._tree.setAcceptDrops(True)
self._tree.setDropIndicatorShown(True)
self._tree.setDragDropMode(QAbstractItemView.DragDropMode.InternalMove)
self._tree.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._tree.customContextMenuRequested.connect(self._show_context_menu)
layout = QHBoxLayout()
layout.addWidget(self._tree)
self.setLayout(layout)
self._tree.setMinimumWidth(220)
[docs]
def set_model(self, model: SteeringTreeModel) -> None:
"""Attach a steering tree model to the view."""
if self._model is not None:
self._disconnect_model_signals(self._model)
self._tree.setModel(model)
self._model = model
self._connect_model_signals(model)
selection_model = self._tree.selectionModel()
if selection_model is not None:
selection_model.selectionChanged.connect(self._on_selection_changed)
root_index = model.index(0, 0, QModelIndex())
if root_index.isValid():
self._tree.setCurrentIndex(root_index)
self._tree.expandAll()
[docs]
def refresh(self) -> None:
"""Refresh expansion and selection after model updates."""
if self._model is None:
return
self._tree.expandAll()
root_index = self._model.index(0, 0, QModelIndex())
if root_index.isValid() and not self._tree.selectionModel().hasSelection():
self._tree.setCurrentIndex(root_index)
[docs]
def _on_selection_changed(self, *_: object) -> None:
"""Emit the section key and pipeline selection for the new tree selection."""
if self._model is None:
return
selection = self._tree.selectionModel()
if selection is None:
return
selected_indexes = selection.selectedIndexes()
pipeline_items: list[PipelineItem] = []
section_key: str | None = None
for index in selected_indexes:
item = self._model.pipeline_item_from_index(index)
if item is not None:
pipeline_items.append(item)
if section_key is None:
section_key = self._model.section_key_from_index(index)
if section_key is not None:
self.section_selected.emit(section_key)
self.pipeline_selection_changed.emit(pipeline_items)
def _connect_model_signals(self, model: SteeringTreeModel) -> None:
if not model.expand_signals_connected:
model.layoutChanged.connect(self._expand_after_change)
model.modelReset.connect(self._expand_after_change)
model._expand_signals_connected = True
def _disconnect_model_signals(self, model: SteeringTreeModel) -> None:
# I do not see any occasions why this method will be ever called.
# the model is created only once at the application startup,
if model.expand_signals_connected:
for signal_name in ('layoutChanged', 'modelReset'):
signal = getattr(model, signal_name)
try:
signal.disconnect(self._expand_after_change)
except (RuntimeError, TypeError):
pass
model.expand_signals_connected = False
def _expand_after_change(self) -> None:
self._tree.expandAll()
def _selected_pipeline_items(self) -> list[PipelineItem]:
if self._model is None:
return []
selection_model = self._tree.selectionModel()
if selection_model is None:
return []
items: list[PipelineItem] = []
seen_ids: set[int] = set()
for index in selection_model.selectedRows():
item = self._model.pipeline_item_from_index(index)
if item is None:
continue
ident = id(item)
if ident in seen_ids:
continue
seen_ids.add(ident)
items.append(item)
return items
def select_section(self, key: str) -> None:
if self._model is None:
return
root_index = self._model.index(0, 0, QModelIndex())
for row in range(self._model.rowCount(root_index)):
candidate = self._model.index(row, 0, root_index)
if self._model.section_key_from_index(candidate) == key:
self._tree.setCurrentIndex(candidate)
return
[docs]
def select_pipeline_item(self, name: str) -> None:
"""Select a pipeline item by its processor/group name."""
if self._model is None:
return
root_index = self._model.index(0, 0, QModelIndex())
processors_index = QModelIndex()
for row in range(self._model.rowCount(root_index)):
candidate = self._model.index(row, 0, root_index)
if self._model.section_key_from_index(candidate) == 'processors':
processors_index = candidate
break
if not processors_index.isValid():
return
index = self._find_pipeline_index(processors_index, name)
if index.isValid():
selection_model = self._tree.selectionModel()
if selection_model is not None:
selection_model.clearSelection()
self._tree.setCurrentIndex(index)
def _find_pipeline_index(self, parent: QModelIndex, name: str) -> QModelIndex:
if self._model is None:
return QModelIndex()
for row in range(self._model.rowCount(parent)):
candidate = self._model.index(row, 0, parent)
item = self._model.pipeline_item_from_index(candidate)
if item is not None and item.name() == name:
return candidate
child = self._find_pipeline_index(candidate, name)
if child.isValid():
return child
return QModelIndex()
def _show_context_menu(self, pos: QPoint) -> None:
if self._model is None:
return
index = self._tree.indexAt(pos)
if not index.isValid():
return
selection_model = self._tree.selectionModel()
if selection_model is not None and not selection_model.isSelected(index):
selection_model.select(
index,
QItemSelectionModel.SelectionFlag.ClearAndSelect | QItemSelectionModel.SelectionFlag.Rows,
)
pipeline_items = self._selected_pipeline_items()
if pipeline_items:
menu = QMenu(self)
delete_action = menu.addAction('Delete')
delete_action.triggered.connect(lambda: self.pipeline_items_delete_requested.emit(pipeline_items))
menu.exec(self._tree.viewport().mapToGlobal(pos))
return
if self._model.pipeline_item_from_index(index) is not None:
return
if self._model.section_key_from_index(index) != 'processors':
return
menu = QMenu(self)
menu.addAction('Add processor').triggered.connect(self.add_processor_requested.emit)
menu.addAction('Add group').triggered.connect(self.add_group_requested.emit)
menu.exec(self._tree.viewport().mapToGlobal(pos))