158 lines
4.6 KiB
Python
158 lines
4.6 KiB
Python
import subprocess, sys
|
|
|
|
""" Connection with the Go Text Protocol of GNU Go.
|
|
You have to have gnugo installed, and in your exec path."""
|
|
import random
|
|
|
|
|
|
class GnuGo:
|
|
|
|
def query(self, s):
|
|
self._stdin.write(s + "\n")
|
|
ret = []
|
|
while True:
|
|
l = self._stdout.readline().rstrip()
|
|
if l == "":
|
|
break
|
|
ret.append(l)
|
|
if len(ret) == 1 and ret[0].startswith("="):
|
|
return ("OK", ret[0][1:])
|
|
elif len(ret) == 0:
|
|
return ("NOK", None)
|
|
else:
|
|
return ("NOK", ret[0])
|
|
|
|
def __str__(self):
|
|
self._stdin.write("showboard\n")
|
|
ret = []
|
|
while True:
|
|
l = self._stdout.readline().rstrip()
|
|
if l == "":
|
|
break
|
|
ret.append(l)
|
|
return "\n".join(ret[1:])
|
|
|
|
def finalScore(self):
|
|
self._stdin.write("final_score\n")
|
|
ret = []
|
|
while True:
|
|
l = self._stdout.readline().rstrip()
|
|
if l == "":
|
|
break
|
|
ret.append(l)
|
|
return ret[0][2:]
|
|
|
|
class Moves:
|
|
|
|
def __init__(self, gnugo):
|
|
self._nextplayer = "black"
|
|
self._gnugo = gnugo
|
|
|
|
def flip(self):
|
|
if self._nextplayer == "black":
|
|
self._nextplayer = "white"
|
|
else:
|
|
self._nextplayer = "black"
|
|
|
|
def player(self):
|
|
return self._nextplayer
|
|
|
|
def getbest(self):
|
|
status, toret = self._gnugo.query("reg_genmove " + self._nextplayer)
|
|
if status == "OK":
|
|
return toret.strip()
|
|
return "ERR"
|
|
|
|
def get_randomized_best(self):
|
|
status, toret = self._gnugo.query("experimental_score " + self._nextplayer)
|
|
if status != "OK":
|
|
return "ERR"
|
|
status, toret = self._gnugo.query("top_moves " + self._nextplayer)
|
|
if status != "OK":
|
|
return "ERR"
|
|
moves = []
|
|
scoremoves = []
|
|
cumulatedscore = []
|
|
cumul = 0
|
|
toread = toret.strip().split()
|
|
if len(toread) == 0:
|
|
return "PASS"
|
|
while len(toread) > 0:
|
|
m = toread.pop(0)
|
|
s = float(toread.pop(0))
|
|
moves.append(m)
|
|
scoremoves.append(s)
|
|
cumul += s
|
|
cumulatedscore.append(cumul)
|
|
r = random.uniform(0, cumul)
|
|
i = 0
|
|
while i < len(moves) and r > cumulatedscore[i]:
|
|
i += 1
|
|
if i >= len(moves):
|
|
i = len(moves) - 1
|
|
return moves[i]
|
|
|
|
def get_history(self):
|
|
status, toret = self._gnugo.query("move_history")
|
|
if status != "OK":
|
|
return "ERR"
|
|
toread = toret.strip().split()
|
|
return toread
|
|
|
|
def playthis(self, move):
|
|
status, toret = self._gnugo.query(
|
|
"play " + self._nextplayer + " " + str(move)
|
|
)
|
|
# print(status, toret)
|
|
self.flip()
|
|
return status
|
|
|
|
def __iter__(self):
|
|
return self
|
|
|
|
def __next__(self):
|
|
status, toret = self._gnugo.query("genmove " + self._nextplayer)
|
|
self.flip()
|
|
if status == "OK":
|
|
return toret.strip()
|
|
return "ERR"
|
|
|
|
def __init__(self, size, seed=12345678):
|
|
self._proc = subprocess.Popen(
|
|
[
|
|
"gnugo",
|
|
"--capture-all-dead",
|
|
"--chinese-rules",
|
|
"--monte-carlo",
|
|
"--boardsize",
|
|
str(size),
|
|
"--mode",
|
|
"gtp",
|
|
"--never-resign",
|
|
"--seed",
|
|
str(seed),
|
|
],
|
|
bufsize=1,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
self._stdin = self._proc.stdin
|
|
self._stdout = self._proc.stdout
|
|
self._size = size
|
|
self._nextplayer = "black"
|
|
(ok, _) = self.query("level 0")
|
|
assert ok == "OK"
|
|
(ok, _) = self.query("boardsize " + str(size))
|
|
assert ok == "OK"
|
|
(ok, _) = self.query("clear_board")
|
|
assert ok == "OK"
|
|
(ok, name) = self.query("name")
|
|
assert ok == "OK"
|
|
(ok, version) = self.query("version")
|
|
assert ok == "OK"
|
|
# print("Connection to", name.strip(), "(" + version.strip() + ")","Ok")
|
|
(ok, legal) = self.query("all_legal black")
|
|
assert ok == "OK"
|
|
# print("Legal moves: ", legal)
|