Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 60 additions & 4 deletions information_gain_aml/core/cnf_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,8 @@ def count_solutions(self, max_solutions: int | None = None) -> int:
Returns:
Number of satisfying assignments
"""
if max_solutions is None and self._cache_valid and self._solution_cache is not None:
return len(self._solution_cache)
return self._enumerate_models(max_count=max_solutions)

def get_all_solutions(self, max_solutions: int | None = None) -> list[set[str]]:
Expand Down Expand Up @@ -556,8 +558,8 @@ def count_models_with_temporary_clause(self, clause_literals: frozenset[str]) ->
"""
Count models with a temporary clause added (Phase 2 enhancement).

Adds clause, counts models, removes clause - no deep copy needed!
Much faster than cnf.copy() for temporary constraints.
When solution cache is available, filters cached solutions in O(|cache| * |clause|)
instead of O(SAT_SOLVE). Falls back to SAT solving when cache is not populated.

Args:
clause_literals: Set of literal strings for the clause
Expand All @@ -569,15 +571,69 @@ def count_models_with_temporary_clause(self, clause_literals: frozenset[str]) ->
if not clause_literals:
return self.count_solutions()

# Use cache-based filtering if available and all fluents are known
if self._cache_valid and self._solution_cache is not None:
result = self._count_models_with_clause_via_filter(clause_literals)
if result >= 0:
return result

clause = self._literals_to_var_clause(clause_literals)

# Add clause temporarily
self.cnf.clauses.append(clause)

try:
# Count models with new clause (creates fresh solver from self.cnf)
count = self.count_solutions()
count = self._enumerate_models()
return count
finally:
# Remove temporary clause (restore original state)
self.cnf.clauses.pop()
self.cnf.clauses.pop()

def _count_models_with_clause_via_filter(self, clause_literals: frozenset[str]) -> int:
"""
Filter cached solutions by a disjunctive clause.

A clause (l₁ ∨ ... ∨ lₖ) is satisfied if ANY literal is true:
- Positive literal p: p ∈ solution
- Negative literal ¬p: base fluent ∉ solution

Only applicable when all fluents in the clause are known to the CNF.
Returns None if unknown fluents are found (caller should fall back to SAT).

Args:
clause_literals: Set of literal strings (prefix '¬' or '-' for negation)

Returns:
Number of cached solutions satisfying the clause, or -1 if
clause references unknown fluents (caller falls back to SAT)
"""
if not self._solution_cache:
return 0

# Parse literals into positive and negative base fluents
positive = set()
negative = set()
for lit in clause_literals:
if lit.startswith('¬'):
negative.add(lit[1:])
elif lit.startswith('-'):
negative.add(lit[1:])
else:
positive.add(lit)

# All fluents must be known to the CNF for filtering to be valid
all_fluents = positive | negative
if not all_fluents.issubset(self.fluent_to_var):
return -1

count = 0
for solution in self._solution_cache:
# Clause satisfied if ANY positive literal is in solution
if positive and not positive.isdisjoint(solution):
count += 1
continue
# Or ANY negative literal's base fluent is absent from solution
if negative and not negative.issubset(solution):
count += 1
return count
1 change: 0 additions & 1 deletion information_gain_aml/experiments/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -323,7 +323,6 @@ def export(self, filepath: str | Path, format: str = 'csv') -> None:
'actions': self.metrics_df.to_dict(orient='records'),
'summary': {
'total_actions': len(self.metrics_df),
'final_mistake_rate': self.compute_mistake_rate(),
'average_runtime': self.compute_average_runtime(),
'action_distribution': self.get_action_distribution()
}
Expand Down
2 changes: 1 addition & 1 deletion information_gain_aml/experiments/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -588,7 +588,7 @@ def _export_results(self, results: Dict[str, Any]) -> None:

# Create timestamp marker file in problem directory
timestamp = self.start_time.strftime('%Y%m%d_%H%M%S') if self.start_time else "unknown"
timestamp_file = base_output_dir / f"time{timestamp}"
timestamp_file = output_dir / f"time{timestamp}"
timestamp_file.touch()

logger.info(f"Exported results to {output_dir}")
Expand Down
48 changes: 48 additions & 0 deletions tests/core/test_cnf_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -427,6 +427,54 @@ def test_count_models_with_temporary_clause(self):
# Original CNF unchanged
assert cnf.count_solutions() == 3

def test_count_models_with_temporary_clause_cached(self):
"""Test that cache-filtered temporary clause gives same results as SAT."""
cnf = CNFManager()
cnf.add_clause(['a', 'b'])

# Populate cache — solutions: {a}, {b}, {a,b}
cnf.get_all_solutions()
assert cnf._cache_valid

# Positive literal clause — (a): solutions where a is true
count = cnf.count_models_with_temporary_clause(frozenset({'a'}))
# {a}: yes. {b}: no. {a,b}: yes.
assert count == 2

# Negative literal clause — (¬a): solutions where a is NOT true
count = cnf.count_models_with_temporary_clause(frozenset({'¬a'}))
# {a}: no. {b}: yes. {a,b}: no.
assert count == 1

# Mixed clause — (a ∨ ¬b): satisfied if a∈solution OR b∉solution
count = cnf.count_models_with_temporary_clause(frozenset({'a', '¬b'}))
# {a}: a∈ → yes. {b}: a∉ and b∈ → no. {a,b}: a∈ → yes.
assert count == 2

# Unknown fluent falls back to SAT (no error)
count = cnf.count_models_with_temporary_clause(frozenset({'c'}))
assert count == 3 # (a∨b) ∧ (c) — all 3 base solutions + c=True

# Original CNF unchanged
assert cnf.count_solutions() == 3
assert len(cnf.cnf.clauses) == 1

def test_count_solutions_uses_cache(self):
"""Test that count_solutions returns len(cache) when cache is valid."""
cnf = CNFManager()
cnf.add_clause(['a', 'b'])

# Without cache — enumerates
assert cnf.count_solutions() == 3

# Populate cache
solutions = cnf.get_all_solutions()
assert len(solutions) == 3
assert cnf._cache_valid

# With cache — returns len(cache) without re-enumerating
assert cnf.count_solutions() == 3

def test_build_from_constraint_sets(self):
"""Test building CNF from constraint sets."""
cnf = CNFManager()
Expand Down
Loading