agents.base_agent 

 1import json
 2import logging
 3import os
 4import sys
 5from typing import Any, Dict, List, Optional
 6
 7from websockets.asyncio.client import connect
 8
 9# Add the parent directory to sys.path to allow running agents as scripts from the root
10sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
11
12logging.basicConfig(level=logging.INFO, format="%(asctime)s - AGENT - %(message)s")
13
14
15class BaseC4Agent:
16    """
17    Abstract base class for Connect Four agents.
18    Subclasses MUST implement the deliberate(valid_actions) method.
19    """
20
21    def __init__(self, server_uri: Optional[str] = None) -> None:
22        """
23        Initializes the agent with the server URI.
24
25        Args:
26            server_uri: The WebSocket URI of the Connect Four server.
27        """
28        self.server_uri: str = server_uri or os.environ.get("SERVER_URI", "ws://localhost:8765")
29        self.player_id: Optional[int] = None
30
31    async def run(self) -> None:
32        """
33        Connects to the server and enters the main communication loop.
34        """
35        try:
36            async with connect(self.server_uri) as websocket:
37                await websocket.send(json.dumps({"client": "agent"}))
38
39                async for message in websocket:
40                    if isinstance(message, bytes):
41                        message = message.decode("utf-8")
42                    data: Dict[str, Any] = json.loads(message)
43
44                    if data.get("type") == "setup":
45                        self.player_id = data.get("player_id")
46                        logging.info(f"Connected! Assigned Player {self.player_id}")
47
48                    elif data.get("type") == "state":
49                        current_turn = data.get("current_turn")
50                        valid_actions = data.get("valid_actions")
51
52                        if current_turn == self.player_id and isinstance(valid_actions, list):
53                            # It's our turn! Ask the subclass to make a decision
54                            action: Optional[int] = await self.deliberate(valid_actions)
55
56                            if action is not None:
57                                await websocket.send(json.dumps({"action": "move", "column": action}))
58
59                    elif data.get("type") == "game_over":
60                        logging.info(f"Round Over: {data.get('message')}")
61                        logging.info("Waiting for next round to start...")
62
63        except Exception as e:
64            logging.error(f"Connection lost: {e}")
65
66    async def deliberate(self, valid_actions: List[int]) -> Optional[int]:
67        """
68        MUST be implemented by subclasses.
69        Returns an integer representing the chosen column [0-6].
70
71        Args:
72            valid_actions: A list of valid column indices where a piece can be dropped.
73
74        Returns:
75            The chosen column index, or None if no move is made.
76        """
77        raise NotImplementedError("Subclasses must implement deliberate()")
class BaseC4Agent: 
16class BaseC4Agent:
17    """
18    Abstract base class for Connect Four agents.
19    Subclasses MUST implement the deliberate(valid_actions) method.
20    """
21
22    def __init__(self, server_uri: Optional[str] = None) -> None:
23        """
24        Initializes the agent with the server URI.
25
26        Args:
27            server_uri: The WebSocket URI of the Connect Four server.
28        """
29        self.server_uri: str = server_uri or os.environ.get("SERVER_URI", "ws://localhost:8765")
30        self.player_id: Optional[int] = None
31
32    async def run(self) -> None:
33        """
34        Connects to the server and enters the main communication loop.
35        """
36        try:
37            async with connect(self.server_uri) as websocket:
38                await websocket.send(json.dumps({"client": "agent"}))
39
40                async for message in websocket:
41                    if isinstance(message, bytes):
42                        message = message.decode("utf-8")
43                    data: Dict[str, Any] = json.loads(message)
44
45                    if data.get("type") == "setup":
46                        self.player_id = data.get("player_id")
47                        logging.info(f"Connected! Assigned Player {self.player_id}")
48
49                    elif data.get("type") == "state":
50                        current_turn = data.get("current_turn")
51                        valid_actions = data.get("valid_actions")
52
53                        if current_turn == self.player_id and isinstance(valid_actions, list):
54                            # It's our turn! Ask the subclass to make a decision
55                            action: Optional[int] = await self.deliberate(valid_actions)
56
57                            if action is not None:
58                                await websocket.send(json.dumps({"action": "move", "column": action}))
59
60                    elif data.get("type") == "game_over":
61                        logging.info(f"Round Over: {data.get('message')}")
62                        logging.info("Waiting for next round to start...")
63
64        except Exception as e:
65            logging.error(f"Connection lost: {e}")
66
67    async def deliberate(self, valid_actions: List[int]) -> Optional[int]:
68        """
69        MUST be implemented by subclasses.
70        Returns an integer representing the chosen column [0-6].
71
72        Args:
73            valid_actions: A list of valid column indices where a piece can be dropped.
74
75        Returns:
76            The chosen column index, or None if no move is made.
77        """
78        raise NotImplementedError("Subclasses must implement deliberate()")

Abstract base class for Connect Four agents. Subclasses MUST implement the deliberate(valid_actions) method.

BaseC4Agent(server_uri: Optional[str] = None) 
22    def __init__(self, server_uri: Optional[str] = None) -> None:
23        """
24        Initializes the agent with the server URI.
25
26        Args:
27            server_uri: The WebSocket URI of the Connect Four server.
28        """
29        self.server_uri: str = server_uri or os.environ.get("SERVER_URI", "ws://localhost:8765")
30        self.player_id: Optional[int] = None

Initializes the agent with the server URI.

Args: server_uri: The WebSocket URI of the Connect Four server.

server_uri: str
player_id: Optional[int]
async def run(self) -> None: 
32    async def run(self) -> None:
33        """
34        Connects to the server and enters the main communication loop.
35        """
36        try:
37            async with connect(self.server_uri) as websocket:
38                await websocket.send(json.dumps({"client": "agent"}))
39
40                async for message in websocket:
41                    if isinstance(message, bytes):
42                        message = message.decode("utf-8")
43                    data: Dict[str, Any] = json.loads(message)
44
45                    if data.get("type") == "setup":
46                        self.player_id = data.get("player_id")
47                        logging.info(f"Connected! Assigned Player {self.player_id}")
48
49                    elif data.get("type") == "state":
50                        current_turn = data.get("current_turn")
51                        valid_actions = data.get("valid_actions")
52
53                        if current_turn == self.player_id and isinstance(valid_actions, list):
54                            # It's our turn! Ask the subclass to make a decision
55                            action: Optional[int] = await self.deliberate(valid_actions)
56
57                            if action is not None:
58                                await websocket.send(json.dumps({"action": "move", "column": action}))
59
60                    elif data.get("type") == "game_over":
61                        logging.info(f"Round Over: {data.get('message')}")
62                        logging.info("Waiting for next round to start...")
63
64        except Exception as e:
65            logging.error(f"Connection lost: {e}")

Connects to the server and enters the main communication loop.

async def deliberate(self, valid_actions: List[int]) -> Optional[int]: 
67    async def deliberate(self, valid_actions: List[int]) -> Optional[int]:
68        """
69        MUST be implemented by subclasses.
70        Returns an integer representing the chosen column [0-6].
71
72        Args:
73            valid_actions: A list of valid column indices where a piece can be dropped.
74
75        Returns:
76            The chosen column index, or None if no move is made.
77        """
78        raise NotImplementedError("Subclasses must implement deliberate()")

MUST be implemented by subclasses. Returns an integer representing the chosen column [0-6].

Args: valid_actions: A list of valid column indices where a piece can be dropped.

Returns: The chosen column index, or None if no move is made.