Merge branch 'master' into hypothesis_pyup

Author: mvcisback

aiger/aig.py

@@ -1,12 +1,12 @@
 from collections import defaultdict
-from itertools import chain
 from functools import reduce
-from typing import Tuple, FrozenSet, NamedTuple, Union, Mapping, List
+from typing import Tuple, FrozenSet, NamedTuple, Union
 
 import funcy as fn
 from toposort import toposort
 
 from aiger import common as cmn
+from aiger import parser
 
 
 def timed_name(name, time):
@@ -73,6 +73,8 @@ class AIG(NamedTuple):
     latch2init: FrozenSet[Tuple[str, Node]] = frozenset()
     comments: Tuple[str] = ()
 
+    _to_aag = parser.aig2aag
+
     def __repr__(self):
         return repr(self._to_aag())
 
@@ -229,34 +231,6 @@ def _unroll():
 
         return unrolled
 
-    def _to_aag(self):
-        aag, max_idx, lit_map = _to_aag(
-            self.cones | self.latch_cones,
-            AAG({}, {}, {}, [], self.comments),
-        )
-
-        # Check that all inputs have a lit.
-        for name in filter(lambda x: x not in aag.inputs, self.inputs):
-            aag.inputs[name] = lit_map[name] = 2 * max_idx
-            max_idx += 1
-
-        # Update cone maps.
-        aag.outputs.update({k: lit_map[cone] for k, cone in self.node_map})
-        latch2init = dict(self.latch2init)
-        for name, cone in self.latch_map:
-            latch = LatchIn(name)
-            if latch not in lit_map:
-                lit = lit_map[latch] = 2 * max_idx
-                max_idx += 1
-            else:
-                lit = lit_map[latch]
-
-            init = int(latch2init[name])
-            ilit = lit_map[cone]
-            aag.latches[name] = lit, ilit, init
-
-        return aag
-
     def write(self, path):
         with open(path, 'w') as f:
             f.write(repr(self))
@@ -293,181 +267,6 @@ def _mod(node):
         )
 
 
-def _to_idx(lit):
-    """AAG format uses least significant bit to encode an inverter.
-    The index is thus the interal literal shifted by one bit."""
-    return lit >> 1
-
-
-def _polarity(i):
-    return Inverter if i & 1 == 1 else lambda x: x
-
-
-class Header(NamedTuple):
-    max_var_index: int
-    num_inputs: int
-    num_latches: int
-    num_outputs: int
-    num_ands: int
-
-
-class AAG(NamedTuple):
-    inputs: Mapping[str, int]
-    latches: Mapping[str, Tuple[int]]
-    outputs: Mapping[str, int]
-    gates: List[Tuple[int]]
-    comments: Tuple[str]
-
-    @property
-    def header(self):
-        literals = chain(
-            self.inputs.values(),
-            self.outputs.values(),
-            fn.pluck(0, self.gates), fn.pluck(0, self.latches.values())
-        )
-        max_idx = max(map(_to_idx, literals), default=0)
-        return Header(max_idx, *map(len, self[:-1]))
-
-    def __repr__(self):
-        if self.inputs:
-            input_names, input_lits = zip(*list(self.inputs.items()))
-        if self.outputs:
-            output_names, output_lits = zip(*list(self.outputs.items()))
-        if self.latches:
-            latch_names, latch_lits = zip(*list(self.latches.items()))
-
-        out = f"aag " + " ".join(map(str, self.header)) + '\n'
-        if self.inputs:
-            out += '\n'.join(map(str, input_lits)) + '\n'
-        if self.latches:
-            out += '\n'.join([' '.join(map(str, xs))
-                              for xs in latch_lits]) + '\n'
-        if self.outputs:
-            out += '\n'.join(map(str, output_lits)) + '\n'
-
-        if self.gates:
-            out += '\n'.join([' '.join(map(str, xs))
-                              for xs in self.gates]) + '\n'
-        if self.inputs:
-            out += '\n'.join(
-                f"i{idx} {name}" for idx, name in enumerate(input_names)
-            ) + '\n'
-        if self.outputs:
-            out += '\n'.join(
-                f"o{idx} {name}" for idx, name in enumerate(output_names)
-            ) + '\n'
-
-        if self.latches:
-            out += '\n'.join(
-                f"l{idx} {name}" for idx, name in enumerate(latch_names)
-            ) + '\n'
-
-        if self.comments:
-            out += 'c\n' + '\n'.join(self.comments)
-            if out[-1] != '\n':
-                out += '\n'
-        return out
-
-    def _to_aig(self):
-        gate_order, latch_order = self.eval_order_and_gate_lookup
-
-        lookup = fn.merge(
-            {0: ConstFalse()},
-            {_to_idx(l): Input(n) for n, l in self.inputs.items()},
-            {
-                _to_idx(l): LatchIn(n)
-                for n, (l, _, init) in self.latches.items()
-            },
-        )
-        latches = set()
-        and_dependencies = {i: (l, r) for i, l, r in self.gates}
-        for gate in fn.cat(gate_order):
-            if _to_idx(gate) in lookup:
-                continue
-
-            inputs = and_dependencies[gate]
-            sources = [_polarity(i)(lookup[_to_idx(i)]) for i in inputs]
-            lookup[_to_idx(gate)] = AndGate(*sources)
-
-        latch_dependencies = {
-            i: (n, dep) for n, (i, dep, _) in self.latches.items()
-        }
-        for gate in fn.cat(latch_order):
-            assert _to_idx(gate) in lookup
-            if not isinstance(lookup[_to_idx(gate)], LatchIn):
-                continue
-
-            name, dep = latch_dependencies[gate]
-            source = _polarity(dep)(lookup[_to_idx(dep)])
-            latches.add((name, source))
-
-        def get_output(v):
-            idx = _to_idx(v)
-            return _polarity(v)(lookup[idx])
-
-        top_level = ((k, get_output(v)) for k, v in self.outputs.items())
-        return AIG(
-            inputs=frozenset(self.inputs),
-            node_map=frozenset(top_level),
-            latch_map=frozenset(latches),
-            latch2init=frozenset(
-                (n, bool(init)) for n, (_, _, init) in self.latches.items()
-            ),
-            comments=self.comments
-        )
-
-    @property
-    def eval_order_and_gate_lookup(self):
-        deps = fn.merge(
-            {a & -2: {b & -2, c & -2} for a, b, c in self.gates},
-            {a & -2: set() for a in self.inputs.values()},
-            {a & -2: set() for a, _, _ in self.latches.values()},  # LatchIn
-        )
-        latch_deps = {a & -2: {b & -2} for a, b, _ in self.latches.values()}
-        return list(toposort(deps)), list(toposort(latch_deps))
-
-
-def _to_aag(gates, aag: AAG = None, *, max_idx=1, lit_map=None):
-    if lit_map is None:
-        lit_map = {}
-
-    if not gates:
-        return aag, max_idx, lit_map
-
-    # Recurse to update get aag for subtrees.
-    for c in fn.mapcat(lambda g: g.children, gates):
-        if c in lit_map:
-            continue
-        aag, max_idx, lit_map = _to_aag(
-            [c], aag, max_idx=max_idx, lit_map=lit_map
-        )
-
-    # Update aag with current level.
-    for gate in gates:
-        if gate in lit_map:
-            continue
-
-        if isinstance(gate, Inverter):
-            input_lit = lit_map[gate.input]
-            lit_map[gate] = (input_lit & -2) | (1 ^ (input_lit & 1))
-            continue
-        elif isinstance(gate, ConstFalse):
-            lit_map[gate] = 0
-            continue
-
-        # Must be And, Latch, or Input
-        lit_map[gate] = 2 * max_idx
-        max_idx += 1
-        if isinstance(gate, AndGate):
-            encoded = tuple(map(lit_map.get, (gate, gate.left, gate.right)))
-            aag.gates.append(encoded)
-
-        elif isinstance(gate, Input):
-            aag.inputs[gate.name] = lit_map[gate]
-
-    return aag, max_idx, lit_map
-
-
 def _dependency_graph(nodes):
     queue, deps, visited = list(nodes), defaultdict(set), set()
     while queue: