Refactor into subclasses

.bumpversion.cfg

@@ -0,0 +1,8 @@
+[bumpversion]
+current_version = 0.0.1
+commit = False
+tag = False
+
+[bumpversion:file:pyproject.toml]
+search = version = "{current_version}"
+replace = version = "{new_version}"

README.md

@@ -33,9 +33,9 @@ The full D-Wave setup instructions are [here](https://docs.ocean.dwavesys.com/en
 
 ## Tangled Game Graph Specification
 
-A Tangled game graph is specified by a graph number, which label specific graphs included here. I've included ten graphs 
-numbered 1 through 10. Each graph requires specification of vertex count (how many vertices the graph has) and an 
-explicit edge list, which are included for these ten graphs. If you'd like to add a new graph, it's simple! Just add 
+A Tangled game graph is specified by a graph number, which label specific graphs included here. I've included eleven 
+graphs numbered 1 through 11. Each graph requires specification of vertex count (how many vertices the graph has) and 
+an explicit edge list, which are included for these 11 graphs. If you'd like to add a new graph, it's simple! Just add 
 it to the GraphProperties class, found in the /utils/game_graph_properties.py file.
 
 ## Tangled Game State Specification: Expected Input Format For Adjudicators

pyproject.toml

@@ -0,0 +1,27 @@
+[tool.poetry]
+name = "tangled-adjudicate"
+version = "0.0.1"
+description = "Tangled adjudicators"
+authors = ["Geordie Rose <[email protected]>"]
+license = "MIT"
+homepage = "https://www.snowdropquantum.com/"
+packages = [
+    { include = "tangled_adjudicate" },
+    { include = "tests" },
+]
+
+[tool.poetry.dependencies]
+python = "^3.8"  # You may want to adjust this based on your needs
+dwave-ocean-sdk = "*"
+dwave-neal = ">=0.6.0"
+matplotlib = "*"
+gdown = "*"
+
+[tool.poetry.group.dev.dependencies]
+# Add development dependencies here if needed
+# pytest = "^7.0.0"
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
+

tangled_adjudicate/__init__.py

@@ -0,0 +1,5 @@
+from .adjudicators.adjudicator import Adjudicator, GameState, AdjudicationResult
+from .adjudicators.lookup_table import LookupTableAdjudicator
+from .adjudicators.schrodinger import SchrodingerEquationAdjudicator
+from .adjudicators.simulated_annealing import SimulatedAnnealingAdjudicator
+from .adjudicators.quantum_annealing import QuantumAnnealingAdjudicator

tangled_adjudicate/adjudicators/adjudicator.py

@@ -0,0 +1,141 @@
+from abc import ABC, abstractmethod
+from typing import Any, TypedDict, List, Tuple, Optional, Dict, Union, Set
+import numpy as np
+import numpy.typing as npt
+
+
+class GameState(TypedDict):
+    num_nodes: int
+    edges: List[Tuple[int, int, int]]  # (node1, node2, edge_label=0,1,2,3)
+    player1_id: str
+    player2_id: str
+    turn_count: int
+    current_player_index: int
+    player1_node: Optional[int]
+    player2_node: Optional[int]
+
+
+class AdjudicationResult(TypedDict):
+    game_state: GameState
+    adjudicator: str
+    winner: Optional[str]  # 'red', 'blue', 'draw', or None
+    score: Optional[float]
+    influence_vector: Optional[npt.NDArray[np.float64]]
+    correlation_matrix: Optional[npt.NDArray[np.float64]]
+    parameters: Dict[str, Union[str, int, float, bool]]
+
+
+class IsingModel(TypedDict):
+    h: Dict[int, float]  # Local fields
+    j: Dict[Tuple[int, int], float]  # Coupling strengths
+
+
+class Adjudicator(ABC):
+    """Base interface for game state adjudication implementations."""
+
+    def __init__(self) -> None:
+        """Initialize base adjudicator."""
+        self._parameters: Dict[str, Any] = {}
+        self.j_map = {0: 0.0,       # edge (i, j) uncolored , J_ij=0
+                      1: 0.0,       # edge (i, j) colored gray, J_ij=0
+                      2: -1.0,      # edge (i, j) colored green, FM coupling, J_ij=-1.0
+                      3: 1.0}       # edge (i, j) colored purple, AFM coupling, J_ij=+1.0
+
+    @abstractmethod
+    def setup(self, **kwargs) -> None:
+        """Optional setup method for implementation-specific initialization."""
+        pass
+
+    @abstractmethod
+    def adjudicate(self, game_state: GameState) -> AdjudicationResult:
+        """Adjudicate the given game state."""
+        pass
+
+    def _validate_game_state(self, game_state: GameState) -> None:
+        """Validate the game state structure and contents."""
+        required_keys = {
+            'num_nodes', 'edges', 'player1_id', 'player2_id',
+            'turn_count', 'current_player_index', 'player1_node', 'player2_node'
+        }
+
+        if not all(key in game_state for key in required_keys):
+            missing_keys = required_keys - set(game_state.keys())
+            raise ValueError(f"Game state missing required keys: {missing_keys}")
+
+        if game_state['num_nodes'] < 1:
+            raise ValueError("Number of nodes must be positive")
+
+        for edge in game_state['edges']:
+            if len(edge) != 3:
+                raise ValueError(f"Invalid edge format: {edge}")
+            if not (0 <= edge[0] < game_state['num_nodes'] and 0 <= edge[1] < game_state['num_nodes']):
+                raise ValueError(f"Edge vertices out of range: {edge}")
+
+    def _game_state_to_ising(self, game_state: GameState) -> IsingModel:
+        """Convert game state to Ising model parameters.
+
+        Args:
+            game_state: The current game state
+
+        Returns:
+            IsingModel containing h (local fields) and j (coupling strengths)
+        """
+        h = {i: 0.0 for i in range(game_state['num_nodes'])}
+        j = {}
+
+        for edge in game_state['edges']:
+            v1, v2, edge_label = edge
+            if v1 > v2:
+                v1, v2 = v2, v1
+            j[(v1, v2)] = float(self.j_map[edge_label])
+
+        return IsingModel(h=h, j=j)
+
+    def _find_isolated_vertices(self, game_state: GameState) -> Set[int]:
+        """Find vertices with no connections in the graph.
+
+        Args:
+            game_state: The current game state
+
+        Returns:
+            Set of isolated vertex indices
+        """
+        connected_vertices = set()
+        for edge in game_state['edges']:
+            connected_vertices.add(edge[0])
+            connected_vertices.add(edge[1])
+
+        all_vertices = set(range(game_state['num_nodes']))
+        return all_vertices - connected_vertices
+
+    def _compute_winner_score_and_influence(
+        self,
+        game_state: GameState,
+        correlation_matrix: npt.NDArray[np.float64],
+        epsilon: float = 0.5
+    ) -> Tuple[Optional[str], Optional[float], npt.NDArray[np.float64]]:
+        """Compute winner, score and influence from correlation matrix."""
+        if not isinstance(correlation_matrix, np.ndarray):
+            raise ValueError("Correlation matrix must be a numpy array")
+
+        if correlation_matrix.shape[0] != correlation_matrix.shape[1]:
+            raise ValueError("Correlation matrix must be square")
+
+        if correlation_matrix.shape[0] != game_state['num_nodes']:
+            raise ValueError("Correlation matrix size must match number of nodes")
+
+        influence_vector = np.sum(correlation_matrix, axis=0)
+
+        if game_state['player1_node'] is None or game_state['player2_node'] is None:
+            return None, None, influence_vector
+
+        score = influence_vector[game_state['player1_node']] - influence_vector[game_state['player2_node']]
+
+        if score > epsilon:
+            winner = 'red'
+        elif score < -epsilon:
+            winner = 'blue'
+        else:
+            winner = 'draw'
+
+        return winner, score, influence_vector

tangled_adjudicate/adjudicators/lookup_table.py

@@ -0,0 +1,117 @@
+import os
+import pickle
+from typing import Dict, Optional
+import numpy as np
+
+from ..utils.utilities import (
+    convert_erik_game_state_to_my_game_state,
+    get_tso,
+    build_results_dict
+)
+from .adjudicator import Adjudicator, GameState, AdjudicationResult
+
+
+class LookupTableAdjudicator(Adjudicator):
+    """Adjudicator implementation using pre-computed lookup tables."""
+
+    def __init__(self) -> None:
+        """Initialize the lookup table adjudicator."""
+        super().__init__()
+        self.data_dir: Optional[str] = None
+        self.results_dict: Optional[Dict[str, str]] = None
+
+    def setup(self, **kwargs) -> None:
+        """Configure lookup table parameters.
+
+        Args:
+            data_dir: Directory containing lookup table data files
+
+        Raises:
+            ValueError: If parameters are invalid or data directory doesn't exist
+        """
+        if 'data_dir' in kwargs:
+            if not isinstance(kwargs['data_dir'], str):
+                raise ValueError("data_dir must be a string")
+            if not os.path.isdir(kwargs['data_dir']):
+                raise ValueError(f"Directory not found: {kwargs['data_dir']}")
+            self.data_dir = kwargs['data_dir']
+
+        self._parameters = {'data_dir': self.data_dir}
+
+    def _load_lookup_table(self, num_nodes: int) -> None:
+        """Load the appropriate lookup table for the given graph size.
+
+        Args:
+            num_nodes: Number of nodes in the graph
+
+        Raises:
+            ValueError: If lookup table is not available for this graph size
+            RuntimeError: If lookup table file cannot be loaded
+        """
+        if num_nodes not in [3, 4]:
+            raise ValueError(
+                "Lookup table only available for complete graphs with 3 or 4 vertices"
+            )
+
+        if not self.data_dir:
+            raise RuntimeError("Data directory not set. Call setup() first.")
+
+        graph_number = num_nodes - 1  # Convert from num_nodes to graph_number
+        file_path = os.path.join(
+            self.data_dir,
+            f'graph_{graph_number}_terminal_state_outcomes.pkl'
+        )
+
+        # Generate lookup table if it doesn't exist
+        if not os.path.exists(file_path):
+            get_tso(graph_number, file_path)
+
+        try:
+            with open(file_path, 'rb') as fp:
+                results = pickle.load(fp)
+            self.results_dict = build_results_dict(results)
+        except Exception as e:
+            raise RuntimeError(f"Failed to load lookup table: {str(e)}")
+
+    def adjudicate(self, game_state: GameState) -> AdjudicationResult:
+        """Adjudicate the game state using the lookup table.
+
+        Args:
+            game_state: The current game state
+
+        Returns:
+            AdjudicationResult containing the adjudication details
+
+        Raises:
+            ValueError: If the game state is invalid or unsupported
+            RuntimeError: If lookup table is not loaded
+        """
+        self._validate_game_state(game_state)
+
+        # Load lookup table if needed
+        if (self.results_dict is None or len(next(iter(self.results_dict.keys()))) != game_state['num_nodes']):
+            self._load_lookup_table(game_state['num_nodes'])
+
+        if not self.results_dict:
+            raise RuntimeError("Failed to load lookup table")
+
+        # Convert game state to lookup format
+        lookup_state = convert_erik_game_state_to_my_game_state(game_state)
+
+        try:
+            winner = self.results_dict[str(lookup_state)]
+        except KeyError:
+            raise ValueError(
+                f"Game state not found in lookup table: {lookup_state}"
+            )
+
+        return AdjudicationResult(
+            game_state=game_state,
+            adjudicator='lookup_table',
+            winner=winner,
+            score=None,  # Lookup table doesn't provide scores
+            influence_vector=None,
+            correlation_matrix=None,
+            parameters=self._parameters
+        )
+