FEAT(OP,.TC): +support node-kws (for predicate narrowing) ...

+ feat(op): +withset() method.
+ refact(netop): merge reworked.
+ test(op, netop): check also node_props  reach network also on merges.

Author: ankostis

graphtik/netop.py

@@ -13,7 +13,7 @@
 from .base import Plotter, aslist, astuple, jetsam
 from .modifiers import optional, sideffect
 from .network import Network, yield_operations
-from .op import Operation, reparse_operation_data
+from .op import FunctionalOperation, Operation, reparse_operation_data
 
 log = logging.getLogger(__name__)
 
@@ -67,6 +67,7 @@ def __init__(
         self.name = name
         self.inputs = inputs
         self.provides = outputs
+        # Prune network
         self.net = net.pruned(inputs, outputs)
         ## Set data asap, for debugging, although `prune()` will reset them.
         self.set_execution_method(method)
@@ -256,6 +257,7 @@ def compose(
     needs=None,
     provides=None,
     merge=False,
+    node_props=None,
     method=None,
     overwrites_collector=None,
 ) -> NetworkOperation:
@@ -283,7 +285,9 @@ def compose(
         this ``compose`` instance).  If any two operations are the same
         (based on name), then that operation is computed only once, instead
         of multiple times (one for each time the operation appears).
-
+    :param node_props:
+        added as-is into NetworkX graph, to provide for filtering
+        by :meth:`.NetworkOperation.narrow()`.
     :param method:
         either `parallel` or None (default);
         if ``"parallel"``, launches multi-threading.
@@ -305,21 +309,36 @@ def compose(
     if not all(isinstance(op, Operation) for op in operations):
         raise ValueError(f"Non-Operation instances given: {operations}")
 
+    def proc_op(op, parent=None):
+        """clone FuncOperation with certain props changed"""
+        assert isinstance(op, FunctionalOperation), op
+
+        ## Convey any node-props specified in the netop here
+        #  to all sub-operations.
+        #
+        if node_props or parent:
+            kw = {}
+            if node_props:
+                op_node_props = op.node_props.copy()
+                op_node_props.update(node_props)
+                kw["node_props"] = op_node_props
+            ## If `merge` asked, leave original `name` to deduplicate operations,
+            #  otherwise rename the op by prefixing them with their parent netop.
+            #
+            if not merge and parent:
+                kw["parents"] = (parent,) + (op.parents or ())
+            op = op.withset(**kw)
+
+        return op
+
     merge_set = iset()  # Preseve given node order.
     for op in operations:
         if isinstance(op, NetworkOperation):
-            # TODO: do we really need sorting ops when merging?
-            netop_nodes = op.net.graph
             merge_set.update(
-                # If merge is desired, set will deduplicate operations
-                s if merge else
-                # else rename added ops by prefixing them with their parent netop.
-                s._adopted_by(op.name)
-                for s in netop_nodes
-                if isinstance(s, Operation)
+                proc_op(s, op.name) for s in op.net.graph if isinstance(s, Operation)
             )
         else:
-            merge_set.add(op)
+            merge_set.add(proc_op(op))
     operations = merge_set
 
     net = Network(*operations)

graphtik/network.py

@@ -565,7 +565,7 @@ def _append_operation(self, graph, operation: Operation):
         """
         Adds the given operation and its data requirements to the network graph.
 
-        - Invoked dyuring constructor only (immutability).
+        - Invoked during constructor only (immutability).
         - Identities are based on the name of the operation, the names of the operation's needs,
           and the names of the data it provides.
 
@@ -584,6 +584,8 @@ def _append_operation(self, graph, operation: Operation):
                 graph.add_node(_DataNode(n), sideffect=True)
             graph.add_edge(_DataNode(n), operation, **kw)
 
+        graph.add_node(operation, **operation.node_props)
+
         # add nodes and edges to graph describing what this layer provides
         for n in operation.provides:
             kw = {}

graphtik/op.py

@@ -5,7 +5,7 @@
 import abc
 import logging
 from collections import abc as cabc
-from typing import Callable, Collection, Tuple, Union
+from typing import Callable, Collection, Mapping, Tuple, Union
 
 from boltons.setutils import IndexedSet as iset
 
@@ -41,7 +41,7 @@ def reparse_operation_data(name, needs, provides):
 
 # TODO: immutable `Operation` by inheriting from `namedtuple`.
 class Operation(abc.ABC):
-    """An abstract class representing a data transformation by :meth:`.compute()`."""
+    """An abstract class representing an action with :meth:`.compute()`."""
 
     @abc.abstractmethod
     def compute(self, named_inputs, outputs=None):
@@ -75,6 +75,7 @@ def __init__(
         provides: Union[Collection, str] = None,
         *,
         parents: Tuple = None,
+        node_props: Mapping = None,
         returns_dict=None,
     ):
         """
@@ -87,18 +88,21 @@ def __init__(
             Names of input data objects this operation requires.
         :param provides:
             Names of output data objects this provides.
-        :param parent:
+        :param parents:
             a tuple wth the names of the parents, prefixing `name`,
             but also kept for equality/hash check.
-
+        :param node_props:
+            added as-is into NetworkX graph
         """
         ## Set op-data early, for repr() to work on errors.
         self.fn = fn
         self.name = name
         self.needs = needs
         self.provides = provides
         self.parents = parents
+        self.node_props = node_props = node_props if node_props else {}
         self.returns_dict = returns_dict
+
         if not fn or not callable(fn):
             raise ValueError(
                 f"Operation was not provided with a callable: {fn}\n  {self}"
@@ -107,6 +111,9 @@ def __init__(
             raise ValueError(
                 f"Operation `parents` must be tuple, was {parents}\n {self}"
             )
+        if node_props is not None and not isinstance(node_props, cabc.Mapping):
+            raise ValueError(f"`node_props` must be a mapping, got: {node_props!r}")
+
         ## Overwrite reparsed op-data.
         name = ".".join(str(pop) for pop in ((parents or ()) + (name,)))
         self.name, self.needs, self.provides = reparse_operation_data(
@@ -133,21 +140,20 @@ def __repr__(self):
         provides = aslist(self.provides, "provides")
         fn_name = self.fn and getattr(self.fn, "__name__", str(self.fn))
         returns_dict_marker = self.returns_dict and "{}" or ""
+        nprops = f", x{len(self.node_props)}props" if self.node_props else ""
         return (
             f"FunctionalOperation(name={self.name!r}, needs={needs!r}, "
-            f"provides={provides!r}, fn{returns_dict_marker}={fn_name!r})"
+            f"provides={provides!r}, fn{returns_dict_marker}={fn_name!r}{nprops})"
         )
 
-    def _adopted_by(self, parent):
-        """Make a clone with the given parrent set."""
-        return FunctionalOperation(
-            self.fn,
-            self.name,
-            self.needs,
-            self.provides,
-            parents=(parent,) + (self.parents or ()),
-            returns_dict=self.returns_dict,
-        )
+    def withset(self, **kw) -> "FunctionalOperation":
+        """Make a clone with the some values replaced."""
+        fn = kw["fn"] if "fn" in kw else self.fn
+        name = kw["name"] if "name" in kw else self.name
+        needs = kw["needs"] if "needs" in kw else self.needs
+        provides = kw["provides"] if "provides" in kw else self.provides
+
+        return FunctionalOperation(fn, name, needs, provides, **kw)
 
     def _zip_results_with_provides(self, results, real_provides: iset) -> dict:
         """Zip results with expected "real" (without sideffects) `provides`."""
@@ -201,7 +207,6 @@ def compute(self, named_inputs, outputs=None) -> dict:
                     if not isinstance(n, (optional, sideffect))
                 ]
             except KeyError:
-                # FIXME:
                 compulsory = iset(
                     n for n in self.needs if not isinstance(n, (optional, sideffect))
                 )
@@ -283,7 +288,8 @@ class operation:
         elements must be returned
     :param bool returns_dict:
         if true, it means the `fn` returns a dictionary with all `provides`,
-        and no further processing is done on them.
+    :param node_props:
+        added as-is into NetworkX graph
 
     :return:
         when called, it returns a :class:`FunctionalOperation`
@@ -316,15 +322,24 @@ def __init__(
         needs=None,
         provides=None,
         returns_dict=None,
+        node_props: Mapping = None,
     ):
         self.fn = fn
         self.name = name
         self.needs = needs
         self.provides = provides
         self.returns_dict = returns_dict
+        self.node_props = node_props
 
     def withset(
-        self, *, fn=None, name=None, needs=None, provides=None, returns_dict=None
+        self,
+        *,
+        fn=None,
+        name=None,
+        needs=None,
+        provides=None,
+        returns_dict=None,
+        node_props: Mapping = None,
     ) -> "operation":
         if fn is not None:
             self.fn = fn
@@ -336,11 +351,20 @@ def withset(
             self.provides = provides
         if returns_dict is not None:
             self.returns_dict = returns_dict
+        if node_props is not None:
+            self.node_props = node_props
 
         return self
 
     def __call__(
-        self, fn=None, *, name=None, needs=None, provides=None, returns_dict=None
+        self,
+        fn=None,
+        *,
+        name=None,
+        needs=None,
+        provides=None,
+        returns_dict=None,
+        node_props: Mapping = None,
     ) -> FunctionalOperation:
         """
         This enables ``operation`` to act as a decorator or as a functional
@@ -365,7 +389,12 @@ def myadd(a, b):
         """
 
         self.withset(
-            fn=fn, name=name, needs=needs, provides=provides, returns_dict=returns_dict
+            fn=fn,
+            name=name,
+            needs=needs,
+            provides=provides,
+            returns_dict=returns_dict,
+            node_props=node_props,
         )
 
         return FunctionalOperation(**vars(self))
@@ -377,4 +406,8 @@ def __repr__(self):
         needs = aslist(self.needs, "needs")
         provides = aslist(self.provides, "provides")
         fn_name = self.fn and getattr(self.fn, "__name__", str(self.fn))
-        return f"operation(name={self.name!r}, needs={needs!r}, provides={provides!r}, fn={fn_name!r})"
+        nprops = f", x{len(self.node_props)}props" if self.node_props else ""
+        return (
+            f"operation(name={self.name!r}, needs={needs!r}, "
+            f"provides={provides!r}, fn={fn_name!r}{nprops})"
+        )

test/test_base.py

@@ -166,6 +166,7 @@ def __init__(self):
         self.name = ("",)
         self.needs = ()
         self.provides = ("a",)
+        self.node_props = {}
 
     def compute(self, named_inputs, outputs=None):
         pass

test/test_op.py

@@ -3,7 +3,8 @@
 
 import pytest
 
-from graphtik import operation, optional, vararg, varargs
+from graphtik import compose, operation, optional, vararg, varargs
+from graphtik.network import yield_operations
 from graphtik.op import Operation, reparse_operation_data
 
 
@@ -133,3 +134,64 @@ def sumall(a, *args, b=0, **kwargs):
     assert op.compute(dict(a=1, arg1=2, arg2=3, b=6, c=7))["sum"] == exp - 4 - 5
     with pytest.raises(ValueError, match="Missing compulsory needs.+'a'"):
         assert op.compute(dict(arg1=2, arg2=3, b=6, c=7))
+
+
+def test_op_node_props_bad():
+    op_factory = operation(lambda: None, name="a", node_props="SHOULD BE DICT")
+    with pytest.raises(ValueError, match="`node_props` must be"):
+        op_factory()
+
+
+def test_op_node_props():
+    op_factory = operation(lambda: None, name="a", node_props=())
+    assert op_factory.node_props == ()
+    assert op_factory().node_props == {}
+
+    np = {"a": 1}
+    op = operation(lambda: None, name="a", node_props=np)()
+    assert op.node_props == np
+
+
+def _collect_op_props(netop):
+    return {
+        k.name: v
+        for k, v in netop.net.graph.nodes(data=True)
+        if isinstance(k, Operation)
+    }
+
+
+def test_netop_node_props():
+    op1 = operation(lambda: None, name="a", node_props={"a": 11, "b": 0, "bb": 2})()
+    op2 = operation(lambda: None, name="b", node_props={"a": 3, "c": 4})()
+    netop = compose("n", op1, op2, node_props={"bb": 22, "c": 44})
+
+    exp = {"a": {"a": 11, "b": 0, "bb": 22, "c": 44}, "b": {"a": 3, "bb": 22, "c": 44}}
+    node_props = _collect_op_props(netop)
+    assert node_props == exp
+
+    # Check node-prop sideffects are not modified
+    #
+    assert op1.node_props == {"a": 11, "b": 0, "bb": 2}
+    assert op2.node_props == {"a": 3, "c": 4}
+
+
+def test_netop_merge_node_props():
+    op1 = operation(lambda: None, name="a", node_props={"a": 1})()
+    netop1 = compose("n1", op1)
+    op2 = operation(lambda: None, name="a", node_props={"a": 11, "b": 0, "bb": 2})()
+    op3 = operation(lambda: None, name="b", node_props={"a": 3, "c": 4})()
+    netop2 = compose("n2", op2, op3)
+
+    netop = compose("n", netop1, netop2, node_props={"bb": 22, "c": 44}, merge=False)
+    exp = {
+        "n1.a": {"a": 1, "bb": 22, "c": 44},
+        "n2.a": {"a": 11, "b": 0, "bb": 22, "c": 44},
+        "n2.b": {"a": 3, "bb": 22, "c": 44},
+    }
+    node_props = _collect_op_props(netop)
+    assert node_props == exp
+
+    netop = compose("n", netop1, netop2, node_props={"bb": 22, "c": 44}, merge=True)
+    exp = {"a": {"a": 1, "bb": 22, "c": 44}, "b": {"a": 3, "bb": 22, "c": 44}}
+    node_props = _collect_op_props(netop)
+    assert node_props == exp