Class Graph

masfactory/components/graphs/graph.py:12–262

Composite graph node with internal entry and exit ports. Graph is both a node and a container. It can be nested inside another graph.

from masfactory.core.node_template import NodeTemplate

class Graph(BaseGraph):
    """
    Composite graph node with internal entry and exit ports.

    Graph is both a node and a container. It can be nested inside another graph.
    """
    def __init__(
        self,
        name,
        pull_keys: dict[str, dict|str] | None = None,
        push_keys: dict[str, dict|str] | None = None,
        attributes: dict[str, object] | None = None,
        edges: list[tuple[str, str] | tuple[str, str, dict[str, dict|str] ]]|None=None,
        nodes: list[tuple[str, NodeTemplate]] | None = None,
        build_func:Callable|None = None,
    ):
        """Create a composite graph node with internal entry/exit ports.

        Args:
            name: Graph name.
            pull_keys: Attribute pull rule for this graph.
            push_keys: Attribute push rule for this graph.
            attributes: Default attributes for this graph.
            edges: Optional declarative edge definitions. Each item is either:
                - `(from_name, to_name)`
                - `(from_name, to_name, edge_keys)`
            nodes: Optional declarative node definitions as `(name, NodeTemplate)` entries.
            build_func: Optional custom build function executed before child build.
                Signature: `(graph: Graph) -> None`.
        """
        if attributes is None:
            attributes = {}
        super().__init__(name, pull_keys, push_keys, attributes, build_func=build_func)
        self._init_nodes = nodes
        self._init_edges = edges
        class EntryNode(InternalGraphNode):
            def __init__(self, name, gate_close_callback:Callable|None=None):
                super().__init__(name, gate_close_callback)
                self.input: dict[str,object] = {}
            @masf_hook(Node.Hook.FORWARD)
            def _forward(self,input:dict[str,object]) -> dict[str,object]:
                return input
            def _message_aggregate_in(self) -> dict[str,object]:     
                return self.input
            def _update_gate_state(self):
                pass
            @masf_hook(Node.Hook.BUILD)
            def build(self):
                self._is_built = True

        class ExitNode(InternalGraphNode):
            def __init__(self, name, gate_close_callback:Callable|None=None):
                super().__init__(name, gate_close_callback)
                self._output = {}
            @masf_hook(Node.Hook.FORWARD)
            def _forward(self,input:dict[str,object]) -> dict[str,object]:
                return input
            def _message_dispatch_out(self,message:dict[str,object]):