# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Qt model for the processor pipeline tree.
:Author: Bulgheroni Antonio
:Description: Provide a hierarchical QAbstractItemModel view of pipeline items for GUI widgets.
"""
from __future__ import annotations
from enum import IntEnum
from typing import no_type_check
from PySide6.QtCore import (
QAbstractItemModel,
QModelIndex,
QObject,
QPersistentModelIndex,
Qt,
)
from .pipeline import PipelineItem, ProcessorPipeline
[docs]
class PipelineRoles(IntEnum):
"""Custom roles exported by the processor pipeline model."""
ITEM = Qt.ItemDataRole.UserRole + 1
NAME = Qt.ItemDataRole.UserRole + 2
IS_GROUP = Qt.ItemDataRole.UserRole + 3
[docs]
class ProcessorPipelineModel(QAbstractItemModel):
"""Expose a ProcessorPipeline as a tree-structured Qt model."""
def __init__(self, pipeline: ProcessorPipeline | None = None, parent: QObject | None = None) -> None:
super().__init__(parent)
self._pipeline = pipeline or ProcessorPipeline()
[docs]
def pipeline(self) -> ProcessorPipeline:
"""Return the currently attached pipeline."""
return self._pipeline
[docs]
def set_pipeline(self, pipeline: ProcessorPipeline) -> None:
"""Replace the pipeline data and reset the model."""
self.beginResetModel()
self._pipeline = pipeline
self.endResetModel()
[docs]
def root_items(self) -> list[PipelineItem]:
"""Return the top-level pipeline items."""
return self._pipeline.items
def index(
self,
row: int,
column: int,
parent: QModelIndex | QPersistentModelIndex = QModelIndex(),
) -> QModelIndex:
if column != 0 or row < 0:
return QModelIndex()
parent_item = self._item_from_index(parent)
children = self._child_items(parent_item)
if row >= len(children):
return QModelIndex()
return self.createIndex(row, column, children[row])
@no_type_check
def parent(
self,
index: QModelIndex | QPersistentModelIndex | None = None,
) -> QModelIndex | QObject | None:
if index is None:
return super().parent()
if not index.isValid():
return QModelIndex()
item = self._item_from_index(index)
if item is None or item.parent is None:
return QModelIndex()
parent_item = item.parent
row = self._row_of_item(parent_item)
if row is None:
return QModelIndex()
return self.createIndex(row, 0, parent_item)
def rowCount(self, parent: QModelIndex | QPersistentModelIndex = QModelIndex()) -> int:
parent_item = self._item_from_index(parent)
return len(self._child_items(parent_item))
def columnCount(self, parent: QModelIndex | QPersistentModelIndex = QModelIndex()) -> int:
return 1
def data(
self,
index: QModelIndex | QPersistentModelIndex,
role: int = Qt.ItemDataRole.DisplayRole,
) -> object:
if not index.isValid():
return None
item = self._item_from_index(index)
if item is None:
return None
if role == Qt.ItemDataRole.DisplayRole:
return item.name()
if role == PipelineRoles.ITEM:
return item
if role == PipelineRoles.NAME:
return item.name()
if role == PipelineRoles.IS_GROUP:
return item.is_group()
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 | Qt.ItemFlag.ItemIsDragEnabled
item = self._item_from_index(index)
if item is not None and item.is_group():
flags |= Qt.ItemFlag.ItemIsDropEnabled
return flags
def _item_from_index(self, index: QModelIndex | QPersistentModelIndex) -> PipelineItem | None:
if not index.isValid():
return None
value = index.internalPointer()
return value if isinstance(value, PipelineItem) else None
def _child_items(self, parent_item: PipelineItem | None) -> list[PipelineItem]:
if parent_item is None:
return self._pipeline.items
return parent_item.children
def _row_of_item(self, item: PipelineItem) -> int | None:
if item.parent is None:
try:
return self._pipeline.items.index(item)
except ValueError:
return None
try:
return item.parent.children.index(item)
except ValueError:
return None