solver.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Author: Nitin Kedia, 160101048\n",
    "Description: A CDCL based SAT solver\n",
    "\"\"\"\n",
    "\n",
    "import time\n",
    "from enum import Enum\n",
    "\n",
    "class CONSTANTS:\n",
    "    # These constant will be explained in context\n",
    "    # used in MINISAT (decision heuristic)\n",
    "    VAR_DECAY_RATE = 0.95 \n",
    "    # used to decide when to restart\n",
    "    THRESHOLD_MULTIPLIER = 1.1\n",
    "    RESTART_LOWER_BOUND = 100\n",
    "    RESTART_UPPER_BOUND_BASE = 1000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ClauseState(Enum):\n",
    "    C_UNRESOLVED = 0\n",
    "    C_SATISFIED = 1\n",
    "    C_CONFLICTING = 2\n",
    "    C_UNIT = 3 # Only one literal is undecided, rest eval to false\n",
    "\n",
    "# Note that LiteralState also functions as VariableState\n",
    "# The state a literal of a variable fixes that of the variable and vice-versa\n",
    "class LiteralState(Enum):\n",
    "    L_UNASSIGNED = -1\n",
    "    L_FALSE = 0\n",
    "    L_TRUE = 1\n",
    "\n",
    "class SolverState(Enum):\n",
    "    S_UNRESOLVED = 0\n",
    "    S_SATISFIED = 1\n",
    "    S_UNSATISFIED = 2\n",
    "    S_CONFLICT = 3\n",
    "\n",
    "\"\"\"\n",
    "Variables are denoted by positive integers 1, 2, ..., VAR_COUNT\n",
    "The positive literal of a variable v is given by 2*v, while negation is 2*v - 1\n",
    "Following four are help functions based on this terminology\n",
    "\"\"\"\n",
    "def get_literal(v: int) -> int:\n",
    "    if (v == 0):\n",
    "        raise Exception(\"Variable cannot be zero.\")\n",
    "    return 2 * v if v > 0 else 2 * (abs(v)) - 1\n",
    "\n",
    "def get_variable(l : int) -> int:\n",
    "    if (l <= 0):\n",
    "        raise Exception(\"Literal must be a positive integer, is {}\".format(l))\n",
    "    return int((l + 1) / 2) if l % 2  else int(l / 2)\n",
    "\n",
    "def get_opposite_literal(l : int) -> int:\n",
    "    if (l <= 0):\n",
    "        raise Exception(\"Literal must be a positive integer, is {}\".format(l))\n",
    "    return (l + 1) if l % 2 else (l - 1)\n",
    "\n",
    "def is_negative(l : int) -> bool:\n",
    "    if (l <= 0):\n",
    "        raise Exception(\"Literal must be a positive integer, is {}\".format(l))\n",
    "    return True if l % 2 else False\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Clause:\n",
    "    def __init__(self, literals: list):\n",
    "        self.literals = literals\n",
    "        self.is_unary = (len(literals) == 1)\n",
    "        self.first_watcher = -1\n",
    "        self.second_watcher = -1\n",
    "\n",
    "    def insert_literal(self, lit):\n",
    "        if lit not in self.literals: # To ensure all literals are unique\n",
    "            self.literals.append(lit)\n",
    "            self.is_unary = (len(self.literals) == 1)\n",
    "\n",
    "    def print(self, clause_id: int):\n",
    "        print(\"Clause {} with literals {}, watchers {} and {}\".format(\n",
    "            clause_id, self.literals, self.first_watcher, self.second_watcher)\n",
    "        )\n",
    "    \n",
    "    def get_first_watcher(self):\n",
    "        return self.literals[self.first_watcher]\n",
    "    def get_second_watcher(self):\n",
    "        return self.literals[self.second_watcher]\n",
    "\n",
    "    \"\"\"\n",
    "    Two watch optimisation used in BCP (unit propagation):\n",
    "    Base case: Initially we keep two watchers pointing to two different (unassigned) literals in the clause.\n",
    "    Invariant: If both the watchers are unassigned, the clause cannot be unit (which requires exactly one assigned)\n",
    "    \"\"\"\n",
    "    def change_watch_location(self, solver, is_first_watcher, other_watcher) -> (ClauseState, int):\n",
    "        # This function is called in bcp(). One of the watchers depending on is_first_watcher is being assigned FALSE.\n",
    "        # First, try to find another unassigned literal\n",
    "        for index, lit in enumerate(self.literals):\n",
    "            if (lit == other_watcher):\n",
    "                continue\n",
    "            lit_state = solver.get_literal_status(lit)\n",
    "            if (lit_state != LiteralState.L_FALSE):\n",
    "                if is_first_watcher:\n",
    "                    self.first_watcher = index\n",
    "                else:\n",
    "                    self.second_watcher = index\n",
    "                return ClauseState.C_UNRESOLVED, index\n",
    "        # Could not find another unassigned, fate of the clause is in the hands of the other watcher\n",
    "        other_watch_status = solver.get_literal_status(other_watcher)\n",
    "        if (other_watch_status == LiteralState.L_FALSE):\n",
    "            # All literals are false\n",
    "            return ClauseState.C_CONFLICTING, None\n",
    "        elif other_watch_status == LiteralState.L_UNASSIGNED:\n",
    "            # Exactly one unassigned (other watcher) remains\n",
    "            return ClauseState.C_UNIT, None\n",
    "        else:\n",
    "            assert(other_watch_status == LiteralState.L_TRUE)\n",
    "            return ClauseState.C_SATISFIED, None\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Solver:\n",
    "    def __init__(self, var_count, clause_count):\n",
    "        self.var_count = var_count\n",
    "        self.clause_count = clause_count\n",
    "        self.clauses = []\n",
    "        self.unary_clauses = []\n",
    "        self.curr_level = 0 # Current depth of the decision tree\n",
    "        self.max_level = 0\n",
    "\n",
    "        # Since variables are 1-indexed, size of these lists if (var_count + 1)\n",
    "        # curr_assignment gives the latest assignment of a variable\n",
    "        self.curr_assignment = [LiteralState.L_UNASSIGNED] * (var_count + 1)\n",
    "        # prev_assignment is used in PHASE SAVING\n",
    "        self.prev_assignment = [-1] * (var_count + 1)\n",
    "        # The level the variable was assigned at (if at all)\n",
    "        self.assignment_level = [-1] * (var_count + 1)\n",
    "        # A stack of all assigned variables in current path,\n",
    "        # most recently assigned variables are at top\n",
    "        self.assigned_till_now = []\n",
    "        # How many assignments had happened upto a level?\n",
    "        self.assignments_upto_level = [0]\n",
    "        # How many conflicts hence clauses learned upto a level?\n",
    "        self.conflicts_upto_level = [0]\n",
    "        self.antecedent = [-1] * (var_count + 1)\n",
    "\n",
    "        self.bcp_stack = []\n",
    "        # watch_map: literal -> list of clauses for which this is the watcher\n",
    "        self.watch_map = {} \n",
    "\n",
    "        # Used in MINISAT decision heuristic explained in decider()    \n",
    "        self.increment_value = 1.0\n",
    "        self.activity = [0.0] * (var_count + 1)\n",
    "\n",
    "        # Used in restart optimisation explained in reset_state()\n",
    "        self.restart_threshold = CONSTANTS.RESTART_LOWER_BOUND\n",
    "        self.restart_upper_bound = CONSTANTS.RESTART_UPPER_BOUND_BASE\n",
    "        \n",
    "        # Statistics\n",
    "        self.restart_count = 0\n",
    "        self.learnt_clauses_count = 0\n",
    "        self.decision_count = 0\n",
    "        self.assignments_count = 0\n",
    "\n",
    "    def print_clauses(self):\n",
    "        print(\"{} variables, {} clauses\".format(self.var_count, self.clause_count))\n",
    "        for clause_id, clause in enumerate(self.clauses):\n",
    "            clause.print(clause_id)\n",
    "    \n",
    "    def print_curr_assignment(self):\n",
    "        assignment = \"State: \"\n",
    "        for var, state in enumerate(self.curr_assignment):\n",
    "            if (var == 0):\n",
    "                continue\n",
    "            assignment += \", {}: {}\".format(var, state.value)\n",
    "        print(assignment)\n",
    "    \n",
    "    # This fn is used to add the given clause to the watchlist of given literal\n",
    "    def watch_this_clause(self, lit, clause_id):\n",
    "        if lit in self.watch_map: \n",
    "            self.watch_map[lit].append(clause_id)\n",
    "        else:\n",
    "            self.watch_map[lit] = [clause_id]\n",
    "    \n",
    "    # Insert a new (input / learned) clause to the cnf\n",
    "    def insert_clause(self, clause : Clause, first_watch, second_watch):\n",
    "        self.clauses.append(clause)\n",
    "        # Setup the two-watch mechanism\n",
    "        # both these literals are guaranteed to be unassigned currently\n",
    "        clause.first_watcher = first_watch\n",
    "        clause.second_watcher = second_watch\n",
    "        clause_id = len(self.clauses) - 1\n",
    "        self.watch_this_clause(clause.get_first_watcher(), clause_id)\n",
    "        self.watch_this_clause(clause.get_second_watcher(), clause_id)\n",
    "        \n",
    "        # In MINISAT decision heusristic:\n",
    "        # Score of a varible is the number of clauses in it\n",
    "        # While inserting a clause, increase the scores of variables in it\n",
    "        for literal in clause.literals:\n",
    "            var = get_variable(literal)\n",
    "            self.activity[var] += self.increment_value\n",
    "    \n",
    "    # Function used to assign a literal TRUE in a unary clause\n",
    "    # These assignments are never reset hence not put in assigned_till_now[]\n",
    "    def assert_unary_literal(self, lit):\n",
    "        self.assignments_count += 1\n",
    "        var = get_variable(lit)\n",
    "        # Set state of the underlying variable\n",
    "        if is_negative(lit):\n",
    "            self.curr_assignment[var] = LiteralState.L_FALSE\n",
    "        else:\n",
    "            self.curr_assignment[var] = LiteralState.L_TRUE\n",
    "        self.assignment_level[var] = 0 # Always done at ground level\n",
    "    \n",
    "    # Function used to assign a literal TRUE in a non-unary clause\n",
    "    # Note that current level is important here\n",
    "    def assert_nonunary_literal(self, lit):\n",
    "        self.assignments_count += 1\n",
    "        self.assigned_till_now.append(lit)\n",
    "        var = get_variable(lit)\n",
    "        if is_negative(lit):\n",
    "            self.prev_assignment[var] = self.curr_assignment[var] = LiteralState.L_FALSE\n",
    "        else:\n",
    "            self.prev_assignment[var] = self.curr_assignment[var] = LiteralState.L_TRUE\n",
    "        self.assignment_level[var] = self.curr_level\n",
    "\n",
    "    # Returns the state of given literal\n",
    "    # States of variables like uassgned, true etc. are tracked, state of literals can be found from it\n",
    "    def get_literal_status(self, lit : int) -> LiteralState:\n",
    "        var_status = self.curr_assignment[get_variable(lit)]\n",
    "        if var_status == LiteralState.L_UNASSIGNED:\n",
    "            return LiteralState.L_UNASSIGNED\n",
    "        elif var_status == LiteralState.L_TRUE:\n",
    "            return LiteralState.L_FALSE if is_negative(lit) else LiteralState.L_TRUE\n",
    "        else:\n",
    "            return LiteralState.L_TRUE if is_negative(lit) else LiteralState.L_FALSE\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Function to implement Boolean Constant Propagation using Two-watcher optimisation:\n",
    "bcp_stack contains all the literals which have been assigned false in current search path.\n",
    "Since we know these literals can now change the state of other clauses.\n",
    "A naive approach of bcp would be iterate every clause to find a unit/unsatisifed clause.\n",
    "If found, repreat the process again else, stop and start guessing some variables in decider()\n",
    "\"\"\"\n",
    "def bcp(self) -> (SolverState, int):\n",
    "    # print(\"Running BCP with stack\", self.bcp_stack)\n",
    "    conflicting_clause_id = -1\n",
    "    while (self.bcp_stack):\n",
    "        # Got a literal with FALSE assignment\n",
    "        lit = self.bcp_stack.pop()\n",
    "        assert self.get_literal_status(lit) == LiteralState.L_FALSE\n",
    "        \n",
    "        if lit not in self.watch_map:\n",
    "            self.watch_map[lit] = []\n",
    "        new_watch_list = [x for x in self.watch_map[lit]] # Backup watch list of lit\n",
    "        \n",
    "        \n",
    "        # Traverse only the watchlist of that clause to save computation\n",
    "        for clause_id in reversed(self.watch_map[lit]):\n",
    "            clause = self.clauses[clause_id]\n",
    "            \n",
    "            # This block determines which watcher (1st / 2nd) was lit\n",
    "            first_watch = clause.get_first_watcher()\n",
    "            second_watch = clause.get_second_watcher()\n",
    "            lit_is_first = (lit == first_watch)\n",
    "            other_watch = second_watch if lit_is_first else first_watch\n",
    "            # Now that we know lit has been assigned FALSE, we need to find another watcher\n",
    "            new_clause_state, new_watch_loc = clause.change_watch_location(self, lit_is_first, other_watch)\n",
    "            \n",
    "            # clause has one more literal FALSE, this might change a state\n",
    "            if (new_clause_state == ClauseState.C_SATISFIED):\n",
    "                pass\n",
    "            elif (new_clause_state == ClauseState.C_UNIT):\n",
    "                # If the clause had become unit, we have got another implication here\n",
    "                self.assert_nonunary_literal(other_watch)\n",
    "                var = get_variable(other_watch)\n",
    "                self.antecedent[var] = clause_id\n",
    "                self.bcp_stack.append(get_opposite_literal(other_watch))\n",
    "            elif (new_clause_state == ClauseState.C_CONFLICTING):\n",
    "                # All the literals of this clause became false, we have a conflict, need to backtrack\n",
    "                # If the conflict occured at ground level, we have a unsatisfiable cnf like (x) ^ (-x)\n",
    "                if self.curr_level == 0:\n",
    "                    return SolverState.S_UNSATISFIED, conflicting_clause_id\n",
    "                conflicting_clause_id = clause_id\n",
    "                # Clear bcp_stack as a backtrack is coming, which will unassign several variables\n",
    "                # As such some information in bcp_state is likely to become stale\n",
    "                self.bcp_stack.clear()\n",
    "                break\n",
    "            elif (new_clause_state == ClauseState.C_UNRESOLVED):\n",
    "                # The clause is still unresolved as we have found another watcher\n",
    "                # Remove this clause from watch list of current lit\n",
    "                new_watch_list.remove(clause_id)\n",
    "                new_watcher = clause.literals[new_watch_loc]\n",
    "                self.watch_this_clause(new_watcher, clause_id)\n",
    "        \n",
    "        # new_watch_list contains the clauses for which lit is still the watcher\n",
    "        # Note that in case of backtrack, we dot need to revert the watchers in two-watcher method\n",
    "        # since in backtracking, some variables will be unassigned, enforcing the two-watch invariant\n",
    "        self.watch_map[lit].clear()\n",
    "        self.watch_map[lit] += new_watch_list\n",
    "        if (conflicting_clause_id >= 0):\n",
    "            return SolverState.S_CONFLICT, conflicting_clause_id\n",
    "    return SolverState.S_UNRESOLVED, conflicting_clause_id\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "This function is for the PHASE-SAVING heuristic\n",
    "In decider() after the variable to be guessed has been selected, we then\n",
    "need it set it to TRUE of FALSE. Phase-saving says we should set it to our\n",
    "previous assignment if any.  \n",
    "\"\"\"\n",
    "def get_lit_memoised(self, var: int) -> int:\n",
    "    prev_state = self.prev_assignment[var]\n",
    "    if (prev_state == LiteralState.L_TRUE):\n",
    "        return get_literal(var)\n",
    "    else:\n",
    "        return get_literal(-1 * var)\n",
    "\n",
    "\"\"\"\n",
    "decide() function selects the next variable to be guessed and the guessed value.\n",
    "Based on MINISAT decision heuristic. Results in increment of current level.  \n",
    "Score of a varible is the number of clauses in it.\n",
    "\"\"\"\n",
    "def decide(self) -> SolverState: # MINISAT based decision heuristic\n",
    "    # print(\"Running decider\")\n",
    "    # self.print_curr_assignment()\n",
    "    # print(\"Activity: \", self.activity)\n",
    "    # selected_var = 0\n",
    "\n",
    "    # Iterate through all variables to find an unassgned one with maximum score\n",
    "    # Some inputs have unused variables, so we select only those with positive score.\n",
    "    selected_lit = 0\n",
    "    max_activity_till_now = 0.0\n",
    "    unassigned_var_found = False\n",
    "    for var, state in enumerate(self.curr_assignment):\n",
    "        if (var == 0):\n",
    "            continue\n",
    "        if (state == LiteralState.L_UNASSIGNED and self.activity[var] > 0.0):\n",
    "            unassigned_var_found = True\n",
    "            if (self.activity[var] > max_activity_till_now):\n",
    "                # selected_var = var\n",
    "                selected_lit = self.get_lit_memoised(var)\n",
    "                max_activity_till_now = self.activity[var]\n",
    "    if not unassigned_var_found:\n",
    "        return SolverState.S_SATISFIED\n",
    "    # print(selected_lit, selected_var, max_activity_till_now)\n",
    "    assert selected_lit != 0\n",
    "    self.decision_count += 1\n",
    "    self.curr_level += 1\n",
    "    # We need to track this new assignment\n",
    "    if (self.curr_level > self.max_level):\n",
    "        # This branch is separate since we are at a new decision level,\n",
    "        # so push_back is required instead of update\n",
    "        self.max_level = self.curr_level\n",
    "        self.assignments_upto_level.append(len(self.assigned_till_now))\n",
    "        self.conflicts_upto_level.append(self.learnt_clauses_count)\n",
    "    else:\n",
    "        self.assignments_upto_level[self.curr_level] = len(self.assigned_till_now) \n",
    "        self.conflicts_upto_level[self.curr_level] = self.learnt_clauses_count       \n",
    "    \n",
    "    # Now we assign the literal as TRUE, and since put the (FALSE) opposite literal to bcp stack\n",
    "    self.assert_nonunary_literal(selected_lit)\n",
    "    self.bcp_stack.append(get_opposite_literal(selected_lit))\n",
    "    return SolverState.S_UNRESOLVED\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "analyse_conflict() takes a conflicting clause and returns the level to backtrack to, and a learned clause\n",
    "We use the nearest UIP (Unique Implication Point) finding method as highlighted in Kroening's book.\n",
    "\"\"\"\n",
    "def analyze_conflict(self, conflicting_clause: Clause) -> (int, int):\n",
    "    # print(\"Running analyse_conflict\")\n",
    "    curr_literals = [lit for lit in conflicting_clause.literals]\n",
    "    learned_clause = Clause([])\n",
    "    backtrack_level = 0 # to be returned by this function\n",
    "    to_resolve_count = 0\n",
    "    watch_lit = 0 # a watcher for the new learned literal\n",
    "\n",
    "    marked = [False] * (self.var_count + 1)\n",
    "    trail_index = len(self.assigned_till_now) - 1\n",
    "    resolve_lit = 0\n",
    "    resolve_var = 0\n",
    "    iter = 0\n",
    "    \"\"\"\n",
    "    This loop outputs the learned clause, it works as follows:\n",
    "    Invariant 1: curr_literals is the clause to be fused into learned_clause\n",
    "    Invariant 2: learned caluse contains exactly one variable assigned at current level (UIP)\n",
    "    All other literals are assigned before.\n",
    "    \"\"\"\n",
    "    while (iter == 0 or to_resolve_count > 0):\n",
    "        iter += 1\n",
    "        for lit in curr_literals:\n",
    "            var = get_variable(lit)\n",
    "            if marked[var]:\n",
    "                continue\n",
    "            marked[var] = True\n",
    "            if (self.assignment_level[var] == self.curr_level):\n",
    "                to_resolve_count += 1\n",
    "            else:\n",
    "                learned_clause.insert_literal(lit)\n",
    "                if (self.assignment_level[var] > backtrack_level):\n",
    "                    # watch_lit: 2nd highest assigment level, first is UIP\n",
    "                    backtrack_level = self.assignment_level[var]\n",
    "                    watch_lit = len(learned_clause.literals) - 1\n",
    "        # Find a variable to be resolved by traversing the recently assigned literals first\n",
    "        while (trail_index >= 0):\n",
    "            resolve_lit = self.assigned_till_now[trail_index]\n",
    "            resolve_var = get_variable(resolve_lit)\n",
    "            trail_index -= 1\n",
    "            if marked[resolve_var]:\n",
    "                break\n",
    "        marked[resolve_var] = False\n",
    "        to_resolve_count -= 1\n",
    "        if not to_resolve_count:\n",
    "            # Just one literal remaining with current level assignment, we are done\n",
    "            continue \n",
    "        antecedent_id = self.antecedent[resolve_var]\n",
    "        curr_literals = [lit for lit in self.clauses[antecedent_id].literals if lit != resolve_lit]\n",
    "    \n",
    "    # The learned clause becomes an unit clause after backtracking\n",
    "    # This is because every other literal in the learned clause was assigned before\n",
    "    # the backtrack level\n",
    "    # resolve_lit is an UIP\n",
    "    self.learnt_clauses_count += 1\n",
    "    opposite_resolv_lit = get_opposite_literal(resolve_lit)\n",
    "    learned_clause.insert_literal(opposite_resolv_lit)\n",
    "    self.increment_value /= CONSTANTS.VAR_DECAY_RATE\n",
    "    if learned_clause.is_unary:\n",
    "        # Not that we are inserting to bcp_stack without asserting UIP\n",
    "        # Asserting will be done immediately after backtrack (see backtrack())\n",
    "        self.bcp_stack.append(resolve_lit)\n",
    "        self.unary_clauses.append(learned_clause)\n",
    "    else:\n",
    "        self.bcp_stack.append(resolve_lit)\n",
    "        self.insert_clause(learned_clause, watch_lit, len(learned_clause.literals) - 1)\n",
    "    # for lit in learned_clause.literals:\n",
    "    #     var = get_variable(lit)\n",
    "    #     print(\"({}, {})\".format(var, self.assignment_level[var]))\n",
    "    return backtrack_level, opposite_resolv_lit\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "# RESTART heuristic, reset all assignments except ground level and start afresh\n",
    "# Note that learned claused are not deleted only we start assignments from the beginning \n",
    "def reset_state(self):\n",
    "    # print(\"Restart\")\n",
    "    \"\"\"\n",
    "    The threshold system works as follows eg. (chainsaw graph)\n",
    "    1. We have a range [1, 10]. Threshold increases after every restart till it crosses the ub\n",
    "    2. At that point the threshold is rest and the range is also increased to let it go even higher\n",
    "    \"\"\"\n",
    "    self.restart_count += 1\n",
    "    self.restart_threshold = int(self.restart_threshold * CONSTANTS.THRESHOLD_MULTIPLIER)\n",
    "    if (self.restart_threshold > self.restart_upper_bound):\n",
    "        self.restart_threshold = CONSTANTS.RESTART_LOWER_BOUND\n",
    "        self.restart_upper_bound = int(self.restart_upper_bound * CONSTANTS.THRESHOLD_MULTIPLIER)\n",
    "\n",
    "    # Resets are similar to backtrack() function below, except that it resets evrything to ground level  \n",
    "    for var in range(1, self.var_count + 1):\n",
    "        if (self.assignment_level[var] > 0):\n",
    "            self.curr_assignment[var] = LiteralState.L_UNASSIGNED\n",
    "    \n",
    "    self.bcp_stack.clear()\n",
    "    self.assigned_till_now.clear()\n",
    "    self.assignments_upto_level = [0]\n",
    "    self.conflicts_upto_level = [0]\n",
    "    self.curr_level = 0\n",
    "    self.max_level = 0\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to backtrack based on the output of analyse_conflict() \n",
    "def backtrack(self, k: int, uip_lit):\n",
    "    # print(\"Running backtrack\")\n",
    "    \n",
    "    # if (k < 0 or k >= len(self.conflicts_upto_level)):\n",
    "    #     print(\"Current level {}, backtrack to {} and max level {}\".format(self.curr_level, k, self.max_level))\n",
    "    #     print(self.conflicts_upto_level)\n",
    "    \n",
    "    # Invoke restart heuristic if too many clauses have been learnt after backtrack target level\n",
    "    if k > 0 and (self.learnt_clauses_count - self.conflicts_upto_level[k] > self.restart_threshold):\n",
    "        self.reset_state()\n",
    "        return\n",
    "    # Iterate over the variables assigned at level >= k + 1 and unassign them\n",
    "    for index in range(self.assignments_upto_level[k+1], len(self.assigned_till_now)):\n",
    "        var = get_variable(self.assigned_till_now[index])\n",
    "        if (self.assignment_level[var] > k):\n",
    "            self.curr_assignment[var] = LiteralState.L_UNASSIGNED\n",
    "\n",
    "    # analyse_function() returns an asserting clause with the UIP just ready for assignment\n",
    "    # This helps to immediately put the learnt clause into practice\n",
    "    self.assigned_till_now = self.assigned_till_now[:self.assignments_upto_level[k+1]]\n",
    "    self.curr_level = k\n",
    "    if k == 0:\n",
    "        # We had learnt a unary clause\n",
    "        self.assert_unary_literal(uip_lit)\n",
    "    else:\n",
    "        self.assert_nonunary_literal(uip_lit)\n",
    "    self.antecedent[get_variable(uip_lit)] = len(self.clauses) - 1\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Function implementing the standard CDCL framework:\n",
    "    1. [Outer Loop] Run BCP and decide() alternately, bcp first because of unary clauses\n",
    "    2. [INNER LOOP] Run till BCP gives UNRESOLVED result on which point guesswork must be done.\n",
    "    If bcp encounters conflict a analyse, backtrack pair is done\n",
    "\"\"\"     \n",
    "def run_cdcl(self) -> SolverState:\n",
    "    result: SolverState\n",
    "    while (True):\n",
    "        while (True):\n",
    "            result, conflicting_clause_id = self.bcp()\n",
    "            # print(\"BCP result was {}\".format(result))\n",
    "            if (result == SolverState.S_UNSATISFIED):\n",
    "                return result\n",
    "            if (result == SolverState.S_CONFLICT):\n",
    "                assert conflicting_clause_id != -1\n",
    "                backtrack_level, uip_lit = self.analyze_conflict(self.clauses[conflicting_clause_id])\n",
    "                # print(\"Analyze result was k = {}, uip = {}\".format(backtrack_level, uip_lit))\n",
    "                self.backtrack(backtrack_level, uip_lit)\n",
    "            else:\n",
    "                break\n",
    "        result = self.decide()\n",
    "        # print(\"Decide result was {}\".format(result))\n",
    "        if (result == SolverState.S_UNSATISFIED or result == SolverState.S_SATISFIED):\n",
    "            return result\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to verify output assignment if any\n",
    "def verify_assignment(self):\n",
    "    non_true_clauses = []\n",
    "    all_clauses = self.clauses + self.unary_clauses\n",
    "    # Every clause including learnt and unary must have atleast one TRUE literal\n",
    "    for clause in all_clauses:\n",
    "        true_literal_found = False\n",
    "        for lit in clause.literals:\n",
    "            if (self.get_literal_status(lit) == LiteralState.L_TRUE):\n",
    "                true_literal_found = True\n",
    "                break\n",
    "        if not true_literal_found:\n",
    "            non_true_clauses.append(clause)\n",
    "    if not non_true_clauses:\n",
    "        print(\"AC, All clauses evaluate to true under given assignment\")\n",
    "    else:\n",
    "        print(\"WA, {} unsatisfied clauses found\".format(len(non_true_clauses)))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Wrapper function to print the result of the CDCL framework\n",
    "def solve(self):\n",
    "    # print(\"Solving\")\n",
    "    result : SolverState = self.run_cdcl()\n",
    "    if (result == SolverState.S_SATISFIED):\n",
    "        # print(\"Assignment of {} variables:\".format(self.var_count))\n",
    "        # for var, state in enumerate(self.curr_assignment):\n",
    "        #     if (var == 0):\n",
    "        #         continue\n",
    "        #     print(\"{}: {},\".format(var, state.value), end = \" \")\n",
    "        # print()\n",
    "        print(\"SATISFIABLE\")\n",
    "    else:\n",
    "        print(\"UNSATISFIABLE\")\n",
    "\n",
    "def print_statistics(self, solve_time):\n",
    "    print(\"## Statistics: \")\n",
    "    print(\"# Restarts: \", self.restart_count)\n",
    "    print(\"# Learned cluases: \", self.learnt_clauses_count)\n",
    "    print(\"# Decisions: \", self.decision_count)\n",
    "    print(\"# Implications: \", self.assignments_count - self.decision_count)\n",
    "    print(\"# Time (s): \", solve_time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "Solver.bcp = bcp\n",
    "Solver.get_lit_memoised = get_lit_memoised\n",
    "Solver.decide = decide\n",
    "Solver.analyze_conflict = analyze_conflict\n",
    "Solver.reset_state = reset_state\n",
    "Solver.backtrack = backtrack\n",
    "Solver.verify_assignment = verify_assignment\n",
    "Solver.run_cdcl = run_cdcl\n",
    "Solver.solve = solve\n",
    "Solver.print_statistics = print_statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Main function\n",
    "def read_and_solve_cnf(input_file):\n",
    "    # Read the specified input cnf file\n",
    "    input_file = open(input_file, 'r')\n",
    "    current_line = input_file.readline()\n",
    "    tokens = current_line.split()\n",
    "    while (tokens[0] != \"p\"):\n",
    "        # Some inputs have redundant starting lines like \"c 1 = P9_0[0]\"\n",
    "        current_line = input_file.readline()\n",
    "        tokens = current_line.split()\n",
    "    var_count = int(tokens[-2])\n",
    "    clause_count = int(tokens[-1])\n",
    "    solver = Solver(var_count, clause_count)\n",
    "    # Read each clause of the CNF\n",
    "    for _ in range(clause_count):\n",
    "        current_line = input_file.readline()\n",
    "        tokens = current_line.split()\n",
    "        assert(tokens.pop() == \"0\")\n",
    "        \n",
    "        # set removes duplicate literals from the clause\n",
    "        literals = set([get_literal(int(literal)) for literal in tokens])\n",
    "        curr_clause = Clause(list(literals))\n",
    "        # unary clauses are processed at ground level\n",
    "        # this also gives some false literals to process in the bcp_stack\n",
    "        if (curr_clause.is_unary):\n",
    "            lit = curr_clause.literals[0]\n",
    "            solver.assert_unary_literal(lit)\n",
    "            solver.bcp_stack.append(get_opposite_literal(lit))\n",
    "            solver.unary_clauses.append(curr_clause)\n",
    "        else:\n",
    "            solver.insert_clause(curr_clause, 0, 1)\n",
    "\n",
    "    # solver.print_clauses()\n",
    "    start_time = time.process_time()\n",
    "    solver.solve()\n",
    "    finish_time = time.process_time()\n",
    "    solver.verify_assignment()\n",
    "    solver.print_statistics(finish_time - start_time)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SATISFIABLE\n",
      "AC, All clauses evaluate to true under given assignment\n",
      "## Statistics: \n",
      "# Restarts:  0\n",
      "# Learned cluases:  50\n",
      "# Decisions:  243\n",
      "# Implications:  11615\n",
      "# Time (s):  0.48908218200000064\n"
     ]
    }
   ],
   "source": [
    "# Use this cell to run the solver\n",
    "input_file = \"input/sat/bmc-2.cnf\"\n",
    "read_and_solve_cnf(input_file)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.2 64-bit",
   "language": "python",
   "name": "python38264bit0fc2e25e52884affaeb7d18f1db7795a"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}