ProgrammingWithRafay/wumpus-backend

1# The Knowledge-Based Wumpus Agent using propositional logic.2 3from typing import Dict, List, Optional, Set, Tuple4 5from knowledge_base import Clause, neg, pl_resolution6 7 8class WumpusAgent:9    def __init__(self, rows: int, cols: int):10        self.rows = rows11        self.cols = cols12 13        # Current agent position14        self.position: Tuple[int, int] = (0, 0)15 16        # Cells the agent has visited17        self.visited: Set[Tuple[int, int]] = set()18 19        # Cells known to be safe (proven by resolution or directly observed)20        self.safe_cells: Set[Tuple[int, int]] = set()21 22        # Cells confirmed to have hazards (from KB unit clauses)23        self.confirmed_pits: Set[Tuple[int, int]] = set()24        self.confirmed_wumpus: Set[Tuple[int, int]] = set()25 26        # The knowledge base: list of CNF clauses27        self.kb: List[Clause] = []28 29        # Tracking metrics30        self.total_inference_steps: int = 031        self.last_percepts: Dict[str, bool] = {}32        self.last_safe_inferred: bool = False33        self.status_message: str = "Agent initialized."34 35        # Initialize: start cell (0,0) is safe (agent spawned there alive)36        self._mark_safe(0, 0)37        self.visited.add((0, 0))38 39    # -----------------------------------------------------------------40    # Helper: cell ID strings for propositional variables41    # -----------------------------------------------------------------42 43    def _pit_id(self, r: int, c: int) -> str:44        return f"Pit_{r}_{c}"45 46    def _wumpus_id(self, r: int, c: int) -> str:47        return f"Wumpus_{r}_{c}"48 49    def _get_neighbors(self, r: int, c: int) -> List[Tuple[int, int]]:50        neighbors = []51        for dr, dc in [(-1, 0), (1, 0), (0, -1), (0, 1)]:52            nr, nc = r + dr, c + dc53            if 0 <= nr < self.rows and 0 <= nc < self.cols:54                neighbors.append((nr, nc))55        return neighbors56 57    def _mark_safe(self, r: int, c: int):58        """Add unit clauses asserting no pit and no wumpus at (r,c)."""59        self.safe_cells.add((r, c))60        # Add as unit clauses: -Pit_r_c and -Wumpus_r_c61        no_pit = frozenset({neg(self._pit_id(r, c))})62        no_wumpus = frozenset({neg(self._wumpus_id(r, c))})63        if no_pit not in self.kb:64            self.kb.append(no_pit)65        if no_wumpus not in self.kb:66            self.kb.append(no_wumpus)67 68    # -----------------------------------------------------------------69    # TELL: update KB from percepts at current position70    # -----------------------------------------------------------------71 72    def tell(self, pos: Tuple[int, int], percepts: Dict[str, bool]):73        # Encode percepts as rules and add to KB74        r, c = pos75        neighbors = self._get_neighbors(r, c)76        self.last_percepts = percepts77 78        # -- Breeze rules --79        if not percepts.get("breeze", False):80            # No breeze => no pit in any adjacent cell81            for nr, nc in neighbors:82                clause = frozenset({neg(self._pit_id(nr, nc))})83                if clause not in self.kb:84                    self.kb.append(clause)85                    # Also directly mark the neighbor's pit status86                    self.safe_cells.add((nr, nc))87        else:88            # Breeze => at least one neighbor has a pit (disjunctive clause)89            pit_lits = frozenset(self._pit_id(nr, nc) for nr, nc in neighbors)90            if pit_lits not in self.kb:91                self.kb.append(pit_lits)92 93        # -- Stench rules --94        if not percepts.get("stench", False):95            # No stench => no wumpus in any adjacent cell96            for nr, nc in neighbors:97                clause = frozenset({neg(self._wumpus_id(nr, nc))})98                if clause not in self.kb:99                    self.kb.append(clause)100        else:101            # Stench => at least one neighbor has the wumpus102            wumpus_lits = frozenset(self._wumpus_id(nr, nc) for nr, nc in neighbors)103            if wumpus_lits not in self.kb:104                self.kb.append(wumpus_lits)105 106        # The current cell is safe (agent is alive here)107        self._mark_safe(r, c)108 109    # -----------------------------------------------------------------110    # ASK: query KB using resolution refutation111    # -----------------------------------------------------------------112 113    def ask_is_safe(self, pos: Tuple[int, int]) -> Tuple[bool, int]:114        # Query: Is cell 'pos' safe? (does KB entail NOT Pit AND NOT Wumpus?)115        r, c = pos116 117        # --- Prove NOT Pit(r,c) ---118        # Negated query: Pit(r,c) is true (unit clause)119        negated_not_pit = [frozenset({self._pit_id(r, c)})]120        no_pit_proved, steps1 = pl_resolution(self.kb, negated_not_pit)121 122        # --- Prove NOT Wumpus(r,c) ---123        negated_not_wumpus = [frozenset({self._wumpus_id(r, c)})]124        no_wumpus_proved, steps2 = pl_resolution(self.kb, negated_not_wumpus)125 126        total_steps = steps1 + steps2127        return (no_pit_proved and no_wumpus_proved), total_steps128 129    # -----------------------------------------------------------------130    # Decision: choose next move131    # -----------------------------------------------------------------132 133    def decide_move(self) -> Tuple[Optional[Tuple[int, int]], bool, int]:134        # Decide which cell to move to next.135        r, c = self.position136        neighbors = self._get_neighbors(r, c)137        unvisited_neighbors = [n for n in neighbors if n not in self.visited]138 139        total_steps = 0140 141        # Check each unvisited neighbor142        for candidate in unvisited_neighbors:143            # First check if it's already in the safe set (from previous no-breeze rules)144            if candidate in self.safe_cells:145                self.last_safe_inferred = True146                self.status_message = f"Safe move to {candidate} (already known safe)"147                return candidate, True, total_steps148 149            # Use resolution to check150            is_safe, steps = self.ask_is_safe(candidate)151            total_steps += steps152            self.total_inference_steps += steps153 154            if is_safe:155                self.safe_cells.add(candidate)156                self.last_safe_inferred = True157                self.status_message = f"Safe move to {candidate} (proven by resolution)"158                return candidate, True, total_steps159 160        # No safe neighbor found via resolution.161        # Look for any unvisited safe cell we can navigate to (simple fallback).162        reachable_safe = [163            cell for cell in self.safe_cells164            if cell not in self.visited165        ]166        if reachable_safe:167            target = reachable_safe[0]168            self.last_safe_inferred = True169            self.status_message = f"Backtracking to safe cell {target}"170            return target, True, total_steps171 172        # Truly stuck: no provably safe move available173        self.last_safe_inferred = False174        self.status_message = "No safe move can be proven. Agent is uncertain."175        return None, False, total_steps176 177    def move_to(self, pos: Tuple[int, int]):178        """Update agent position and mark cell as visited."""179        self.position = pos180        self.visited.add(pos)181