Battleship board generator and samplers.

modified:   ../../../../.gitignore
	new file:   battleship_boards.py
	modified:   battleship_genetic.py
	deleted:    battleship_genetic2.py
	modified:   battleship_learning.json
	modified:   compare_battleship.sh
	new file:   elites.json
	new file:   sample_from_tarball.sh
	new file:   sample_one_from_tarball.sh
This commit is contained in:
Russell Ballestrini 2024-09-14 15:42:10 -04:00
parent ef127236e1
commit e365749c37
9 changed files with 3220 additions and 2083 deletions

2
.gitignore vendored
View file

@ -5,6 +5,8 @@ __pycache__/
# C extensions # C extensions
*.so *.so
*.tar.gz
.DS_Store .DS_Store
*.swp *.swp

View file

@ -0,0 +1,63 @@
import json
import multiprocessing
import random
from tqdm import tqdm
# Define ship configurations
ships = {"Carrier": 5, "Battleship": 4, "Cruiser": 3, "Submarine": 3, "Destroyer": 2}
def place_ships(grid_size=100):
"""Randomly place ships on a grid."""
grid = [""] * grid_size
for ship, size in ships.items():
placed = False
while not placed:
orientation = random.choice(["horizontal", "vertical"])
if orientation == "horizontal":
start = random.randint(0, grid_size - size)
if all(grid[start + i] == "" for i in range(size)):
for i in range(size):
grid[start + i] = ship
placed = True
else:
start = random.randint(0, grid_size - size * 10)
if all(grid[start + i * 10] == "" for i in range(size)):
for i in range(size):
grid[start + i * 10] = ship
placed = True
return grid
def worker(num_boards, output_queue):
"""Worker function to generate boards."""
for _ in range(num_boards):
board = place_ships()
output_queue.put(board)
def main(total_boards=1000000, num_processes=4):
"""Main function to manage multiprocessing and progress tracking."""
boards_per_process = total_boards // num_processes
output_queue = multiprocessing.Queue()
processes = []
for _ in range(num_processes):
p = multiprocessing.Process(
target=worker, args=(boards_per_process, output_queue)
)
processes.append(p)
p.start()
with open("battleship_boards.json", "w") as f:
for _ in tqdm(range(total_boards), desc="Generating Boards"):
board = output_queue.get()
json.dump(board, f)
f.write("\n")
for p in processes:
p.join()
if __name__ == "__main__":
main()

View file

@ -3,116 +3,229 @@ import json
import os import os
import sys import sys
import time import time
from tqdm import tqdm
from multiprocessing import Pool
GRID_SIZE = 10 GRID_SIZE = 10
SHIPS = {"Carrier": 5, "Battleship": 4, "Cruiser": 3, "Submarine": 3, "Destroyer": 2} SHIPS = {"Carrier": 5, "Battleship": 4, "Cruiser": 3, "Submarine": 3, "Destroyer": 2}
POPULATION_SIZE = 200 POPULATION_SIZE = 200
GENERATIONS = 100 GENERATIONS = 50
MUTATION_RATE = 0.1 BASE_MUTATION_RATE = 0.1
GAMES_PER_INDIVIDUAL = 10 GAMES_PER_INDIVIDUAL = 20
LEARNING_FILE = "battleship_learning.json" LEARNING_FILE = "battleship_learning.json"
MUTATION_MULTIPLIERS = {
"adjacent_hit_weight": 0.5,
"ship_size_weight": 1.5,
"parity_weight": 1.5,
"hunt_mode_threshold": 1.0,
"prob_grid_weight": 1.0,
"avoid_edges_weight": 1.0,
}
class Board:
def __init__(self):
self.grid = [[None for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)]
self.ships_left = set(SHIPS.keys())
self.place_all_ships()
def place_all_ships(self):
for ship, size in SHIPS.items():
while True:
row = random.randint(0, GRID_SIZE - 1)
col = random.randint(0, GRID_SIZE - 1)
horizontal = random.choice([True, False])
if self.is_valid_placement(row, col, size, horizontal):
self.place_ship(row, col, size, horizontal, ship)
break
def is_valid_placement(self, row, col, size, horizontal):
if horizontal:
if col + size > GRID_SIZE:
return False
return all(self.grid[row][c] is None for c in range(col, col + size))
else:
if row + size > GRID_SIZE:
return False
return all(self.grid[r][col] is None for r in range(row, row + size))
def place_ship(self, row, col, size, horizontal, ship):
if horizontal:
for c in range(col, col + size):
self.grid[row][c] = ship
else:
for r in range(row, row + size):
self.grid[r][col] = ship
def is_sunk(self, ship, hits):
return all(
(r, c) in hits
for r in range(GRID_SIZE)
for c in range(GRID_SIZE)
if self.grid[r][c] == ship
)
def update_sunk_ships(self, hits):
for ship in list(self.ships_left):
if self.is_sunk(ship, hits):
self.ships_left.remove(ship)
def get_remaining_ships(self):
return [size for ship, size in SHIPS.items() if ship in self.ships_left]
class BattleshipStrategy: class BattleshipStrategy:
ADJACENT_HIT_WEIGHT_RANGE = (1, 20)
SHIP_SIZE_WEIGHT_RANGE = (0, 10)
PARITY_WEIGHT_RANGE = (0, 10)
HUNT_MODE_THRESHOLD_RANGE = (0.1, 1)
PROB_GRID_WEIGHT_RANGE = (0, 1)
AVOID_EDGES_WEIGHT_RANGE = (0, 1)
def __init__(self): def __init__(self):
self.adjacent_hit_weight = random.uniform(1, 5) self.adjacent_hit_weight = random.uniform(*self.ADJACENT_HIT_WEIGHT_RANGE)
self.checkerboard_weight = random.uniform(0, 2) self.ship_size_weight = random.uniform(*self.SHIP_SIZE_WEIGHT_RANGE)
self.avoid_edges_weight = random.uniform(0, 1) self.parity_weight = random.uniform(*self.PARITY_WEIGHT_RANGE)
self.hunt_mode_threshold = random.uniform(0.1, 0.5) self.hunt_mode_threshold = random.uniform(*self.HUNT_MODE_THRESHOLD_RANGE)
self.prob_grid_weight = random.uniform(*self.PROB_GRID_WEIGHT_RANGE)
self.avoid_edges_weight = random.uniform(*self.AVOID_EDGES_WEIGHT_RANGE)
self._probability_grid = None
def serialize(self, include_grid=False):
data = {
"adjacent_hit_weight": self.adjacent_hit_weight,
"ship_size_weight": self.ship_size_weight,
"parity_weight": self.parity_weight,
"hunt_mode_threshold": self.hunt_mode_threshold,
"prob_grid_weight": self.prob_grid_weight,
"avoid_edges_weight": self.avoid_edges_weight,
}
if include_grid:
data["probability_grid"] = self.probability_grid
return data
@property
def probability_grid(self):
if self._probability_grid is None:
self.reset_probability_grid()
return self._probability_grid
def reset_probability_grid(self):
self._probability_grid = [
[1 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)
]
def update_probability_grid(self, hits, misses):
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if (i, j) in hits:
self.probability_grid[i][j] = 0
elif (i, j) in misses:
self.probability_grid[i][j] = -1
else:
self.probability_grid[i][j] = 1
for i, j in hits:
for di, dj in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
ni, nj = i + di, j + dj
if (
0 <= ni < GRID_SIZE
and 0 <= nj < GRID_SIZE
and (ni, nj) not in hits
and (ni, nj) not in misses
):
self.probability_grid[ni][nj] *= self.adjacent_hit_weight
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if i == 0 or i == GRID_SIZE - 1 or j == 0 or j == GRID_SIZE - 1:
self.probability_grid[i][j] *= 1 - self.avoid_edges_weight
max_prob = max(max(row) for row in self.probability_grid if max(row) > 0)
if max_prob > 0:
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if self.probability_grid[i][j] > 0:
self.probability_grid[i][j] = int(
(self.probability_grid[i][j] / max_prob) * 100
)
def get_next_move(self, board, hits, misses): def get_next_move(self, board, hits, misses):
scores = [[0 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)] self.update_probability_grid(hits, misses)
probabilities = [
[self.probability_grid[i][j] for j in range(GRID_SIZE)]
for i in range(GRID_SIZE)
]
remaining_ships = [size for ship, size in SHIPS.items() if size > len(hits)]
for i in range(GRID_SIZE): for i in range(GRID_SIZE):
for j in range(GRID_SIZE): for j in range(GRID_SIZE):
if (i, j) in hits or (i, j) in misses: if (i, j) in hits or (i, j) in misses:
probabilities[i][j] = 0
continue continue
# Adjacent to hit for size in remaining_ships:
for di, dj in [(-1, 0), (1, 0), (0, -1), (0, 1)]: if self.can_fit_ship(board, i, j, size):
if ( probabilities[i][j] *= self.ship_size_weight
0 <= i + di < GRID_SIZE
and 0 <= j + dj < GRID_SIZE
and (i + di, j + dj) in hits
):
scores[i][j] += self.adjacent_hit_weight
# Checkerboard pattern
if (i + j) % 2 == 0: if (i + j) % 2 == 0:
scores[i][j] += self.checkerboard_weight probabilities[i][j] *= self.parity_weight
# Avoid edges max_prob = max(max(row) for row in probabilities)
if i in [0, GRID_SIZE - 1] or j in [0, GRID_SIZE - 1]: for i in range(GRID_SIZE):
scores[i][j] -= self.avoid_edges_weight for j in range(GRID_SIZE):
probabilities[i][j] *= self.prob_grid_weight
if max_prob > 0:
probabilities[i][j] += (1 - self.prob_grid_weight) * (
probabilities[i][j] / max_prob
)
max_score = max(max(row) for row in scores) max_prob = max(max(row) for row in probabilities)
candidates = [ candidates = [
(i, j) (i, j)
for i in range(GRID_SIZE) for i in range(GRID_SIZE)
for j in range(GRID_SIZE) for j in range(GRID_SIZE)
if scores[i][j] == max_score and (i, j) not in hits and (i, j) not in misses if probabilities[i][j] == max_prob
] ]
return ( return random.choice(candidates)
random.choice(candidates)
if candidates def can_fit_ship(self, board, row, col, size):
else random.choice( if col + size <= GRID_SIZE and all(
[ board[row][c] == 0 for c in range(col, col + size)
(i, j) ):
for i in range(GRID_SIZE) return True
for j in range(GRID_SIZE) if row + size <= GRID_SIZE and all(
if (i, j) not in hits and (i, j) not in misses board[r][col] == 0 for r in range(row, row + size)
] ):
) return True
) return False
def create_random_board(): def play_game(strategy, use_ensemble=False):
board = [[0 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)] board = Board()
for ship, size in SHIPS.items():
while True:
row = random.randint(0, GRID_SIZE - 1)
col = random.randint(0, GRID_SIZE - 1)
horizontal = random.choice([True, False])
if is_valid_placement(board, row, col, size, horizontal):
place_ship(board, row, col, size, horizontal)
break
return board
def is_valid_placement(board, row, col, size, horizontal):
if horizontal:
if col + size > GRID_SIZE:
return False
return all(board[row][c] == 0 for c in range(col, col + size))
else:
if row + size > GRID_SIZE:
return False
return all(board[r][col] == 0 for r in range(row, row + size))
def place_ship(board, row, col, size, horizontal):
if horizontal:
for c in range(col, col + size):
board[row][c] = 1
else:
for r in range(row, row + size):
board[r][col] = 1
def play_game(strategy):
board = create_random_board()
hits = set() hits = set()
misses = set() misses = set()
turns = 0 turns = 0
ships_left = sum(SHIPS.values())
while ships_left > 0 and turns < 100: elite_strategies = load_elite_strategies() if use_ensemble else None
i, j = strategy.get_next_move(board, hits, misses)
if strategy is not None:
strategy.reset_probability_grid()
while board.ships_left and turns < 100:
if use_ensemble:
i, j = get_ensemble_move(elite_strategies, board, hits, misses)
else:
i, j = strategy.get_next_move(board.grid, hits, misses)
strategy.update_probability_grid(hits, misses)
turns += 1 turns += 1
if board[i][j] == 1: if board.grid[i][j] is not None:
hits.add((i, j)) hits.add((i, j))
ships_left -= 1 ship = board.grid[i][j]
board.update_sunk_ships(hits)
else: else:
misses.add((i, j)) misses.add((i, j))
@ -121,38 +234,69 @@ def play_game(strategy):
def crossover(parent1, parent2): def crossover(parent1, parent2):
child = BattleshipStrategy() child = BattleshipStrategy()
child.avoid_edges_weight = random.choice( child.adjacent_hit_weight = random.choice(
[parent1.avoid_edges_weight, parent2.avoid_edges_weight] [parent1.adjacent_hit_weight, parent2.adjacent_hit_weight]
) )
child.ship_size_weight = random.choice(
[parent1.ship_size_weight, parent2.ship_size_weight]
)
child.parity_weight = random.choice([parent1.parity_weight, parent2.parity_weight])
child.hunt_mode_threshold = random.choice( child.hunt_mode_threshold = random.choice(
[parent1.hunt_mode_threshold, parent2.hunt_mode_threshold] [parent1.hunt_mode_threshold, parent2.hunt_mode_threshold]
) )
child.prob_grid_weight = random.choice(
if parent1.adjacent_hit_weight < parent2.adjacent_hit_weight: [parent1.prob_grid_weight, parent2.prob_grid_weight]
child.adjacent_hit_weight = parent1.adjacent_hit_weight )
else: child.avoid_edges_weight = random.choice(
child.adjacent_hit_weight = parent2.adjacent_hit_weight [parent1.avoid_edges_weight, parent2.avoid_edges_weight]
)
if parent1.checkerboard_weight < parent2.checkerboard_weight:
child.checkerboard_weight = parent1.checkerboard_weight
else:
child.checkerboard_weight = parent2.checkerboard_weight
return child return child
def mutate(individual): def mutate(individual):
if random.random() < MUTATION_RATE * 2: if (
individual.avoid_edges_weight = random.uniform(0, 1) random.random()
if random.random() < MUTATION_RATE * 1.5: < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["adjacent_hit_weight"]
individual.hunt_mode_threshold = random.uniform(0.1, 0.5) ):
if random.random() < MUTATION_RATE * 0.8: individual.adjacent_hit_weight = random.uniform(
individual.adjacent_hit_weight = random.uniform(1, 5) *BattleshipStrategy.ADJACENT_HIT_WEIGHT_RANGE
if random.random() < MUTATION_RATE * 0.8: )
individual.checkerboard_weight = random.uniform(0, 2) if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["ship_size_weight"]:
individual.ship_size_weight = random.uniform(
*BattleshipStrategy.SHIP_SIZE_WEIGHT_RANGE
)
if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["parity_weight"]:
individual.parity_weight = random.uniform(
*BattleshipStrategy.PARITY_WEIGHT_RANGE
)
if (
random.random()
< BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["hunt_mode_threshold"]
):
individual.hunt_mode_threshold = random.uniform(
*BattleshipStrategy.HUNT_MODE_THRESHOLD_RANGE
)
if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["prob_grid_weight"]:
individual.prob_grid_weight = random.uniform(
*BattleshipStrategy.PROB_GRID_WEIGHT_RANGE
)
if (
random.random()
< BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["avoid_edges_weight"]
):
individual.avoid_edges_weight = random.uniform(
*BattleshipStrategy.AVOID_EDGES_WEIGHT_RANGE
)
return individual return individual
def evaluate_strategy(strategy):
total_turns = 0
for _ in range(GAMES_PER_INDIVIDUAL):
total_turns += play_game(strategy)
return total_turns
def genetic_algorithm(continue_training=False): def genetic_algorithm(continue_training=False):
if continue_training and os.path.exists(LEARNING_FILE): if continue_training and os.path.exists(LEARNING_FILE):
with open(LEARNING_FILE, "r") as f: with open(LEARNING_FILE, "r") as f:
@ -160,23 +304,26 @@ def genetic_algorithm(continue_training=False):
population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)] population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)]
for i, strategy_data in enumerate(data): for i, strategy_data in enumerate(data):
population[i].__dict__.update(strategy_data) population[i].__dict__.update(strategy_data)
print("Continuing training from existing population.") tqdm.write("Continuing training from existing population.")
else: else:
if os.path.exists(LEARNING_FILE): if os.path.exists(LEARNING_FILE):
timestamp = time.strftime("%Y%m%d-%H%M%S") timestamp = time.strftime("%Y%m%d-%H%M%S")
os.rename(LEARNING_FILE, f"{LEARNING_FILE}.{timestamp}") os.rename(LEARNING_FILE, f"{LEARNING_FILE}.{timestamp}")
print(f"Moved existing model to {LEARNING_FILE}.{timestamp}") tqdm.write(f"Moved existing model to {LEARNING_FILE}.{timestamp}")
population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)] population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)]
print("Starting with a new population.") tqdm.write("Starting with a new population.")
best_fitness = float("inf") best_fitness = float("inf")
generations_without_improvement = 0 generations_without_improvement = 0
progress_bar = tqdm(
total=GENERATIONS * POPULATION_SIZE, file=sys.stderr, desc="Overall Progress"
)
for generation in range(GENERATIONS): for generation in range(GENERATIONS):
fitness_scores = [ with Pool() as pool:
sum(play_game(strategy) for _ in range(GAMES_PER_INDIVIDUAL)) fitness_scores = list(pool.imap(evaluate_strategy, population))
for strategy in population progress_bar.update(POPULATION_SIZE)
]
population = [ population = [
x x
@ -194,25 +341,51 @@ def genetic_algorithm(continue_training=False):
else: else:
generations_without_improvement += 1 generations_without_improvement += 1
print( tqdm.write(
f"Generation {generation + 1}: Best fitness = {current_best_fitness}, Avg fitness = {avg_fitness:.2f}, Best overall = {best_fitness}" f"Generation {generation + 1}: Best fitness = {current_best_fitness}, Avg fitness = {avg_fitness:.2f}, Best overall = {best_fitness}"
) )
if generations_without_improvement >= 20: elitism_count = POPULATION_SIZE // 20
print("No improvement for 20 generations. Stopping early.") new_population = population[:elitism_count]
break
new_population = population[:2]
while len(new_population) < POPULATION_SIZE: while len(new_population) < POPULATION_SIZE:
parent1, parent2 = random.sample(population[:50], 2) tournament_size = 5
parent1 = min(
random.sample(population, tournament_size),
key=lambda x: fitness_scores[population.index(x)],
)
parent2 = min(
random.sample(population, tournament_size),
key=lambda x: fitness_scores[population.index(x)],
)
child = crossover(parent1, parent2) child = crossover(parent1, parent2)
new_population.append(mutate(child)) new_population.append(mutate(child))
population = new_population population = new_population
with open(LEARNING_FILE, "w") as f: elite_strategies = population[:elitism_count]
json.dump([strategy.__dict__ for strategy in population], f) elite_data = [
strategy.serialize(include_grid=True) for strategy in elite_strategies
]
elite_file = "elites.json"
with open(elite_file, "w") as f:
json.dump(elite_data, f, indent=4)
try:
with open(LEARNING_FILE, "w") as f:
json.dump(
[strategy.serialize() for strategy in population], f, indent=4
)
except Exception as e:
tqdm.write(f"Error saving generation {generation + 1}: {e}")
if generations_without_improvement >= 20:
tqdm.write("No improvement for 20 generations. Stopping early.")
break
progress_bar.close()
return population[0] return population[0]
@ -228,10 +401,67 @@ def load_best_strategy():
return BattleshipStrategy() return BattleshipStrategy()
def load_elite_strategies(num_elites=10):
if os.path.exists(LEARNING_FILE):
with open(LEARNING_FILE, "r") as f:
data = json.load(f)
elite_strategies = []
for strategy_data in data[:num_elites]:
strategy = BattleshipStrategy()
strategy.__dict__.update(strategy_data)
elite_strategies.append(strategy)
return elite_strategies
else:
return [BattleshipStrategy()]
def get_ensemble_move(elite_strategies, board, hits, misses):
probabilities = [[0 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)]
remaining_ships = board.get_remaining_ships()
for strategy in elite_strategies:
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if (i, j) in hits or (i, j) in misses:
continue
prob = 1
for di, dj in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
if (
0 <= i + di < GRID_SIZE
and 0 <= j + dj < GRID_SIZE
and (i + di, j + dj) in hits
):
prob *= strategy.adjacent_hit_weight
for size in remaining_ships:
if strategy.can_fit_ship(board.grid, i, j, size):
prob *= strategy.ship_size_weight
if (i + j) % 2 == 0:
prob *= strategy.parity_weight
if i == 0 or i == GRID_SIZE - 1 or j == 0 or j == GRID_SIZE - 1:
prob *= 1 - strategy.avoid_edges_weight
probabilities[i][j] += prob * strategy.prob_grid_weight
max_prob = max(max(row) for row in probabilities)
candidates = [
(i, j)
for i in range(GRID_SIZE)
for j in range(GRID_SIZE)
if probabilities[i][j] == max_prob
]
return random.choice(candidates)
def main(): def main():
if len(sys.argv) < 2: if len(sys.argv) < 2:
print( print(
"Usage: python battleship_genetic.py [train|train_continue|sample|sample_multi]" "Usage: python battleship_genetic2.py [train|train_continue|sample|sample_multi|sample_ensemble]"
) )
sys.exit(1) sys.exit(1)
@ -253,9 +483,12 @@ def main():
total_turns = sum(play_game(best_strategy) for _ in range(num_games)) total_turns = sum(play_game(best_strategy) for _ in range(num_games))
avg_turns = total_turns / num_games avg_turns = total_turns / num_games
print(f"Average turns over {num_games} games: {avg_turns:.2f}") print(f"Average turns over {num_games} games: {avg_turns:.2f}")
elif mode == "sample_ensemble":
turns = play_game(None, use_ensemble=True)
print(turns)
else: else:
print( print(
"Invalid mode. Use 'train', 'train_continue', 'sample', or 'sample_multi'." "Invalid mode. Use 'train', 'train_continue', 'sample', 'sample_multi', or 'sample_ensemble'."
) )
sys.exit(1) sys.exit(1)

View file

@ -1,487 +0,0 @@
import random
import json
import os
import sys
import time
from tqdm import tqdm
from multiprocessing import Pool
GRID_SIZE = 10
SHIPS = {"Carrier": 5, "Battleship": 4, "Cruiser": 3, "Submarine": 3, "Destroyer": 2}
POPULATION_SIZE = 200
GENERATIONS = 50
BASE_MUTATION_RATE = 0.1
GAMES_PER_INDIVIDUAL = 20
LEARNING_FILE = "battleship_learning.json"
MUTATION_MULTIPLIERS = {
"adjacent_hit_weight": 0.5,
"ship_size_weight": 1.5,
"parity_weight": 1.5,
"hunt_mode_threshold": 1.0,
"prob_grid_weight": 1.0,
"avoid_edges_weight": 1.0, # Added for avoid_edges_weight
}
class BattleshipStrategy:
# Define ranges as class attributes
ADJACENT_HIT_WEIGHT_RANGE = (1, 20)
SHIP_SIZE_WEIGHT_RANGE = (0, 10)
PARITY_WEIGHT_RANGE = (0, 10)
HUNT_MODE_THRESHOLD_RANGE = (0.1, 1)
PROB_GRID_WEIGHT_RANGE = (0, 1)
AVOID_EDGES_WEIGHT_RANGE = (0, 1) # New weight range
def __init__(self):
self.adjacent_hit_weight = random.uniform(*self.ADJACENT_HIT_WEIGHT_RANGE)
self.ship_size_weight = random.uniform(*self.SHIP_SIZE_WEIGHT_RANGE)
self.parity_weight = random.uniform(*self.PARITY_WEIGHT_RANGE)
self.hunt_mode_threshold = random.uniform(*self.HUNT_MODE_THRESHOLD_RANGE)
self.prob_grid_weight = random.uniform(*self.PROB_GRID_WEIGHT_RANGE)
self.avoid_edges_weight = random.uniform(
*self.AVOID_EDGES_WEIGHT_RANGE
) # Initialize new weight
self._probability_grid = None
def serialize(self, include_grid=False):
data = {
"adjacent_hit_weight": self.adjacent_hit_weight,
"ship_size_weight": self.ship_size_weight,
"parity_weight": self.parity_weight,
"hunt_mode_threshold": self.hunt_mode_threshold,
"prob_grid_weight": self.prob_grid_weight,
"avoid_edges_weight": self.avoid_edges_weight, # Include in serialization
}
if include_grid:
data["probability_grid"] = self.probability_grid
return data
@property
def probability_grid(self):
if self._probability_grid is None:
self.reset_probability_grid()
return self._probability_grid
def reset_probability_grid(self):
self._probability_grid = [
[1 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)
]
def update_probability_grid(self, hits, misses):
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if (i, j) in hits:
self.probability_grid[i][j] = 0
elif (i, j) in misses:
self.probability_grid[i][j] = -1
else:
self.probability_grid[i][j] = 1
for i, j in hits:
for di, dj in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
ni, nj = i + di, j + dj
if (
0 <= ni < GRID_SIZE
and 0 <= nj < GRID_SIZE
and (ni, nj) not in hits
and (ni, nj) not in misses
):
self.probability_grid[ni][nj] *= self.adjacent_hit_weight
# Apply avoid_edges_weight
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if i == 0 or i == GRID_SIZE - 1 or j == 0 or j == GRID_SIZE - 1:
self.probability_grid[i][j] *= 1 - self.avoid_edges_weight
max_prob = max(max(row) for row in self.probability_grid if max(row) > 0)
if max_prob > 0:
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if self.probability_grid[i][j] > 0:
self.probability_grid[i][j] = int(
(self.probability_grid[i][j] / max_prob) * 100
)
def get_next_move(self, board, hits, misses):
self.update_probability_grid(hits, misses)
probabilities = [
[self.probability_grid[i][j] for j in range(GRID_SIZE)]
for i in range(GRID_SIZE)
]
remaining_ships = [size for ship, size in SHIPS.items() if size > len(hits)]
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if (i, j) in hits or (i, j) in misses:
probabilities[i][j] = 0
continue
for size in remaining_ships:
if self.can_fit_ship(board, i, j, size):
probabilities[i][j] *= self.ship_size_weight
if (i + j) % 2 == 0:
probabilities[i][j] *= self.parity_weight
max_prob = max(max(row) for row in probabilities)
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
probabilities[i][j] *= self.prob_grid_weight
if max_prob > 0:
probabilities[i][j] += (1 - self.prob_grid_weight) * (
probabilities[i][j] / max_prob
)
max_prob = max(max(row) for row in probabilities)
candidates = [
(i, j)
for i in range(GRID_SIZE)
for j in range(GRID_SIZE)
if probabilities[i][j] == max_prob
]
return random.choice(candidates)
def can_fit_ship(self, board, row, col, size):
if col + size <= GRID_SIZE and all(
board[row][c] == 0 for c in range(col, col + size)
):
return True
if row + size <= GRID_SIZE and all(
board[r][col] == 0 for r in range(row, row + size)
):
return True
return False
def create_random_board():
board = [[0 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)]
for ship, size in SHIPS.items():
while True:
row = random.randint(0, GRID_SIZE - 1)
col = random.randint(0, GRID_SIZE - 1)
horizontal = random.choice([True, False])
if is_valid_placement(board, row, col, size, horizontal):
place_ship(board, row, col, size, horizontal)
break
return board
def is_valid_placement(board, row, col, size, horizontal):
if horizontal:
if col + size > GRID_SIZE:
return False
return all(board[row][c] == 0 for c in range(col, col + size))
else:
if row + size > GRID_SIZE:
return False
return all(board[r][col] == 0 for r in range(row, row + size))
def place_ship(board, row, col, size, horizontal):
if horizontal:
for c in range(col, col + size):
board[row][c] = 1
else:
for r in range(row, row + size):
board[r][col] = 1
def play_game(strategy, use_ensemble=False):
board = create_random_board()
hits = set()
misses = set()
turns = 0
ships_left = sum(SHIPS.values())
elite_strategies = load_elite_strategies() if use_ensemble else None
# Reset the probability grid at the start of each game
strategy.reset_probability_grid()
while ships_left > 0 and turns < 100:
if use_ensemble:
i, j = get_ensemble_move(elite_strategies, board, hits, misses)
else:
i, j = strategy.get_next_move(board, hits, misses)
# Update the probability grid after each move
strategy.update_probability_grid(hits, misses)
turns += 1
if board[i][j] == 1:
hits.add((i, j))
ships_left -= 1
else:
misses.add((i, j))
return turns
def crossover(parent1, parent2):
child = BattleshipStrategy()
child.adjacent_hit_weight = random.choice(
[parent1.adjacent_hit_weight, parent2.adjacent_hit_weight]
)
child.ship_size_weight = random.choice(
[parent1.ship_size_weight, parent2.ship_size_weight]
)
child.parity_weight = random.choice([parent1.parity_weight, parent2.parity_weight])
child.hunt_mode_threshold = random.choice(
[parent1.hunt_mode_threshold, parent2.hunt_mode_threshold]
)
child.prob_grid_weight = random.choice(
[parent1.prob_grid_weight, parent2.prob_grid_weight]
)
child.avoid_edges_weight = random.choice(
[parent1.avoid_edges_weight, parent2.avoid_edges_weight]
)
return child
def mutate(individual):
if (
random.random()
< BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["adjacent_hit_weight"]
):
individual.adjacent_hit_weight = random.uniform(
*BattleshipStrategy.ADJACENT_HIT_WEIGHT_RANGE
)
if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["ship_size_weight"]:
individual.ship_size_weight = random.uniform(
*BattleshipStrategy.SHIP_SIZE_WEIGHT_RANGE
)
if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["parity_weight"]:
individual.parity_weight = random.uniform(
*BattleshipStrategy.PARITY_WEIGHT_RANGE
)
if (
random.random()
< BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["hunt_mode_threshold"]
):
individual.hunt_mode_threshold = random.uniform(
*BattleshipStrategy.HUNT_MODE_THRESHOLD_RANGE
)
if random.random() < BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["prob_grid_weight"]:
individual.prob_grid_weight = random.uniform(
*BattleshipStrategy.PROB_GRID_WEIGHT_RANGE
)
if (
random.random()
< BASE_MUTATION_RATE * MUTATION_MULTIPLIERS["avoid_edges_weight"]
):
individual.avoid_edges_weight = random.uniform(
*BattleshipStrategy.AVOID_EDGES_WEIGHT_RANGE
)
return individual
def evaluate_strategy(strategy):
total_turns = 0
for _ in range(GAMES_PER_INDIVIDUAL):
total_turns += play_game(strategy)
return total_turns
def genetic_algorithm(continue_training=False):
if continue_training and os.path.exists(LEARNING_FILE):
with open(LEARNING_FILE, "r") as f:
data = json.load(f)
population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)]
for i, strategy_data in enumerate(data):
population[i].__dict__.update(strategy_data)
tqdm.write("Continuing training from existing population.")
else:
if os.path.exists(LEARNING_FILE):
timestamp = time.strftime("%Y%m%d-%H%M%S")
os.rename(LEARNING_FILE, f"{LEARNING_FILE}.{timestamp}")
tqdm.write(f"Moved existing model to {LEARNING_FILE}.{timestamp}")
population = [BattleshipStrategy() for _ in range(POPULATION_SIZE)]
tqdm.write("Starting with a new population.")
best_fitness = float("inf")
generations_without_improvement = 0
progress_bar = tqdm(
total=GENERATIONS * POPULATION_SIZE, file=sys.stderr, desc="Overall Progress"
)
for generation in range(GENERATIONS):
with Pool() as pool:
fitness_scores = list(pool.imap(evaluate_strategy, population))
progress_bar.update(POPULATION_SIZE)
population = [
x
for _, x in sorted(
zip(fitness_scores, population), key=lambda pair: pair[0]
)
]
current_best_fitness = min(fitness_scores)
avg_fitness = sum(fitness_scores) / len(fitness_scores)
if current_best_fitness < best_fitness:
best_fitness = current_best_fitness
generations_without_improvement = 0
else:
generations_without_improvement += 1
# Use tqdm.write to output generation results
tqdm.write(
f"Generation {generation + 1}: Best fitness = {current_best_fitness}, Avg fitness = {avg_fitness:.2f}, Best overall = {best_fitness}"
)
elitism_count = POPULATION_SIZE // 20
new_population = population[:elitism_count]
while len(new_population) < POPULATION_SIZE:
tournament_size = 5
parent1 = min(
random.sample(population, tournament_size),
key=lambda x: fitness_scores[population.index(x)],
)
parent2 = min(
random.sample(population, tournament_size),
key=lambda x: fitness_scores[population.index(x)],
)
child = crossover(parent1, parent2)
new_population.append(mutate(child))
population = new_population
# Dump elite strategies with probability grids after all games are played
elite_strategies = population[:elitism_count]
elite_data = [
strategy.serialize(include_grid=True) for strategy in elite_strategies
]
# Reuse the same file for elites
elite_file = "elites.json"
with open(elite_file, "w") as f:
json.dump(elite_data, f, indent=4)
# Serialize the final population without probability grids
try:
with open(LEARNING_FILE, "w") as f:
json.dump(
[strategy.serialize() for strategy in population], f, indent=4
)
# tqdm.write(f"Generation {generation + 1} saved successfully.")
except Exception as e:
tqdm.write(f"Error saving generation {generation + 1}: {e}")
if generations_without_improvement >= 20:
tqdm.write("No improvement for 20 generations. Stopping early.")
break
progress_bar.close()
return population[0]
def load_best_strategy():
if os.path.exists(LEARNING_FILE):
with open(LEARNING_FILE, "r") as f:
data = json.load(f)
best_strategy = BattleshipStrategy()
best_strategy.__dict__.update(data[0])
return best_strategy
else:
return BattleshipStrategy()
def load_elite_strategies(num_elites=10):
if os.path.exists(LEARNING_FILE):
with open(LEARNING_FILE, "r") as f:
data = json.load(f)
elite_strategies = []
for strategy_data in data[:num_elites]:
strategy = BattleshipStrategy()
strategy.__dict__.update(strategy_data)
elite_strategies.append(strategy)
return elite_strategies
else:
return [BattleshipStrategy()]
def get_ensemble_move(elite_strategies, board, hits, misses):
probabilities = [[0 for _ in range(GRID_SIZE)] for _ in range(GRID_SIZE)]
remaining_ships = [size for ship, size in SHIPS.items() if size > len(hits)]
for strategy in elite_strategies:
for i in range(GRID_SIZE):
for j in range(GRID_SIZE):
if (i, j) in hits or (i, j) in misses:
continue
prob = 1
for di, dj in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
if (
0 <= i + di < GRID_SIZE
and 0 <= j + dj < GRID_SIZE
and (i + di, j + dj) in hits
):
prob *= strategy.adjacent_hit_weight
for size in remaining_ships:
if strategy.can_fit_ship(board, i, j, size):
prob *= strategy.ship_size_weight
if (i + j) % 2 == 0:
prob *= strategy.parity_weight
probabilities[i][j] += prob * strategy.prob_grid_weight
max_prob = max(max(row) for row in probabilities)
candidates = [
(i, j)
for i in range(GRID_SIZE)
for j in range(GRID_SIZE)
if probabilities[i][j] == max_prob
]
return random.choice(candidates)
def main():
if len(sys.argv) < 2:
print(
"Usage: python battleship_genetic2.py [train|train_continue|sample|sample_multi|sample_ensemble]"
)
sys.exit(1)
mode = sys.argv[1]
if mode == "train":
best_strategy = genetic_algorithm(continue_training=False)
print("Training completed. Best strategy saved.")
elif mode == "train_continue":
best_strategy = genetic_algorithm(continue_training=True)
print("Continued training completed. Best strategy saved.")
elif mode == "sample":
best_strategy = load_best_strategy()
turns = play_game(best_strategy)
print(turns)
elif mode == "sample_multi":
best_strategy = load_best_strategy()
num_games = 100
total_turns = sum(play_game(best_strategy) for _ in range(num_games))
avg_turns = total_turns / num_games
print(f"Average turns over {num_games} games: {avg_turns:.2f}")
elif mode == "sample_ensemble":
turns = play_game(None, use_ensemble=True)
print(turns)
else:
print(
"Invalid mode. Use 'train', 'train_continue', 'sample', 'sample_multi', or 'sample_ensemble'."
)
sys.exit(1)
if __name__ == "__main__":
main()

View file

@ -2,6 +2,7 @@
RUNS=100 RUNS=100
PROB_FILE="prob_results.txt" PROB_FILE="prob_results.txt"
#PROB_FILE2="prob_results2.txt"
GENETIC_FILE="genetic_results.txt" GENETIC_FILE="genetic_results.txt"
GENETIC_FILE2="genetic_results_ensemble.txt" GENETIC_FILE2="genetic_results_ensemble.txt"
@ -45,16 +46,22 @@ calculate_stats() {
# Run probability grid simulation # Run probability grid simulation
run_simulation "battleship_prob.py" $PROB_FILE "Probability Grid" run_simulation "battleship_prob.py" $PROB_FILE "Probability Grid"
# Run genetic algorithm simulation # Run probability grid simulation
run_simulation "battleship_genetic2.py" $GENETIC_FILE "Genetic Algorithm" #run_simulation "battleship_prob2.py" $PROB_FILE2 "Probability Grid 2"
# Run genetic algorithm simulation # Run genetic algorithm simulation
run_simulation "battleship_genetic2.py" $GENETIC_FILE2 "Genetic Algorithm Ensemble" run_simulation "battleship_genetic.py" $GENETIC_FILE "Genetic Algorithm"
# Run genetic algorithm simulation
# temporarily commented out because the single top elite often works better...
#run_simulation "battleship_genetic.py" $GENETIC_FILE2 "Genetic Algorithm Ensemble"
# Calculate and display statistics # Calculate and display statistics
calculate_stats $PROB_FILE "Probability Grid" calculate_stats $PROB_FILE "Probability Grid"
#calculate_stats $PROB_FILE2 "Probability Grid 2"
calculate_stats $GENETIC_FILE "Genetic Algorithm" calculate_stats $GENETIC_FILE "Genetic Algorithm"
calculate_stats $GENETIC_FILE2 "Genetic Algorithm Ensemble" #calculate_stats $GENETIC_FILE2 "Genetic Algorithm Ensemble"
# Clean up # Clean up
rm $PROB_FILE $GENETIC_FILE $GENETIC_FILE2 #rm $PROB_FILE $PROB_FILE2 $GENETIC_FILE $GENETIC_FILE2
rm $PROB_FILE $GENETIC_FILE

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,10 @@
# Variables
TARBALL="battleship_boards.json.tar.gz"
SAMPLE_FILE="sampled_boards.json"
NUM_SAMPLES=1000 # Number of random samples to extract
# Stream the tarball and sample lines
tar -xzOf "$TARBALL" | shuf -n "$NUM_SAMPLES" > "$SAMPLE_FILE"
# Output the location of the sampled file
echo "Random samples saved to $SAMPLE_FILE"

View file

@ -0,0 +1,7 @@
#!/bin/bash
# Variable
TARBALL="battleship_boards.json.tar.gz"
# Stream the tarball and sample one random line
tar -xzOf "$TARBALL" | shuf -n 1