Updates

- Parses multiple implication tables from magic
- Speed improvements to model closure
- Make use of prior model_closure computations

model.py

@@ -8,7 +8,7 @@ Operation, Conjunction, Disjunction, Implication
 )
 from typing import Set, Dict, Tuple, Optional
 from functools import lru_cache
-from itertools import combinations, chain, product
+from itertools import combinations, chain, product, permutations
 from copy import deepcopy
 
 
@@ -32,6 +32,8 @@ class ModelValue:
     def __lt__(self, other):
         assert isinstance(other, ModelValue)
         return ModelOrderConstraint(self, other)
+    def __deepcopy__(self, memo):
+        return ModelValue(self.name)
 
 
 class ModelFunction:
@@ -195,66 +197,47 @@ def satisfiable(logic: Logic, model: Model, interpretation: Dict[Operation, Mode
 
     return True
 
+from itertools import combinations_with_replacement
+from collections import defaultdict
 
 def model_closure(initial_set: Set[ModelValue], mfunctions: Set[ModelFunction]):
-    last_set: Set[ModelValue] = set()
+    closure_set: Set[ModelValue] = initial_set
-    current_set: Set[ModelValue] = initial_set
+    last_new = initial_set
+    changed = True
 
-    while last_set != current_set:
+    while changed:
-        last_set = deepcopy(current_set)
+        changed = False
+        new_elements = set()
+        old_closure = closure_set - last_new
+
+        # arity -> args
+        cached_args = defaultdict(list)
 
         for mfun in mfunctions:
-            # Get output for every possible input configuration
-            # from last_set
-            for args in product(last_set, repeat=mfun.arity):
-                current_set.add(mfun(*args))
 
-    return current_set
+            # Use cached args if this arity was looked at before
-
+            if mfun.arity in cached_args:
-def has_vsp(model: Model, interpretation: Dict[Operation, ModelFunction]) -> bool:
+                for args in cached_args[mfun.arity]:
-    """
+                    element = mfun(*args)
-    Tells you whether a model violates the
+                    if element not in closure_set:
-    variable sharing property.
+                        new_elements.add(element)
-    """
+                # Move onto next function
-
-    impfunction = interpretation[Implication]
-
-    # Compute I the set of tuples (x, y) where
-    # x -> y does not take a designiated value
-    I: Set[Tuple[ModelValue, ModelValue]] = set()
-
-    for (x, y) in product(model.carrier_set, model.carrier_set):
-        if impfunction(x, y) not in model.designated_values:
-            I.add((x, y))
-
-    # Construct the powerset without the empty set
-    s = list(I)
-    I_power = chain.from_iterable(combinations(s, r) for r in range(1, len(s) + 1))
-    # ((x1, y1)), ((x1, y1), (x2, y2)), ...
-
-    for xys in I_power:
-        # Compute the closure of all operations
-        # with just the xs
-        xs = {xy[0] for xy in xys}
-        carrier_set_left: Set[ModelValue] = model_closure(xs, model.logical_operations)
-
-        # Compute the closure of all operations
-        # with just the ys
-        ys = {xy[1] for xy in xys}
-        carrier_set_right: Set[ModelValue] = model_closure(ys, model.logical_operations)
-
-        # If the carrier set intersects, then we violate VSP
-        if len(carrier_set_left & carrier_set_right) > 0:
                 continue
 
-        invalid = False
+            # Iterate over how many new elements would be within the arguments
-        for (x2, y2) in product(carrier_set_left, carrier_set_right):
+            # NOTE: To not repeat work, there must be at least one new element
-            if impfunction(x2, y2) in model.designated_values:
+            for num_new in range(1, mfun.arity + 1):
-                invalid = True
+                new_args = combinations_with_replacement(last_new, r=num_new)
-                break
+                old_args = combinations_with_replacement(old_closure, r=mfun.arity - num_new)
+                for new_arg, old_arg in product(new_args, old_args):
+                    for args in permutations(new_arg + old_arg):
+                        cached_args[mfun.arity].append(args)
+                        element = mfun(*args)
+                        if element not in closure_set:
+                            new_elements.add(element)
 
-        if not invalid:
+        closure_set.update(new_elements)
-            return True
+        changed = len(new_elements) > 0
-
+        last_new = new_elements
-    return False
 
+    return closure_set

parse_magic.py

@@ -13,6 +13,7 @@ from logic import (
     Negation,
     Disjunction
 )
+from vsp import has_vsp
 
 def parse_matrices(infile: TextIO) -> List[Tuple[Model, Dict]]:
     next(infile) # Skip header line
@@ -42,12 +43,11 @@ def parse_matrices(infile: TextIO) -> List[Tuple[Model, Dict]]:
                     if designated_values is None:
                         break
 
-                    while True:
+                    results = parse_implication(infile, size)
-                        result = parse_implication(infile, size)
                     if result is None:
                         break
-                        mimplication, hasnext = result
 
+                    for mimplication in results:
                         logical_operations = {
                             mnegation, mconjunction, mdisjunction,
                             mimplication
@@ -60,10 +60,7 @@ def parse_matrices(infile: TextIO) -> List[Tuple[Model, Dict]]:
                             Implication: mimplication
                         }
                         solutions.append((model, interpretation))
-                        print(f"Parsed {len(solutions)} so far")
 
-                        if not hasnext:
-                            break
     return solutions
 
 def carrier_set_from_size(size: int):
@@ -200,22 +197,21 @@ def parse_designated(infile: TextIO, size: int) -> Optional[Set[ModelValue]]:
     return designated_values
 
 
-def parse_implication(infile: TextIO, size: int) -> Optional[Tuple[ModelFunction, bool]]:
+def parse_implication(infile: TextIO, size: int) -> Optional[List[ModelFunction]]:
     line = next(infile).strip()
     if line == '-1':
         return None
 
-    table = line.split(" ")
+    # Split and remove the last '-1' character
-    has_next = True
+    table = line.split(" ")[:-1]
-    if table[-1] == '-1':
-        has_next = False
-        table = table[:-1] 
 
-    assert len(table) == (size + 1)**2
+    assert len(table) % (size + 1)**2 == 0
-
-    mapping = {}
 
     table_i = 0
+    mimplications: List[ModelFunction] = []
+
+    for _ in range(len(table) // (size + 1)**2):
+        mapping = {}
 
         for i in range(size + 1):
             x = mvalue_from_index(i)
@@ -228,16 +224,18 @@ def parse_implication(infile: TextIO, size: int) -> Optional[Tuple[ModelFunction
                 mapping[(x, y)] = r
 
         mimplication = ModelFunction(2, mapping, "Implication")
-    return mimplication, has_next
+        mimplications.append(mimplication)
+
+    return mimplications
 
 
 if __name__ == "__main__":
-    from model import has_vsp
-
     solutions: List[Model] = parse_matrices(sys.stdin)
     print(f"Parsed {len(solutions)} matrices")
-    for model, interpretation in solutions:
+    for i, (model, interpretation) in enumerate(solutions):
         # print(model)
         if has_vsp(model, interpretation):
             print(model)
             print("Has VSP")
+        else:
+            print("Model", i, "does not have VSP")

vsp.py

@@ -0,0 +1,114 @@
+"""
+Check to see if the model has the variable
+sharing property.
+"""
+from itertools import chain, combinations, product
+from typing import Dict, Set, Tuple, List
+from model import (
+    Model, ModelFunction, ModelValue, model_closure
+)
+from logic import Implication, Operation
+
+def preseed(initial_set: Set[ModelValue], cache:List[Tuple[Set[ModelValue], Set[ModelValue]]]):
+    """
+    Cache contains caches of model closure calls:
+    Ex:
+    {1, 2, 3} -> {....}
+
+    If {1,2,3} is a subset of initial set,
+    then {....} is the subset of the output of model_closure.
+
+    We'll use the output to speed up the saturation procedure
+    """
+    candidate_preseed: Tuple[Set[ModelValue], int] = (None, None)
+
+    for i, o in cache:
+        if i < initial_set:
+            cost = len(initial_set - i)
+            if candidate_preseed[1] is None or cost < candidate_preseed[1]:
+                candidate_preseed = o, cost
+
+    same_set = candidate_preseed[1] == 0
+    return candidate_preseed[0], same_set
+
+def has_vsp(model: Model, interpretation: Dict[Operation, ModelFunction]) -> bool:
+    """
+    Tells you whether a model has the
+    variable sharing property.
+    """
+
+    impfunction = interpretation[Implication]
+
+    # Compute I the set of tuples (x, y) where
+    # x -> y does not take a designiated value
+    I: Set[Tuple[ModelValue, ModelValue]] = set()
+
+    for (x, y) in product(model.carrier_set, model.carrier_set):
+        if impfunction(x, y) not in model.designated_values:
+            I.add((x, y))
+
+    # Construct the powerset of I without the empty set
+    s = list(I)
+    I_power = chain.from_iterable(combinations(s, r) for r in range(1, len(s) + 1))
+    # ((x1, y1)), ((x1, y1), (x2, y2)), ...
+
+    # Closure cache
+    closure_cache: List[Tuple[Set[ModelValue], Set[ModelValue]]] = []
+
+    # Find the subalgebras which falsify implication
+    for xys in I_power:
+
+        xs = {xy[0] for xy in xys}
+        orig_xs = xs
+        cached_xs = preseed(xs, closure_cache)
+        if cached_xs[0] is not None:
+            xs |= cached_xs[0]
+
+        ys = {xy[1] for xy in xys}
+        orig_ys = ys
+        cached_ys = preseed(ys, closure_cache)
+        if cached_ys[0] is not None:
+            ys |= cached_ys[0]
+
+
+        # NOTE: Optimziation before model_closure
+        # If the carrier set intersects, then move on to the next
+        # subalgebra
+        if len(xs & ys) > 0:
+            continue
+
+        # Compute the closure of all operations
+        # with just the xs
+        carrier_set_left: Set[ModelValue] = model_closure(xs, model.logical_operations)
+
+        # Save to cache
+        if cached_xs[0] is not None and not cached_ys[1]:
+            closure_cache.append((orig_xs, carrier_set_left))
+
+
+        # Compute the closure of all operations
+        # with just the ys
+        carrier_set_right: Set[ModelValue] = model_closure(ys, model.logical_operations)
+
+        # Save to cache
+        if cached_ys[0] is not None and not cached_ys[1]:
+            closure_cache.append((orig_ys, carrier_set_right))
+
+
+        # If the carrier set intersects, then move on to the next
+        # subalgebra
+        if len(carrier_set_left & carrier_set_right) > 0:
+            continue
+
+        # See if for all pairs in the subalgebras, that
+        # implication is falsified
+        falsified = True
+        for (x2, y2) in product(carrier_set_left, carrier_set_right):
+            if impfunction(x2, y2) in model.designated_values:
+                falsified = False
+                break
+
+        if falsified:
+            return True
+
+    return False