This commit is contained in:
2023-05-23 15:52:09 +08:00
parent e8f9c8a287
commit 663edbb5e2
88 changed files with 6250 additions and 0 deletions
+3
View File
@@ -0,0 +1,3 @@
train_data/*
__pycache__
View File
+7
View File
@@ -0,0 +1,7 @@
from .alphago import *
from .base import *
#from .pg import *
from .predict import *
#from .naive import *
#from .naive_fast import *
from .termination import *
+171
View File
@@ -0,0 +1,171 @@
# tag::alphago_imports[]
import numpy as np
from tugo.agent.base import Agent
from tugo.goboard_fast import Move
from tugo import kerasutil
import operator
# end::alphago_imports[]
__all__ = [
'AlphaGoNode',
'AlphaGoMCTS'
]
# tag::init_alphago_node[]
class AlphaGoNode:
def __init__(self, parent=None, probability=1.0):
self.parent = parent # <1>
self.children = {} # <1>
self.visit_count = 0
self.q_value = 0
self.prior_value = probability # <2>
self.u_value = probability # <3>
# <1> Tree nodes have one parent and potentially many children.
# <2> A node is initialized with a prior probability.
# <3> The utility function will be updated during search.
# end::init_alphago_node[]
# tag::select_node[]
def select_child(self):
return max(self.children.items(),
key=lambda child: child[1].q_value + \
child[1].u_value)
# end::select_node[]
# tag::expand_children[]
def expand_children(self, moves, probabilities):
for move, prob in zip(moves, probabilities):
if move not in self.children:
self.children[move] = AlphaGoNode(parent=self, probability=prob)
# end::expand_children[]
# tag::update_values[]
def update_values(self, leaf_value):
if self.parent is not None:
self.parent.update_values(leaf_value) # <1>
self.visit_count += 1 # <2>
self.q_value += leaf_value / self.visit_count # <3>
if self.parent is not None:
c_u = 5
self.u_value = c_u * np.sqrt(self.parent.visit_count) \
* self.prior_value / (1 + self.visit_count) # <4>
# <1> We update parents first to ensure we traverse the tree top to bottom.
# <2> Increment the visit count for this node.
# <3> Add the specified leaf value to the Q-value, normalized by visit count.
# <4> Update utility with current visit counts.
# end::update_values[]
# tag::alphago_mcts_init[]
class AlphaGoMCTS(Agent):
# def __init__(self, policy_agent, fast_policy_agent, value_agent,
# lambda_value=0.5, num_simulations=1000,
# depth=50, rollout_limit=100):
def __init__(self, policy_agent, fast_policy_agent, value_agent,
lambda_value=0.5, num_simulations=100,
depth=10, rollout_limit=10):
self.policy = policy_agent
self.rollout_policy = fast_policy_agent
self.value = value_agent
self.lambda_value = lambda_value
self.num_simulations = num_simulations
self.depth = depth
self.rollout_limit = rollout_limit
self.root = AlphaGoNode()
# end::alphago_mcts_init[]
# tag::alphago_mcts_rollout[]
def select_move(self, game_state):
for simulation in range(self.num_simulations): # <1>
current_state = game_state
node = self.root
for depth in range(self.depth): # <2>
if not node.children: # <3>
if current_state.is_over():
break
moves, probabilities = self.policy_probabilities(current_state) # <4>
node.expand_children(moves, probabilities) # <4>
move, node = node.select_child() # <5>
current_state = current_state.apply_move(move) # <5>
value = self.value.predict(current_state) # <6>
rollout = self.policy_rollout(current_state) # <6>
weighted_value = (1 - self.lambda_value) * value + \
self.lambda_value * rollout # <7>
node.update_values(weighted_value) # <8>
# <1> From current state play out a number of simulations
# <2> Play moves until the specified depth is reached.
# <3> If the current node doesn't have any children...
# <4> ... expand them with probabilities from the strong policy.
# <5> If there are children, we can select one and play the corresponding move.
# <6> Compute output of value network and a rollout by the fast policy.
# <7> Determine the combined value function.
# <8> Update values for this node in the backup phase
# end::alphago_mcts_rollout[]
# tag::alphago_mcts_selection[]
move = max(self.root.children, key=lambda move: # <1>
self.root.children.get(move).visit_count) # <1>
self.root = AlphaGoNode()
if move in self.root.children: # <2>
self.root = self.root.children[move]
self.root.parent = None
return move
# <1> Pick most visited child of the root as next move.
# <2> If the picked move is a child, set new root to this child node.
# end::alphago_mcts_selection[]
# tag::alphago_policy_probs[]
def policy_probabilities(self, game_state):
encoder = self.policy._encoder
outputs = self.policy.predict(game_state)
legal_moves = game_state.legal_moves()
if not legal_moves:
return [], []
encoded_points = [encoder.encode_point(move.point) for move in legal_moves if move.point]
legal_outputs = outputs[encoded_points]
normalized_outputs = legal_outputs / np.sum(legal_outputs)
return legal_moves, normalized_outputs
# end::alphago_policy_probs[]
# tag::alphago_policy_rollout[]
def policy_rollout(self, game_state):
for step in range(self.rollout_limit):
if game_state.is_over():
break
move_probabilities = self.rollout_policy.predict(game_state)
encoder = self.rollout_policy.encoder
for idx in np.argsort(move_probabilities)[::-1]:
max_point = encoder.decode_point_index(idx)
greedy_move = Move(max_point)
if greedy_move in game_state.legal_moves():
game_state = game_state.apply_move(greedy_move)
break
next_player = game_state.next_player
winner = game_state.winner()
if winner is not None:
return 1 if winner == next_player else -1
else:
return 0
# end::alphago_policy_rollout[]
def serialize(self, h5file):
raise IOError("AlphaGoMCTS agent can\'t be serialized" +
"consider serializing the three underlying" +
"neural networks instad.")
+16
View File
@@ -0,0 +1,16 @@
__all__ = [
'Agent',
]
# tag::agent[]
class Agent:
def __init__(self):
pass
def select_move(self, game_state):
raise NotImplementedError()
# end::agent[]
def diagnostics(self):
return {}
+44
View File
@@ -0,0 +1,44 @@
# tag::helpersimport[]
from tugo.gotypes import Point
# end::helpersimport[]
__all__ = [
'is_point_an_eye',
]
# tag::eye[]
def is_point_an_eye(board, point, color):
if board.get(point) is not None: # <1>
return False
for neighbor in point.neighbors(): # <2>
if board.is_on_grid(neighbor):
neighbor_color = board.get(neighbor)
if neighbor_color != color:
return False
friendly_corners = 0 # <3>
off_board_corners = 0
corners = [
Point(point.row - 1, point.col - 1),
Point(point.row - 1, point.col + 1),
Point(point.row + 1, point.col - 1),
Point(point.row + 1, point.col + 1),
]
for corner in corners:
if board.is_on_grid(corner):
corner_color = board.get(corner)
if corner_color == color:
friendly_corners += 1
else:
off_board_corners += 1
if off_board_corners > 0:
return off_board_corners + friendly_corners == 4 # <4>
return friendly_corners >= 3 # <5>
# <1> An eye is an empty point.
# <2> All adjacent points must contain friendly stones.
# <3> We must control 3 out of 4 corners if the point is in the middle of the board; on the edge we must control all corners.
# <4> Point is on the edge or corner.
# <5> Point is in the middle.
# end::eye[]
+37
View File
@@ -0,0 +1,37 @@
from tugo.gotypes import Point
__all__ = [
'is_point_an_eye',
]
def is_point_an_eye(board, point, color):
if board.get(point) is not None:
return False
# All adjacent points must contain friendly stones.
for neighbor in board.neighbors(point):
neighbor_color = board.get(neighbor)
if neighbor_color != color:
return False
# We must control 3 out of 4 corners if the point is in the middle
# of the board; on the edge we must control all corners.
friendly_corners = 0
off_board_corners = 0
corners = [
Point(point.row - 1, point.col - 1),
Point(point.row - 1, point.col + 1),
Point(point.row + 1, point.col - 1),
Point(point.row + 1, point.col + 1),
]
for corner in corners:
if board.is_on_grid(corner):
corner_color = board.get(corner)
if corner_color == color:
friendly_corners += 1
else:
off_board_corners += 1
if off_board_corners > 0:
# Point is on the edge or corner.
return off_board_corners + friendly_corners == 4
# Point is in the middle.
return friendly_corners >= 3
+29
View File
@@ -0,0 +1,29 @@
# tag::randombotimports[]
import random
from tugo.agent.base import Agent
from tugo.agent.helpers import is_point_an_eye
from tugo.goboard_slow import Move
from tugo.gotypes import Point
# end::randombotimports[]
__all__ = ['RandomBot']
# tag::random_bot[]
class RandomBot(Agent):
def select_move(self, game_state):
"""Choose a random valid move that preserves our own eyes."""
candidates = []
for r in range(1, game_state.board.num_rows + 1):
for c in range(1, game_state.board.num_cols + 1):
candidate = Point(row=r, col=c)
if game_state.is_valid_move(Move.play(candidate)) and \
not is_point_an_eye(game_state.board,
candidate,
game_state.next_player):
candidates.append(candidate)
if not candidates:
return Move.pass_turn()
return Move.play(random.choice(candidates))
# end::random_bot[]
+41
View File
@@ -0,0 +1,41 @@
import numpy as np
from tugo.agent.base import Agent
from tugo.agent.helpers_fast import is_point_an_eye
from tugo.goboard import Move
from tugo.gotypes import Point
__all__ = ['FastRandomBot']
class FastRandomBot(Agent):
def __init__(self):
Agent.__init__(self)
self.dim = None
self.point_cache = []
def _update_cache(self, dim):
self.dim = dim
rows, cols = dim
self.point_cache = []
for r in range(1, rows + 1):
for c in range(1, cols + 1):
self.point_cache.append(Point(row=r, col=c))
def select_move(self, game_state):
"""Choose a random valid move that preserves our own eyes."""
dim = (game_state.board.num_rows, game_state.board.num_cols)
if dim != self.dim:
self._update_cache(dim)
idx = np.arange(len(self.point_cache))
np.random.shuffle(idx)
for i in idx:
p = self.point_cache[i]
if game_state.is_valid_move(Move.play(p)) and \
not is_point_an_eye(game_state.board,
p,
game_state.next_player):
return Move.play(p)
return Move.pass_turn()
+57
View File
@@ -0,0 +1,57 @@
# tag::dl_agent_imports[]
import numpy as np
from tugo.agent.base import Agent
from tugo.agent.helpers import is_point_an_eye
from tugo import encoders
from tugo import goboard
from tugo import kerasutil
from tugo.networks import AlphaGoModel
# end::dl_agent_imports[]
__all__ = [
'DeepLearningAgent',
]
import numpy as np
import torch
class DeepLearningAgent(Agent):
def __init__(self, model, encoder):
super().__init__()
self.model = model
self.encoder = encoder
def predict(self, game_state):
encoded_state = self.encoder.encode(game_state)
input_tensor = torch.tensor([encoded_state], dtype=torch.float32).to('cuda')
with torch.no_grad():
output_tensor = self.model(input_tensor)
return output_tensor.cpu().numpy()[0]
def select_move(self, game_state):
num_moves = self.encoder.board_width * self.encoder.board_height
move_probs = self.predict(game_state)
move_probs = move_probs ** 3
eps = 1e-6
move_probs = np.clip(move_probs, eps, 1 - eps)
move_probs = move_probs / np.sum(move_probs)
candidates = np.arange(num_moves)
ranked_moves = np.random.choice(
candidates, num_moves, replace=False, p=move_probs)
for point_idx in ranked_moves:
point = self.encoder.decode_point_index(point_idx)
if game_state.is_valid_move(goboard.Move.play(point)) and \
not is_point_an_eye(game_state.board, point, game_state.next_player):
return goboard.Move.play(point)
return goboard.Move.pass_turn()
def save(self, path):
torch.save(self.model.state_dict(), path)
@classmethod
def load(cls, path, encoder):
model = AlphaGoModel(encoder.get_input_shape(), is_policy_net=True).to('cuda')
model.load_state_dict(torch.load(path))
return cls(model, encoder)
+81
View File
@@ -0,0 +1,81 @@
# tag::termination_imports[]
from tugo import goboard
from tugo.agent.base import Agent
from tugo import scoring
# end::termination_imports[]
# tag::termination_strategy[]
class TerminationStrategy:
def __init__(self):
pass
def should_pass(self, game_state):
return False
def should_resign(self, game_state):
return False
# end::termination_strategy[]
# tag::opponent_passes[]
class PassWhenOpponentPasses(TerminationStrategy):
def should_pass(self, game_state):
if game_state.last_move is not None:
return True if game_state.last_move.is_pass else False
# end::opponent_passes[]
# tag::resign_margin[]
class ResignLargeMargin(TerminationStrategy):
def __init__(self, own_color, cut_off_move, margin):
TerminationStrategy.__init__(self)
self.own_color = own_color
self.cut_off_move = cut_off_move
self.margin = margin
self.moves_played = 0
def should_pass(self, game_state):
return False
def should_resign(self, game_state):
self.moves_played += 1
if self.moves_played:
game_result = scoring.compute_game_result(self)
if game_result.winner != self.own_color and game_result.winning_margin >= self.margin:
return True
return False
# end::resign_margin[]
# tag::termination_agent[]
class TerminationAgent(Agent):
def __init__(self, agent, strategy=None):
Agent.__init__(self)
self.agent = agent
self.strategy = strategy if strategy is not None \
else TerminationStrategy()
def select_move(self, game_state):
if self.strategy.should_pass(game_state):
return goboard.Move.pass_turn()
elif self.strategy.should_resign(game_state):
return goboard.Move.resign()
else:
return self.agent.select_move(game_state)
# end::termination_agent[]
# tag::get_termination[]
def get(termination):
if termination == 'opponent_passes':
return PassWhenOpponentPasses()
else:
raise ValueError("Unsupported termination strategy: {}"
.format(termination))
# end::get_termination[]
+72
View File
@@ -0,0 +1,72 @@
import torch
from tqdm import tqdm
from tugo.agent.pg import PolicyAgent
from tugo.agent.predict import DeepLearningAgent
from tugo.encoders.alphago import AlphaGoEncoder
from tugo.rl.simulate import experience_simulation
encoder = AlphaGoEncoder()
# Load opponents
sl_agent = DeepLearningAgent.load('checkpoints/alphago_sl_policy.pt', encoder)
sl_opponent = DeepLearningAgent.load('checkpoints/alphago_sl_policy.pt', encoder)
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
alphago_rl_agent = PolicyAgent(sl_agent.model.to(device), encoder)
opponent = PolicyAgent(sl_opponent.model.to(device), encoder)
# Run simulation
num_games = 1000
with tqdm(total=num_games) as pbar:
experience = experience_simulation(num_games, alphago_rl_agent, opponent, progress_callback=lambda: pbar.update(1))
alphago_rl_agent.train(experience)
# # Serialize RL agent
# alphago_rl_agent.save('checkpoints/alphago_rl_policy.pt')
# # Serialize experience
# experience.save('checkpoints/alphago_rl_experience.pt')
def generate_incremental_filepath(file_path):
import os
directory, filename = os.path.split(file_path)
filename_base, file_extension = os.path.splitext(filename)
# 查找保存目录下的最后一个文件序号
last_index = 0
file_exists = False
for file_name in os.listdir(directory):
if file_name.startswith(filename_base) and file_name.endswith(file_extension):
try:
index = int(file_name[len(filename_base):-len(file_extension)])
last_index = max(last_index, index)
file_exists = True
except ValueError:
continue
if not file_exists:
last_index = 0
# 生成带序号的新文件名
new_index = last_index + 1
new_filename = f"{filename_base}_{new_index}{file_extension}"
# 生成新文件的完整路径
new_file_path = os.path.join(directory, new_filename)
return new_file_path
# Serialize RL agent
alphago_rl_agent.save(generate_incremental_filepath('checkpoints/alphago_rl_policy.pt'))
# Serialize experience
experience.save(generate_incremental_filepath('checkpoints/alphago_rl_experience.pt'))
+95
View File
@@ -0,0 +1,95 @@
from tugo.data.parallel_processor import GoDataProcessor
from tugo.encoders.alphago import AlphaGoEncoder
from tugo.agent.predict import DeepLearningAgent
from tugo.networks.alphago import AlphaGoModel
import torch
from torch import nn
import torch.optim as optim
from torch.utils.data import DataLoader
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
rows, cols = 19, 19
num_classes = rows * cols
# num_games = 10000
num_games = 1000
encoder = AlphaGoEncoder()
processor = GoDataProcessor(encoder=encoder.name())
train_dataset = processor.load_go_data('train', num_games, use_generator=False)
test_dataset = processor.load_go_data('test', num_games, use_generator=False)
input_shape = (encoder.num_planes, rows, cols)
# alphago_sl_policy = AlphaGoModel(input_shape, is_policy_net=True)
alphago_sl_policy = AlphaGoModel(input_shape, is_policy_net=True, num_classes=361)
optimizer = optim.SGD(alphago_sl_policy.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()
# Set up TensorBoard logging
summary_writer = SummaryWriter(log_dir='./logs')
epochs = 200
# batch_size = 128
batch_size = 1024
# epochs = 1
# batch_size = 32
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)
# 检查是否有可用的 GPU
if torch.cuda.is_available():
print("cuda is available, will use gpu!")
device = torch.device('cuda')
else:
print("cuda is unavailable, will use cpu")
device = torch.device('cpu')
alphago_sl_policy.to(device)
alphago_sl_policy.train()
for epoch in range(epochs):
print(f"Epoch: {epoch+1}/{epochs}")
for step, (inputs, targets) in enumerate(tqdm(train_loader)):
# print("Inputs shape:", inputs.shape)
# print("Targets shape:", targets.shape)
# print("step:", step)
inputs, targets = inputs.to(device), targets.to(device)
optimizer.zero_grad()
outputs = alphago_sl_policy(inputs)
loss = criterion(outputs, targets)
loss.backward()
optimizer.step()
summary_writer.add_scalar('Training Loss', loss.item(), epoch * len(train_loader) + step)
alphago_sl_agent = DeepLearningAgent(alphago_sl_policy, encoder)
alphago_sl_agent.save('checkpoints/alphago_sl_policy.pt')
# Test set evaluation
# 评估测试集上的模型性能
alphago_sl_policy.eval()
test_loss = 0.0
correct = 0
total = 0
with torch.no_grad():
for inputs, targets in test_loader:
inputs, targets = inputs.to(device), targets.to(device)
outputs = alphago_sl_policy(inputs)
loss = criterion(outputs, targets)
test_loss += loss.item()
_, predicted = outputs.max(1)
total += targets.size(0)
correct += predicted.eq(targets).sum().item()
print(f"Test Loss: {test_loss/total:.4f}")
print(f"Test Accuracy: {correct/total:.4f}")
summary_writer.close()
Binary file not shown.
Binary file not shown.
View File
+40
View File
@@ -0,0 +1,40 @@
import glob
import torch
from torch.nn.functional import one_hot
class DataGenerator:
def __init__(self, data_directory, samples):
self.data_directory = data_directory
self.samples = samples
self.files = set(file_name for file_name, index in samples)
self.num_samples = None
def get_num_samples(self, batch_size=128, num_classes=19 * 19):
if self.num_samples is not None:
return self.num_samples
else:
self.num_samples = 0
for X, y in self._generate(batch_size=batch_size, num_classes=num_classes):
self.num_samples += X.shape[0]
return self.num_samples
def _generate(self, batch_size, num_classes):
for zip_file_name in self.files:
file_name = zip_file_name.replace('.tar.gz', '') + 'train'
base = self.data_directory + '/' + file_name + '_features_*.npy'
for feature_file in glob.glob(base):
label_file = feature_file.replace('features', 'labels')
x = torch.from_numpy(np.load(feature_file)).float()
y = torch.from_numpy(np.load(label_file)).long()
y = one_hot(y, num_classes=num_classes)
while x.shape[0] >= batch_size:
x_batch, x = x[:batch_size], x[batch_size:]
y_batch, y = y[:batch_size], y[batch_size:]
yield x_batch, y_batch
def generate(self, batch_size=128, num_classes=19 * 19):
while True:
for item in self._generate(batch_size, num_classes):
yield item
+107
View File
@@ -0,0 +1,107 @@
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file, You can
# obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import print_function
from __future__ import absolute_import
import os
import sys
import multiprocessing
import six
if sys.version_info[0] == 3:
from urllib.request import urlopen, urlretrieve
else:
from urllib import urlopen, urlretrieve
def worker(url_and_target): # Parallelize data download via multiprocessing
try:
(url, target_path) = url_and_target
print('>>> Downloading ' + target_path)
urlretrieve(url, target_path)
except (KeyboardInterrupt, SystemExit):
print('>>> Exiting child process')
class KGSIndex:
def __init__(self,
kgs_url='http://u-go.net/gamerecords/',
index_page='train_data/kgs_index.html',
data_directory='data'):
"""Create an index of zip files containing SGF data of actual Go Games on KGS.
Parameters:
-----------
kgs_url: URL with links to zip files of games
index_page: Name of local html file of kgs_url
data_directory: name of directory relative to current path to store SGF data
"""
self.kgs_url = kgs_url
self.index_page = index_page
self.data_directory = data_directory
self.file_info = []
self.urls = []
self.load_index() # Load index on creation
def download_files(self):
"""Download zip files by distributing work on all available CPUs"""
if not os.path.isdir(self.data_directory):
os.makedirs(self.data_directory)
urls_to_download = []
for file_info in self.file_info:
url = file_info['url']
file_name = file_info['filename']
if not os.path.isfile(self.data_directory + '/' + file_name):
urls_to_download.append((url, self.data_directory + '/' + file_name))
cores = multiprocessing.cpu_count()
pool = multiprocessing.Pool(processes=cores)
try:
it = pool.imap(worker, urls_to_download)
for _ in it:
pass
pool.close()
pool.join()
except KeyboardInterrupt:
print(">>> Caught KeyboardInterrupt, terminating workers")
pool.terminate()
pool.join()
sys.exit(-1)
def create_index_page(self):
"""If there is no local html containing links to files, create one."""
if os.path.isfile(self.index_page):
print('>>> Reading cached index page')
index_file = open(self.index_page, 'r')
index_contents = index_file.read()
index_file.close()
else:
print('>>> Downloading index page')
fp = urlopen(self.kgs_url)
data = six.text_type(fp.read())
fp.close()
index_contents = data
index_file = open(self.index_page, 'w')
index_file.write(index_contents)
index_file.close()
return index_contents
def load_index(self):
"""Create the actual index representation from the previously downloaded or cached html."""
index_contents = self.create_index_page()
split_page = [item for item in index_contents.split('<a href="') if item.startswith("https://")]
for item in split_page:
download_url = item.split('">Download')[0]
if download_url.endswith('.tar.gz'):
self.urls.append(download_url)
for url in self.urls:
filename = os.path.basename(url)
split_file_name = filename.split('-')
num_games = int(split_file_name[len(split_file_name) - 2])
print(filename + ' ' + str(num_games))
self.file_info.append({'url': url, 'filename': filename, 'num_games': num_games})
if __name__ == '__main__':
index = KGSIndex()
index.download_files()
+237
View File
@@ -0,0 +1,237 @@
# from __future__ import print_function
# from __future__ import absolute_import
import os
import glob
import os.path
import tarfile
import gzip
import shutil
import numpy as np
import multiprocessing
from os import sys
import torch
from tugo.gosgf import Sgf_game
from tugo.goboard_fast import Board, GameState, Move
from tugo.gotypes import Player, Point
from tugo.data.index_processor import KGSIndex
from tugo.data.sampling import Sampler
from tugo.data.generator import DataGenerator
from tugo.encoders.base import get_encoder_by_name
from torch.utils.data import TensorDataset
def worker(jobinfo):
try:
clazz, encoder, zip_file, data_file_name, game_list = jobinfo
clazz(encoder=encoder).process_zip(zip_file, data_file_name, game_list)
except (KeyboardInterrupt, SystemExit):
raise Exception('>>> Exiting child process.')
class GoDataProcessor:
def __init__(self, encoder='simple', data_directory='train_data'):
self.encoder_string = encoder
self.encoder = get_encoder_by_name(encoder, 19)
self.data_dir = data_directory
def load_go_data(self, data_type='train', num_samples=1000,
use_generator=False):
index = KGSIndex(data_directory=self.data_dir)
index.download_files()
sampler = Sampler(data_dir=self.data_dir)
data = sampler.draw_data(data_type, num_samples)
self.map_to_workers(data_type, data)
if use_generator:
generator = DataGenerator(self.data_dir, data)
return generator
else:
features_and_labels = self.consolidate_games(data_type, data)
return features_and_labels
def unzip_data(self, zip_file_name):
this_gz = gzip.open(self.data_dir + '/' + zip_file_name)
tar_file = zip_file_name[0:-3]
this_tar = open(self.data_dir + '/' + tar_file, 'wb')
shutil.copyfileobj(this_gz, this_tar)
this_tar.close()
return tar_file
def process_zip(self, zip_file_name, data_file_name, game_list):
print(">>> Processing zip file:", zip_file_name, "Data file name:", data_file_name)
tar_file = self.unzip_data(zip_file_name)
zip_file = tarfile.open(self.data_dir + '/' + tar_file)
name_list = zip_file.getnames()
total_examples = self.num_total_examples(zip_file, game_list, name_list)
print(">>> Total examples:", total_examples)
shape = self.encoder.shape()
feature_shape = np.insert(shape, 0, np.asarray([total_examples]))
features = np.zeros(feature_shape)
labels = np.zeros((total_examples,))
counter = 0
for index in game_list:
name = name_list[index + 1]
if not name.endswith('.sgf'):
raise ValueError(name + ' is not a valid sgf')
sgf_content = zip_file.extractfile(name).read()
sgf = Sgf_game.from_string(sgf_content)
if sgf.get_handicap() is not None and sgf.get_handicap() != 0:
print('<func process_zip>', 'get_handicap:', sgf.get_handicap(), ' => skipping handicaped game ...')
continue
game_state, first_move_done = self.get_handicap(sgf)
for item in sgf.main_sequence_iter():
color, move_tuple = item.get_move()
point = None
if color is not None:
if move_tuple is not None:
row, col = move_tuple
point = Point(row + 1, col + 1)
move = Move.play(point)
else:
move = Move.pass_turn()
if first_move_done and point is not None:
features[counter] = self.encoder.encode(game_state)
labels[counter] = self.encoder.encode_point(point)
counter += 1
game_state = game_state.apply_move(move)
first_move_done = True
feature_file_base = self.data_dir + '/' + data_file_name + '_features_%d'
label_file_base = self.data_dir + '/' + data_file_name + '_labels_%d'
chunk = 0 # Due to files with large content, split up after chunksize
chunksize = 1024
while features.shape[0] > 0:
feature_file = feature_file_base % chunk
label_file = label_file_base % chunk
chunk += 1
# current_features, features = features[:chunksize], features[chunksize:]
# current_labels, labels = labels[:chunksize], labels[chunksize:]
current_chunksize = min(chunksize, features.shape[0])
current_features, features = features[:current_chunksize], features[current_chunksize:]
current_labels, labels = labels[:current_chunksize], labels[current_chunksize:]
np.save(feature_file, current_features)
# print(">>>>>> save feature_file:", feature_file)
np.save(label_file, current_labels)
def consolidate_games(self, name, samples):
files_needed = set(file_name for file_name, index in samples)
file_names = []
for zip_file_name in files_needed:
file_name = zip_file_name.replace('.tar.gz', '') + name
file_names.append(file_name)
feature_list = []
label_list = []
for file_name in file_names:
file_prefix = file_name.replace('.tar.gz', '')
base = self.data_dir + '/' + file_prefix + '_features_*.npy'
# print("base:", base)
# print(glob.glob(base))
for feature_file in glob.glob(base):
# print(f"Processing feature file: {feature_file}")
label_file = feature_file.replace('features', 'labels')
x = np.load(feature_file)
y = np.load(label_file)
x = x.astype('float32')
# y = torch.nn.functional.one_hot(torch.tensor(y.astype(int)), 19 * 19)
y = torch.tensor(y.astype(int))
# print("Feature shape:", x.shape)
# print("Label shape:", y.shape)
feature_list.append(x)
label_list.append(y)
assert feature_list, "No features found. Please check the data files."
features = torch.tensor(np.concatenate(feature_list, axis=0)).float()
labels = torch.cat(label_list, axis=0)
feature_file = self.data_dir + '/' + name + '_feature.pt'
label_file = self.data_dir + '/' + name + '_label.pt'
torch.save(features, feature_file)
torch.save(labels, label_file)
dataset = TensorDataset(torch.tensor(features), labels)
return dataset
@staticmethod
def get_handicap(sgf): # Get handicap stones
go_board = Board(19, 19)
first_move_done = False
move = None
game_state = GameState.new_game(19)
if sgf.get_handicap() is not None and sgf.get_handicap() != 0:
# print('get_handicap2:', sgf.get_handicap())
for setup in sgf.get_root().get_setup_stones():
# print("------- setup:", setup, type(setup))
for move in setup:
# print("------- move:", move, type(move))
row, col = move
point = Point(row + 1, col + 1)
move = Move.play(point)
go_board.place_stone(Player.black, Point(row + 1, col + 1)) # black gets handicap
first_move_done = True
game_state = GameState(go_board, Player.white, None, move)
return game_state, first_move_done
def map_to_workers(self, data_type, samples):
zip_names = set()
indices_by_zip_name = {}
for filename, index in samples:
zip_names.add(filename)
if filename not in indices_by_zip_name:
indices_by_zip_name[filename] = []
indices_by_zip_name[filename].append(index)
zips_to_process = []
for zip_name in zip_names:
base_name = zip_name.replace('.tar.gz', '')
data_file_name = base_name + data_type
if not os.path.isfile(self.data_dir + '/' + data_file_name):
zips_to_process.append((self.__class__, self.encoder_string, zip_name,
data_file_name, indices_by_zip_name[zip_name]))
cores = multiprocessing.cpu_count() # Determine number of CPU cores and split work load among them
pool = multiprocessing.Pool(processes=cores)
p = pool.map_async(worker, zips_to_process)
try:
_ = p.get()
except KeyboardInterrupt: # Caught keyboard interrupt, terminating workers
pool.terminate()
pool.join()
sys.exit(-1)
def num_total_examples(self, zip_file, game_list, name_list):
total_examples = 0
for index in game_list:
name = name_list[index + 1]
if name.endswith('.sgf'):
sgf_content = zip_file.extractfile(name).read()
sgf = Sgf_game.from_string(sgf_content)
if sgf.get_handicap() is not None and sgf.get_handicap() != 0:
print('get_handicap:', sgf.get_handicap(), ' => skipping handicaped game ...')
continue
game_state, first_move_done = self.get_handicap(sgf)
num_moves = 0
for item in sgf.main_sequence_iter():
color, move = item.get_move()
if color is not None:
if first_move_done:
num_moves += 1
first_move_done = True
total_examples = total_examples + num_moves
else:
raise ValueError(name + ' is not a valid sgf')
return total_examples
+137
View File
@@ -0,0 +1,137 @@
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file, You can
# obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import print_function
from __future__ import absolute_import
import os
import random
from tugo.data.index_processor import KGSIndex
from six.moves import range
class Sampler:
"""Sample training and test data from zipped sgf files such that test data is kept stable."""
def __init__(self, data_dir='data', num_test_games=100, cap_year=2015, seed=1337):
self.data_dir = data_dir
self.num_test_games = num_test_games
self.test_games = []
self.train_games = []
self.test_folder = 'train_data/test_samples.py'
self.cap_year = cap_year
random.seed(seed)
self.compute_test_samples()
def draw_data(self, data_type, num_samples):
if data_type == 'test':
return self.test_games
elif data_type == 'train' and num_samples is not None:
return self.draw_training_samples(num_samples)
elif data_type == 'train' and num_samples is None:
return self.draw_all_training()
else:
raise ValueError(data_type + " is not a valid data type, choose from 'train' or 'test'")
def draw_samples(self, num_sample_games):
"""Draw num_sample_games many training games from index."""
available_games = []
index = KGSIndex(data_directory=self.data_dir)
for fileinfo in index.file_info:
filename = fileinfo['filename']
year = int(filename.split('-')[1].split('_')[0])
if year > self.cap_year:
continue
num_games = fileinfo['num_games']
for i in range(num_games):
available_games.append((filename, i))
print('>>> Total number of games used: ' + str(len(available_games)))
sample_set = set()
while len(sample_set) < num_sample_games:
sample = random.choice(available_games)
if sample not in sample_set:
sample_set.add(sample)
print('Drawn ' + str(num_sample_games) + ' samples:')
return list(sample_set)
def draw_training_games(self):
"""Get list of all non-test games, that are no later than dec 2014
Ignore games after cap_year to keep training data stable
"""
index = KGSIndex(data_directory=self.data_dir)
for file_info in index.file_info:
filename = file_info['filename']
year = int(filename.split('-')[1].split('_')[0])
if year > self.cap_year:
continue
num_games = file_info['num_games']
for i in range(num_games):
sample = (filename, i)
if sample not in self.test_games:
self.train_games.append(sample)
print('total num training games: ' + str(len(self.train_games)))
def compute_test_samples(self):
"""If not already existing, create local file to store fixed set of test samples"""
if not os.path.isfile(self.test_folder):
test_games = self.draw_samples(self.num_test_games)
test_sample_file = open(self.test_folder, 'w')
for sample in test_games:
test_sample_file.write(str(sample) + "\n")
test_sample_file.close()
test_sample_file = open(self.test_folder, 'r')
sample_contents = test_sample_file.read()
test_sample_file.close()
for line in sample_contents.split('\n'):
if line != "":
(filename, index) = eval(line)
self.test_games.append((filename, index))
def draw_training_samples(self, num_sample_games):
"""Draw training games, not overlapping with any of the test games."""
available_games = []
index = KGSIndex(data_directory=self.data_dir)
for fileinfo in index.file_info:
filename = fileinfo['filename']
year = int(filename.split('-')[1].split('_')[0])
if year > self.cap_year:
continue
num_games = fileinfo['num_games']
for i in range(num_games):
available_games.append((filename, i))
print('total num games: ' + str(len(available_games)))
sample_set = set()
while len(sample_set) < num_sample_games:
sample = random.choice(available_games)
if sample not in self.test_games:
sample_set.add(sample)
print('Drawn ' + str(num_sample_games) + ' samples:')
return list(sample_set)
def draw_all_training(self):
"""Draw all available training games."""
available_games = []
index = KGSIndex(data_directory=self.data_dir)
for fileinfo in index.file_info:
filename = fileinfo['filename']
year = int(filename.split('-')[1].split('_')[0])
if year > self.cap_year:
continue
if 'num_games' in fileinfo.keys():
num_games = fileinfo['num_games']
else:
continue
for i in range(num_games):
available_games.append((filename, i))
print('total num games: ' + str(len(available_games)))
sample_set = set()
for sample in available_games:
if sample not in self.test_games:
sample_set.add(sample)
print('Drawn all samples, ie ' + str(len(sample_set)) + ' samples:')
return list(sample_set)
+6
View File
@@ -0,0 +1,6 @@
from tugo.encoders.base import *
#from tugo.encoders.alphago import *
#from tugo.encoders.betago import *
from tugo.encoders.oneplane import *
from tugo.encoders.sevenplane import *
from tugo.encoders.simple import *
+141
View File
@@ -0,0 +1,141 @@
from tugo.encoders.base import Encoder
from tugo.encoders.utils import is_ladder_escape, is_ladder_capture
from tugo.gotypes import Point, Player
from tugo.goboard_fast import Move
from tugo.agent.helpers_fast import is_point_an_eye
import numpy as np
import torch
"""
Feature name num of planes Description
Stone colour 3 Player stone / opponent stone / empty
Ones 1 A constant plane filled with 1
Zeros 1 A constant plane filled with 0
Sensibleness 1 Whether a move is legal and does not fill its own eyes
Turns since 8 How many turns since a move was played
Liberties 8 Number of liberties (empty adjacent points)
Liberties after move 8 Number of liberties after this move is played
Capture size 8 How many opponent stones would be captured
Self-atari size 8 How many of own stones would be captured
Ladder capture 1 Whether a move at this point is a successful ladder capture
Ladder escape 1 Whether a move at this point is a successful ladder escape
"""
FEATURE_OFFSETS = {
"stone_color": 0,
"ones": 3,
"zeros": 4,
"sensibleness": 5,
"turns_since": 6,
"liberties": 14,
"liberties_after": 22,
"capture_size": 30,
"self_atari_size": 38,
"ladder_capture": 46,
"ladder_escape": 47,
"current_player_color": 48,
}
def offset(feature):
return FEATURE_OFFSETS[feature]
class AlphaGoEncoder(Encoder):
def __init__(self, board_size=(19, 19), use_player_plane=True):
self.board_width, self.board_height = board_size
self.use_player_plane = use_player_plane
self.num_planes = 48 + use_player_plane
def name(self):
return "alphago"
def encode(self, game_state):
board_tensor = torch.zeros((self.num_planes, self.board_height, self.board_width))
for r in range(self.board_height):
for c in range(self.board_width):
point = Point(row=r + 1, col=c + 1)
go_string = game_state.board.get_go_string(point)
if go_string and go_string.color == game_state.next_player:
board_tensor[offset("stone_color"), r, c] = 1
elif go_string and go_string.color == game_state.next_player.other:
board_tensor[offset("stone_color") + 1, r, c] = 1
else:
board_tensor[offset("stone_color") + 2, r, c] = 1
board_tensor[offset("ones")] = self.ones()
board_tensor[offset("zeros")] = self.zeros()
if not is_point_an_eye(game_state.board, point, game_state.next_player):
board_tensor[offset("sensibleness"), r, c] = 1
ages = min(game_state.board.move_ages.get(r, c), 8)
if ages > 0:
board_tensor[offset("turns_since") + int(ages), r, c] = 1
if game_state.board.get_go_string(point):
liberties = min(game_state.board.get_go_string(point).num_liberties, 8)
board_tensor[offset("liberties") + liberties, r, c] = 1
move = Move(point)
if game_state.is_valid_move(move):
new_state = game_state.apply_move(move)
liberties = min(new_state.board.get_go_string(point).num_liberties, 8)
board_tensor[offset("liberties_after") + liberties, r, c] = 1
adjacent_strings = [game_state.board.get_go_string(nb) for nb in point.neighbors()]
capture_count = 0
for go_string in adjacent_strings:
other_player = game_state.next_player.other
if go_string and go_string.num_liberties == 1 and go_string.color == other_player:
capture_count += len(go_string.stones)
capture_count = min(capture_count, 8)
board_tensor[offset("capture_size") + capture_count, r, c] = 1
if go_string and go_string.num_liberties == 1:
if go_string := game_state.board.get_go_string(point):
num_atari_stones = min(len(go_string.stones), 8)
board_tensor[offset("self_atari_size") + num_atari_stones, r, c] = 1
if is_ladder_capture(game_state, point):
board_tensor[offset("ladder_capture"), r, c] = 1
if is_ladder_escape(game_state, point):
board_tensor[offset("ladder_escape"), r, c] = 1
if self.use_player_plane:
if game_state.next_player == Player.black:
board_tensor[offset("ones")] = self.ones()
else:
board_tensor[offset("zeros")] = self.zeros()
return board_tensor
def ones(self):
return torch.ones((1, self.board_height, self.board_width))
def zeros(self):
return torch.zeros((1, self.board_height, self.board_width))
def capture_size(self, game_state, num_planes=8):
pass
def encode_point(self, point):
return self.board_width * (point.row - 1) + (point.col - 1)
def decode_point_index(self, index):
row = index // self.board_width
col = index % self.board_width
return Point(row=row + 1, col=col + 1)
def num_points(self):
return self.board_width * self.board_height
def shape(self):
return self.num_planes, self.board_height, self.board_width
def create(board_size):
return AlphaGoEncoder(board_size)
+26
View File
@@ -0,0 +1,26 @@
import unittest
from tugo.agent.helpers import is_point_an_eye
from tugo.goboard_fast import Board, GameState, Move
from tugo.gotypes import Player, Point
from tugo.encoders.alphago import AlphaGoEncoder
class AlphaGoEncoderTest(unittest.TestCase):
def test_encoder(self):
alphago = AlphaGoEncoder()
start = GameState.new_game(19)
next_state = start.apply_move(Move.play(Point(16, 16)))
alphago.encode(next_state)
self.assertEquals(alphago.name(), 'alphago')
self.assertEquals(alphago.board_height, 19)
self.assertEquals(alphago.board_width, 19)
self.assertEquals(alphago.num_planes, 49)
self.assertEquals(alphago.shape(), (49, 19, 19))
if __name__ == '__main__':
unittest.main()
+51
View File
@@ -0,0 +1,51 @@
# tag::importlib[]
import importlib
# end::importlib[]
__all__ = [
'Encoder',
'get_encoder_by_name',
]
# tag::base_encoder[]
class Encoder:
def name(self): # <1>
raise NotImplementedError()
def encode(self, game_state): # <2>
raise NotImplementedError()
def encode_point(self, point): # <3>
raise NotImplementedError()
def decode_point_index(self, index): # <4>
raise NotImplementedError()
def num_points(self): # <5>
raise NotImplementedError()
def shape(self): # <6>
raise NotImplementedError()
# <1> Lets us support logging or saving the name of the encoder our model is using.
# <2> Turn a Go board into a numeric data.
# <3> Turn a Go board point into an integer index.
# <4> Turn an integer index back into a Go board point.
# <5> Number of points on the board, i.e. board width times board height.
# <6> Shape of the encoded board structure.
# end::base_encoder[]
# tag::encoder_by_name[]
def get_encoder_by_name(name, board_size): # <1>
if isinstance(board_size, int):
board_size = (board_size, board_size) # <2>
module = importlib.import_module('tugo.encoders.' + name)
constructor = getattr(module, 'create') # <3>
return constructor(board_size)
# <1> We can create encoder instances by referencing their name.
# <2> If board_size is one integer, we create a square board from it.
# <3> Each encoder implementation will have to provide a "create" function that provides an instance.
# end::encoder_by_name[]
+60
View File
@@ -0,0 +1,60 @@
# tag::oneplane_imports[]
import numpy as np
from tugo.encoders.base import Encoder
from tugo.goboard import Point
# end::oneplane_imports[]
# tag::oneplane_encoder[]
class OnePlaneEncoder(Encoder):
def __init__(self, board_size):
self.board_width, self.board_height = board_size
self.num_planes = 1
def name(self): # <1>
return 'oneplane'
def encode(self, game_state): # <2>
board_matrix = np.zeros(self.shape())
next_player = game_state.next_player
for r in range(self.board_height):
for c in range(self.board_width):
p = Point(row=r + 1, col=c + 1)
go_string = game_state.board.get_go_string(p)
if go_string is None:
continue
if go_string.color == next_player:
board_matrix[0, r, c] = 1
else:
board_matrix[0, r, c] = -1
return board_matrix
# <1> We can reference this encoder by the name "oneplane".
# <2> To encode, we fill a matrix with 1 if the point contains one of the current player's stones, -1 if the point contains the opponent's stones and 0 if the point is empty.
# end::oneplane_encoder[]
# tag::oneplane_encoder_2[]
def encode_point(self, point): # <1>
return self.board_width * (point.row - 1) + (point.col - 1)
def decode_point_index(self, index): # <2>
row = index // self.board_width
col = index % self.board_width
return Point(row=row + 1, col=col + 1)
def num_points(self):
return self.board_width * self.board_height
def shape(self):
return self.num_planes, self.board_height, self.board_width
# <1> Turn a board point into an integer index.
# <2> Turn an integer index into a board point.
# end::oneplane_encoder_2[]
# tag::oneplane_create[]
def create(board_size):
return OnePlaneEncoder(board_size)
# end::oneplane_create[]
+57
View File
@@ -0,0 +1,57 @@
# tag::sevenplane_init[]
import numpy as np
from tugo.encoders.base import Encoder
from tugo.goboard import Move, Point
class SevenPlaneEncoder(Encoder):
def __init__(self, board_size):
self.board_width, self.board_height = board_size
self.num_planes = 7
def name(self):
return 'sevenplane'
# end::sevenplane_init[]
# tag::sevenplane_encode[]
def encode(self, game_state):
board_tensor = np.zeros(self.shape())
base_plane = {game_state.next_player: 0,
game_state.next_player.other: 3}
for row in range(self.board_height):
for col in range(self.board_width):
p = Point(row=row + 1, col=col + 1)
go_string = game_state.board.get_go_string(p)
if go_string is None:
if game_state.does_move_violate_ko(game_state.next_player,
Move.play(p)):
board_tensor[6][row][col] = 1 # <1>
else:
liberty_plane = min(3, go_string.num_liberties) - 1
liberty_plane += base_plane[go_string.color]
board_tensor[liberty_plane][row][col] = 1 # <2>
return board_tensor
# <1> Encoding moves prohibited by the ko rule
# <2> Encoding black and white stones with 1, 2 or more liberties.
# end::sevenplane_encode[]
# tag::sevenplane_rest[]
def encode_point(self, point):
return self.board_width * (point.row - 1) + (point.col - 1)
def decode_point_index(self, index):
row = index // self.board_width
col = index % self.board_width
return Point(row=row + 1, col=col + 1)
def num_points(self):
return self.board_width * self.board_height
def shape(self):
return self.num_planes, self.board_height, self.board_width
def create(board_size):
return SevenPlaneEncoder(board_size)
# end::sevenplane_rest[]
+67
View File
@@ -0,0 +1,67 @@
import numpy as np
from tugo.encoders.base import Encoder
from tugo.goboard import Move
from tugo.gotypes import Player, Point
class SimpleEncoder(Encoder):
def __init__(self, board_size):
"""
Args:
board_size: tuple of (width, height)
"""
self.board_width, self.board_height = board_size
# 0 - 3. black stones with 1, 2, 3, 4+ liberties
# 4 - 7. white stones with 1, 2, 3, 4+ liberties
# 8. black plays next
# 9. white plays next
# 10. move would be illegal due to ko
self.num_planes = 11
def name(self):
return 'simple'
def encode(self, game_state):
board_tensor = np.zeros(self.shape())
if game_state.next_player == Player.black:
board_tensor[8] = 1
else:
board_tensor[9] = 1
for r in range(self.board_height):
for c in range(self.board_width):
p = Point(row=r + 1, col=c + 1)
go_string = game_state.board.get_go_string(p)
if go_string is None:
if game_state.does_move_violate_ko(game_state.next_player,
Move.play(p)):
board_tensor[10][r][c] = 1
else:
liberty_plane = min(4, go_string.num_liberties) - 1
if go_string.color == Player.white:
liberty_plane += 4
board_tensor[liberty_plane][r][c] = 1
return board_tensor
def encode_point(self, point):
"""Turn a board point into an integer index."""
# Points are 1-indexed
return self.board_width * (point.row - 1) + (point.col - 1)
def decode_point_index(self, index):
"""Turn an integer index into a board point."""
row = index // self.board_width
col = index % self.board_width
return Point(row=row + 1, col=col + 1)
def num_points(self):
return self.board_width * self.board_height
def shape(self):
return self.num_planes, self.board_height, self.board_width
def create(board_size):
return SimpleEncoder(board_size)
+107
View File
@@ -0,0 +1,107 @@
from tugo.goboard import Move
def is_ladder_capture(game_state, candidate, recursion_depth=50):
return is_ladder(True, game_state, candidate, None, recursion_depth)
def is_ladder_escape(game_state, candidate, recursion_depth=50):
return is_ladder(False, game_state, candidate, None, recursion_depth)
def is_ladder(try_capture, game_state, candidate,
ladder_stones=None, recursion_depth=50):
"""Ladders are played out in reversed roles, one player tries to capture,
the other to escape. We determine the ladder status by recursively calling
is_ladder in opposite roles, providing suitable capture or escape candidates.
Arguments:
try_capture: boolean flag to indicate if you want to capture or escape the ladder
game_state: current game state, instance of GameState
candidate: a move that potentially leads to escaping the ladder or capturing it, instance of Move
ladder_stones: the stones to escape or capture, list of Point. Will be inferred if not provided.
recursion_depth: when to stop recursively calling this function, integer valued.
Returns True if game state is a ladder and try_capture is true (the ladder captures)
or if game state is not a ladder and try_capture is false (you can successfully escape)
and False otherwise.
"""
if not game_state.is_valid_move(Move(candidate)) or not recursion_depth:
return False
next_player = game_state.next_player
capture_player = next_player if try_capture else next_player.other
escape_player = capture_player.other
if ladder_stones is None:
ladder_stones = guess_ladder_stones(game_state, candidate, escape_player)
for ladder_stone in ladder_stones:
current_state = game_state.apply_move(candidate)
if try_capture:
candidates = determine_escape_candidates(
game_state, ladder_stone, capture_player)
attempted_escapes = [ # now try to escape
is_ladder(False, current_state, escape_candidate,
ladder_stone, recursion_depth - 1)
for escape_candidate in candidates]
if not any(attempted_escapes):
return True # if at least one escape fails, we capture
else:
if count_liberties(current_state, ladder_stone) >= 3:
return True # successful escape
if count_liberties(current_state, ladder_stone) == 1:
continue # failed escape, others might still do
candidates = liberties(current_state, ladder_stone)
attempted_captures = [ # now try to capture
is_ladder(True, current_state, capture_candidate,
ladder_stone, recursion_depth - 1)
for capture_candidate in candidates]
if any(attempted_captures):
continue # failed escape, try others
return True # candidate can't be caught in a ladder, escape.
return False # no captures / no escapes
def is_candidate(game_state, move, player):
return game_state.next_player == player and \
count_liberties(game_state, move) == 2
def guess_ladder_stones(game_state, move, escape_player):
adjacent_strings = [game_state.board.get_go_string(nb) for nb in move.neighbors() if game_state.board.get_go_string(nb)]
if adjacent_strings:
string = adjacent_strings[0]
neighbors = []
for string in adjacent_strings:
stones = string.stones
for stone in stones:
neighbors.append(stone)
return [Move(nb) for nb in neighbors if is_candidate(game_state, Move(nb), escape_player)]
else:
return []
def determine_escape_candidates(game_state, move, capture_player):
escape_candidates = move.neighbors()
for other_ladder_stone in game_state.board.get_go_string(move).stones:
for neighbor in other_ladder_stone.neighbors():
right_color = game_state.color(neighbor) == capture_player
one_liberty = count_liberties(game_state, neighbor) == 1
if right_color and one_liberty:
escape_candidates.append(liberties(game_state, neighbor))
return escape_candidates
def count_liberties(game_state, move):
if game_state.board.get_go_string(move):
return game_state.board.get_go_string(move).num_liberties
else:
return 0
def liberties(game_state, move):
return list(game_state.board.get_go_string(move).liberties)
+307
View File
@@ -0,0 +1,307 @@
import copy
from tugo.gotypes import Player, Point
from tugo.scoring import compute_game_result
# tag::import_zobrist[]
from tugo import zobrist
# end::import_zobrist[]
__all__ = [
'Board',
'GameState',
'Move',
]
class IllegalMoveError(Exception):
pass
# tag::fast_go_strings[]
class GoString:
def __init__(self, color, stones, liberties):
self.color = color
self.stones = frozenset(stones)
self.liberties = frozenset(liberties) # <1>
def without_liberty(self, point): # <2>
new_liberties = self.liberties - set([point])
return GoString(self.color, self.stones, new_liberties)
def with_liberty(self, point):
new_liberties = self.liberties | set([point])
return GoString(self.color, self.stones, new_liberties)
# <1> `stones` and `liberties` are now immutable `frozenset` instances
# <2> The `without_liberty` methods replaces the previous `remove_liberty` method...
# <3> ... and `with_liberty` replaces `add_liberty`.
# end::fast_go_strings[]
def merged_with(self, string):
"""Return a new string containing all stones in both strings."""
assert string.color == self.color
combined_stones = self.stones | string.stones
return GoString(
self.color,
combined_stones,
(self.liberties | string.liberties) - combined_stones)
@property
def num_liberties(self):
return len(self.liberties)
def __eq__(self, other):
return isinstance(other, GoString) and \
self.color == other.color and \
self.stones == other.stones and \
self.liberties == other.liberties
def __deepcopy__(self, memodict={}):
return GoString(self.color, self.stones, copy.deepcopy(self.liberties))
# tag::init_zobrist[]
class Board:
def __init__(self, num_rows, num_cols):
self.num_rows = num_rows
self.num_cols = num_cols
self._grid = {}
self._hash = zobrist.EMPTY_BOARD
# end::init_zobrist[]
def place_stone(self, player, point):
assert self.is_on_grid(point)
if self._grid.get(point) is not None:
print('Illegal play on %s' % str(point))
assert self._grid.get(point) is None
# 0. Examine the adjacent points.
adjacent_same_color = []
adjacent_opposite_color = []
liberties = []
for neighbor in point.neighbors():
if not self.is_on_grid(neighbor):
continue
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
liberties.append(neighbor)
elif neighbor_string.color == player:
if neighbor_string not in adjacent_same_color:
adjacent_same_color.append(neighbor_string)
else:
if neighbor_string not in adjacent_opposite_color:
adjacent_opposite_color.append(neighbor_string)
new_string = GoString(player, [point], liberties)
# tag::apply_zobrist[]
new_string = GoString(player, [point], liberties) # <1>
for same_color_string in adjacent_same_color: # <2>
new_string = new_string.merged_with(same_color_string)
for new_string_point in new_string.stones:
self._grid[new_string_point] = new_string
self._hash ^= zobrist.HASH_CODE[point, player] # <3>
for other_color_string in adjacent_opposite_color:
replacement = other_color_string.without_liberty(point) # <4>
if replacement.num_liberties:
self._replace_string(other_color_string.without_liberty(point))
else:
self._remove_string(other_color_string) # <5>
# <1> Until this line `place_stone` remains the same.
# <2> You merge any adjacent strings of the same color.
# <3> Next, you apply the hash code for this point and player
# <4> Then you reduce liberties of any adjacent strings of the opposite color.
# <5> If any opposite color strings now have zero liberties, remove them.
# end::apply_zobrist[]
# tag::unapply_zobrist[]
def _replace_string(self, new_string): # <1>
for point in new_string.stones:
self._grid[point] = new_string
def _remove_string(self, string):
for point in string.stones:
for neighbor in point.neighbors(): # <2>
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
continue
if neighbor_string is not string:
self._replace_string(neighbor_string.with_liberty(point))
self._grid[point] = None
self._hash ^= zobrist.HASH_CODE[point, string.color] # <3>
# <1> This new helper method updates our Go board grid.
# <2> Removing a string can create liberties for other strings.
# <3> With Zobrist hashing, you need to unapply the hash for this move.
# end::unapply_zobrist[]
def is_on_grid(self, point):
return 1 <= point.row <= self.num_rows and \
1 <= point.col <= self.num_cols
def get(self, point):
"""Return the content of a point on the board.
Returns None if the point is empty, or a Player if there is a
stone on that point.
"""
string = self._grid.get(point)
if string is None:
return None
return string.color
def get_go_string(self, point):
"""Return the entire string of stones at a point.
Returns None if the point is empty, or a GoString if there is
a stone on that point.
"""
string = self._grid.get(point)
if string is None:
return None
return string
def __eq__(self, other):
return isinstance(other, Board) and \
self.num_rows == other.num_rows and \
self.num_cols == other.num_cols and \
self._hash() == other._hash()
def __deepcopy__(self, memodict={}):
copied = Board(self.num_rows, self.num_cols)
# Can do a shallow copy b/c the dictionary maps tuples
# (immutable) to GoStrings (also immutable)
copied._grid = copy.copy(self._grid)
copied._hash = self._hash
return copied
# tag::return_zobrist[]
def zobrist_hash(self):
return self._hash
# end::return_zobrist[]
class Move:
"""Any action a player can play on a turn.
Exactly one of is_play, is_pass, is_resign will be set.
"""
def __init__(self, point=None, is_pass=False, is_resign=False):
assert (point is not None) ^ is_pass ^ is_resign
self.point = point
self.is_play = (self.point is not None)
self.is_pass = is_pass
self.is_resign = is_resign
@classmethod
def play(cls, point):
"""A move that places a stone on the board."""
return Move(point=point)
@classmethod
def pass_turn(cls):
return Move(is_pass=True)
@classmethod
def resign(cls):
return Move(is_resign=True)
def __str__(self):
if self.is_pass:
return 'pass'
if self.is_resign:
return 'resign'
return '(r %d, c %d)' % (self.point.row, self.point.col)
# tag::init_state_zobrist[]
class GameState:
def __init__(self, board, next_player, previous, move):
self.board = board
self.next_player = next_player
self.previous_state = previous
if self.previous_state is None:
self.previous_states = frozenset()
else:
self.previous_states = frozenset(
previous.previous_states |
{(previous.next_player, previous.board.zobrist_hash())})
self.last_move = move
# end::init_state_zobrist[]
def apply_move(self, move):
"""Return the new GameState after applying the move."""
if move.is_play:
next_board = copy.deepcopy(self.board)
next_board.place_stone(self.next_player, move.point)
else:
next_board = self.board
return GameState(next_board, self.next_player.other, self, move)
@classmethod
def new_game(cls, board_size):
if isinstance(board_size, int):
board_size = (board_size, board_size)
board = Board(*board_size)
return GameState(board, Player.black, None, None)
def is_move_self_capture(self, player, move):
if not move.is_play:
return False
next_board = copy.deepcopy(self.board)
next_board.place_stone(player, move.point)
new_string = next_board.get_go_string(move.point)
return new_string.num_liberties == 0
@property
def situation(self):
return (self.next_player, self.board)
# tag::ko_zobrist[]
def does_move_violate_ko(self, player, move):
if not move.is_play:
return False
next_board = copy.deepcopy(self.board)
next_board.place_stone(player, move.point)
next_situation = (player.other, next_board.zobrist_hash())
return next_situation in self.previous_states
# end::ko_zobrist[]
def is_valid_move(self, move):
if self.is_over():
return False
if move.is_pass or move.is_resign:
return True
return (
self.board.get(move.point) is None and
not self.is_move_self_capture(self.next_player, move) and
not self.does_move_violate_ko(self.next_player, move))
def is_over(self):
if self.last_move is None:
return False
if self.last_move.is_resign:
return True
second_last_move = self.previous_state.last_move
if second_last_move is None:
return False
return self.last_move.is_pass and second_last_move.is_pass
def legal_moves(self):
moves = []
for row in range(1, self.board.num_rows + 1):
for col in range(1, self.board.num_cols + 1):
move = Move.play(Point(row, col))
if self.is_valid_move(move):
moves.append(move)
# These two moves are always legal.
moves.append(Move.pass_turn())
moves.append(Move.resign())
return moves
def winner(self):
if not self.is_over():
return None
if self.last_move.is_resign:
return self.next_player
game_result = compute_game_result(self)
return game_result.winner
+400
View File
@@ -0,0 +1,400 @@
import copy
from tugo.gotypes import Player, Point
from tugo.scoring import compute_game_result
from tugo import zobrist
from tugo.utils import MoveAge
__all__ = [
'Board',
'GameState',
'Move',
]
neighbor_tables = {}
corner_tables = {}
def init_neighbor_table(dim):
rows, cols = dim
new_table = {}
for r in range(1, rows + 1):
for c in range(1, cols + 1):
p = Point(row=r, col=c)
full_neighbors = p.neighbors()
true_neighbors = [
n for n in full_neighbors
if 1 <= n.row <= rows and 1 <= n.col <= cols]
new_table[p] = true_neighbors
neighbor_tables[dim] = new_table
def init_corner_table(dim):
rows, cols = dim
new_table = {}
for r in range(1, rows + 1):
for c in range(1, cols + 1):
p = Point(row=r, col=c)
full_corners = [
Point(row=p.row - 1, col=p.col - 1),
Point(row=p.row - 1, col=p.col + 1),
Point(row=p.row + 1, col=p.col - 1),
Point(row=p.row + 1, col=p.col + 1),
]
true_corners = [
n for n in full_corners
if 1 <= n.row <= rows and 1 <= n.col <= cols]
new_table[p] = true_corners
corner_tables[dim] = new_table
class IllegalMoveError(Exception):
pass
class GoString():
"""Stones that are linked by a chain of connected stones of the
same color.
"""
def __init__(self, color, stones, liberties):
self.color = color
self.stones = frozenset(stones)
self.liberties = frozenset(liberties)
def without_liberty(self, point):
new_liberties = self.liberties - set([point])
return GoString(self.color, self.stones, new_liberties)
def with_liberty(self, point):
new_liberties = self.liberties | set([point])
return GoString(self.color, self.stones, new_liberties)
def merged_with(self, string):
"""Return a new string containing all stones in both strings."""
assert string.color == self.color
combined_stones = self.stones | string.stones
return GoString(
self.color,
combined_stones,
(self.liberties | string.liberties) - combined_stones)
@property
def num_liberties(self):
return len(self.liberties)
def __eq__(self, other):
return isinstance(other, GoString) and \
self.color == other.color and \
self.stones == other.stones and \
self.liberties == other.liberties
def __deepcopy__(self, memodict={}):
return GoString(self.color, self.stones, copy.deepcopy(self.liberties))
class Board():
def __init__(self, num_rows, num_cols):
self.num_rows = num_rows
self.num_cols = num_cols
self._grid = {}
self._hash = zobrist.EMPTY_BOARD
global neighbor_tables
dim = (num_rows, num_cols)
if dim not in neighbor_tables:
init_neighbor_table(dim)
if dim not in corner_tables:
init_corner_table(dim)
self.neighbor_table = neighbor_tables[dim]
self.corner_table = corner_tables[dim]
self.move_ages = MoveAge(self)
def neighbors(self, point):
return self.neighbor_table[point]
def corners(self, point):
return self.corner_table[point]
def place_stone(self, player, point):
assert self.is_on_grid(point)
if self._grid.get(point) is not None:
print('Illegal play on %s' % str(point))
assert self._grid.get(point) is None
# 0. Examine the adjacent points.
adjacent_same_color = []
adjacent_opposite_color = []
liberties = []
self.move_ages.increment_all()
self.move_ages.add(point)
for neighbor in self.neighbor_table[point]:
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
liberties.append(neighbor)
elif neighbor_string.color == player:
if neighbor_string not in adjacent_same_color:
adjacent_same_color.append(neighbor_string)
else:
if neighbor_string not in adjacent_opposite_color:
adjacent_opposite_color.append(neighbor_string)
new_string = GoString(player, [point], liberties)
# tag::apply_zobrist[]
# 1. Merge any adjacent strings of the same color.
for same_color_string in adjacent_same_color:
new_string = new_string.merged_with(same_color_string)
for new_string_point in new_string.stones:
self._grid[new_string_point] = new_string
# Remove empty-point hash code.
self._hash ^= zobrist.HASH_CODE[point, None]
# Add filled point hash code.
self._hash ^= zobrist.HASH_CODE[point, player]
# end::apply_zobrist[]
# 2. Reduce liberties of any adjacent strings of the opposite
# color.
# 3. If any opposite color strings now have zero liberties,
# remove them.
for other_color_string in adjacent_opposite_color:
replacement = other_color_string.without_liberty(point)
if replacement.num_liberties:
self._replace_string(other_color_string.without_liberty(point))
else:
self._remove_string(other_color_string)
def _replace_string(self, new_string):
for point in new_string.stones:
self._grid[point] = new_string
def _remove_string(self, string):
for point in string.stones:
self.move_ages.reset_age(point)
# Removing a string can create liberties for other strings.
for neighbor in self.neighbor_table[point]:
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
continue
if neighbor_string is not string:
self._replace_string(neighbor_string.with_liberty(point))
self._grid[point] = None
# Remove filled point hash code.
self._hash ^= zobrist.HASH_CODE[point, string.color]
# Add empty point hash code.
self._hash ^= zobrist.HASH_CODE[point, None]
def is_self_capture(self, player, point):
friendly_strings = []
for neighbor in self.neighbor_table[point]:
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
# This point has a liberty. Can't be self capture.
return False
elif neighbor_string.color == player:
# Gather for later analysis.
friendly_strings.append(neighbor_string)
else:
if neighbor_string.num_liberties == 1:
# This move is real capture, not a self capture.
return False
if all(neighbor.num_liberties == 1 for neighbor in friendly_strings):
return True
return False
def will_capture(self, player, point):
for neighbor in self.neighbor_table[point]:
neighbor_string = self._grid.get(neighbor)
if neighbor_string is None:
continue
elif neighbor_string.color == player:
continue
else:
if neighbor_string.num_liberties == 1:
# This move would capture.
return True
return False
def is_on_grid(self, point):
return 1 <= point.row <= self.num_rows and \
1 <= point.col <= self.num_cols
def get(self, point):
"""Return the content of a point on the board.
Returns None if the point is empty, or a Player if there is a
stone on that point.
"""
string = self._grid.get(point)
if string is None:
return None
return string.color
def get_go_string(self, point):
"""Return the entire string of stones at a point.
Returns None if the point is empty, or a GoString if there is
a stone on that point.
"""
string = self._grid.get(point)
if string is None:
return None
return string
def __eq__(self, other):
return isinstance(other, Board) and \
self.num_rows == other.num_rows and \
self.num_cols == other.num_cols and \
self._hash() == other._hash()
def __deepcopy__(self, memodict={}):
copied = Board(self.num_rows, self.num_cols)
# Can do a shallow copy b/c the dictionary maps tuples
# (immutable) to GoStrings (also immutable)
copied._grid = copy.copy(self._grid)
copied._hash = self._hash
return copied
# tag::return_zobrist[]
def zobrist_hash(self):
return self._hash
# end::return_zobrist[]
class Move():
"""Any action a player can play on a turn.
Exactly one of is_play, is_pass, is_resign will be set.
"""
def __init__(self, point=None, is_pass=False, is_resign=False):
assert (point is not None) ^ is_pass ^ is_resign
self.point = point
self.is_play = (self.point is not None)
self.is_pass = is_pass
self.is_resign = is_resign
@classmethod
def play(cls, point):
"""A move that places a stone on the board."""
return Move(point=point)
@classmethod
def pass_turn(cls):
return Move(is_pass=True)
@classmethod
def resign(cls):
return Move(is_resign=True)
def __str__(self):
if self.is_pass:
return 'pass'
if self.is_resign:
return 'resign'
return '(r %d, c %d)' % (self.point.row, self.point.col)
def __hash__(self):
return hash((
self.is_play,
self.is_pass,
self.is_resign,
self.point))
def __eq__(self, other):
return (
self.is_play,
self.is_pass,
self.is_resign,
self.point) == (
other.is_play,
other.is_pass,
other.is_resign,
other.point)
class GameState():
def __init__(self, board, next_player, previous, move):
self.board = board
self.next_player = next_player
self.previous_state = previous
if previous is None:
self.previous_states = frozenset()
else:
self.previous_states = frozenset(
previous.previous_states |
{(previous.next_player, previous.board.zobrist_hash())})
self.last_move = move
def apply_move(self, move):
"""Return the new GameState after applying the move."""
if move.is_play:
next_board = copy.deepcopy(self.board)
next_board.place_stone(self.next_player, move.point)
else:
next_board = self.board
return GameState(next_board, self.next_player.other, self, move)
@classmethod
def new_game(cls, board_size):
if isinstance(board_size, int):
board_size = (board_size, board_size)
board = Board(*board_size)
return GameState(board, Player.black, None, None)
def is_move_self_capture(self, player, move):
if not move.is_play:
return False
return self.board.is_self_capture(player, move.point)
@property
def situation(self):
return (self.next_player, self.board)
def does_move_violate_ko(self, player, move):
if not move.is_play:
return False
if not self.board.will_capture(player, move.point):
return False
next_board = copy.deepcopy(self.board)
next_board.place_stone(player, move.point)
next_situation = (player.other, next_board.zobrist_hash())
return next_situation in self.previous_states
def is_valid_move(self, move):
if self.is_over():
return False
if move.is_pass or move.is_resign:
return True
return (
self.board.get(move.point) is None and
not self.is_move_self_capture(self.next_player, move) and
not self.does_move_violate_ko(self.next_player, move))
def is_over(self):
if self.last_move is None:
return False
if self.last_move.is_resign:
return True
if self.previous_state is None:
return False
second_last_move = self.previous_state.last_move
if second_last_move is None:
return False
return self.last_move.is_pass and second_last_move.is_pass
def legal_moves(self):
moves = []
for row in range(1, self.board.num_rows + 1):
for col in range(1, self.board.num_cols + 1):
move = Move.play(Point(row, col))
if self.is_valid_move(move):
moves.append(move)
# These two moves are always legal.
moves.append(Move.pass_turn())
moves.append(Move.resign())
return moves
def winner(self):
if not self.is_over():
return None
if self.last_move.is_resign:
return self.next_player
game_result = compute_game_result(self)
return game_result.winner
+1
View File
@@ -0,0 +1 @@
from .sgf import *
+824
View File
@@ -0,0 +1,824 @@
"""Represent SGF games.
This is intended for use with SGF FF[4]; see http://www.red-bean.com/sgf/
Adapted from gomill by Matthew Woodcraft, https://github.com/mattheww/gomill
"""
from __future__ import absolute_import
import datetime
import six
from . import sgf_grammar
from . import sgf_properties
__all__ = [
'Node',
'Sgf_game',
'Tree_node',
]
class Node:
"""An SGF node.
Instantiate with a raw property map (see sgf_grammar) and an
sgf_properties.Presenter.
A Node doesn't belong to a particular game (cf Tree_node below), but it
knows its board size (in order to interpret move values) and the encoding
to use for the raw property strings.
Changing the SZ property isn't allowed.
"""
def __init__(self, property_map, presenter):
# Map identifier (PropIdent) -> nonempty list of raw values
self._property_map = property_map
self._presenter = presenter
def get_size(self):
"""Return the board size used to interpret property values."""
return self._presenter.size
def get_encoding(self):
"""Return the encoding used for raw property values.
Returns a string (a valid Python codec name, eg "UTF-8").
"""
return self._presenter.encoding
def get_presenter(self):
"""Return the node's sgf_properties.Presenter."""
return self._presenter
def has_property(self, identifier):
"""Check whether the node has the specified property."""
return identifier in self._property_map
def properties(self):
"""Find the properties defined for the node.
Returns a list of property identifiers, in unspecified order.
"""
return list(self._property_map.keys())
def get_raw_list(self, identifier):
"""Return the raw values of the specified property.
Returns a nonempty list of 8-bit strings, in the raw property encoding.
The strings contain the exact bytes that go between the square brackets
(without interpreting escapes or performing any whitespace conversion).
Raises KeyError if there was no property with the given identifier.
(If the property is an empty elist, this returns a list containing a
single empty string.)
"""
return self._property_map[identifier]
def get_raw(self, identifier):
"""Return a single raw value of the specified property.
Returns an 8-bit string, in the raw property encoding.
The string contains the exact bytes that go between the square brackets
(without interpreting escapes or performing any whitespace conversion).
Raises KeyError if there was no property with the given identifier.
If the property has multiple values, this returns the first (if the
value is an empty elist, this returns an empty string).
"""
return self._property_map[identifier][0]
def get_raw_property_map(self):
"""Return the raw values of all properties as a dict.
Returns a dict mapping property identifiers to lists of raw values
(see get_raw_list()).
Returns the same dict each time it's called.
Treat the returned dict as read-only.
"""
return self._property_map
def _set_raw_list(self, identifier, values):
if identifier == b"SZ" and \
values != [str(self._presenter.size).encode(self._presenter.encoding)]:
raise ValueError("changing size is not permitted")
self._property_map[identifier] = values
def unset(self, identifier):
"""Remove the specified property.
Raises KeyError if the property isn't currently present.
"""
if identifier == b"SZ" and self._presenter.size != 19:
raise ValueError("changing size is not permitted")
del self._property_map[identifier]
def set_raw_list(self, identifier, values):
"""Set the raw values of the specified property.
identifier -- ascii string passing is_valid_property_identifier()
values -- nonempty iterable of 8-bit strings in the raw property
encoding
The values specify the exact bytes to appear between the square
brackets in the SGF file; you must perform any necessary escaping
first.
(To specify an empty elist, pass a list containing a single empty
string.)
"""
if not sgf_grammar.is_valid_property_identifier(identifier):
raise ValueError("ill-formed property identifier")
values = list(values)
if not values:
raise ValueError("empty property list")
for value in values:
if not sgf_grammar.is_valid_property_value(value):
raise ValueError("ill-formed raw property value")
self._set_raw_list(identifier, values)
def set_raw(self, identifier, value):
"""Set the specified property to a single raw value.
identifier -- ascii string passing is_valid_property_identifier()
value -- 8-bit string in the raw property encoding
The value specifies the exact bytes to appear between the square
brackets in the SGF file; you must perform any necessary escaping
first.
"""
if not sgf_grammar.is_valid_property_identifier(identifier):
raise ValueError("ill-formed property identifier")
if not sgf_grammar.is_valid_property_value(value):
raise ValueError("ill-formed raw property value")
self._set_raw_list(identifier, [value])
def get(self, identifier):
"""Return the interpreted value of the specified property.
Returns the value as a suitable Python representation.
Raises KeyError if the node does not have a property with the given
identifier.
Raises ValueError if it cannot interpret the value.
See sgf_properties.Presenter.interpret() for details.
"""
return self._presenter.interpret(
identifier, self._property_map[identifier])
def set(self, identifier, value):
"""Set the value of the specified property.
identifier -- ascii string passing is_valid_property_identifier()
value -- new property value (in its Python representation)
For properties with value type 'none', use value True.
Raises ValueError if it cannot represent the value.
See sgf_properties.Presenter.serialise() for details.
"""
self._set_raw_list(
identifier, self._presenter.serialise(identifier, value))
def get_raw_move(self):
"""Return the raw value of the move from a node.
Returns a pair (colour, raw value)
colour is 'b' or 'w'.
Returns None, None if the node contains no B or W property.
"""
values = self._property_map.get(b"B")
if values is not None:
colour = "b"
else:
values = self._property_map.get(b"W")
if values is not None:
colour = "w"
else:
return None, None
return colour, values[0]
def get_move(self):
"""Retrieve the move from a node.
Returns a pair (colour, move)
colour is 'b' or 'w'.
move is (row, col), or None for a pass.
Returns None, None if the node contains no B or W property.
"""
colour, raw = self.get_raw_move()
if colour is None:
return None, None
return (colour,
sgf_properties.interpret_go_point(raw, self._presenter.size))
def get_setup_stones(self):
"""Retrieve Add Black / Add White / Add Empty properties from a node.
Returns a tuple (black_points, white_points, empty_points)
Each value is a set of pairs (row, col).
"""
try:
bp = self.get(b"AB")
except KeyError:
bp = set()
try:
wp = self.get(b"AW")
except KeyError:
wp = set()
try:
ep = self.get(b"AE")
except KeyError:
ep = set()
return bp, wp, ep
def has_setup_stones(self):
"""Check whether the node has any AB/AW/AE properties."""
d = self._property_map
return (b"AB" in d or b"AW" in d or b"AE" in d)
def set_move(self, colour, move):
"""Set the B or W property.
colour -- 'b' or 'w'.
move -- (row, col), or None for a pass.
Replaces any existing B or W property in the node.
"""
if colour not in ('b', 'w'):
raise ValueError
if b'B' in self._property_map:
del self._property_map[b'B']
if b'W' in self._property_map:
del self._property_map[b'W']
self.set(colour.upper().encode('ascii'), move)
def set_setup_stones(self, black, white, empty=None):
"""Set Add Black / Add White / Add Empty properties.
black, white, empty -- list or set of pairs (row, col)
Removes any existing AB/AW/AE properties from the node.
"""
if b'AB' in self._property_map:
del self._property_map[b'AB']
if b'AW' in self._property_map:
del self._property_map[b'AW']
if b'AE' in self._property_map:
del self._property_map[b'AE']
if black:
self.set(b'AB', black)
if white:
self.set(b'AW', white)
if empty:
self.set(b'AE', empty)
def add_comment_text(self, text):
"""Add or extend the node's comment.
If the node doesn't have a C property, adds one with the specified
text.
Otherwise, adds the specified text to the existing C property value
(with two newlines in front).
"""
if self.has_property(b'C'):
self.set(b'C', self.get(b'C') + b"\n\n" + text)
else:
self.set(b'C', text)
def __str__(self):
encoding = self.get_encoding()
def format_property(ident, values):
return ident.decode(encoding) + "".join(
"[%s]" % s.decode(encoding) for s in values)
return "\n".join(
format_property(ident, values)
for (ident, values) in sorted(self._property_map.items())) \
+ "\n"
class Tree_node(Node):
"""A node embedded in an SGF game.
A Tree_node is a Node that also knows its position within an Sgf_game.
Do not instantiate directly; retrieve from an Sgf_game or another Tree_node.
A Tree_node is a list-like container of its children: it can be indexed,
sliced, and iterated over like a list, and supports index().
A Tree_node with no children is treated as having truth value false.
Public attributes (treat as read-only):
owner -- the node's Sgf_game
parent -- the nodes's parent Tree_node (None for the root node)
"""
def __init__(self, parent, properties):
self.owner = parent.owner
self.parent = parent
self._children = []
Node.__init__(self, properties, parent._presenter)
def _add_child(self, node):
self._children.append(node)
def __len__(self):
return len(self._children)
def __getitem__(self, key):
return self._children[key]
def index(self, child):
return self._children.index(child)
def new_child(self, index=None):
"""Create a new Tree_node and add it as this node's last child.
If 'index' is specified, the new node is inserted in the child list at
the specified index instead (behaves like list.insert).
Returns the new node.
"""
child = Tree_node(self, {})
if index is None:
self._children.append(child)
else:
self._children.insert(index, child)
return child
def delete(self):
"""Remove this node from its parent."""
if self.parent is None:
raise ValueError("can't remove the root node")
self.parent._children.remove(self)
def reparent(self, new_parent, index=None):
"""Move this node to a new place in the tree.
new_parent -- Tree_node from the same game.
Raises ValueError if the new parent is this node or one of its
descendants.
If 'index' is specified, the node is inserted in the new parent's child
list at the specified index (behaves like list.insert); otherwise it's
placed at the end.
"""
if new_parent.owner != self.owner:
raise ValueError("new parent doesn't belong to the same game")
n = new_parent
while True:
if n == self:
raise ValueError("would create a loop")
n = n.parent
if n is None:
break
# self.parent is not None because moving the root would create a loop.
self.parent._children.remove(self)
self.parent = new_parent
if index is None:
new_parent._children.append(self)
else:
new_parent._children.insert(index, self)
def find(self, identifier):
"""Find the nearest ancestor-or-self containing the specified property.
Returns a Tree_node, or None if there is no such node.
"""
node = self
while node is not None:
if node.has_property(identifier):
return node
node = node.parent
return None
def find_property(self, identifier):
"""Return the value of a property, defined at this node or an ancestor.
This is intended for use with properties of type 'game-info', and with
properties with the 'inherit' attribute.
This returns the interpreted value, in the same way as get().
It searches up the tree, in the same way as find().
Raises KeyError if no node defining the property is found.
"""
node = self.find(identifier)
if node is None:
raise KeyError
return node.get(identifier)
class _Root_tree_node(Tree_node):
"""Variant of Tree_node used for a game root."""
def __init__(self, property_map, owner):
self.owner = owner
self.parent = None
self._children = []
Node.__init__(self, property_map, owner.presenter)
class _Unexpanded_root_tree_node(_Root_tree_node):
"""Variant of _Root_tree_node used with 'loaded' Sgf_games."""
def __init__(self, owner, coarse_tree):
_Root_tree_node.__init__(self, coarse_tree.sequence[0], owner)
self._coarse_tree = coarse_tree
def _expand(self):
sgf_grammar.make_tree(
self._coarse_tree, self, Tree_node, Tree_node._add_child)
delattr(self, '_coarse_tree')
self.__class__ = _Root_tree_node
def __len__(self):
self._expand()
return self.__len__()
def __getitem__(self, key):
self._expand()
return self.__getitem__(key)
def index(self, child):
self._expand()
return self.index(child)
def new_child(self, index=None):
self._expand()
return self.new_child(index)
def _main_sequence_iter(self):
presenter = self._presenter
for properties in sgf_grammar.main_sequence_iter(self._coarse_tree):
yield Node(properties, presenter)
class Sgf_game:
"""An SGF game tree.
The complete game tree is represented using Tree_nodes. The various methods
which return Tree_nodes will always return the same object for the same
node.
Instantiate with
size -- int (board size), in range 1 to 26
encoding -- the raw property encoding (default "UTF-8")
'encoding' must be a valid Python codec name.
The following root node properties are set for newly-created games:
FF[4]
GM[1]
SZ[size]
CA[encoding]
Changing FF and GM is permitted (but this library will carry on using the
FF[4] and GM[1] rules). Changing SZ is not permitted (unless the change
leaves the effective value unchanged). Changing CA is permitted; this
controls the encoding used by serialise().
"""
def __new__(cls, size, encoding="UTF-8", *args, **kwargs):
# To complete initialisation after this, you need to set 'root'.
if not 1 <= size <= 26:
raise ValueError("size out of range: %s" % size)
game = super(Sgf_game, cls).__new__(cls)
game.size = size
game.presenter = sgf_properties.Presenter(size, encoding)
return game
def __init__(self, *args, **kwargs):
self.root = _Root_tree_node({}, self)
self.root.set_raw(b'FF', b"4")
self.root.set_raw(b'GM', b"1")
self.root.set_raw(b'SZ', str(self.size).encode(self.presenter.encoding))
# Read the encoding back so we get the normalised form
self.root.set_raw(b'CA', self.presenter.encoding.encode('ascii'))
@classmethod
def from_coarse_game_tree(cls, coarse_game, override_encoding=None):
"""Alternative constructor: create an Sgf_game from the parser output.
coarse_game -- Coarse_game_tree
override_encoding -- encoding name, eg "UTF-8" (optional)
The nodes' property maps (as returned by get_raw_property_map()) will
be the same dictionary objects as the ones from the Coarse_game_tree.
The board size and raw property encoding are taken from the SZ and CA
properties in the root node (defaulting to 19 and "ISO-8859-1",
respectively).
If override_encoding is specified, the source data is assumed to be in
the specified encoding (no matter what the CA property says), and the
CA property is set to match.
"""
try:
size_s = coarse_game.sequence[0][b'SZ'][0]
except KeyError:
size = 19
else:
try:
size = int(size_s)
except ValueError:
raise ValueError("bad SZ property: %s" % size_s)
if override_encoding is None:
try:
encoding = coarse_game.sequence[0][b'CA'][0]
except KeyError:
encoding = b"ISO-8859-1"
else:
encoding = override_encoding
game = cls.__new__(cls, size, encoding)
game.root = _Unexpanded_root_tree_node(game, coarse_game)
if override_encoding is not None:
game.root.set_raw(b"CA", game.presenter.encoding.encode('ascii'))
return game
@classmethod
def from_string(cls, s, override_encoding=None):
"""Alternative constructor: read a single Sgf_game from a string.
s -- 8-bit string
Raises ValueError if it can't parse the string. See parse_sgf_game()
for details.
See from_coarse_game_tree for details of size and encoding handling.
"""
if not isinstance(s, six.binary_type):
s = s.encode('ascii')
coarse_game = sgf_grammar.parse_sgf_game(s)
return cls.from_coarse_game_tree(coarse_game, override_encoding)
def serialise(self, wrap=79):
"""Serialise the SGF data as a string.
wrap -- int (default 79), or None
Returns an 8-bit string, in the encoding specified by the CA property
in the root node (defaulting to "ISO-8859-1").
If the raw property encoding and the target encoding match (which is
the usual case), the raw property values are included unchanged in the
output (even if they are improperly encoded.)
Otherwise, if any raw property value is improperly encoded,
UnicodeDecodeError is raised, and if any property value can't be
represented in the target encoding, UnicodeEncodeError is raised.
If the target encoding doesn't identify a Python codec, ValueError is
raised. Behaviour is unspecified if the target encoding isn't
ASCII-compatible (eg, UTF-16).
If 'wrap' is not None, makes some effort to keep output lines no longer
than 'wrap'.
"""
try:
encoding = self.get_charset()
except ValueError:
raise ValueError("unsupported charset: %r" %
self.root.get_raw_list(b"CA"))
coarse_tree = sgf_grammar.make_coarse_game_tree(
self.root, lambda node: node, Node.get_raw_property_map)
serialised = sgf_grammar.serialise_game_tree(coarse_tree, wrap)
if encoding == self.root.get_encoding():
return serialised
else:
return serialised.decode(self.root.get_encoding()).encode(encoding)
def get_property_presenter(self):
"""Return the property presenter.
Returns an sgf_properties.Presenter.
This can be used to customise how property values are interpreted and
serialised.
"""
return self.presenter
def get_root(self):
"""Return the root node (as a Tree_node)."""
return self.root
def get_last_node(self):
"""Return the last node in the 'leftmost' variation (as a Tree_node)."""
node = self.root
while node:
node = node[0]
return node
def get_main_sequence(self):
"""Return the 'leftmost' variation.
Returns a list of Tree_nodes, from the root to a leaf.
"""
node = self.root
result = [node]
while node:
node = node[0]
result.append(node)
return result
def get_main_sequence_below(self, node):
"""Return the 'leftmost' variation below the specified node.
node -- Tree_node
Returns a list of Tree_nodes, from the first child of 'node' to a leaf.
"""
if node.owner is not self:
raise ValueError("node doesn't belong to this game")
result = []
while node:
node = node[0]
result.append(node)
return result
def get_sequence_above(self, node):
"""Return the partial variation leading to the specified node.
node -- Tree_node
Returns a list of Tree_nodes, from the root to the parent of 'node'.
"""
if node.owner is not self:
raise ValueError("node doesn't belong to this game")
result = []
while node.parent is not None:
node = node.parent
result.append(node)
result.reverse()
return result
def main_sequence_iter(self):
"""Provide the 'leftmost' variation as an iterator.
Returns an iterator providing Node instances, from the root to a leaf.
The Node instances may or may not be Tree_nodes.
It's OK to use these Node instances to modify properties: even if they
are not the same objects as returned by the main tree navigation
methods, they share the underlying property maps.
If you know the game has no variations, or you're only interested in
the 'leftmost' variation, you can use this function to retrieve the
nodes without building the entire game tree.
"""
if isinstance(self.root, _Unexpanded_root_tree_node):
return self.root._main_sequence_iter()
return iter(self.get_main_sequence())
def extend_main_sequence(self):
"""Create a new Tree_node and add to the 'leftmost' variation.
Returns the new node.
"""
return self.get_last_node().new_child()
def get_size(self):
"""Return the board size as an integer."""
return self.size
def get_charset(self):
"""Return the effective value of the CA root property.
This applies the default, and returns the normalised form.
Raises ValueError if the CA property doesn't identify a Python codec.
"""
try:
s = self.root.get(b"CA")
except KeyError:
return "ISO-8859-1"
try:
return sgf_properties.normalise_charset_name(s)
except LookupError:
raise ValueError("no codec available for CA %s" % s)
def get_komi(self):
"""Return the komi as a float.
Returns 0.0 if the KM property isn't present in the root node.
Raises ValueError if the KM property is malformed.
"""
try:
return self.root.get(b"KM")
except KeyError:
return 0.0
def get_handicap(self):
"""Return the number of handicap stones as a small integer.
Returns None if the HA property isn't present, or has (illegal) value
zero.
Raises ValueError if the HA property is otherwise malformed.
"""
try:
handicap = self.root.get(b"HA")
except KeyError:
return None
if handicap == 0:
handicap = None
elif handicap == 1:
raise ValueError
return handicap
def get_player_name(self, colour):
"""Return the name of the specified player.
Returns None if there is no corresponding 'PB' or 'PW' property.
"""
try:
return self.root.get(
{'b': b'PB', 'w': b'PW'}[colour]).decode(self.presenter.encoding)
except KeyError:
return None
def get_winner(self):
"""Return the colour of the winning player.
Returns None if there is no RE property, or if neither player won.
"""
try:
colour = self.root.get(b"RE").decode(self.presenter.encoding)[0].lower()
except LookupError:
return None
if colour not in ("b", "w"):
return None
return colour
def set_date(self, date=None):
"""Set the DT property to a single date.
date -- datetime.date (defaults to today)
(SGF allows dates to be rather more complicated than this, so there's
no corresponding get_date() method.)
"""
if date is None:
date = datetime.date.today()
self.root.set('DT', date.strftime("%Y-%m-%d"))
+533
View File
@@ -0,0 +1,533 @@
"""Parse and serialise SGF data.
This is intended for use with SGF FF[4]; see http://www.red-bean.com/sgf/
Nothing in this module is Go-specific.
This module is encoding-agnostic: it works with 8-bit strings in an arbitrary
'ascii-compatible' encoding.
In the documentation below, a _property map_ is a dict mapping a PropIdent to a
nonempty list of raw property values.
A raw property value is an 8-bit string containing a PropValue without its
enclosing brackets, but with backslashes and line endings left untouched.
So a property map's keys should pass is_valid_property_identifier(), and its
values should pass is_valid_property_value().
Adapted from gomill by Matthew Woodcraft, https://github.com/mattheww/gomill
"""
from __future__ import absolute_import
import re
import string
import six
_propident_re = re.compile(r"\A[A-Z]{1,8}\Z".encode('ascii'))
_propvalue_re = re.compile(r"\A [^\\\]]* (?: \\. [^\\\]]* )* \Z".encode('ascii'),
re.VERBOSE | re.DOTALL)
_find_start_re = re.compile(r"\(\s*;".encode('ascii'))
_tokenise_re = re.compile(r"""
\s*
(?:
\[ (?P<V> [^\\\]]* (?: \\. [^\\\]]* )* ) \] # PropValue
|
(?P<I> [A-Z]{1,8} ) # PropIdent
|
(?P<D> [;()] ) # delimiter
)
""".encode('ascii'), re.VERBOSE | re.DOTALL)
def is_valid_property_identifier(s):
"""Check whether 's' is a well-formed PropIdent.
s -- 8-bit string
This accepts the same values as the tokeniser.
Details:
- it doesn't permit lower-case letters (these are allowed in some ancient
SGF variants)
- it accepts at most 8 letters (there is no limit in the spec; no standard
property has more than 2)
"""
return bool(_propident_re.search(s))
def is_valid_property_value(s):
"""Check whether 's' is a well-formed PropValue.
s -- 8-bit string
This accepts the same values as the tokeniser: any string that doesn't
contain an unescaped ] or end with an unescaped \ .
"""
return bool(_propvalue_re.search(s))
def tokenise(s, start_position=0):
"""Tokenise a string containing SGF data.
s -- 8-bit string
start_position -- index into 's'
Skips leading junk.
Returns a list of pairs of strings (token type, contents), and also the
index in 's' of the start of the unprocessed 'tail'.
token types and contents:
I -- PropIdent: upper-case letters
V -- PropValue: raw value, without the enclosing brackets
D -- delimiter: ';', '(', or ')'
Stops when it has seen as many closing parens as open ones, at the end of
the string, or when it first finds something it can't tokenise.
The first two tokens are always '(' and ';' (otherwise it won't find the
start of the content).
"""
result = []
m = _find_start_re.search(s, start_position)
if not m:
return [], 0
i = m.start()
depth = 0
while True:
m = _tokenise_re.match(s, i)
if not m:
break
group = m.lastgroup
token = m.group(m.lastindex)
result.append((group, token))
i = m.end()
if group == 'D':
if token == b'(':
depth += 1
elif token == b')':
depth -= 1
if depth == 0:
break
return result, i
class Coarse_game_tree:
"""An SGF GameTree.
This is a direct representation of the SGF parse tree. It's 'coarse' in the
sense that the objects in the tree structure represent node sequences, not
individual nodes.
Public attributes
sequence -- nonempty list of property maps
children -- list of Coarse_game_trees
The sequence represents the nodes before the variations.
"""
def __init__(self):
self.sequence = [] # must be at least one node
self.children = [] # may be empty
def _parse_sgf_game(s, start_position):
"""Common implementation for parse_sgf_game and parse_sgf_games."""
tokens, end_position = tokenise(s, start_position)
if not tokens:
return None, None
stack = []
game_tree = None
sequence = None
properties = None
index = 0
try:
while True:
token_type, token = tokens[index]
index += 1
if token_type == 'V':
raise ValueError("unexpected value")
if token_type == 'D':
if token == b';':
if sequence is None:
raise ValueError("unexpected node")
properties = {}
sequence.append(properties)
else:
if sequence is not None:
if not sequence:
raise ValueError("empty sequence")
game_tree.sequence = sequence
sequence = None
if token == b'(':
stack.append(game_tree)
game_tree = Coarse_game_tree()
sequence = []
else:
# token == ')'
variation = game_tree
game_tree = stack.pop()
if game_tree is None:
break
game_tree.children.append(variation)
properties = None
else:
# token_type == 'I'
prop_ident = token
prop_values = []
while True:
token_type, token = tokens[index]
if token_type != 'V':
break
index += 1
prop_values.append(token)
if not prop_values:
raise ValueError("property with no values")
try:
if prop_ident in properties:
properties[prop_ident] += prop_values
else:
properties[prop_ident] = prop_values
except TypeError:
raise ValueError("property value outside a node")
except IndexError:
raise ValueError("unexpected end of SGF data")
assert index == len(tokens)
return variation, end_position
def parse_sgf_game(s):
"""Read a single SGF game from a string, returning the parse tree.
s -- 8-bit string
Returns a Coarse_game_tree.
Applies the rules for FF[4].
Raises ValueError if can't parse the string.
If a property appears more than once in a node (which is not permitted by
the spec), treats it the same as a single property with multiple values.
Identifies the start of the SGF content by looking for '(;' (with possible
whitespace between); ignores everything preceding that. Ignores everything
following the first game.
"""
game_tree, _ = _parse_sgf_game(s, 0)
if game_tree is None:
raise ValueError("no SGF data found")
return game_tree
def parse_sgf_collection(s):
"""Read an SGF game collection, returning the parse trees.
s -- 8-bit string
Returns a nonempty list of Coarse_game_trees.
Raises ValueError if no games were found in the string.
Raises ValueError if there is an error parsing a game. See
parse_sgf_game() for details.
Ignores non-SGF data before the first game, between games, and after the
final game. Identifies the start of each game in the same way as
parse_sgf_game().
"""
position = 0
result = []
while True:
try:
game_tree, position = _parse_sgf_game(s, position)
except ValueError as e:
raise ValueError("error parsing game %d: %s" % (len(result), e))
if game_tree is None:
break
result.append(game_tree)
if not result:
raise ValueError("no SGF data found")
return result
def block_format(pieces, width=79):
"""Concatenate bytestrings, adding newlines.
pieces -- iterable of strings
width -- int (default 79)
Returns "".join(pieces), with added newlines between pieces as necessary to
avoid lines longer than 'width'.
Leaves newlines inside 'pieces' untouched, and ignores them in its width
calculation. If a single piece is longer than 'width', it will become a
single long line in the output.
"""
lines = []
line = b""
for s in pieces:
if len(line) + len(s) > width:
lines.append(line)
line = b""
line += s
if line:
lines.append(line)
return b"\n".join(lines)
def serialise_game_tree(game_tree, wrap=79):
"""Serialise an SGF game as a string.
game_tree -- Coarse_game_tree
wrap -- int (default 79), or None
Returns an 8-bit string, ending with a newline.
If 'wrap' is not None, makes some effort to keep output lines no longer
than 'wrap'.
"""
l = []
to_serialise = [game_tree]
while to_serialise:
game_tree = to_serialise.pop()
if game_tree is None:
l.append(b")")
continue
l.append(b"(")
for properties in game_tree.sequence:
l.append(b";")
# Force FF to the front, largely to work around a Quarry bug which
# makes it ignore the first few bytes of the file.
for prop_ident, prop_values in sorted(
list(properties.items()),
key=lambda pair: (-(pair[0] == b"FF"), pair[0])):
# Make a single string for each property, to get prettier
# block_format output.
m = [prop_ident]
for value in prop_values:
m.append(b"[" + value + b"]")
l.append(b"".join(m))
to_serialise.append(None)
to_serialise.extend(reversed(game_tree.children))
l.append(b"\n")
if wrap is None:
return b"".join(l)
else:
return block_format(l, wrap)
def make_tree(game_tree, root, node_builder, node_adder):
"""Construct a node tree from a Coarse_game_tree.
game_tree -- Coarse_game_tree
root -- node
node_builder -- function taking parameters (parent node, property map)
returning a node
node_adder -- function taking a pair (parent node, child node)
Builds a tree of nodes corresponding to this GameTree, calling
node_builder() to make new nodes and node_adder() to add child nodes to
their parent.
Makes no further assumptions about the node type.
"""
to_build = [(root, game_tree, 0)]
while to_build:
node, game_tree, index = to_build.pop()
if index < len(game_tree.sequence) - 1:
child = node_builder(node, game_tree.sequence[index + 1])
node_adder(node, child)
to_build.append((child, game_tree, index + 1))
else:
node._children = []
for child_tree in game_tree.children:
child = node_builder(node, child_tree.sequence[0])
node_adder(node, child)
to_build.append((child, child_tree, 0))
def make_coarse_game_tree(root, get_children, get_properties):
"""Construct a Coarse_game_tree from a node tree.
root -- node
get_children -- function taking a node, returning a sequence of nodes
get_properties -- function taking a node, returning a property map
Returns a Coarse_game_tree.
Walks the node tree based at 'root' using get_children(), and uses
get_properties() to extract the raw properties.
Makes no further assumptions about the node type.
Doesn't check that the property maps have well-formed keys and values.
"""
result = Coarse_game_tree()
to_serialise = [(result, root)]
while to_serialise:
game_tree, node = to_serialise.pop()
while True:
game_tree.sequence.append(get_properties(node))
children = get_children(node)
if len(children) != 1:
break
node = children[0]
for child in children:
child_tree = Coarse_game_tree()
game_tree.children.append(child_tree)
to_serialise.append((child_tree, child))
return result
def main_sequence_iter(game_tree):
"""Provide the 'leftmost' complete sequence of a Coarse_game_tree.
game_tree -- Coarse_game_tree
Returns an iterable of property maps.
If the game has no variations, this provides the complete game. Otherwise,
it chooses the first variation each time it has a choice.
"""
while True:
for properties in game_tree.sequence:
yield properties
if not game_tree.children:
break
game_tree = game_tree.children[0]
_split_compose_re = re.compile(
r"( (?: [^\\:] | \\. )* ) :".encode('ascii'),
re.VERBOSE | re.DOTALL)
def parse_compose(s):
"""Split the parts of an SGF Compose value.
If the value is a well-formed Compose, returns a pair of strings.
If it isn't (ie, there is no delimiter), returns the complete string and
None.
Interprets backslash escapes in order to find the delimiter, but leaves
backslash escapes unchanged in the returned strings.
"""
m = _split_compose_re.match(s)
if not m:
return s, None
return m.group(1), s[m.end():]
def compose(s1, s2):
"""Construct a value of Compose value type.
s1, s2 -- serialised form of a property value
(This is only needed if the type of the first value permits colons.)
"""
return s1.replace(b":", b"\\:") + b":" + s2
_newline_re = re.compile(r"\n\r|\r\n|\n|\r".encode('ascii'))
if six.PY2:
_binary_maketrans = string.maketrans
else:
_binary_maketrans = bytes.maketrans
_whitespace_table = _binary_maketrans(b"\t\f\v", b" ")
_chunk_re = re.compile(r" [^\n\\]+ | [\n\\] ".encode('ascii'), re.VERBOSE)
def simpletext_value(s):
"""Convert a raw SimpleText property value to the string it represents.
Returns an 8-bit string, in the encoding of the original SGF string.
This interprets escape characters, and does whitespace mapping:
- backslash followed by linebreak (LF, CR, LFCR, or CRLF) disappears
- any other linebreak is replaced by a space
- any other whitespace character is replaced by a space
- other backslashes disappear (but double-backslash -> single-backslash)
"""
s = _newline_re.sub(b"\n", s)
s = s.translate(_whitespace_table)
is_escaped = False
result = []
for chunk in _chunk_re.findall(s):
if is_escaped:
if chunk != b"\n":
result.append(chunk)
is_escaped = False
elif chunk == b"\\":
is_escaped = True
elif chunk == b"\n":
result.append(b" ")
else:
result.append(chunk)
return b"".join(result)
def text_value(s):
"""Convert a raw Text property value to the string it represents.
Returns an 8-bit string, in the encoding of the original SGF string.
This interprets escape characters, and does whitespace mapping:
- linebreak (LF, CR, LFCR, or CRLF) is converted to \n
- any other whitespace character is replaced by a space
- backslash followed by linebreak disappears
- other backslashes disappear (but double-backslash -> single-backslash)
"""
s = _newline_re.sub(b"\n", s)
s = s.translate(_whitespace_table)
is_escaped = False
result = []
for chunk in _chunk_re.findall(s):
if is_escaped:
if chunk != b"\n":
result.append(chunk)
is_escaped = False
elif chunk == b"\\":
is_escaped = True
else:
result.append(chunk)
return b"".join(result)
def escape_text(s):
"""Convert a string to a raw Text property value that represents it.
s -- 8-bit string, in the desired output encoding.
Returns an 8-bit string which passes is_valid_property_value().
Normally text_value(escape_text(s)) == s, but there are the following
exceptions:
- all linebreaks are are normalised to \n
- whitespace other than line breaks is converted to a single space
"""
return s.replace(b"\\", b"\\\\").replace(b"]", b"\\]")
+763
View File
@@ -0,0 +1,763 @@
"""Interpret SGF property values.
This is intended for use with SGF FF[4]; see http://www.red-bean.com/sgf/
This supports all general properties and Go-specific properties, but not
properties for other games. Point, Move and Stone values are interpreted as Go
points.
Adapted from gomill by Matthew Woodcraft, https://github.com/mattheww/gomill
"""
from __future__ import absolute_import
import codecs
from math import isinf, isnan
import six
from tugo.gosgf import sgf_grammar
from six.moves import range
# In python 2, indexing a str gives one-character strings.
# In python 3, indexing a bytes gives ints.
if six.PY2:
_bytestring_ord = ord
else:
def identity(x):
return x
_bytestring_ord = identity
def normalise_charset_name(s):
"""Convert an encoding name to the form implied in the SGF spec.
In particular, normalises to 'ISO-8859-1' and 'UTF-8'.
Raises LookupError if the encoding name isn't known to Python.
"""
if not isinstance(s, six.text_type):
s = s.decode('ascii')
return (codecs.lookup(s).name.replace("_", "-").upper()
.replace("ISO8859", "ISO-8859"))
def interpret_go_point(s, size):
"""Convert a raw SGF Go Point, Move, or Stone value to coordinates.
s -- 8-bit string
size -- board size (int)
Returns a pair (row, col), or None for a pass.
Raises ValueError if the string is malformed or the coordinates are out of
range.
Only supports board sizes up to 26.
The returned coordinates are in the GTP coordinate system (as in the rest
of gomill), where (0, 0) is the lower left.
"""
if s == b"" or (s == b"tt" and size <= 19):
return None
# May propagate ValueError
col_s, row_s = s
col = _bytestring_ord(col_s) - 97 # 97 == ord("a")
row = size - _bytestring_ord(row_s) + 96
if not ((0 <= col < size) and (0 <= row < size)):
raise ValueError
return row, col
def serialise_go_point(move, size):
"""Serialise a Go Point, Move, or Stone value.
move -- pair (row, col), or None for a pass
Returns an 8-bit string.
Only supports board sizes up to 26.
The move coordinates are in the GTP coordinate system (as in the rest of
gomill), where (0, 0) is the lower left.
"""
if not 1 <= size <= 26:
raise ValueError
if move is None:
# Prefer 'tt' where possible, for the sake of older code
if size <= 19:
return b"tt"
else:
return b""
row, col = move
if not ((0 <= col < size) and (0 <= row < size)):
raise ValueError
col_s = "abcdefghijklmnopqrstuvwxy"[col].encode('ascii')
row_s = "abcdefghijklmnopqrstuvwxy"[size - row - 1].encode('ascii')
return col_s + row_s
class _Context:
def __init__(self, size, encoding):
self.size = size
self.encoding = encoding
def interpret_none(s, context=None):
"""Convert a raw None value to a boolean.
That is, unconditionally returns True.
"""
return True
def serialise_none(b, context=None):
"""Serialise a None value.
Ignores its parameter.
"""
return b""
def interpret_number(s, context=None):
"""Convert a raw Number value to the integer it represents.
This is a little more lenient than the SGF spec: it permits leading and
trailing spaces, and spaces between the sign and the numerals.
"""
return int(s, 10)
def serialise_number(i, context=None):
"""Serialise a Number value.
i -- integer
"""
return ("%d" % i).encode('ascii')
def interpret_real(s, context=None):
"""Convert a raw Real value to the float it represents.
This is more lenient than the SGF spec: it accepts strings accepted as a
float by the platform libc. It rejects infinities and NaNs.
"""
result = float(s)
if isinf(result):
raise ValueError("infinite")
if isnan(result):
raise ValueError("not a number")
return result
def serialise_real(f, context=None):
"""Serialise a Real value.
f -- real number (int or float)
If the absolute value is too small to conveniently express as a decimal,
returns "0" (this currently happens if abs(f) is less than 0.0001).
"""
f = float(f)
try:
i = int(f)
except OverflowError:
# infinity
raise ValueError
if f == i:
# avoid trailing '.0'; also avoid scientific notation for large numbers
return str(i).encode('ascii')
s = repr(f)
if 'e-' in s:
return "0".encode('ascii')
return s.encode('ascii')
def interpret_double(s, context=None):
"""Convert a raw Double value to an integer.
Returns 1 or 2 (unknown values are treated as 1).
"""
if s.strip() == b"2":
return 2
else:
return 1
def serialise_double(i, context=None):
"""Serialise a Double value.
i -- integer (1 or 2)
(unknown values are treated as 1)
"""
if i == 2:
return "2"
return "1"
def interpret_colour(s, context=None):
"""Convert a raw Color value to a gomill colour.
Returns 'b' or 'w'.
"""
colour = s.decode('ascii').lower()
if colour not in ('b', 'w'):
raise ValueError
return colour
def serialise_colour(colour, context=None):
"""Serialise a Colour value.
colour -- 'b' or 'w'
"""
if colour not in ('b', 'w'):
raise ValueError
return colour.upper().encode('ascii')
def _transcode(s, encoding):
"""Common implementation for interpret_text and interpret_simpletext."""
# If encoding is UTF-8, we don't need to transcode, but we still want to
# report an error if it's not properly encoded.
u = s.decode(encoding)
if encoding == "UTF-8":
return s
else:
return u.encode("utf-8")
def interpret_simpletext(s, context):
"""Convert a raw SimpleText value to a string.
See sgf_grammar.simpletext_value() for details.
s -- raw value
Returns an 8-bit utf-8 string.
"""
return _transcode(sgf_grammar.simpletext_value(s), context.encoding)
def serialise_simpletext(s, context):
"""Serialise a SimpleText value.
See sgf_grammar.escape_text() for details.
s -- 8-bit utf-8 string
"""
if context.encoding != "UTF-8":
s = s.decode("utf-8").encode(context.encoding)
return sgf_grammar.escape_text(s)
def interpret_text(s, context):
"""Convert a raw Text value to a string.
See sgf_grammar.text_value() for details.
s -- raw value
Returns an 8-bit utf-8 string.
"""
return _transcode(sgf_grammar.text_value(s), context.encoding)
def serialise_text(s, context):
"""Serialise a Text value.
See sgf_grammar.escape_text() for details.
s -- 8-bit utf-8 string
"""
if context.encoding != "UTF-8":
s = s.decode("utf-8").encode(context.encoding)
return sgf_grammar.escape_text(s)
def interpret_point(s, context):
"""Convert a raw SGF Point or Stone value to coordinates.
See interpret_go_point() above for details.
Returns a pair (row, col).
"""
result = interpret_go_point(s, context.size)
if result is None:
raise ValueError
return result
def serialise_point(point, context):
"""Serialise a Point or Stone value.
point -- pair (row, col)
See serialise_go_point() above for details.
"""
if point is None:
raise ValueError
return serialise_go_point(point, context.size)
def interpret_move(s, context):
"""Convert a raw SGF Move value to coordinates.
See interpret_go_point() above for details.
Returns a pair (row, col), or None for a pass.
"""
return interpret_go_point(s, context.size)
def serialise_move(move, context):
"""Serialise a Move value.
move -- pair (row, col), or None for a pass
See serialise_go_point() above for details.
"""
return serialise_go_point(move, context.size)
def interpret_point_list(values, context):
"""Convert a raw SGF list of Points to a set of coordinates.
values -- list of strings
Returns a set of pairs (row, col).
If 'values' is empty, returns an empty set.
This interprets compressed point lists.
Doesn't complain if there is overlap, or if a single point is specified as
a 1x1 rectangle.
Raises ValueError if the data is otherwise malformed.
"""
result = set()
for s in values:
# No need to use parse_compose(), as \: would always be an error.
p1, is_rectangle, p2 = s.partition(b":")
if is_rectangle:
top, left = interpret_point(p1, context)
bottom, right = interpret_point(p2, context)
if not (bottom <= top and left <= right):
raise ValueError
for row in range(bottom, top + 1):
for col in range(left, right + 1):
result.add((row, col))
else:
pt = interpret_point(p1, context)
result.add(pt)
return result
def serialise_point_list(points, context):
"""Serialise a list of Points, Moves, or Stones.
points -- iterable of pairs (row, col)
Returns a list of strings.
If 'points' is empty, returns an empty list.
Doesn't produce a compressed point list.
"""
result = [serialise_point(point, context) for point in points]
result.sort()
return result
def interpret_AP(s, context):
"""Interpret an AP (application) property value.
Returns a pair of strings (name, version number)
Permits the version number to be missing (which is forbidden by the SGF
spec), in which case the second returned value is an empty string.
"""
application, version = sgf_grammar.parse_compose(s)
if version is None:
version = b""
return (interpret_simpletext(application, context),
interpret_simpletext(version, context))
def serialise_AP(value, context):
"""Serialise an AP (application) property value.
value -- pair (application, version)
application -- string
version -- string
Note this takes a single parameter (which is a pair).
"""
application, version = value
return sgf_grammar.compose(serialise_simpletext(application, context),
serialise_simpletext(version, context))
def interpret_ARLN_list(values, context):
"""Interpret an AR (arrow) or LN (line) property value.
Returns a list of pairs (point, point), where point is a pair (row, col)
"""
result = []
for s in values:
p1, p2 = sgf_grammar.parse_compose(s)
result.append((interpret_point(p1, context),
interpret_point(p2, context)))
return result
def serialise_ARLN_list(values, context):
"""Serialise an AR (arrow) or LN (line) property value.
values -- list of pairs (point, point), where point is a pair (row, col)
"""
return [b":".join((serialise_point(p1, context), serialise_point(p2, context)))
for p1, p2 in values]
def interpret_FG(s, context):
"""Interpret an FG (figure) property value.
Returns a pair (flags, string), or None.
flags is an integer; see http://www.red-bean.com/sgf/properties.html#FG
"""
if s == b"":
return None
flags, name = sgf_grammar.parse_compose(s)
return int(flags), interpret_simpletext(name, context)
def serialise_FG(value, context):
"""Serialise an FG (figure) property value.
value -- pair (flags, name), or None
flags -- int
name -- string
Use serialise_FG(None) to produce an empty value.
"""
if value is None:
return b""
flags, name = value
return str(flags).encode('ascii') + b":" + serialise_simpletext(name, context)
def interpret_LB_list(values, context):
"""Interpret an LB (label) property value.
Returns a list of pairs ((row, col), string).
"""
result = []
for s in values:
point, label = sgf_grammar.parse_compose(s)
result.append((interpret_point(point, context),
interpret_simpletext(label, context)))
return result
def serialise_LB_list(values, context):
"""Serialise an LB (label) property value.
values -- list of pairs ((row, col), string)
"""
return [b":".join((serialise_point(point, context), serialise_simpletext(text, context)))
for point, text in values]
class Property_type:
"""Description of a property type."""
def __init__(self, interpreter, serialiser, uses_list,
allows_empty_list=False):
self.interpreter = interpreter
self.serialiser = serialiser
self.uses_list = bool(uses_list)
self.allows_empty_list = bool(allows_empty_list)
def _make_property_type(type_name, allows_empty_list=False):
return Property_type(
globals()["interpret_" + type_name],
globals()["serialise_" + type_name],
uses_list=(type_name.endswith("_list")),
allows_empty_list=allows_empty_list)
_property_types_by_name = {
'none': _make_property_type('none'),
'number': _make_property_type('number'),
'real': _make_property_type('real'),
'double': _make_property_type('double'),
'colour': _make_property_type('colour'),
'simpletext': _make_property_type('simpletext'),
'text': _make_property_type('text'),
'point': _make_property_type('point'),
'move': _make_property_type('move'),
'point_list': _make_property_type('point_list'),
'point_elist': _make_property_type('point_list', allows_empty_list=True),
'stone_list': _make_property_type('point_list'),
'AP': _make_property_type('AP'),
'ARLN_list': _make_property_type('ARLN_list'),
'FG': _make_property_type('FG'),
'LB_list': _make_property_type('LB_list'),
}
P = _property_types_by_name
_property_types_by_ident = {
b'AB': P['stone_list'], # setup Add Black
b'AE': P['point_list'], # setup Add Empty
b'AN': P['simpletext'], # game-info Annotation
b'AP': P['AP'], # root Application
b'AR': P['ARLN_list'], # - Arrow
b'AW': P['stone_list'], # setup Add White
b'B': P['move'], # move Black
b'BL': P['real'], # move Black time left
b'BM': P['double'], # move Bad move
b'BR': P['simpletext'], # game-info Black rank
b'BT': P['simpletext'], # game-info Black team
b'C': P['text'], # - Comment
b'CA': P['simpletext'], # root Charset
b'CP': P['simpletext'], # game-info Copyright
b'CR': P['point_list'], # - Circle
b'DD': P['point_elist'], # - [inherit] Dim points
b'DM': P['double'], # - Even position
b'DO': P['none'], # move Doubtful
b'DT': P['simpletext'], # game-info Date
b'EV': P['simpletext'], # game-info Event
b'FF': P['number'], # root Fileformat
b'FG': P['FG'], # - Figure
b'GB': P['double'], # - Good for Black
b'GC': P['text'], # game-info Game comment
b'GM': P['number'], # root Game
b'GN': P['simpletext'], # game-info Game name
b'GW': P['double'], # - Good for White
b'HA': P['number'], # game-info Handicap
b'HO': P['double'], # - Hotspot
b'IT': P['none'], # move Interesting
b'KM': P['real'], # game-info Komi
b'KO': P['none'], # move Ko
b'LB': P['LB_list'], # - Label
b'LN': P['ARLN_list'], # - Line
b'MA': P['point_list'], # - Mark
b'MN': P['number'], # move set move number
b'N': P['simpletext'], # - Nodename
b'OB': P['number'], # move OtStones Black
b'ON': P['simpletext'], # game-info Opening
b'OT': P['simpletext'], # game-info Overtime
b'OW': P['number'], # move OtStones White
b'PB': P['simpletext'], # game-info Player Black
b'PC': P['simpletext'], # game-info Place
b'PL': P['colour'], # setup Player to play
b'PM': P['number'], # - [inherit] Print move mode
b'PW': P['simpletext'], # game-info Player White
b'RE': P['simpletext'], # game-info Result
b'RO': P['simpletext'], # game-info Round
b'RU': P['simpletext'], # game-info Rules
b'SL': P['point_list'], # - Selected
b'SO': P['simpletext'], # game-info Source
b'SQ': P['point_list'], # - Square
b'ST': P['number'], # root Style
b'SZ': P['number'], # root Size
b'TB': P['point_elist'], # - Territory Black
b'TE': P['double'], # move Tesuji
b'TM': P['real'], # game-info Timelimit
b'TR': P['point_list'], # - Triangle
b'TW': P['point_elist'], # - Territory White
b'UC': P['double'], # - Unclear pos
b'US': P['simpletext'], # game-info User
b'V': P['real'], # - Value
b'VW': P['point_elist'], # - [inherit] View
b'W': P['move'], # move White
b'WL': P['real'], # move White time left
b'WR': P['simpletext'], # game-info White rank
b'WT': P['simpletext'], # game-info White team
}
_text_property_type = P['text']
del P
class Presenter(_Context):
"""Convert property values between Python and SGF-string representations.
Instantiate with:
size -- board size (int)
encoding -- encoding for the SGF strings
Public attributes (treat as read-only):
size -- int
encoding -- string (normalised form)
See the _property_types_by_ident table above for a list of properties
initially known, and their types.
Initially, treats unknown (private) properties as if they had type Text.
"""
def __init__(self, size, encoding):
try:
encoding = normalise_charset_name(encoding)
except LookupError:
raise ValueError("unknown encoding: %s" % encoding)
_Context.__init__(self, size, encoding)
self.property_types_by_ident = _property_types_by_ident.copy()
self.default_property_type = _text_property_type
def get_property_type(self, identifier):
"""Return the Property_type for the specified PropIdent.
Rasies KeyError if the property is unknown.
"""
return self.property_types_by_ident[identifier]
def register_property(self, identifier, property_type):
"""Specify the Property_type for a PropIdent."""
self.property_types_by_ident[identifier] = property_type
def deregister_property(self, identifier):
"""Forget the type for the specified PropIdent."""
del self.property_types_by_ident[identifier]
def set_private_property_type(self, property_type):
"""Specify the Property_type to use for unknown properties.
Pass property_type = None to make unknown properties raise an error.
"""
self.default_property_type = property_type
def _get_effective_property_type(self, identifier):
try:
return self.property_types_by_ident[identifier]
except KeyError:
result = self.default_property_type
if result is None:
raise ValueError("unknown property")
return result
def interpret_as_type(self, property_type, raw_values):
"""Variant of interpret() for explicitly specified type.
property_type -- Property_type
"""
if not raw_values:
raise ValueError("no raw values")
if property_type.uses_list:
if raw_values == [b""]:
raw = []
else:
raw = raw_values
else:
if len(raw_values) > 1:
raise ValueError("multiple values")
raw = raw_values[0]
return property_type.interpreter(raw, self)
def interpret(self, identifier, raw_values):
"""Return a Python representation of a property value.
identifier -- PropIdent
raw_values -- nonempty list of 8-bit strings in the presenter's encoding
See the interpret_... functions above for details of how values are
represented as Python types.
Raises ValueError if it cannot interpret the value.
Note that in some cases the interpret_... functions accept values which
are not strictly permitted by the specification.
elist handling: if the property's value type is a list type and
'raw_values' is a list containing a single empty string, passes an
empty list to the interpret_... function (that is, this function treats
all lists like elists).
Doesn't enforce range restrictions on values with type Number.
"""
return self.interpret_as_type(
self._get_effective_property_type(identifier), raw_values)
def serialise_as_type(self, property_type, value):
"""Variant of serialise() for explicitly specified type.
property_type -- Property_type
"""
serialised = property_type.serialiser(value, self)
if property_type.uses_list:
if serialised == []:
if property_type.allows_empty_list:
return [b""]
else:
raise ValueError("empty list")
return serialised
else:
return [serialised]
def serialise(self, identifier, value):
"""Serialise a Python representation of a property value.
identifier -- PropIdent
value -- corresponding Python value
Returns a nonempty list of 8-bit strings in the presenter's encoding,
suitable for use as raw PropValues.
See the serialise_... functions above for details of the acceptable
values for each type.
elist handling: if the property's value type is an elist type and the
serialise_... function returns an empty list, this returns a list
containing a single empty string.
Raises ValueError if it cannot serialise the value.
In general, the serialise_... functions try not to produce an invalid
result, but do not try to prevent garbage input happening to produce a
valid result.
"""
return self.serialise_as_type(
self._get_effective_property_type(identifier), value)
+37
View File
@@ -0,0 +1,37 @@
# tag::enumimport[]
import enum
# end::enumimport[]
# tag::namedtuple[]
from collections import namedtuple
# end::namedtuple[]
__all__ = [
'Player',
'Point',
]
# tag::color[]
class Player(enum.Enum):
black = 1
white = 2
@property
def other(self):
return Player.black if self == Player.white else Player.white
# end::color[]
# tag::points[]
class Point(namedtuple('Point', 'row col')):
def neighbors(self):
return [
Point(self.row - 1, self.col),
Point(self.row + 1, self.col),
Point(self.row, self.col - 1),
Point(self.row, self.col + 1),
]
# end::points[]
def __deepcopy__(self, memodict={}):
# These are very immutable.
return self
+30
View File
@@ -0,0 +1,30 @@
import tempfile
import os
import torch
def save_model_to_hdf5_group(model, f):
tempfd, tempfname = tempfile.mkstemp(prefix='tmp-torchmodel')
try:
os.close(tempfd)
torch.save(model, tempfname)
with open(tempfname, 'rb') as model_file:
model_data = model_file.read()
f.create_dataset('torchmodel', data=model_data)
finally:
os.unlink(tempfname)
def load_model_from_hdf5_group(f, map_location=None):
tempfd, tempfname = tempfile.mkstemp(prefix='tmp-torchmodel')
try:
os.close(tempfd)
with open(tempfname, 'wb') as model_file:
model_file.write(f['torchmodel'][()])
model = torch.load(tempfname, map_location=map_location)
return model
finally:
os.unlink(tempfname)
def set_gpu_memory_target(device, frac):
# This function is not required in PyTorch since it doesn't pre-allocate all GPU memory by default.
pass
Binary file not shown.
+6
View File
@@ -0,0 +1,6 @@
#from .fullyconnected import *
#from .large import *
#from .leaky import *
#from .medium import *
#from .small import *
from .alphago import *
+52
View File
@@ -0,0 +1,52 @@
import torch
import torch.nn as nn
class AlphaGoModel(nn.Module):
def __init__(self, input_shape, is_policy_net=False,
# num_filters=192, first_kernel_size=5, other_kernel_size=3):
num_filters=192, first_kernel_size=5, other_kernel_size=3, num_classes=None):
super(AlphaGoModel, self).__init__()
layers = [
nn.Conv2d(input_shape[0], num_filters, first_kernel_size, padding=first_kernel_size//2),
nn.ReLU()
]
for i in range(2, 12):
layers.extend([
nn.Conv2d(num_filters, num_filters, other_kernel_size, padding=other_kernel_size//2),
nn.ReLU()
])
if is_policy_net:
assert num_classes is not None, "num_classes must be provided for policy network"
layers.extend([
nn.Conv2d(num_filters, 1, kernel_size=1, padding=0),
# nn.Softmax(dim=1),
nn.Flatten()
])
else:
layers.extend([
nn.Conv2d(num_filters, num_filters, other_kernel_size, padding=other_kernel_size//2),
nn.ReLU(),
nn.Conv2d(num_filters, 1, kernel_size=1, padding=0),
nn.ReLU(),
nn.Flatten(),
nn.Linear(input_shape[1] * input_shape[2], 256),
nn.ReLU(),
nn.Linear(256, 1),
nn.Tanh()
])
self.model = nn.Sequential(*layers)
def forward(self, x):
return self.model(x)
# Instantiate policy network
# alphago_policy_model = AlphaGoModel(input_shape=(3, 19, 19), is_policy_net=True)
alphago_policy_model = AlphaGoModel(input_shape=(3, 19, 19), is_policy_net=True, num_classes=361)
# Instantiate value network
alphago_value_model = AlphaGoModel(input_shape=(3, 19, 19), is_policy_net=False)
+140
View File
@@ -0,0 +1,140 @@
from keras.layers import *
from keras.models import Model
'''The dual residual architecture is the strongest
of the architectures tested by DeepMind for AlphaGo
Zero. It consists of an initial convolutional block,
followed by a number (40 for the strongest, 20 as
baseline) of residual blocks. The network is topped
off by two "heads", one to predict policies and one
for value functions.
'''
def dual_residual_network(input_shape, blocks=20):
inputs = Input(shape=input_shape)
first_conv = conv_bn_relu_block(name="init")(inputs)
res_tower = residual_tower(blocks=blocks)(first_conv)
policy = policy_head()(res_tower)
value = value_head()(res_tower)
return Model(inputs=inputs, outputs=[policy, value])
'''The dual convolutional architecture replaces residual
blocks from the dual residual architecture with batch-normalized
convolution layers. The default block size is 12.
'''
def dual_conv_network(input_shape, blocks=12):
inputs = Input(shape=input_shape)
first_conv = conv_bn_relu_block(name="init")(inputs)
conv_tower = convolutional_tower(blocks=blocks)(first_conv)
policy = policy_head()(conv_tower)
value = value_head()(conv_tower)
return Model(inputs=inputs, outputs=[policy, value])
''' In the separate residual architecture policy and value
head don't share a common "tail", i.e. there's two sets of
residual blocks for policy and value networks, respectively.
'''
def separate_residual_network(input_shape, blocks=20):
inputs_pol = Input(shape=input_shape)
first_conv_pol = conv_bn_relu_block(name="init")(inputs_pol)
res_tower_pol = residual_tower(blocks=blocks)(first_conv_pol)
policy = policy_head()(res_tower_pol)
policy_model = Model(inputs=inputs_pol, outputs=policy)
inputs_val = Input(shape=input_shape)
first_conv_val = conv_bn_relu_block(name="init")(inputs_val)
res_tower_val = residual_tower(blocks=blocks)(first_conv_val)
value = value_head()(res_tower_val)
value_model = Model(inputs=inputs_val, outputs=value)
return policy_model, value_model
'''The separate convolutional network is structurally identical
to the separate residual network, except that residual blocks
are replaced by convolutional blocks.
'''
def separate_conv_network(input_shape, blocks=20):
inputs_pol = Input(shape=input_shape)
first_conv_pol = conv_bn_relu_block(name="init")(inputs_pol)
conv_tower_pol = convolutional_tower(blocks=blocks)(first_conv_pol)
policy = policy_head()(conv_tower_pol)
policy_model = Model(inputs=inputs_pol, outputs=policy)
inputs_val = Input(shape=input_shape)
first_conv_val = conv_bn_relu_block(name="init")(inputs_val)
conv_tower_val = convolutional_tower(blocks=blocks)(first_conv_val)
value = value_head()(conv_tower_val)
value_model = Model(inputs=inputs_val, outputs=value)
return policy_model, value_model
def conv_bn_relu_block(name, activation=True, filters=256, kernel_size=(3,3),
strides=(1,1), padding="same", init="he_normal"):
def f(inputs):
conv = Conv2D(filters=filters,
kernel_size=kernel_size,
strides=strides,
padding=padding,
kernel_initializer=init,
data_format='channels_first',
name="{}_conv_block".format(name))(inputs)
batch_norm = BatchNormalization(axis=1, name="{}_batch_norm".format(name))(conv)
return Activation("relu", name="{}_relu".format(name))(batch_norm) if activation else batch_norm
return f
def residual_block(block_num, **args):
def f(inputs):
res = conv_bn_relu_block(name="residual_1_{}".format(block_num), activation=True, **args)(inputs)
res = conv_bn_relu_block(name="residual_2_{}".format(block_num) , activation=False, **args)(res)
res = add([inputs, res], name="add_{}".format(block_num))
return Activation("relu", name="{}_relu".format(block_num))(res)
return f
def residual_tower(blocks, **args):
def f(inputs):
x = inputs
for i in range(blocks):
x = residual_block(block_num=i)(x)
return x
return f
def convolutional_tower(blocks, **args):
def f(inputs):
x = inputs
for i in range(blocks):
x = conv_bn_relu_block(name=i)(x)
return x
return f
def policy_head():
def f(inputs):
conv = Conv2D(filters=2,
kernel_size=(3, 3),
strides=(1, 1),
padding="same",
name="policy_head_conv_block")(inputs)
batch_norm = BatchNormalization(axis=1, name="policy_head_batch_norm")(conv)
activation = Activation("relu", name="policy_head_relu")(batch_norm)
return Dense(units= 19*19 +1, name="policy_head_dense")(activation)
return f
def value_head():
def f(inputs):
conv = Conv2D(filters=1,
kernel_size=(1, 1),
strides=(1, 1),
padding="same",
name="value_head_conv_block")(inputs)
batch_norm = BatchNormalization(axis=1, name="value_head_batch_norm")(conv)
activation = Activation("relu", name="value_head_relu")(batch_norm)
dense = Dense(units= 256, name="value_head_dense", activation="relu")(activation)
return Dense(units= 1, name="value_head_output", activation="tanh")(dense)
return f
+14
View File
@@ -0,0 +1,14 @@
from __future__ import absolute_import
from keras.layers.core import Dense, Activation, Flatten
def layers(input_shape):
return [
Dense(128, input_shape=input_shape),
Activation('relu'),
Dense(128, input_shape=input_shape),
Activation('relu'),
Flatten(),
Dense(128),
Activation('relu'),
]
+39
View File
@@ -0,0 +1,39 @@
from __future__ import absolute_import
from keras.layers.core import Dense, Activation, Flatten
from keras.layers.convolutional import Conv2D, ZeroPadding2D
def layers(input_shape):
return [
ZeroPadding2D((3, 3), input_shape=input_shape, data_format='channels_first'),
Conv2D(64, (7, 7), padding='valid', data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(64, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(64, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(48, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(48, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
Activation('relu'),
Flatten(),
Dense(1024),
Activation('relu'),
]
+40
View File
@@ -0,0 +1,40 @@
from __future__ import absolute_import
from keras.layers import LeakyReLU
from keras.layers.core import Dense, Flatten
from keras.layers.convolutional import Conv2D, ZeroPadding2D
def layers(input_shape):
return [
ZeroPadding2D((3, 3), input_shape=input_shape, data_format='channels_first'),
Conv2D(64, (7, 7), padding='valid', data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(64, (5, 5), data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(64, (5, 5), data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(48, (5, 5), data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(48, (5, 5), data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
LeakyReLU(),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
LeakyReLU(),
Flatten(),
Dense(1024),
LeakyReLU(),
]
+31
View File
@@ -0,0 +1,31 @@
from __future__ import absolute_import
from keras.layers.core import Dense, Activation, Flatten
from keras.layers.convolutional import Conv2D, ZeroPadding2D
def layers(input_shape):
return [
ZeroPadding2D((2, 2), input_shape=input_shape, data_format='channels_first'),
Conv2D(64, (5, 5), padding='valid', data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((2, 2), data_format='channels_first'),
Conv2D(64, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((1, 1), data_format='channels_first'),
Conv2D(64, (3, 3), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((1, 1), data_format='channels_first'),
Conv2D(64, (3, 3), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D((1, 1), data_format='channels_first'),
Conv2D(64, (3, 3), data_format='channels_first'),
Activation('relu'),
Flatten(),
Dense(512),
Activation('relu'),
]
+33
View File
@@ -0,0 +1,33 @@
from __future__ import absolute_import
# tag::small_network[]
from keras.layers.core import Dense, Activation, Flatten
from keras.layers.convolutional import Conv2D, ZeroPadding2D
def layers(input_shape):
return [
ZeroPadding2D(padding=3, input_shape=input_shape, data_format='channels_first'), # <1>
Conv2D(48, (7, 7), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D(padding=2, data_format='channels_first'), # <2>
Conv2D(32, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D(padding=2, data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
Activation('relu'),
ZeroPadding2D(padding=2, data_format='channels_first'),
Conv2D(32, (5, 5), data_format='channels_first'),
Activation('relu'),
Flatten(),
Dense(512),
Activation('relu'),
]
# <1> We use zero padding layers to enlarge input images.
# <2> By using `channels_first` we specify that the input plane dimension for our features comes first.
# end::small_network[]
+4
View File
@@ -0,0 +1,4 @@
torch
tqdm
numpy
tensorboard
+138
View File
@@ -0,0 +1,138 @@
# tag::scoring_imports[]
from __future__ import absolute_import
from collections import namedtuple
from tugo.gotypes import Player, Point
# end::scoring_imports[]
# tag::scoring_territory[]
class Territory:
def __init__(self, territory_map): # <1>
self.num_black_territory = 0
self.num_white_territory = 0
self.num_black_stones = 0
self.num_white_stones = 0
self.num_dame = 0
self.dame_points = []
for point, status in territory_map.items(): # <2>
if status == Player.black:
self.num_black_stones += 1
elif status == Player.white:
self.num_white_stones += 1
elif status == 'territory_b':
self.num_black_territory += 1
elif status == 'territory_w':
self.num_white_territory += 1
elif status == 'dame':
self.num_dame += 1
self.dame_points.append(point)
# <1> A `territory_map` splits the board into stones, territory and neutral points (dame).
# <2> Depending on the status of a point, we increment the respective counter.
# end::scoring_territory[]
# tag::scoring_game_result[]
class GameResult(namedtuple('GameResult', 'b w komi')):
@property
def winner(self):
if self.b > self.w + self.komi:
return Player.black
return Player.white
@property
def winning_margin(self):
w = self.w + self.komi
return abs(self.b - w)
def __str__(self):
w = self.w + self.komi
if self.b > w:
return 'B+%.1f' % (self.b - w,)
return 'W+%.1f' % (w - self.b,)
# end::scoring_game_result[]
""" evaluate_territory:
Map a board into territory and dame.
Any points that are completely surrounded by a single color are
counted as territory; it makes no attempt to identify even
trivially dead groups.
"""
# tag::scoring_evaluate_territory[]
def evaluate_territory(board):
status = {}
for r in range(1, board.num_rows + 1):
for c in range(1, board.num_cols + 1):
p = Point(row=r, col=c)
if p in status: # <1>
continue
stone = board.get(p)
if stone is not None: # <2>
status[p] = board.get(p)
else:
group, neighbors = _collect_region(p, board)
if len(neighbors) == 1: # <3>
neighbor_stone = neighbors.pop()
stone_str = 'b' if neighbor_stone == Player.black else 'w'
fill_with = 'territory_' + stone_str
else:
fill_with = 'dame' # <4>
for pos in group:
status[pos] = fill_with
return Territory(status)
# <1> Skip the point, if you already visited this as part of a different group.
# <2> If the point is a stone, add it as status.
# <3> If a point is completely surrounded by black or white stones, count it as territory.
# <4> Otherwise the point has to be a neutral point, so we add it to dame.
# end::scoring_evaluate_territory[]
""" _collect_region:
Find the contiguous section of a board containing a point. Also
identify all the boundary points.
"""
# tag::scoring_collect_region[]
def _collect_region(start_pos, board, visited=None):
if visited is None:
visited = {}
if start_pos in visited:
return [], set()
all_points = [start_pos]
all_borders = set()
visited[start_pos] = True
here = board.get(start_pos)
deltas = [(-1, 0), (1, 0), (0, -1), (0, 1)]
for delta_r, delta_c in deltas:
next_p = Point(row=start_pos.row + delta_r, col=start_pos.col + delta_c)
if not board.is_on_grid(next_p):
continue
neighbor = board.get(next_p)
if neighbor == here:
points, borders = _collect_region(next_p, board, visited)
all_points += points
all_borders |= borders
else:
all_borders.add(neighbor)
return all_points, all_borders
# end::scoring_collect_region[]
# tag::scoring_compute_game_result[]
def compute_game_result(game_state):
territory = evaluate_territory(game_state.board)
return GameResult(
territory.num_black_territory + territory.num_black_stones,
territory.num_white_territory + territory.num_white_stones,
komi=7.5)
# end::scoring_compute_game_result[]
View File
+78
View File
@@ -0,0 +1,78 @@
import platform
import subprocess
import numpy as np
# tag::print_utils[]
from tugo import gotypes
COLS = 'ABCDEFGHJKLMNOPQRST'
STONE_TO_CHAR = {
None: ' . ',
gotypes.Player.black: ' x ',
gotypes.Player.white: ' o ',
}
def print_move(player, move):
if move.is_pass:
move_str = 'passes'
elif move.is_resign:
move_str = 'resigns'
else:
move_str = '%s%d' % (COLS[move.point.col - 1], move.point.row)
print('%s %s' % (player, move_str))
def print_board(board):
for row in range(board.num_rows, 0, -1):
bump = " " if row <= 9 else ""
line = []
for col in range(1, board.num_cols + 1):
stone = board.get(gotypes.Point(row=row, col=col))
line.append(STONE_TO_CHAR[stone])
print('%s%d %s' % (bump, row, ''.join(line)))
print(' ' + ' '.join(COLS[:board.num_cols]))
# end::print_utils[]
# tag::human_coordinates[]
def point_from_coords(coords):
col = COLS.index(coords[0]) + 1
row = int(coords[1:])
return gotypes.Point(row=row, col=col)
# end::human_coordinates[]
def coords_from_point(point):
return '%s%d' % (
COLS[point.col - 1],
point.row
)
def clear_screen():
# see https://stackoverflow.com/a/23075152/323316
if platform.system() == "Windows":
subprocess.Popen("cls", shell=True).communicate()
else: # Linux and Mac
# the link uses print("\033c", end=""), but this is the original sequence given in the book.
print(chr(27) + "[2J")
# NOTE: MoveAge is only used in chapter 13, and doesn't make it to the main text.
# This feature will only be implemented in goboard_fast.py so as not to confuse
# readers in early chapters.
class MoveAge():
def __init__(self, board):
self.move_ages = - np.ones((board.num_rows, board.num_cols))
def get(self, row, col):
return self.move_ages[row, col]
def reset_age(self, point):
self.move_ages[point.row - 1, point.col - 1] = -1
def add(self, point):
self.move_ages[point.row - 1, point.col - 1] = 0
def increment_all(self):
self.move_ages[self.move_ages > -1] += 1
+1091
View File
File diff suppressed because it is too large Load Diff