#186: implement graph builder

sxpat/converting/legacy.py

@@ -1,45 +1,63 @@
+from typing import Type, Union
+
+import functools
+import networkx as nx
+
 from sxpat.annotatedGraph import AnnotatedGraph
 
 from sxpat.graph import IOGraph, SGraph
-from sxpat.graph.node import BoolVariable, BoolConstant, And, Not, Identity
+from sxpat.graph.builder import GraphBuilder
+from sxpat.graph.node import BoolVariable, BoolConstant, And, Extras, Node, Not, Identity
+
 from sxpat.utils.functions import str_to_bool
 
 
 __all__ = ['iograph_from_legacy', 'sgraph_from_legacy']
 
 
-def _nodes_from_inner_legacy(inner_graph):
-    nodes = list()
-    for (name, value) in inner_graph.nodes(True):
+def _add_nodes_from_legacy(builder: GraphBuilder, digraph: nx.digraph.DiGraph) -> GraphBuilder:
+    def get_type(label: str) -> Type[Union[Node, Extras]]:
+        if label.startswith('in'): return BoolVariable
+        elif label.startswith('out'): return Identity
+        elif label == 'and': return And
+        elif label == 'not': return Not
+        elif label in ('FALSE', 'TRUE'): return functools.partial(BoolConstant, value=str_to_bool(label))
+        else: raise RuntimeError(f'Unable to parse node {name} from AnnotatedGraph ({value})')
+
+    # add all nodes (and edges)
+    for (name, value) in digraph.nodes(True):
         # get features
         label = value.get('label')
         weight = value.get('weight', None)
         in_subgraph = bool(value.get('subgraph', False))
-        operands = inner_graph.predecessors(name)
-
-        # create node
-        if label.startswith('in'):  # input
-            nodes.append(BoolVariable(name, weight, in_subgraph))
-        elif label.startswith('out'):  # output
-            nodes.append(Identity(name, operands, weight, in_subgraph))
-        elif label in ('and', 'not'):  # and/not
-            cls = {'not': Not, 'and': And}[label]
-            nodes.append(cls(name, operands, weight, in_subgraph))
-        elif label in ('FALSE', 'TRUE'):  # constant
-            nodes.append(BoolConstant(name, str_to_bool(label), weight, in_subgraph))
-        else:
-            raise RuntimeError(f'Unable to parse node {name} from AnnotatedGraph ({value})')
+        operands = digraph.predecessors(name)
 
-    return nodes
+        # add node
+        builder.add_node(name, get_type(label), weight=weight, in_subgraph=in_subgraph)
+        if digraph.in_degree(name) > 0: builder.add_operands(operands)
 
 
 def iograph_from_legacy(l_graph: AnnotatedGraph) -> IOGraph:
-    return IOGraph(_nodes_from_inner_legacy(l_graph.graph),
-                   l_graph.input_dict.values(),
-                   l_graph.output_dict.values())
+    return (
+        GraphBuilder()
+        # add nodes
+        .update_with(_add_nodes_from_legacy, l_graph.graph)
+        # mark inputs/outputs
+        .mark_inputs(l_graph.input_dict.values())
+        .mark_outputs(l_graph.output_dict.values())
+        #
+        .build(IOGraph)
+    )
 
 
 def sgraph_from_legacy(l_graph: AnnotatedGraph) -> SGraph:
-    return SGraph(_nodes_from_inner_legacy(l_graph.subgraph),
-                  l_graph.input_dict.values(),
-                  l_graph.output_dict.values())
+    return (
+        GraphBuilder()
+        # add nodes
+        .update_with(_add_nodes_from_legacy, l_graph.subgraph)
+        # mark inputs/outputs
+        .mark_inputs(l_graph.input_dict.values())
+        .mark_outputs(l_graph.output_dict.values())
+        #
+        .build(SGraph)
+    )

sxpat/converting/porters.py

@@ -1,14 +1,14 @@
-from abc import abstractmethod
+from __future__ import annotations
+from abc import ABC, abstractmethod
 from typing import Type, Callable, Mapping, Optional, Union, Generic
 import dataclasses as dc
 
 from bidict import bidict
-
 import itertools as it
 import re
 import json
 
-from sxpat.graph import *
+from sxpat.graph import Graph, IOGraph, SGraph, PGraph, CGraph, T_Graph
 from sxpat.graph.node import *
 from sxpat.utils.inheritance import get_all_subclasses, get_all_leaves_subclasses
 from sxpat.utils.functions import str_to_bool
@@ -29,10 +29,11 @@
 
 
 @make_utility_class
-class GraphImporter(Generic[T_Graph]):
+class GraphImporter(Generic[T_Graph], ABC):
     """Abstract class for importing a Graph from a string/file."""
 
     @classmethod
+    @abstractmethod
     def from_string(cls, string: str) -> T_Graph:
         raise NotImplementedError(f'{cls.__name__}.from_string(...) is abstract.')
 
@@ -44,7 +45,7 @@ def from_file(cls, filename: str) -> T_Graph:
 
 
 @make_utility_class
-class GraphExporter(Generic[T_Graph]):
+class GraphExporter(Generic[T_Graph], ABC):
     """Abstract class for exporting a Graph to a string/file."""
 
     @classmethod
@@ -384,8 +385,8 @@ class VerilogExporter(GraphExporter[IOGraph]):
         Not: lambda n: f'(~{n.operand})',
         And: lambda n: f'({" & ".join(n.operands)})',
         Or: lambda n: f'({" | ".join(n.operands)})',
-        Xor: lambda n : f'({" ^ ".join(n.operands)})',
-        Xnor: lambda n : f'(~({n.left} ^ {n.right}))',
+        Xor: lambda n: f'({" ^ ".join(n.operands)})',
+        Xnor: lambda n: f'(~({n.left} ^ {n.right}))',
         Implies: lambda n: f'(~{n.left} | {n.right})',
         # int-int operations
         Sum: lambda n: f'({" + ".join(n.operands)})',

sxpat/definitions/distances/AbsoluteDifferenceOfInteger.py

@@ -1,6 +1,8 @@
 from typing import Sequence, Tuple
 from typing_extensions import override
 
+from sxpat.graph.builder import GraphBuilder
+
 from .DistanceSpecification import DistanceSpecification
 
 from sxpat.graph import CGraph
@@ -20,30 +22,28 @@ class AbsoluteDifferenceOfInteger(DistanceSpecification):
     @override
     @classmethod
     def _define(cls, _0, _1,
-                wanted_a: Sequence[str], wanted_b: Sequence[str],
-                ) -> Tuple[CGraph, str]:
+                  wanted_a: Sequence[str], wanted_b: Sequence[str],
+                  ) -> Tuple[CGraph, str]:
+        # prepare builder
+        builder = GraphBuilder()
+
+        # add placeholders
+        builder.add_placeholders(wanted_a).add_placeholders(wanted_b)
 
         # define outputs of a and of b as integers
-        int_a = ToInt('dist_int_a_adoi', operands=wanted_a)
-        int_b = ToInt('dist_int_b_adoi', operands=wanted_b)
+        builder.push_recording()
+        builder.add_node('dist_int_a_adoi', ToInt, operands=wanted_a)
+        builder.add_node('dist_int_b_adoi', ToInt, operands=wanted_b)
 
         # distance
-        distance = AbsDiff('dist_distance', operands=[int_a, int_b])
-
-        # construct CGraph
-        dist_func = CGraph((
-            *(PlaceHolder(name) for name in wanted_a),
-            int_a,
-            *(PlaceHolder(name) for name in wanted_b),
-            int_b,
-            distance,
-        ))
+        distance_name = 'dist_distance'
+        builder.add_node(distance_name, AbsDiff, operands=builder.pop_recording())
 
-        return (dist_func, distance.name)
+        return (builder.build(CGraph), distance_name)
 
     @override
     @classmethod
     def _minimum_distance(cls, _0,
-                wanted_a: Sequence[str]
-                ) -> int:
-        return 1
+                          wanted_a: Sequence[str]
+                          ) -> int:
+        return 1

sxpat/definitions/distances/DistanceSpecification.py

@@ -16,9 +16,11 @@ class DistanceSpecification(metaclass=ABCMeta):
 
     @classmethod
     @overload
-    def define(cls, graph_a: IOGraph, graph_b: IOGraph,
-               wanted_a: Sequence[str], wanted_b: Optional[Sequence[str]] = None
-               ) -> Tuple[CGraph, str]:
+    def define(
+        cls,
+        graph_a: IOGraph, graph_b: IOGraph,
+        wanted_a: Sequence[str], wanted_b: Optional[Sequence[str]] = None
+    ) -> Tuple[CGraph, str]:
         """
             Defines a distance between two circuits (given as graph), given a specific sequence (or one per circuit) of nodes to use.
 
@@ -27,7 +29,10 @@ def define(cls, graph_a: IOGraph, graph_b: IOGraph,
 
     @classmethod
     @overload
-    def define(cls, graph_a: IOGraph, graph_b: IOGraph) -> Tuple[CGraph, str]:
+    def define(
+        cls,
+        graph_a: IOGraph, graph_b: IOGraph
+    ) -> Tuple[CGraph, str]:
         """
             Defines a distance between the outputs of two circuits (given as graphs).
 
@@ -36,9 +41,11 @@ def define(cls, graph_a: IOGraph, graph_b: IOGraph) -> Tuple[CGraph, str]:
 
     @classmethod
     @final
-    def define(cls, graph_a: IOGraph, graph_b: IOGraph,
-               wanted_a: Optional[Sequence[str]] = None, wanted_b: Optional[Sequence[str]] = None,
-               ) -> Tuple[CGraph, str]:
+    def define(
+        cls,
+        graph_a: IOGraph, graph_b: IOGraph,
+        wanted_a: Optional[Sequence[str]] = None, wanted_b: Optional[Sequence[str]] = None,
+    ) -> Tuple[CGraph, str]:
 
         # > generate distance function graph
 
@@ -47,12 +54,7 @@ def define(cls, graph_a: IOGraph, graph_b: IOGraph,
             wanted_a = graph_a.outputs_names
             wanted_b = graph_b.outputs_names
 
-            # delegate computation
-            dist_func, root_name = cls._define(
-                graph_a, graph_b,
-                graph_a.outputs_names, graph_b.outputs_names,
-            )
-
+        # some wanted names given
         elif wanted_a is not None:
             # default
             if wanted_b is None: wanted_b = wanted_a
@@ -63,13 +65,14 @@ def define(cls, graph_a: IOGraph, graph_b: IOGraph,
             if (missing := first(lambda n: n not in graph_b, wanted_b, None)) is not None:
                 raise g_error.MissingNodeError(f'Node {missing} is not in graph_b ({graph_b}).')
 
-            # delegate computation
-            dist_func, root_name = cls._define(
-                graph_a, graph_b,
-                wanted_a, wanted_b,
-            )
+        else:
+            raise ValueError('Illegal call with `wanted_b` without `wanted_a`.')
 
-        else: raise ValueError(f'Illegal call with `wanted_b` without `wanted_a`.')
+        # delegate computation
+        dist_func, root_name = cls._define(
+            graph_a, graph_b,
+            wanted_a, wanted_b,
+        )
 
         # > assign rolling prefix
         prefix = get_rolling_code() + '_'
@@ -80,22 +83,27 @@ def define(cls, graph_a: IOGraph, graph_b: IOGraph,
 
     @classmethod
     @abstractmethod
-    def _define(cls, graph_a: IOGraph, graph_b: IOGraph,
-                wanted_a: Sequence[str], wanted_b: Sequence[str]
-                ) -> Tuple[CGraph, str]: ...
+    def _define(
+        cls,
+        graph_a: IOGraph, graph_b: IOGraph,
+        wanted_a: Sequence[str], wanted_b: Sequence[str]
+    ) -> Tuple[CGraph, str]: ...
 
     @classmethod
-    def minimum_distance(cls, graph_a: SGraph,
-                wanted_a: Optional[Sequence[str]] = None
-                ) -> int:
-
+    def minimum_distance(
+        cls,
+        graph_a: SGraph,
+        wanted_a: Optional[Sequence[str]] = None
+    ) -> int:
         if wanted_a is None:
             wanted_a = (n.name for n in graph_a.subgraph_outputs)
-        
+
         return cls._minimum_distance(graph_a, wanted_a)
-        
+
     @classmethod
     @abstractmethod
-    def _minimum_distance(cls, graph_a: SGraph,
-                wanted_a: Sequence[str]
-                ) -> int: ...
+    def _minimum_distance(
+        cls,
+        graph_a: SGraph,
+        wanted_a: Sequence[str]
+    ) -> int: ...

sxpat/definitions/distances/HammingDistance.py

@@ -3,8 +3,9 @@
 
 from .DistanceSpecification import DistanceSpecification
 
-from sxpat.graph import CGraph, IOGraph
-from sxpat.graph.node import If, IntConstant, PlaceHolder, Sum, Xor
+from sxpat.graph import CGraph
+from sxpat.graph.builder import GraphBuilder
+from sxpat.graph.node import If, IntConstant, Sum, Xor
 from sxpat.utils.collections import formatted_int_range
 
 
@@ -23,52 +24,46 @@ class HammingDistance(DistanceSpecification):
     def _define(cls, _0, _1,
                 wanted_a: Sequence[str], wanted_b: Sequence[str],
                 ) -> Tuple[CGraph, str]:
-
-        # guard
+        # guards
         if len(wanted_a) != len(wanted_b):
             raise ValueError('The sequences of wanted nodes have different lengths (or the graphs have different number of outputs).')
         if len(_0.outputs_names) != len(_1.outputs_names):
             raise ValueError('The sequences of wanted nodes have different lengths (or the graphs have different number of outputs).')
 
+        # prepare builder
+        builder = GraphBuilder()
+
+        # add placeholders
+        builder.add_placeholders(wanted_a).add_placeholders(wanted_b)
+
         # bit flips to int
-        consts = []
-        flipped_bits = []
-        int_bits = []
+        builder.push_recording()
         for (i, out_a, out_b) in zip(
             formatted_int_range(len(wanted_a)),
             wanted_a,
             wanted_b,
         ):
-            # create constants
-            consts.extend([
-                const_0 := IntConstant(f'dist_a{i}_const_0', 0),
-                const_1 := IntConstant(f'dist_a{i}_const_1', 1),
-            ])
+            builder.push_recording()
 
             # create node reflecting if a bit is flipped
-            flipped_bits.append(bit := Xor(f'dist_is_different_{i}', operands=[out_a, out_b]))
+            builder.add_node(f'dist_is_different_{i}', Xor, operands=[out_a, out_b])
+            # create constants
+            builder \
+                .add_node(f'dist_a{i}_const_1', IntConstant, value=1) \
+                .add_node(f'dist_a{i}_const_0', IntConstant, value=0)
 
             # create node that reflects 1 if the bit is flipped, or 0
-            int_bits.append(If(f'dist_value_{i}', operands=[bit, const_1, const_0]))
+            builder.add_node(f'dist_value_{i}', If, operands=builder.pop_recording())
 
         # distance
-        distance = Sum('dist_distance', operands=int_bits)
-
-        # construct CGraph
-        dist_func = CGraph((
-            *(PlaceHolder(name) for name in wanted_a),
-            *(PlaceHolder(name) for name in wanted_b),
-            *consts,
-            *flipped_bits,
-            *int_bits,
-            distance,
-        ))
+        distance_name = 'dist_distance'
+        builder.add_node(distance_name, Sum, operands=builder.pop_recording())
 
-        return (dist_func, distance.name)
+        return (builder.build(CGraph), distance_name)
 
     @override
     @classmethod
     def minimum_distance(cls, _0,
-                wanted_a: Sequence[str]
-                ) -> int:
+                         wanted_a: Sequence[str]
+                         ) -> int:
         return 1