|
| 1 | +from abc import ABC, abstractmethod |
| 2 | +from types import MappingProxyType |
| 3 | +from collections import OrderedDict, defaultdict |
| 4 | + |
| 5 | +from unification import unify, reify, Var |
| 6 | +from unification.core import _unify, _reify |
| 7 | + |
| 8 | + |
| 9 | +class KanrenConstraintStore(ABC): |
| 10 | + """A class that enforces constraints between logic variables in a miniKanren state.""" |
| 11 | + |
| 12 | + @abstractmethod |
| 13 | + def pre_check(self, state, key=None, value=None): |
| 14 | + """Check a key-value pair before they're added to a KanrenState.""" |
| 15 | + raise NotImplementedError() |
| 16 | + |
| 17 | + @abstractmethod |
| 18 | + def post_check(self, new_state, key=None, value=None, old_state=None): |
| 19 | + """Check a key-value pair after they're added to a KanrenState.""" |
| 20 | + raise NotImplementedError() |
| 21 | + |
| 22 | + @abstractmethod |
| 23 | + def update(self, *args, **kwargs): |
| 24 | + """Add a new constraint.""" |
| 25 | + raise NotImplementedError() |
| 26 | + |
| 27 | + @abstractmethod |
| 28 | + def constraints_str(self, var): |
| 29 | + """Print the constraints on a logic variable.""" |
| 30 | + raise NotImplementedError() |
| 31 | + |
| 32 | + |
| 33 | +class KanrenState(dict): |
| 34 | + """A miniKanren state that holds unifications of logic variables and upholds constraints on logic variables.""" |
| 35 | + |
| 36 | + __slots__ = ("constraints",) |
| 37 | + |
| 38 | + def __init__(self, *s, constraints=None): |
| 39 | + super().__init__(*s) |
| 40 | + self.constraints = OrderedDict(constraints or []) |
| 41 | + |
| 42 | + def pre_checks(self, key, value): |
| 43 | + return all(cstore.pre_check(self, key, value) for cstore in self.constraints.values()) |
| 44 | + |
| 45 | + def post_checks(self, new_state, key, value): |
| 46 | + return all( |
| 47 | + cstore.post_check(new_state, key, value, old_state=self) |
| 48 | + for cstore in self.constraints.values() |
| 49 | + ) |
| 50 | + |
| 51 | + def add_constraint(self, constraint): |
| 52 | + assert isinstance(constraint, KanrenConstraintStore) |
| 53 | + self.constraints[type(constraint)] = constraint |
| 54 | + |
| 55 | + def __eq__(self, other): |
| 56 | + if isinstance(other, KanrenState): |
| 57 | + return super().__eq__(other) |
| 58 | + |
| 59 | + # When comparing with a plain dict, disregard the constraints. |
| 60 | + if isinstance(other, dict): |
| 61 | + return super().__eq__(other) |
| 62 | + return False |
| 63 | + |
| 64 | + def __repr__(self): |
| 65 | + return f"KanrenState({super().__repr__()}, {self.constraints})" |
| 66 | + |
| 67 | + |
| 68 | +class Disequality(KanrenConstraintStore): |
| 69 | + """A disequality constraint (i.e. two things do not unify).""" |
| 70 | + |
| 71 | + def __init__(self, mappings=None): |
| 72 | + # Unallowed mappings |
| 73 | + self.mappings = mappings or defaultdict(set) |
| 74 | + |
| 75 | + def post_check(self, new_state, key=None, value=None, old_state=None): |
| 76 | + return not any( |
| 77 | + any(new_state == unify(lvar, val, new_state) for val in vals) |
| 78 | + for lvar, vals in self.mappings.items() |
| 79 | + ) |
| 80 | + |
| 81 | + def pre_check(self, state, key=None, value=None): |
| 82 | + return True |
| 83 | + |
| 84 | + def update(self, key, value): |
| 85 | + self.mappings[key].add(value) |
| 86 | + |
| 87 | + def constraints_str(self, var): |
| 88 | + if var in self.mappings: |
| 89 | + return f"=/= {self.mappings[var]}" |
| 90 | + else: |
| 91 | + return "" |
| 92 | + |
| 93 | + def __repr__(self): |
| 94 | + return ",".join([f"{k} =/= {v}" for k, v in self.mappings.items()]) |
| 95 | + |
| 96 | + |
| 97 | +def unify_KanrenState(u, v, S): |
| 98 | + if S.pre_checks(u, v): |
| 99 | + s = unify(u, v, MappingProxyType(S)) |
| 100 | + if s is not False and S.post_checks(s, u, v): |
| 101 | + return KanrenState(s, constraints=S.constraints) |
| 102 | + |
| 103 | + return False |
| 104 | + |
| 105 | + |
| 106 | +unify.add((object, object, KanrenState), unify_KanrenState) |
| 107 | +unify.add( |
| 108 | + (object, object, MappingProxyType), |
| 109 | + lambda u, v, d: unify.dispatch(type(u), type(v), dict)(u, v, d), |
| 110 | +) |
| 111 | +_unify.add( |
| 112 | + (object, object, MappingProxyType), |
| 113 | + lambda u, v, d: _unify.dispatch(type(u), type(v), dict)(u, v, d), |
| 114 | +) |
| 115 | + |
| 116 | + |
| 117 | +class ConstrainedVar(Var): |
| 118 | + """A logic variable that tracks its own constraints. |
| 119 | +
|
| 120 | + Currently, this is only for display/reification purposes. |
| 121 | +
|
| 122 | + """ |
| 123 | + |
| 124 | + def __new__(cls, var, S): |
| 125 | + obj = super().__new__(cls, var.token) |
| 126 | + obj.S = S |
| 127 | + obj.var = var |
| 128 | + return obj |
| 129 | + |
| 130 | + def __repr__(self): |
| 131 | + u_constraints = ",".join([c.constraints_str(self.var) for c in self.S.constraints.values()]) |
| 132 | + return f"{self.var}: {{{u_constraints}}}" |
| 133 | + |
| 134 | + |
| 135 | +def reify_KanrenState(u, S): |
| 136 | + u_res = reify(u, MappingProxyType(S)) |
| 137 | + if isinstance(u_res, Var): |
| 138 | + return ConstrainedVar(u_res, S) |
| 139 | + else: |
| 140 | + return u_res |
| 141 | + |
| 142 | + |
| 143 | +_reify.add((tuple(p[0] for p in _reify.ordering if p[1] == dict), KanrenState), reify_KanrenState) |
| 144 | +_reify.add((object, MappingProxyType), lambda u, s: _reify.dispatch(type(u), dict)(u, s)) |
| 145 | + |
| 146 | + |
| 147 | +def neq(u, v): |
| 148 | + """Construct a disequality goal.""" |
| 149 | + |
| 150 | + def neq_goal(S): |
| 151 | + if not isinstance(S, KanrenState): |
| 152 | + S = KanrenState(S) |
| 153 | + |
| 154 | + diseq_constraint = S.constraints.setdefault(Disequality, Disequality()) |
| 155 | + |
| 156 | + diseq_constraint.update(u, v) |
| 157 | + |
| 158 | + if diseq_constraint.post_check(S): |
| 159 | + return iter([S]) |
| 160 | + else: |
| 161 | + return iter([]) |
| 162 | + |
| 163 | + return neq_goal |
0 commit comments