qtpynodeeditorというパッケージの中身を確認する過程のメモ

背景

• mindmapのような、GUIベースのアプリケーションを作りたいと思ったけど、いままでGUIの処理なんていじったことないのでとりあえず参考になりそうなqtpynodeeditorをみて勉強しようと思う。

• ついでに、備忘録として既存パッケージの読み方みたなのがまとめてみる。（まとまるかは不明）

• 環境

  • macOS Mojave 10.14.6
  • python=3.7

setup on commund line

qtpynodeeditorは、それなりにコード量のあるpackage（Python初心者の私にとっては）なので、１日で終わるはずがないと踏んで初期化方法をメモっておく。

• pip install graphviz
• conda remove -n qtpy --all
• conda create -n qtpy python=3.7
• conda activate qtpy
• cd /project/*Qt*/
• pip install PyQt5 -t ./packages/PyQt5/
• pip install qtpy -t ./packages/qtpy/
• pip install qtpynodeeditor -t ./packages/qtpynodeeditor/
• pip install pytest
• pip install graphviz
• pip install pycallgraph
• pip install jupyter

↑のように、"pip install -t" tオプションでinstallしておくと実験folder内にpackageを置けて、中身をいじりながらのコードリーディングをやりやすくなる. （eオプションなしだと、 .../site-package/...以下にpackageをinstallしてしまうので、わちゃわちゃいじる実験folderと分離してしまい不便）

symlinkをsys.path内のdirectoryに発行してpathを追加しなくても参照できるようにしておく。

import os
import sys

cwd = os.getcwd()
tmp_path =  [i for i in sys.path if 'site-packages' in i][0]

os.symlink(cwd+'/packages/qtpy/',tmp_path+'/qtpy')
os.symlink(cwd+'/packages/PyQt5',tmp_path+'/PyQt5')
os.symlink(cwd+'/packages/qtpynodeeditor',tmp_path+'/qtpynodeeditor')

• target packageがinstallされたことを確認

>> import qtpynodeeditor
>> qtpynodeeditor.__file__
'/project/20200211_QtForPython/20200307_qtpy/qtpynodeeditor/qtpynodeeditor/__init__.py'

コードリーディング開始

共通tool

• ファイル検索方法

  • grep -rl [検索文字] [探索directory]
    • 例） grep -rl pytestqt ./

• コード検索方法

  • grep -A 2 -B 2 -rn [検索文字] [探索directory]
    • grep -A 2 -B 2 -rn pytestqt ./

step1 とりあえず動かしてみる

• python run_tests.py
  • error発生 pytestqt というパッケージをimport できないっぽい。 先行きが不安すぎる

E   _pytest.config.ConftestImportFailure: (local('/project/20200211_QtForPython/20200307_qtpy/qtpynodeeditor/qtpynodeeditor/tests/conftest.py'), (<class 'ModuleNotFoundError'>, ModuleNotFoundError("No module named 'pytestqt'"), <traceback object at 0x1058ad910>))

コード検索方法に書いてある方法"grep -rl pytestqt ./" でpytestqtなるものを探したが、みつからず　testを通すことはできずorz

• python ./qtpynodeeditor/qtpynodeeditor/examples/style.py

  • 

  • それっぽいwindowが出現！！　現状の機能をまとめると

    • Captionとかかれたnodeを右クリックで新規作成可能
    • nodeの移動可能
    • nodeの新規作成
    • node間の線の描画
    • node間の線の再描画
    • node間の線の削除
    • windowの拡大縮小（感度が高すぎて少し調整を変えたくても極大極小になる笑）
    • windowの表示範囲の移動

  • できないこと

    • node内のtextの編集
    • node

step1.2 コード全体を俯瞰する

step2 documentを読む

• documentを読む（gitのreadme）

Python Qt node editor
Pure Python port of NodeEditor, supporting PyQt5 and PySide through qtpy.

↓　google英和
Python Qtノードエディター
NodeEditorの純粋なPythonポート。qtpyを介してPyQt5およびPySideをサポートします。

portがなんなのかよくわからないが、qtpyを使っていることがわかった（）

# Requirements
Python 3.6+
qtpy
PyQt5 / PySide

#Documentation
Sphinx-generated documentation

#Screenshots
Style example

他にも、必要な依存package(Requirements)やDocの存在があるみたい　Docを見てみる。

• documentを読む(APIの説明ページ)
  • API
    • Scene and View
      • FlowScene
      • FlowView
    • Styles
      • StyleCollection
      • StyleConnectionStyle
      • FlowViewStyle
      • NodeStyle
    • Nodes
      • Node
      • NodeData
      • NodeDataModel
      • NodeState
      • NodeDataType
      • NodeGeometry
      • NodeGraphicsObject
      • DataModelRegistry
    • Connections
      • Connection
      • ConnectionGeometry
      • ConnectionGraphicsObject
    • Ports
      • Port
      • PortType
    • Other / Internal
      • ConnectionPainter
      • ConnectionPolicy
      • NodeConnectionInteraction
      • NodePainter
      • NodePainterDelegate
      • NodeValidationState

正直にいってまったくわけがわからないが、粒度別にまとめてみた。
とりあえず現段階で予想をまとめると

• 現段階の予想
  • Scene and View
    • 画面の描画全体に関することをつかさどるのでは？　アプリケーションとしての起動から終了までだったり、イベントを受け取る方法を定義するのかなー
  • style
    • 画面の色とかフォント、nodeの色とかフォントを制御、
  • Nodes
    • ノードの形とか、点から点へ矢印を繋げるみたいなのでその点の数や、outputをoutputへつなげたらだめだよ的なことを定義？　
  • Connections
    • ノードとノードをつなぐ矢印のいい感じの曲がり具合とか、nodeを動かした時にちゃんと同期して矢印を動かしたりとかを定義?
  • Ports
    • まったくわからん
  • Other / Internal
    • まったくわからん

とりあえず、"Scene and View"が一番上にきているしもっとも基本的な部分じゃないかなーと想像して進めようと思う

step3　コード概要の確認

step3.1 コードのブラウジング

• とりあえず動かしたコードの中身をざっとみる

./qtpynodeeditor/qtpynodeeditor/examples/style.py

import logging

from qtpy import QtWidgets

import qtpynodeeditor
from qtpynodeeditor import (NodeData, NodeDataModel, NodeDataType, PortType,
                            StyleCollection)

style_json = '''
    {
      "FlowViewStyle": {
        "BackgroundColor": [255, 255, 240],
        "FineGridColor": [245, 245, 230],
        "CoarseGridColor": [235, 235, 220]
      },
      "NodeStyle": {
        "NormalBoundaryColor": "darkgray",
        "SelectedBoundaryColor": "deepskyblue",
        "GradientColor0": "mintcream",
        "GradientColor1": "mintcream",
        "GradientColor2": "mintcream",
        "GradientColor3": "mintcream",
        "ShadowColor": [200, 200, 200],
        "FontColor": [10, 10, 10],
        "FontColorFaded": [100, 100, 100],
        "ConnectionPointColor": "white",
        "PenWidth": 2.0,
        "HoveredPenWidth": 2.5,
        "ConnectionPointDiameter": 10.0,
        "Opacity": 1.0
      },
      "ConnectionStyle": {
        "ConstructionColor": "gray",
        "NormalColor": "black",
        "SelectedColor": "gray",
        "SelectedHaloColor": "deepskyblue",
        "HoveredColor": "deepskyblue",

        "LineWidth": 3.0,
        "ConstructionLineWidth": 2.0,
        "PointDiameter": 10.0,

        "UseDataDefinedColors": false
      }
  }
'''


class MyNodeData(NodeData):
    data_type = NodeDataType(id='MyNodeData', name='My Node Data')


class MyDataModel(NodeDataModel):
    name = 'MyDataModel'
    caption = 'Caption'
    caption_visible = True
    num_ports = {PortType.input: 3,
                 PortType.output: 3,
                 }
    data_type = MyNodeData.data_type

    def out_data(self, port):
        return MyNodeData()

    def set_in_data(self, node_data, port):
        ...

    def embedded_widget(self):
        return None


def main(app):
    style = StyleCollection.from_json(style_json)

    registry = qtpynodeeditor.DataModelRegistry()
    registry.register_model(MyDataModel, category='My Category', style=style)
    scene = qtpynodeeditor.FlowScene(style=style, registry=registry)

    view = qtpynodeeditor.FlowView(scene)
    view.setWindowTitle("Style example")
    view.resize(800, 600)

    node = scene.create_node(MyDataModel)
    return scene, view, [node]


if __name__ == '__main__':
    logging.basicConfig(level='DEBUG')
    app = QtWidgets.QApplication([])
    scene, view, nodes = main(app)
    view.show()
    app.exec_()

step3.2 使用moduleを確認

• package
  • logging
  • QtWidgets
  • qtpynodeeditor.DataModelRegistry
  • qtpynodeeditor.FlowScene
  • qtpynodeeditor.FlowView
  • qtpynodeeditor
    • NodeData
    • NodeDataModel
    • NodeDataType
    • PortType
    • StyleCollection

↓ APIの説明順番に並びかえてみる(loggingとQtWidgetsは他ライブラリのため除外)

• API
  • Scene and View
    • FlowScene
    • FlowView
  • Styles
    • StyleCollection
  • Nodes
    • NodeDataModel
    • NodeState
    • NodeDataType
    • DataModelRegistry
  • Connections
  • Ports
    • PortType
  • Other / Internal

step3.3 複数の観点でmoduleをカテゴライズ、ソート

• 観点

  • import形式
    • 仮定：　作者がimport形式を変えているのには用途が異なる
  • moduleの実行場所
    • 仮定： module単位で利用方法（methodを使うのか？、classを生成するのか？, 継承に利用するのか？ importするだけで効果を発揮するのか？）が異なる
  • 実行順番でmoduleをそーと
    • 仮定： moduleの役割が異なるため、moduleの利用回数は実行順番に対してかたよる

• import形式

  • qtpynodeeditor.~ の形式
    • Scene and View
      • FlowScene
      • FlowView
    • Nodes
      • DataModelRegistry
  • from qtpynodeeditor import ~
    • Styles
      • StyleCollection
    • Nodes
      • NodeData
      • NodeDataModel
      • NodeDataType
    • Ports
      • PortType

• moduleの実行場所

  • main内
    • FlowScene
    • FlowView
    • StyleCollection
  • class継承
    • NodeData
    • NodeDataModel
  • class内
    • NodeDataType <- NodeData内
    • PortType　<- NodeDataModel内

• 実行順番でmoduleをそーと

  1. app = QtWidgets.QApplication([])
  2. scene, view, nodes = main(app)
     3. StyleCollection
     4. DataModelRegistry
     5. NodeDataModel
     6. PortType
     7. NodeData
     8. NodeDataType
     6. FlowScene
     7. FlowView
  3. view.show()
  4. app.exec_()

test code

top down

DataModelRegistry()

import qtpynodeeditor
registry = qtpynodeeditor.DataModelRegistry()
registry.register_model(MyDataModel, category='My Category', style=style)

botom up

a

source code

init.py

print('helloworld')

from .connection import Connection
from .connection_geometry import ConnectionGeometry
from .connection_graphics_object import ConnectionGraphicsObject
from .connection_painter import ConnectionPainter
from .data_model_registry import DataModelRegistry
from .enums import ConnectionPolicy, NodeValidationState, PortType
from .exceptions import (ConnectionPointFailure, ConnectionPortNotEmptyFailure,
                         ConnectionRequiresPortFailure, ConnectionSelfFailure,
                         NodeConnectionFailure)
from .flow_scene import FlowScene
from .flow_view import FlowView
from .node import Node, NodeDataType
from .node_connection_interaction import NodeConnectionInteraction
from .node_data import NodeData, NodeDataModel
from .node_geometry import NodeGeometry
from .node_graphics_object import NodeGraphicsObject
from .node_painter import NodePainter, NodePainterDelegate
from .node_state import NodeState
from .port import Port, opposite_port
from .style import (ConnectionStyle, FlowViewStyle, NodeStyle, Style,
                    StyleCollection)

__all__ = [
    'Connection',
    'ConnectionGeometry',
    'ConnectionGraphicsObject',
    'ConnectionPainter',
    'ConnectionPolicy',
    'ConnectionStyle',
    'ConnectionRequiresPortFailure',
    'ConnectionSelfFailure',
    'ConnectionPointFailure',
    'ConnectionPortNotEmptyFailure',
    'DataModelRegistry',
    'FlowScene',
    'FlowView',
    'FlowViewStyle',
    'Node',
    'NodeConnectionInteraction',
    'NodeConnectionFailure',
    'NodeData',
    'NodeDataModel',
    'NodeDataType',
    'NodeGeometry',
    'NodeGraphicsObject',
    'NodePainter',
    'NodePainterDelegate',
    'NodeState',
    'NodeStyle',
    'NodeValidationState',
    'Port',
    'PortType',
    'Style',
    'StyleCollection',
    'opposite_port',
]

data_model_registory.py

import logging

from .node_data import NodeDataModel, NodeDataType
from .type_converter import TypeConverter

logger = logging.getLogger(__name__)


class DataModelRegistry:
    def __init__(self):
        self.type_converters = {}
        self._models_category = {}
        self._item_creators = {}
        self._categories = set()

    def register_model(self, creator, category='', *, style=None, **init_kwargs):
        name = creator.name
        self._item_creators[name] = (creator, {'style': style, **init_kwargs})
        self._categories.add(category)
        self._models_category[name] = category

    def register_type_converter(self, type_in: NodeDataType, type_out:
                                NodeDataType, type_converter: TypeConverter):
        """
        Register type converter

        Parameters
        ----------
        id_ : NodeData subclass or TypeConverterId
        type_converter : TypeConverter
        """
        # TODO typing annotation
        if hasattr(type_in, 'type'):
            type_in = type_in.type
        if hasattr(type_out, 'type'):
            type_out = type_out.type

        self.type_converters[(type_in, type_out)] = type_converter

    def create(self, model_name: str) -> NodeDataModel:
        """
        Create

        Parameters
        ----------
        model_name : str

        Returns
        -------
        value : (NodeDataModel, init_kwargs)
        """
        cls, kwargs = self._item_creators[model_name]
        return cls(**kwargs)

    def registered_model_creators(self) -> dict:
        """
        Registered model creators

        Returns
        -------
        value : dict
        """
        return dict(self._item_creators)

    def registered_models_category_association(self) -> dict:
        """
        Registered models category association

        Returns
        -------
        value : DataModelRegistry.RegisteredModelsCategoryMap
        """
        return self._models_category

    def categories(self) -> set:
        """
        Categories

        Returns
        -------
        value : DataModelRegistry.CategoriesSet
        """
        return self._categories

    def get_type_converter(self, d1: NodeDataType, d2: NodeDataType) -> TypeConverter:
        """
        Get type converter

        Parameters
        ----------
        d1 : NodeDataType
        d2 : NodeDataType

        Returns
        -------
        value : TypeConverter
        """
        return self.type_converters.get((d1, d2), None)

flow_scene.py

import contextlib
import json
import os

from qtpy.QtCore import QDir, QObject, QPoint, QPointF, Qt, Signal
from qtpy.QtWidgets import QFileDialog, QGraphicsScene

from . import style as style_module
from .connection import Connection
from .connection_graphics_object import ConnectionGraphicsObject
from .data_model_registry import DataModelRegistry
from .node import Node
from .node_data import NodeDataModel, NodeDataType
from .node_graphics_object import NodeGraphicsObject
from .port import Port, PortType
from .type_converter import DefaultTypeConverter, TypeConverter


def locate_node_at(scene_point, scene, view_transform):
    items = scene.items(scene_point, Qt.IntersectsItemShape,
                        Qt.DescendingOrder, view_transform)
    filtered_items = [item for item in items
                      if isinstance(item, NodeGraphicsObject)]
    return filtered_items[0].node if filtered_items else None


class _FlowSceneModel:
    def __init__(self, registry=None):
        super().__init__()
        self._connections = []
        self._nodes = {}

        if registry is None:
            registry = DataModelRegistry()

        self._registry = registry

        # this connection should come first
        self.connection_created.connect(self._setup_connection_signals)
        self.connection_created.connect(self._send_connection_created_to_nodes)
        self.connection_deleted.connect(self._send_connection_deleted_to_nodes)

    @property
    def registry(self) -> DataModelRegistry:
        """
        Registry

        Returns
        -------
        value : DataModelRegistry
        """
        return self._registry

    @registry.setter
    def registry(self, registry: DataModelRegistry):
        self._registry = registry

    @property
    def nodes(self) -> dict:
        """
        All nodes in the scene

        Returns
        -------
        value : dict
            Key: uuid
            Value: Node
        """
        return dict(self._nodes)

    @property
    def connections(self) -> list:
        """
        All connections in the scene

        Returns
        -------
        conn : list of Connection
        """
        return list(self._connections)

    def clear_scene(self):
        # Manual node cleanup. Simply clearing the holding datastructures
        # doesn't work, the code crashes when there are both nodes and
        # connections in the scene. (The data propagation internal logic tries
        # to propagate data through already freed connections.)
        for conn in list(self._connections):
            self.delete_connection(conn)

        for node in list(self._nodes.values()):
            self.remove_node(node)

    def save(self, file_name=None):
        if file_name is None:
            file_name, _ = QFileDialog.getSaveFileName(
                None, "Open Flow Scene", QDir.homePath(),
                "Flow Scene Files (.flow)")

        if file_name:
            file_name = str(file_name)
            if not file_name.endswith(".flow"):
                file_name += ".flow"

            with open(file_name, 'wt') as f:
                json.dump(self.__getstate__(), f)

    def load(self, file_name=None):
        if file_name is None:
            file_name, _ = QFileDialog.getOpenFileName(
                None, "Open Flow Scene", QDir.homePath(),
                "Flow Scene Files (.flow)")

        if not os.path.exists(file_name):
            return

        with open(file_name, 'rt') as f:
            doc = json.load(f)

        self.__setstate__(doc)

    def __getstate__(self) -> dict:
        """
        Save scene state to a dictionary

        Returns
        -------
        value : dict
        """
        scene_json = {}
        nodes_json_array = []
        connection_json_array = []
        for node in self._nodes.values():
            nodes_json_array.append(node.__getstate__())

        scene_json["nodes"] = nodes_json_array
        for connection in self._connections:
            connection_json = connection.__getstate__()
            if connection_json:
                connection_json_array.append(connection_json)

        scene_json["connections"] = connection_json_array
        return scene_json

    def __setstate__(self, doc: dict):
        """
        Load scene state from a dictionary

        Parameters
        ----------
        doc : dict
            Dictionary of settings
        """
        self.clear_scene()

        for node in doc["nodes"]:
            self.restore_node(node)

        for connection in doc["connections"]:
            self.restore_connection(connection)

    def _setup_connection_signals(self, conn: Connection):
        """
        Setup connection signals

        Parameters
        ----------
        conn : Connection
        """
        conn.connection_made_incomplete.connect(
            self.connection_deleted.emit, Qt.UniqueConnection)

    def _send_connection_created_to_nodes(self, conn: Connection):
        """
        Send connection created to nodes

        Parameters
        ----------
        conn : Connection
        """
        input_node, output_node = conn.nodes
        assert input_node is not None
        assert output_node is not None
        output_node.model.output_connection_created(conn)
        input_node.model.input_connection_created(conn)

    def _send_connection_deleted_to_nodes(self, conn: Connection):
        """
        Send connection deleted to nodes

        Parameters
        ----------
        conn : Connection
        """
        input_node, output_node = conn.nodes
        assert input_node is not None
        assert output_node is not None
        output_node.model.output_connection_deleted(conn)
        input_node.model.input_connection_deleted(conn)

    def iterate_over_nodes(self):
        """
        Generator: Iterate over nodes
        """
        for node in self._nodes.values():
            yield node

    def iterate_over_node_data(self):
        """
        Generator: Iterate over node data
        """
        for node in self._nodes.values():
            yield node.model

    def iterate_over_node_data_dependent_order(self):
        """
        Generator: Iterate over node data dependent order
        """
        visited_nodes = []

        # A leaf node is a node with no input ports, or all possible input ports empty
        def is_node_leaf(node, model):
            for port in node[PortType.input].values():
                if not port.connections:
                    return False

            return True

        # Iterate over "leaf" nodes
        for node in self._nodes.values():
            model = node.model
            if is_node_leaf(node, model):
                yield model
                visited_nodes.append(node)

        def are_node_inputs_visited_before(node, model):
            for port in node[PortType.input].values():
                for conn in port.connections:
                    other = conn.get_node(PortType.output)
                    if visited_nodes and other == visited_nodes[-1]:
                        return False
            return True

        # Iterate over dependent nodes
        while len(self._nodes) != len(visited_nodes):
            for node in self._nodes.values():
                if node in visited_nodes and node is not visited_nodes[-1]:
                    continue

                model = node.model
                if are_node_inputs_visited_before(node, model):
                    yield model
                    visited_nodes.append(node)

    def to_digraph(self):
        '''
        Create a networkx digraph

        Returns
        -------
        digraph : networkx.DiGraph
            The generated DiGraph

        Raises
        ------
        ImportError
            If networkx is unavailable
        '''
        import networkx
        graph = networkx.DiGraph()
        for node in self._nodes.values():
            graph.add_node(node)

        for node in self._nodes.values():
            graph.add_edges_from(conn.nodes
                                 for conn in node.state.all_connections)

        return graph

    def remove_node(self, node: Node):
        """
        Remove node

        Parameters
        ----------
        node : Node
        """
        self.node_deleted.emit(node)
        for conn in list(node.state.all_connections):
            self.delete_connection(conn)

        node._cleanup()
        del self._nodes[node.id]

    def _restore_node(self, node_json: dict) -> Node:
        """
        Restore a node from a state dictionary

        Parameters
        ----------
        node_json : dict

        Returns
        -------
        value : Node
        """
        with self._new_node_context(node_json["model"]["name"]) as node:
            ...

        return node

    @contextlib.contextmanager
    def _new_node_context(self, data_model_name, *, emit_placed=False):
        'Context manager: creates Node/yields it, handling necessary Signals'
        data_model = self._registry.create(data_model_name)
        if not data_model:
            raise ValueError("No registered model with name {}"
                             "".format(data_model_name))

        node = Node(data_model)
        yield node
        self._nodes[node.id] = node
        if emit_placed:
            self.node_placed.emit(node)
        self.node_created.emit(node)

    def restore_node(self, node_json: dict) -> Node:
        """
        Restore a node from a state dictionary

        Parameters
        ----------
        node_json : dict

        Returns
        -------
        value : Node
        """
        name = node_json["model"]["name"]
        with self._new_node_context(name, emit_placed=True) as node:
            node.__setstate__(node_json)

        return node

    def delete_connection(self, connection: Connection):
        """
        Delete connection

        Parameters
        ----------
        connection : Connection
        """
        try:
            self._connections.remove(connection)
        except ValueError:
            ...
        else:
            connection.remove_from_nodes()
            connection._cleanup()


class FlowSceneModel(QObject):
    '''
    A model representing a flow scene

    Emits the following signals upon connection/node creation/deletion::

        connection_created : Signal(Connection)
        connection_deleted : Signal(Connection)
        node_created : Signal(Node)
        node_deleted : Signal(Node)
    '''
    connection_created = Signal(Connection)
    connection_deleted = Signal(Connection)

    node_created = Signal(Node)
    node_deleted = Signal(Node)


class FlowScene(QGraphicsScene, _FlowSceneModel):
    # Implement the FlowSceneModel signals:
    connection_created = Signal(Connection)
    connection_deleted = Signal(Connection)
    node_created = Signal(Node)
    node_deleted = Signal(Node)
    # End of re-implemented signals

    connection_hover_left = Signal(Connection)
    connection_hovered = Signal(Connection, QPoint)

    #  Node has been added to the scene.
    #  Connect to self signal if need a correct position of node.
    node_placed = Signal(Node)

    # node_context_menu(node, scene_position, screen_position)
    node_context_menu = Signal(Node, QPointF, QPoint)
    node_double_clicked = Signal(Node)
    node_hover_left = Signal(Node)
    node_hovered = Signal(Node, QPoint)
    node_moved = Signal(Node, QPointF)

    def __init__(self, registry=None, style=None, parent=None,
                 allow_node_creation=True, allow_node_deletion=True):
        '''
        Create a new flow scene

        Parameters
        ----------
        registry : DataModelRegistry, optional
        style : StyleCollection, optional
        parent : QObject, optional
        '''
        # Note: PySide2 does not support a cooperative __init__, meaning we
        # cannot use super().__init__ here.
        # super().__init__(registry=registry, parent=parent)
        QGraphicsScene.__init__(self, parent=parent)
        _FlowSceneModel.__init__(self, registry=registry)

        if style is None:
            style = style_module.default_style

        self._style = style
        self.allow_node_deletion = allow_node_creation
        self.allow_node_creation = allow_node_deletion

        self.setItemIndexMethod(QGraphicsScene.NoIndex)

    def _cleanup(self):
        self.clear_scene()

    def __del__(self):
        try:
            self._cleanup()
        except Exception:
            ...

    @property
    def allow_node_creation(self):
        return self._allow_node_creation

    @allow_node_creation.setter
    def allow_node_creation(self, allow):
        self._allow_node_creation = bool(allow)

    @property
    def allow_node_deletion(self):
        return self._allow_node_deletion

    @allow_node_deletion.setter
    def allow_node_deletion(self, allow):
        self._allow_node_deletion = bool(allow)

    @property
    def style_collection(self) -> style_module.StyleCollection:
        'The style collection for the scene'
        return self._style

    def locate_node_at(self, point, transform):
        return locate_node_at(point, self, transform)

    def create_connection(self, port_a: Port, port_b: Port = None, *,
                          converter: TypeConverter = None) -> Connection:
        """
        Create a connection

        Parameters
        ----------
        port_a : Port
            The first port, either input or output
        port_b : Port, optional
            The second port, opposite of the type of port_a
        converter : TypeConverter, optional
            The type converter to use for data propagation

        Returns
        -------
        value : Connection
        """
        connection = Connection(port_a=port_a, port_b=port_b, style=self._style)
        if port_a is not None:
            port_a.add_connection(connection)
        if port_b is not None:
            port_b.add_connection(connection)

        cgo = ConnectionGraphicsObject(self, connection)
        # after self function connection points are set to node port
        connection.graphics_object = cgo
        self._connections.append(connection)

        if not port_a or not port_b:
            # This connection isn't truly created yet. It's only partially
            # created.  Thus, don't send the connection_created(...) signal.
            connection.connection_completed.connect(self.connection_created.emit)
        else:
            in_port, out_port = connection.ports
            out_port.node.on_data_updated(out_port)
            self.connection_created.emit(connection)

        return connection

    def create_connection_by_index(
            self, node_in: Node, port_index_in: int,
            node_out: Node, port_index_out: int,
            converter: TypeConverter) -> Connection:
        """
        Create connection

        Parameters
        ----------
        node_in : Node
        port_index_in : int
        node_out : Node
        port_index_out : int
        converter : TypeConverter

        Returns
        -------
        value : Connection
        """
        port_in = node_in[PortType.input][port_index_in]
        port_out = node_out[PortType.output][port_index_out]
        return self.create_connection(port_out, port_in, converter=converter)

    def restore_connection(self, connection_json: dict) -> Connection:
        """
        Restore connection

        Parameters
        ----------
        connection_json : dict

        Returns
        -------
        value : Connection
        """
        node_in_id = connection_json["in_id"]
        node_out_id = connection_json["out_id"]

        port_index_in = connection_json["in_index"]
        port_index_out = connection_json["out_index"]
        node_in = self._nodes[node_in_id]
        node_out = self._nodes[node_out_id]

        def get_converter():
            converter = connection_json.get("converter", None)
            if converter is None:
                return DefaultTypeConverter

            in_type = NodeDataType(
                id=converter["in"]["id"],
                name=converter["in"]["name"],
            )

            out_type = NodeDataType(
                id=converter["out"]["id"],
                name=converter["out"]["name"],
            )

            return self._registry.get_type_converter(out_type, in_type)

        connection = self.create_connection_by_index(
            node_in, port_index_in,
            node_out, port_index_out,
            converter=get_converter())

        # Note: the connection_created(...) signal has already been sent by
        # create_connection(...)
        return connection

    def create_node(self, data_model: NodeDataModel) -> Node:
        """
        Create a node in the scene

        Parameters
        ----------
        data_model : NodeDataModel

        Returns
        -------
        value : Node
        """
        with self._new_node_context(data_model.name) as node:
            ngo = NodeGraphicsObject(self, node)
            node.graphics_object = ngo

        return node

    def restore_node(self, node_json: dict) -> Node:
        """
        Restore a node from a state dictinoary

        Parameters
        ----------
        node_json : dict

        Returns
        -------
        value : Node
        """
        # NOTE: Overrides FlowSceneModel.restore_node
        with self._new_node_context(node_json["model"]["name"]) as node:
            node.graphics_object = NodeGraphicsObject(self, node)
            node.__setstate__(node_json)
        return node

    def auto_arrange(self, layout='bipartite', scale=700, align='horizontal',
                     **kwargs):
        '''
        Automatically arrange nodes with networkx, if available

        Raises
        ------
        ImportError
            If networkx is unavailable
        '''
        import networkx
        dig = self.to_digraph()

        layouts = {
            name: getattr(networkx.layout, '{}_layout'.format(name))
            for name in ('bipartite', 'circular', 'kamada_kawai', 'random',
                         'shell', 'spring', 'spectral')
        }

        try:
            layout_func = layouts[layout]
        except KeyError:
            raise ValueError('Unknown layout type {}'.format(layout)) from None

        layout = layout_func(dig, **kwargs)
        for node, pos in layout.items():
            pos_x, pos_y = pos
            node.position = (pos_x * scale, pos_y * scale)

    def selected_nodes(self) -> list:
        """
        Selected nodes

        Returns
        -------
        value : list of Node
        """
        return [item.node for item in self.selectedItems()
                if isinstance(item, NodeGraphicsObject)]

flow_view.py

import logging
import math

from qtpy.QtCore import QLineF, QPoint, QRectF, Qt
from qtpy.QtGui import (QContextMenuEvent, QKeyEvent, QMouseEvent, QPainter,
                        QPen, QShowEvent, QWheelEvent, QKeySequence)
from qtpy.QtWidgets import (QAction, QGraphicsView, QLineEdit, QMenu,
                            QTreeWidget, QTreeWidgetItem, QWidgetAction)

from .connection_graphics_object import ConnectionGraphicsObject
from .flow_scene import FlowScene
from .node_graphics_object import NodeGraphicsObject

logger = logging.getLogger(__name__)


class FlowView(QGraphicsView):
    def __init__(self, scene, parent=None):
        super().__init__(parent=parent)

        self._clear_selection_action = None
        self._delete_selection_action = None
        self._scene = None
        self._click_pos = None

        self.setDragMode(QGraphicsView.ScrollHandDrag)
        self.setRenderHint(QPainter.Antialiasing)

        # setViewportUpdateMode(QGraphicsView.FullViewportUpdate)
        # setViewportUpdateMode(QGraphicsView.MinimalViewportUpdate)
        self.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
        self.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)

        self.setTransformationAnchor(QGraphicsView.AnchorUnderMouse)

        self.setCacheMode(QGraphicsView.CacheBackground)
        # setViewport(new QGLWidget(QGLFormat(QGL.SampleBuffers)))
        if scene is not None:
            self.setScene(scene)

        self._style = self._scene.style_collection
        self.setBackgroundBrush(self._style.flow_view.background_color)

    def clear_selection_action(self) -> QAction:
        """
        Clear selection action

        Returns
        -------
        value : QAction
        """
        return self._clear_selection_action

    def delete_selection_action(self) -> QAction:
        """
        Delete selection action

        Returns
        -------
        value : QAction
        """
        return self._delete_selection_action

    def setScene(self, scene: FlowScene):
        """
        setScene

        Parameters
        ----------
        scene : FlowScene
        """
        self._scene = scene
        super().setScene(self._scene)

        # setup actions
        del self._clear_selection_action
        self._clear_selection_action = QAction("Clear Selection", self)
        self._clear_selection_action.setShortcut(QKeySequence.Cancel)
        self._clear_selection_action.triggered.connect(self._scene.clearSelection)

        self.addAction(self._clear_selection_action)
        del self._delete_selection_action
        self._delete_selection_action = QAction("Delete Selection", self)
        self._delete_selection_action.setShortcut(QKeySequence.Backspace)
        self._delete_selection_action.setShortcut(QKeySequence.Delete)
        self._delete_selection_action.triggered.connect(self.delete_selected)
        self.addAction(self._delete_selection_action)

    def scale_up(self):
        step = 1.2
        factor = step ** 1.0
        t = self.transform()
        if t.m11() <= 2.0:
            self.scale(factor, factor)

    def scale_down(self):
        step = 1.2
        factor = step ** -1.0
        self.scale(factor, factor)

    def delete_selected(self):
        # Delete the selected connections first, ensuring that they won't be
        # automatically deleted when selected nodes are deleted (deleting a node
        # deletes some connections as well)
        for item in self._scene.selectedItems():
            if isinstance(item, ConnectionGraphicsObject):
                self._scene.delete_connection(item.connection)

        if not self._scene.allow_node_deletion:
            return

        # Delete the nodes; self will delete many of the connections.
        # Selected connections were already deleted prior to self loop, otherwise
        # qgraphicsitem_cast<NodeGraphicsObject*>(item) could be a use-after-free
        # when a selected connection is deleted by deleting the node.
        for item in self._scene.selectedItems():
            if isinstance(item, NodeGraphicsObject):
                self._scene.remove_node(item.node)

    def generate_context_menu(self, pos: QPoint):
        """
        Generate a context menu for contextMenuEvent

        Parameters
        ----------
        pos : QPoint
            The point where the context menu was requested
        """
        model_menu = QMenu()
        skip_text = "skip me"

        # Add filterbox to the context menu
        txt_box = QLineEdit(model_menu)
        txt_box.setPlaceholderText("Filter")
        txt_box.setClearButtonEnabled(True)
        txt_box_action = QWidgetAction(model_menu)
        txt_box_action.setDefaultWidget(txt_box)
        model_menu.addAction(txt_box_action)

        # Add result treeview to the context menu
        tree_view = QTreeWidget(model_menu)
        tree_view.header().close()
        tree_view_action = QWidgetAction(model_menu)
        tree_view_action.setDefaultWidget(tree_view)
        model_menu.addAction(tree_view_action)

        top_level_items = {}
        for cat in self._scene.registry.categories():
            item = QTreeWidgetItem(tree_view)
            item.setText(0, cat)
            item.setData(0, Qt.UserRole, skip_text)
            top_level_items[cat] = item

        registry = self._scene.registry
        for model, category in registry.registered_models_category_association().items():
            self.parent = top_level_items[category]
            item = QTreeWidgetItem(self.parent)
            item.setText(0, model)
            item.setData(0, Qt.UserRole, model)

        tree_view.expandAll()

        def click_handler(item):
            model_name = item.data(0, Qt.UserRole)
            if model_name == skip_text:
                return

            type_ = self._scene.registry.create(model_name)
            if type_:
                node = self._scene.create_node(type_)
                pos_view = self.mapToScene(pos)
                node.graphics_object.setPos(pos_view)
                self._scene.node_placed.emit(node)
            else:
                logger.debug("Model not found")

            model_menu.close()

        tree_view.itemClicked.connect(click_handler)

        # Setup filtering
        def filter_handler(text):
            for name, top_lvl_item in top_level_items.items():
                for i in range(top_lvl_item.childCount()):
                    child = top_lvl_item.child(i)
                    model_name = child.data(0, Qt.UserRole)
                    child.setHidden(text not in model_name)

        txt_box.textChanged.connect(filter_handler)

        # make sure the text box gets focus so the user doesn't have to click on it
        txt_box.setFocus()
        return model_menu

    def contextMenuEvent(self, event: QContextMenuEvent):
        """
        contextMenuEvent

        Parameters
        ----------
        event : QContextMenuEvent
        """
        if self.itemAt(event.pos()):
            super().contextMenuEvent(event)
            return
        elif not self._scene.allow_node_creation:
            return

        menu = self.generate_context_menu(event.pos())
        menu.exec_(event.globalPos())

    def wheelEvent(self, event: QWheelEvent):
        """
        wheelEvent

        Parameters
        ----------
        event : QWheelEvent
        """
        delta = event.angleDelta()
        if delta.y() == 0:
            event.ignore()
            return

        d = delta.y() / abs(delta.y())
        if d > 0.0:
            self.scale_up()
        else:
            self.scale_down()

    def keyPressEvent(self, event: QKeyEvent):
        """
        keyPressEvent

        Parameters
        ----------
        event : QKeyEvent
        """
        if event.key() == Qt.Key_Shift:
            self.setDragMode(QGraphicsView.RubberBandDrag)

        super().keyPressEvent(event)

    def keyReleaseEvent(self, event: QKeyEvent):
        """
        keyReleaseEvent

        Parameters
        ----------
        event : QKeyEvent
        """
        if event.key() == Qt.Key_Shift:
            self.setDragMode(QGraphicsView.ScrollHandDrag)
        super().keyReleaseEvent(event)

    def mousePressEvent(self, event: QMouseEvent):
        """
        mousePressEvent

        Parameters
        ----------
        event : QMouseEvent
        """
        super().mousePressEvent(event)
        if event.button() == Qt.LeftButton:
            self._click_pos = self.mapToScene(event.pos())

    def mouseMoveEvent(self, event: QMouseEvent):
        """
        mouseMoveEvent

        Parameters
        ----------
        event : QMouseEvent
        """
        super().mouseMoveEvent(event)
        if self._scene.mouseGrabberItem() is None and event.buttons() == Qt.LeftButton:
            # Make sure shift is not being pressed
            if not (event.modifiers() & Qt.ShiftModifier):
                difference = self._click_pos - self.mapToScene(event.pos())
                self.setSceneRect(self.sceneRect().translated(difference.x(), difference.y()))

    def drawBackground(self, painter: QPainter, r: QRectF):
        """
        drawBackground

        Parameters
        ----------
        painter : QPainter
        r : QRectF
        """
        super().drawBackground(painter, r)

        def draw_grid(grid_step):
            window_rect = self.rect()
            tl = self.mapToScene(window_rect.topLeft())
            br = self.mapToScene(window_rect.bottomRight())
            left = math.floor(tl.x() / grid_step - 0.5)
            right = math.floor(br.x() / grid_step + 1.0)
            bottom = math.floor(tl.y() / grid_step - 0.5)
            top = math.floor(br.y() / grid_step + 1.0)

            # vertical lines
            lines = [
                QLineF(xi * grid_step, bottom * grid_step, xi * grid_step, top * grid_step)
                for xi in range(int(left), int(right) + 1)
            ]

            # horizontal lines
            lines.extend(
                [QLineF(left * grid_step, yi * grid_step, right * grid_step, yi * grid_step)
                 for yi in range(int(bottom), int(top) + 1)
                 ]
            )

            painter.drawLines(lines)

        style = self._style.flow_view
        # brush = self.backgroundBrush()
        pfine = QPen(style.fine_grid_color, 1.0)
        painter.setPen(pfine)
        draw_grid(15)
        p = QPen(style.coarse_grid_color, 1.0)
        painter.setPen(p)
        draw_grid(150)

    def showEvent(self, event: QShowEvent):
        """
        showEvent

        Parameters
        ----------
        event : QShowEvent
        """
        self._scene.setSceneRect(QRectF(self.rect()))
        super().showEvent(event)

    @property
    def scene(self) -> FlowScene:
        """
        Scene

        Returns
        -------
        value : FlowScene
        """
        return self._scene

node_data.py

import inspect
from collections import namedtuple

from qtpy.QtCore import QObject, Signal
from qtpy.QtWidgets import QWidget

from . import style as style_module
from .base import Serializable
from .enums import ConnectionPolicy, NodeValidationState, PortType
from .port import Port

NodeDataType = namedtuple('NodeDataType', ('id', 'name'))


class NodeData:
    """
    Class represents data transferred between nodes.

    The actual data is stored in subtypes
    """

    data_type = NodeDataType(None, None)

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        if cls.data_type is None:
            raise ValueError('Subclasses must set the `data_type` attribute')

    def same_type(self, other) -> bool:
        """
        Is another NodeData instance of the same type?

        Parameters
        ----------
        other : NodeData

        Returns
        -------
        value : bool
        """
        return self.data_type.id == other.data_type.id


class NodeDataModel(QObject, Serializable):
    name = None
    caption = None
    caption_visible = True
    num_ports = {PortType.input: 1,
                 PortType.output: 1,
                 }

    # data_updated and data_invalidated refer to the port index that has
    # changed:
    data_updated = Signal(int)
    data_invalidated = Signal(int)

    computing_started = Signal()
    computing_finished = Signal()
    embedded_widget_size_updated = Signal()

    def __init__(self, style=None, parent=None):
        super().__init__(parent=parent)
        if style is None:
            style = style_module.default_style
        self._style = style

    def __init_subclass__(cls, verify=True, **kwargs):
        super().__init_subclass__(**kwargs)
        # For all subclasses, if no name is defined, default to the class name
        if cls.name is None:
            cls.name = cls.__name__
        if cls.caption is None and cls.caption_visible:
            cls.caption = cls.name

        num_ports = cls.num_ports
        if isinstance(num_ports, property):
            # Dynamically defined - that's OK, but we can't verify it.
            return

        if verify:
            cls._verify()

    @classmethod
    def _verify(cls):
        '''
        Verify the data model won't crash in strange spots
        Ensure valid dictionaries:
            - num_ports
            - data_type
            - port_caption
            - port_caption_visible
        '''
        num_ports = cls.num_ports
        if isinstance(num_ports, property):
            # Dynamically defined - that's OK, but we can't verify it.
            return

        assert set(num_ports.keys()) == {'input', 'output'}

        # TODO while the end result is nicer, this is ugly; refactor away...

        def new_dict(value):
            return {
                PortType.input: {i: value
                                 for i in range(num_ports[PortType.input])
                                 },
                PortType.output: {i: value
                                  for i in range(num_ports[PortType.output])
                                  },
            }

        def get_default(attr, default, valid_type):
            current = getattr(cls, attr, None)
            if current is None:
                # Unset - use the default
                return default

            if valid_type is not None:
                if isinstance(current, valid_type):
                    # Fill in the dictionary with the user-provided value
                    return current

            if attr == 'data_type' and inspect.isclass(current):
                if issubclass(current, NodeData):
                    return current.data_type

            if inspect.ismethod(current) or inspect.isfunction(current):
                raise ValueError('{} should not be a function; saw: {}\n'
                                 'Did you forget a @property decorator?'
                                 ''.format(attr, current))

            try:
                type(default)(current)
            except TypeError:
                raise ValueError('{} is of an unexpected type: {}'
                                 ''.format(attr, current)) from None

            # Fill in the dictionary with the given value
            return current

        def fill_defaults(attr, default, valid_type=None):
            if isinstance(getattr(cls, attr, None), dict):
                return

            default = get_default(attr, default, valid_type)
            if default is None:
                raise ValueError('Cannot leave {} unspecified'.format(attr))

            setattr(cls, attr, new_dict(default))

        fill_defaults('port_caption', '')
        fill_defaults('port_caption_visible', False)
        fill_defaults('data_type', None, valid_type=NodeDataType)

        reasons = []
        for attr in ('data_type', 'port_caption', 'port_caption_visible'):
            try:
                dct = getattr(cls, attr)
            except AttributeError:
                reasons.append('{} is missing dictionary: {}'
                               ''.format(cls.__name__, attr))
                continue

            if isinstance(dct, property):
                continue

            for port_type in {'input', 'output'}:
                if port_type not in dct:
                    if num_ports[port_type] == 0:
                        dct[port_type] = {}
                    else:
                        reasons.append('Port type key {}[{!r}] missing'
                                       ''.format(attr, port_type))
                        continue

                for i in range(num_ports[port_type]):
                    if i not in dct[port_type]:
                        reasons.append('Port key {}[{!r}][{}] missing'
                                       ''.format(attr, port_type, i))

        if reasons:
            reason_text = '\n'.join('* {}'.format(reason)
                                    for reason in reasons)
            raise ValueError(
                'Verification of NodeDataModel class failed:\n{}'
                ''.format(reason_text)
            )

    @property
    def style(self):
        'Style collection for drawing this data model'
        return self._style

    def save(self) -> dict:
        """
        Subclasses may implement this to save additional state for
        pickling/saving to JSON.

        Returns
        -------
        value : dict
        """
        return {}

    def restore(self, doc: dict):
        """
        Subclasses may implement this to load additional state from
        pickled or saved-to-JSON data.

        Parameters
        ----------
        value : dict
        """
        return {}

    def __setstate__(self, doc: dict):
        """
        Set the state of the NodeDataModel

        Parameters
        ----------
        doc : dict
        """
        self.restore(doc)
        return doc

    def __getstate__(self) -> dict:
        """
        Get the state of the NodeDataModel for saving/pickling

        Returns
        -------
        value : QJsonObject
        """
        doc = {'name': self.name}
        doc.update(**self.save())
        return doc

    @property
    def data_type(self):
        """
        Data type placeholder - to be implemented by subclass.

        Parameters
        ----------
        port_type : PortType
        port_index : int

        Returns
        -------
        value : NodeDataType
        """
        raise NotImplementedError(f'Subclass {self.__class__.__name__} must '
                                  f'implement `data_type`')

    def port_out_connection_policy(self, port_index: int) -> ConnectionPolicy:
        """
        Port out connection policy

        Parameters
        ----------
        port_index : int

        Returns
        -------
        value : ConnectionPolicy
        """
        return ConnectionPolicy.many

    @property
    def node_style(self) -> style_module.NodeStyle:
        """
        Node style

        Returns
        -------
        value : NodeStyle
        """
        return self._style.node

    def set_in_data(self, node_data: NodeData, port: Port):
        """
        Triggers the algorithm; to be overridden by subclasses

        Parameters
        ----------
        node_data : NodeData
        port : Port
        """
        ...

    def out_data(self, port: int) -> NodeData:
        """
        Out data

        Parameters
        ----------
        port : int

        Returns
        -------
        value : NodeData
        """
        ...

    def embedded_widget(self) -> QWidget:
        """
        Embedded widget

        Returns
        -------
        value : QWidget
        """
        ...

    def resizable(self) -> bool:
        """
        Resizable

        Returns
        -------
        value : bool
        """
        return False

    def validation_state(self) -> NodeValidationState:
        """
        Validation state

        Returns
        -------
        value : NodeValidationState
        """
        return NodeValidationState.valid

    def validation_message(self) -> str:
        """
        Validation message

        Returns
        -------
        value : str
        """
        return ""

    def painter_delegate(self):
        """
        Painter delegate

        Returns
        -------
        value : NodePainterDelegate
        """
        return None

    def input_connection_created(self, connection):
        """
        Input connection created

        Parameters
        ----------
        connection : Connection
        """
        ...

    def input_connection_deleted(self, connection):
        """
        Input connection deleted

        Parameters
        ----------
        connection : Connection
        """
        ...

    def output_connection_created(self, connection):
        """
        Output connection created

        Parameters
        ----------
        connection : Connection
        """
        ...

    def output_connection_deleted(self, connection):
        """
        Output connection deleted

        Parameters
        ----------
        connection : Connection
        """
        ...

node.py

import uuid

from qtpy.QtCore import QObject, QPointF, QSizeF

from .base import NodeBase, Serializable
from .enums import ReactToConnectionState
from .node_data import NodeData, NodeDataModel, NodeDataType
from .node_geometry import NodeGeometry
from .node_graphics_object import NodeGraphicsObject
from .node_state import NodeState
from .port import Port, PortType
from .style import NodeStyle


class Node(QObject, Serializable, NodeBase):
    def __init__(self, data_model: NodeDataModel):
        '''
        A single Node in the scene

        Parameters
        ----------
        data_model : NodeDataModel
        '''
        super().__init__()
        self._model = data_model
        self._uid = str(uuid.uuid4())
        self._style = data_model.node_style
        self._state = NodeState(self)
        self._geometry = NodeGeometry(self)
        self._graphics_obj = None
        self._geometry.recalculate_size()

        # propagate data: model => node
        self._model.data_updated.connect(self._on_port_index_data_updated)
        self._model.embedded_widget_size_updated.connect(self.on_node_size_updated)

    def __getitem__(self, key):
        return self._state[key]

    def _cleanup(self):
        if self._graphics_obj is not None:
            self._graphics_obj._cleanup()
            self._graphics_obj = None
            self._geometry = None

    def __del__(self):
        try:
            self._cleanup()
        except Exception:
            ...

    def __getstate__(self) -> dict:
        """
        Save

        Returns
        -------
        value : dict
        """
        return {
            "id": self._uid,
            "model": self._model.__getstate__(),
            "position": {"x": self._graphics_obj.pos().x(),
                         "y": self._graphics_obj.pos().y()}
        }

    def __setstate__(self, state: dict):
        """
        Restore

        Parameters
        ----------
        state : dict
        """
        self._uid = state["id"]
        if self._graphics_obj:
            pos = state["position"]
            self.position = (pos["x"], pos["y"])
        self._model.__setstate__(state["model"])

    @property
    def id(self) -> str:
        """
        Node unique identifier (uuid)

        Returns
        -------
        value : str
        """
        return self._uid

    def react_to_possible_connection(self, reacting_port_type: PortType,
                                     reacting_data_type: NodeDataType,
                                     scene_point: QPointF
                                     ):
        """
        React to possible connection

        Parameters
        ----------
        port_type : PortType
        node_data_type : NodeDataType
        scene_point : QPointF
        """
        transform = self._graphics_obj.sceneTransform()
        inverted, invertible = transform.inverted()
        if invertible:
            pos = inverted.map(scene_point)
            self._geometry.dragging_position = pos
        self._graphics_obj.update()
        self._state.set_reaction(ReactToConnectionState.reacting,
                                 reacting_port_type, reacting_data_type)

    def reset_reaction_to_connection(self):
        self._state.set_reaction(ReactToConnectionState.not_reacting)
        self._graphics_obj.update()

    @property
    def graphics_object(self) -> NodeGraphicsObject:
        """
        Node graphics object

        Returns
        -------
        value : NodeGraphicsObject
        """
        return self._graphics_obj

    @graphics_object.setter
    def graphics_object(self, graphics: NodeGraphicsObject):
        """
        Set graphics object

        Parameters
        ----------
        graphics : NodeGraphicsObject
        """
        self._graphics_obj = graphics
        self._geometry.recalculate_size()

    @property
    def geometry(self) -> NodeGeometry:
        """
        Node geometry

        Returns
        -------
        value : NodeGeometry
        """
        return self._geometry

    @property
    def model(self) -> NodeDataModel:
        """
        Node data model

        Returns
        -------
        value : NodeDataModel
        """
        return self._model

    def propagate_data(self, node_data: NodeData, input_port: Port):
        """
        Propagates incoming data to the underlying model.

        Parameters
        ----------
        node_data : NodeData
        input_port : int
        """
        if input_port.node is not self:
            raise ValueError('Port does not belong to this Node')
        elif input_port.port_type != PortType.input:
            raise ValueError('Port is not an input port')

        self._model.set_in_data(node_data, input_port)

        # Recalculate the nodes visuals. A data change can result in the node
        # taking more space than before, so self forces a recalculate+repaint
        # on the affected node
        self._graphics_obj.set_geometry_changed()
        self._geometry.recalculate_size()
        self._graphics_obj.update()
        self._graphics_obj.move_connections()

    def _on_port_index_data_updated(self, port_index: int):
        """
        Data has been updated on this Node's output port port_index;
        propagate it to any connections.

        Parameters
        ----------
        index : int
        """
        port = self[PortType.output][port_index]
        self.on_data_updated(port)

    def on_data_updated(self, port: Port):
        """
        Fetches data from model's output port and propagates it along the
        connection

        Parameters
        ----------
        port : Port
        """
        node_data = port.data
        for conn in port.connections:
            conn.propagate_data(node_data)

    def on_node_size_updated(self):
        """
        update the graphic part if the size of the embeddedwidget changes
        """
        widget = self.model.embedded_widget()
        if widget:
            widget.adjustSize()

        self.geometry.recalculate_size()
        for conn in self.state.all_connections:
            conn.graphics_object.move()

    @property
    def size(self) -> QSizeF:
        """
        Get the node size

        Parameters
        ----------
        node : Node

        Returns
        -------
        value : QSizeF
        """
        return self._geometry.size

    @property
    def position(self) -> QPointF:
        """
        Get the node position

        Parameters
        ----------
        node : Node

        Returns
        -------
        value : QPointF
        """
        return self._graphics_obj.pos()

    @position.setter
    def position(self, pos):
        if not isinstance(pos, QPointF):
            px, py = pos
            pos = QPointF(px, py)

        self._graphics_obj.setPos(pos)
        self._graphics_obj.move_connections()

    @property
    def style(self) -> NodeStyle:
        'Node style'
        return self._style

    @property
    def state(self) -> NodeState:
        """
        Node state

        Returns
        -------
        value : NodeState
        """
        return self._state

    def __repr__(self):
        return (f'<{self.__class__.__name__} model={self._model} '
                f'uid={self._uid!r}>')

enums.py

from enum import Enum


class NodeValidationState(str, Enum):
    valid = 'valid'
    warning = 'warning'
    error = 'error'


class PortType(str, Enum):
    none = 'none'
    input = 'input'
    output = 'output'


class ConnectionPolicy(str, Enum):
    one = 'one'
    many = 'many'


class ReactToConnectionState(str, Enum):
    reacting = 'reacting'
    not_reacting = 'not_reacting'

port.py

from qtpy.QtCore import QObject, Signal

from .base import ConnectionBase
from .enums import ConnectionPolicy, PortType


def opposite_port(port: PortType):
    return {PortType.input: PortType.output,
            PortType.output: PortType.input}.get(port, PortType.none)


class Port(QObject):
    connection_created = Signal(ConnectionBase)
    connection_deleted = Signal(ConnectionBase)
    data_updated = Signal(QObject)
    data_invalidated = Signal(QObject)

    def __init__(self, node, *, port_type: PortType, index: int):
        super().__init__(parent=node)
        self.node = node
        self.port_type = port_type
        self.index = index
        self._connections = []
        self.opposite_port = {PortType.input: PortType.output,
                              PortType.output: PortType.input}[self.port_type]

    @property
    def connections(self):
        return list(self._connections)

    @property
    def model(self):
        'The data model associated with the Port'
        return self.node.model

    @property
    def data(self):
        'The NodeData associated with the Port, if an output port'
        if self.port_type == PortType.input:
            # return self.model.in_data(self.index)
            # TODO
            return
        else:
            return self.model.out_data(self.index)

    @property
    def can_connect(self):
        'Can this port be connected to?'
        return (not self._connections or
                self.connection_policy == ConnectionPolicy.many)

    @property
    def caption(self):
        'Data model-specified caption for the port'
        return self.model.port_caption[self.port_type][self.index]

    @property
    def caption_visible(self):
        'Show the data model-specified caption?'
        return self.model.port_caption_visible[self.port_type][self.index]

    @property
    def data_type(self):
        'The NodeData type associated with the Port'
        return self.model.data_type[self.port_type][self.index]

    @property
    def display_text(self):
        'The text to show on the label caption'
        return (self.caption
                if self.caption_visible
                else self.data_type.name)

    @property
    def connection_policy(self):
        'The connection policy (one/many) for the port'
        if self.port_type == PortType.input:
            return ConnectionPolicy.one
        else:
            return self.model.port_out_connection_policy(self.index)

    def add_connection(self, connection: ConnectionBase):
        'Add a Connection to the Port'
        if connection in self._connections:
            raise ValueError('Connection already in list')

        self._connections.append(connection)
        self.connection_created.emit(connection)

    def remove_connection(self, connection: ConnectionBase):
        'Remove a Connection from the Port'
        try:
            self._connections.remove(connection)
        except ValueError:
            # TODO: should not be reaching this
            ...
        else:
            self.connection_deleted.emit(connection)

    @property
    def scene_position(self):
        '''
        The position in the scene of the Port

        Returns
        -------
        value : QPointF

        See also
        --------
        get_mapped_scene_position
        '''
        return self.node.geometry.port_scene_position(self.port_type,
                                                      self.index)

    def get_mapped_scene_position(self, transform):
        """
        Node port scene position after a transform

        Parameters
        ----------
        port_type : PortType
        port_index : int

        Returns
        -------
        value : QPointF
        """
        ngo = self.node.graphics_object
        return ngo.sceneTransform().map(self.scene_position)

    def __repr__(self):
        return (f'<{self.__class__.__name__} port_type={self.port_type} '
                f'index={self.index} connections={len(self._connections)}>')