agents.dummy_agent

  1"""
  2Random Walker Agent implementation.
  3An agent that moves randomly among valid actions.
  4"""
  5
  6import asyncio
  7import json
  8import random
  9from typing import Any, Dict, List, Optional
 10
 11try:
 12    from base_agent import BaseAgent
 13except ImportError:
 14    from agents.base_agent import BaseAgent
 15
 16
 17class RandomWalkerAgent(BaseAgent):
 18    """
 19    A baseline agent that moves completely randomly among valid actions.
 20    Useful for testing server physics, map loading, and UI rendering.
 21    """
 22
 23    def __init__(self, server_uri: str = "ws://localhost:8765") -> None:
 24        """Initialize the random walker agent.
 25
 26        Args:
 27            server_uri (str): URI of the simulation server.
 28        """
 29        super().__init__(server_uri)
 30        self.visited_tiles: set[str] = set()
 31
 32    def reset_memory(self) -> None:
 33        """Clears memory when the user resets the map from the UI."""
 34        self.visited_tiles.clear()
 35
 36    async def deliberate_maze(self) -> Optional[str]:
 37        """Logic for maps where 'target' is defined.
 38
 39        Returns:
 40            Optional[str]: The chosen direction or None.
 41        """
 42        if not self.current_state:
 43            return None
 44
 45        # Record current position
 46        pos_str = f"{self.current_state['position'][0]},{self.current_state['position'][1]}"
 47        self.visited_tiles.add(pos_str)
 48
 49        # Get valid actions from the server state
 50        valid_actions: List[str] = self.current_state.get("valid_actions", [])
 51
 52        if not valid_actions:
 53            return None
 54
 55        # Choose randomly
 56        return random.choice(valid_actions)
 57
 58    async def deliberate_room(self) -> Optional[str]:
 59        """Logic for room clearing (no target).
 60
 61        Uses the same random walking strategy as maze mode.
 62
 63        Returns:
 64            Optional[str]: The chosen direction or None.
 65        """
 66        return await self.deliberate_maze()
 67
 68    async def send_telemetry(self, websocket: Any) -> None:
 69        """Calculates arbitrary probabilities (since it's random) and sends internal map memory to the frontend UI.
 70
 71        Args:
 72            websocket (Any): The current WebSocket connection.
 73        """
 74        if not self.current_state:
 75            return
 76
 77        valid_actions: List[str] = self.current_state.get("valid_actions", [])
 78
 79        # Build equal probability distribution for valid moves
 80        prob_per_move: float = 1.0 / len(valid_actions) if valid_actions else 0.0
 81        probs: Dict[str, float] = {
 82            "N": prob_per_move if "N" in valid_actions else 0.0,
 83            "S": prob_per_move if "S" in valid_actions else 0.0,
 84            "E": prob_per_move if "E" in valid_actions else 0.0,
 85            "W": prob_per_move if "W" in valid_actions else 0.0,
 86        }
 87
 88        telemetry_data = {
 89            "action": "telemetry",
 90            "data": {"visited": list(self.visited_tiles), "current_probs": probs},
 91        }
 92
 93        await websocket.send(json.dumps(telemetry_data))
 94
 95
 96if __name__ == "__main__":
 97    import sys
 98    uri = sys.argv[1] if len(sys.argv) > 1 else "ws://localhost:8765"
 99    agent = RandomWalkerAgent(uri)
100    asyncio.run(agent.run())