- refacored state
This commit is contained in:
@@ -0,0 +1 @@
|
|||||||
|
__pycache__/
|
||||||
+58
@@ -0,0 +1,58 @@
|
|||||||
|
import numpy as np
|
||||||
|
from numpy.random import uniform
|
||||||
|
|
||||||
|
|
||||||
|
class APlayer(object):
|
||||||
|
def __init__(self, mark):
|
||||||
|
self.mark = mark
|
||||||
|
self.other_mark = 'O'
|
||||||
|
if 'O' in mark:
|
||||||
|
self.other_mark = 'X'
|
||||||
|
|
||||||
|
self.state = None
|
||||||
|
self.state_last = None
|
||||||
|
|
||||||
|
def move(self, state: np.array):
|
||||||
|
return state, False
|
||||||
|
|
||||||
|
def new_game(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def f_state_slices():
|
||||||
|
nr = 3
|
||||||
|
nc = 3
|
||||||
|
result = []
|
||||||
|
# Create row finishing states
|
||||||
|
d1 = ()
|
||||||
|
d2_r = ()
|
||||||
|
for r in range(0, nr):
|
||||||
|
d1 += (r,)
|
||||||
|
for c in range(0, nc):
|
||||||
|
d2 = (c,) * nc
|
||||||
|
result.append((d1, d2))
|
||||||
|
|
||||||
|
# Create column finishing states
|
||||||
|
for c in range(0, nc):
|
||||||
|
d2 = (c,) * nc
|
||||||
|
result.append((d2, d1))
|
||||||
|
for c in range(0, nc):
|
||||||
|
d2_r += (nc - c - 1,)
|
||||||
|
|
||||||
|
# Create diagonal finishing states #1
|
||||||
|
result.append((d1, d1))
|
||||||
|
|
||||||
|
# Create diagonal finishing states #2
|
||||||
|
result.append((d1, d2_r))
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
def has_won(self, state: np.array):
|
||||||
|
result = False
|
||||||
|
ref = [self.mark, self.mark, self.mark]
|
||||||
|
for f_state_slice in APlayer.f_state_slices():
|
||||||
|
if np.all(state[f_state_slice] == ref):
|
||||||
|
result = True
|
||||||
|
break
|
||||||
|
|
||||||
|
return result
|
||||||
@@ -0,0 +1,86 @@
|
|||||||
|
import numpy as np
|
||||||
|
from numpy.random import uniform
|
||||||
|
|
||||||
|
|
||||||
|
def create_test_state() -> np.array:
|
||||||
|
return np.array([['1', '2', '3'], ['4', '5', '6'], ['7', '8', '9']])
|
||||||
|
|
||||||
|
|
||||||
|
def create_empty_state() -> np.array:
|
||||||
|
return np.array([['-', '-', '-'], ['-', '-', '-'], ['-', '-', '-']])
|
||||||
|
|
||||||
|
|
||||||
|
def softmax(values: np.array, eps=1.0E-6) -> np.array:
|
||||||
|
result = (values + eps) / (sum(values + eps))
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def sample(values: np.array, with_debug=False):
|
||||||
|
# Normalize and sort
|
||||||
|
probs = softmax(values)
|
||||||
|
sorted_indices = np.argsort(probs)
|
||||||
|
sorted_probs = probs[sorted_indices]
|
||||||
|
z = uniform()
|
||||||
|
if with_debug:
|
||||||
|
print(f"Probs={probs}")
|
||||||
|
print(f"Z={z:.3f}")
|
||||||
|
index = None
|
||||||
|
p_sum = 0
|
||||||
|
for idx, p in enumerate(sorted_probs):
|
||||||
|
p_sum += p
|
||||||
|
if z <= p_sum:
|
||||||
|
index = sorted_indices[idx]
|
||||||
|
break
|
||||||
|
|
||||||
|
return index
|
||||||
|
|
||||||
|
|
||||||
|
def test_sample(data):
|
||||||
|
p = np.array(data)
|
||||||
|
c = np.array([0]*len(data))
|
||||||
|
|
||||||
|
for i in range(0, 1000):
|
||||||
|
index = sample(p, with_debug=False)
|
||||||
|
c[index] += 1
|
||||||
|
|
||||||
|
print(c)
|
||||||
|
|
||||||
|
|
||||||
|
def to_state_string(state, state_nex=None):
|
||||||
|
sp = ' '
|
||||||
|
sp_arrow = ' => '
|
||||||
|
sp_ = [sp, sp_arrow, sp]
|
||||||
|
|
||||||
|
def col(str_in, state):
|
||||||
|
str_out = str_in
|
||||||
|
for c in range(0, 3):
|
||||||
|
char = state[r][c]
|
||||||
|
if char == '-':
|
||||||
|
char = ' '
|
||||||
|
str_out += "|" + char
|
||||||
|
str_out += '|'
|
||||||
|
return str_out
|
||||||
|
|
||||||
|
state_str = ''
|
||||||
|
for r in range(0, 3):
|
||||||
|
if state is not None:
|
||||||
|
state_str = col(state_str, state)
|
||||||
|
|
||||||
|
if state_nex is not None:
|
||||||
|
state_str += sp_[r]
|
||||||
|
state_str = col(state_str, state_nex)
|
||||||
|
|
||||||
|
if r != 2:
|
||||||
|
state_str += '\x0A'
|
||||||
|
|
||||||
|
return state_str
|
||||||
|
|
||||||
|
print("Testing")
|
||||||
|
test_sample([0.7, 0.1, 0.1, 0.1])
|
||||||
|
test_sample([0.1, 0.1, 0.2, 0.1])
|
||||||
|
test_sample([0.2, 0.8])
|
||||||
|
test_sample([0.1, 0.1])
|
||||||
|
test_sample([0.3, 0.0, 0.2, 0.0])
|
||||||
|
test_sample([0.0, 0.0, 0.0, 0.0])
|
||||||
|
test_sample([0.5, 0.0, 0.2, 0.3])
|
||||||
|
test_sample([0.3, 0.1, 0.4, 0.2])
|
||||||
+49
-145
@@ -1,163 +1,70 @@
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
from numpy.random import uniform
|
from numpy.random import uniform
|
||||||
|
from a_player import APlayer
|
||||||
|
from helper import sample, create_empty_state, to_state_string
|
||||||
|
|
||||||
float_formatter = "{:.3f}".format
|
float_formatter = "{:.3f}".format
|
||||||
np.set_printoptions(formatter={'float_kind': float_formatter})
|
np.set_printoptions(formatter={'float_kind': float_formatter})
|
||||||
|
|
||||||
|
|
||||||
def softmax(values: np.array, eps=1.0E-6) -> np.array:
|
|
||||||
result = (values + eps) / (sum(values + eps))
|
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
def sample(values: np.array, with_debug=False):
|
|
||||||
# Normalize and sort
|
|
||||||
probs = softmax(values)
|
|
||||||
sorted_indices = np.argsort(probs)
|
|
||||||
sorted_probs = probs[sorted_indices]
|
|
||||||
z = uniform()
|
|
||||||
if with_debug:
|
|
||||||
print(f"Probs={probs}")
|
|
||||||
print(f"Z={z:.3f}")
|
|
||||||
index = None
|
|
||||||
p_sum = 0
|
|
||||||
for idx, p in enumerate(sorted_probs):
|
|
||||||
p_sum += p
|
|
||||||
if z <= p_sum:
|
|
||||||
index = sorted_indices[idx]
|
|
||||||
break
|
|
||||||
|
|
||||||
return index
|
|
||||||
|
|
||||||
|
|
||||||
def test_sample(data):
|
|
||||||
p = np.array(data)
|
|
||||||
c = np.array([0]*len(data))
|
|
||||||
|
|
||||||
for i in range(0, 1000):
|
|
||||||
index = sample(p, with_debug=False)
|
|
||||||
c[index] += 1
|
|
||||||
|
|
||||||
print(c)
|
|
||||||
|
|
||||||
|
|
||||||
def to_state_string(state, state_nex=None):
|
|
||||||
sp = ' '
|
|
||||||
sp_arrow = ' => '
|
|
||||||
sp_ = [sp, sp_arrow, sp]
|
|
||||||
|
|
||||||
def col(str_in, state):
|
|
||||||
str_out = str_in
|
|
||||||
for c in range(0, 3):
|
|
||||||
char = state[3*r+c]
|
|
||||||
if char == '-':
|
|
||||||
char = ' '
|
|
||||||
str_out += "|" + char
|
|
||||||
str_out += '|'
|
|
||||||
return str_out
|
|
||||||
|
|
||||||
state_str = ''
|
|
||||||
for r in range(0, 3):
|
|
||||||
state_str = col(state_str, state)
|
|
||||||
|
|
||||||
if state_nex is not None:
|
|
||||||
state_str += sp_[r]
|
|
||||||
state_str = col(state_str, state_nex)
|
|
||||||
|
|
||||||
if r != 2:
|
|
||||||
state_str += '\x0A'
|
|
||||||
|
|
||||||
return state_str
|
|
||||||
|
|
||||||
|
|
||||||
class APlayer(object):
|
|
||||||
def __init__(self, mark):
|
|
||||||
self.mark = mark
|
|
||||||
self.state = None
|
|
||||||
self.state_opp = None
|
|
||||||
self.state_last = None
|
|
||||||
|
|
||||||
def move(self, state):
|
|
||||||
return state, False
|
|
||||||
|
|
||||||
def has_won(self, state):
|
|
||||||
ref = self.mark + self.mark + self.mark
|
|
||||||
|
|
||||||
substr = state[0:3]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[3:6]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[6:9]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[0:9:3]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[1:9:3]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[2:9:3]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[0:9:4]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
substr = state[6:0:-2]
|
|
||||||
if substr in ref:
|
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
class MachinePlayer(APlayer):
|
class MachinePlayer(APlayer):
|
||||||
def __init__(self, mark='X', with_debug=False, values=None):
|
def __init__(self, mark='X', with_debug=False, values=None):
|
||||||
APlayer.__init__(self, mark)
|
APlayer.__init__(self, mark)
|
||||||
self.p_exp = 0.2
|
self.p_exp = 0.2
|
||||||
self.alpha = 0.9
|
self.alpha = 0.1
|
||||||
self.with_debug = with_debug
|
self.with_debug = with_debug
|
||||||
if values is None:
|
if values is None:
|
||||||
self.values = {}
|
self.init_values()
|
||||||
else:
|
else:
|
||||||
self.values = values
|
self.values = values
|
||||||
|
|
||||||
def get_value(self, state):
|
def init_values(self):
|
||||||
|
self.values = {}
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def to_key(state: np.array):
|
||||||
|
key = ''
|
||||||
|
for st in state.reshape(state.size):
|
||||||
|
key += st
|
||||||
|
return key
|
||||||
|
|
||||||
|
def get_value(self, state: np.array):
|
||||||
|
if state is None:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
key = self.to_key(state)
|
||||||
try:
|
try:
|
||||||
result = self.values[state]
|
result = self.values[key]
|
||||||
except KeyError:
|
except KeyError:
|
||||||
result = 0
|
result = 0
|
||||||
|
|
||||||
return result
|
return result
|
||||||
|
|
||||||
def set_value(self, state, value):
|
def set_value(self, state: np.array, value):
|
||||||
self.values[state] = value
|
self.values[self.to_key(state)] = value
|
||||||
|
|
||||||
def reward(self, value):
|
def reward(self, value):
|
||||||
self.set_value(self.state_last, value)
|
self.set_value(self.state_last, value)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_potential_moves(state) -> np.array:
|
def get_potential_moves(state: np.array) -> np.array:
|
||||||
indices = [idx for idx, s in enumerate(state) if '-' in s]
|
st = state.reshape(state.size)
|
||||||
|
indices = [idx for idx, s in enumerate(st) if '-' in s]
|
||||||
return np.array(indices)
|
return np.array(indices)
|
||||||
|
|
||||||
def move(self, state):
|
def new_game(self):
|
||||||
|
self.state = None
|
||||||
|
self.state_last = None
|
||||||
|
|
||||||
|
def move(self, state: np.array):
|
||||||
values = np.array([])
|
values = np.array([])
|
||||||
# get possible move
|
# get possible move
|
||||||
moves = self.get_potential_moves(state)
|
moves = self.get_potential_moves(state)
|
||||||
can_move = moves.size > 0
|
can_move = moves.size > 0
|
||||||
self.state_opp = state
|
|
||||||
if can_move:
|
if can_move:
|
||||||
for move in moves:
|
for move in moves:
|
||||||
# create hypothetical next state
|
# create hypothetical next state
|
||||||
state_next = self.state_from_move(state, move)
|
state_next = self.state_from_move(state.copy(), move)
|
||||||
# evaluate value
|
# evaluate value
|
||||||
value = self.get_value(state_next)
|
value = self.get_value(state_next)
|
||||||
values = np.append(values, value)
|
values = np.append(values, value)
|
||||||
@@ -175,23 +82,26 @@ class MachinePlayer(APlayer):
|
|||||||
print(f"{self.mark}: Moves = {moves+1}")
|
print(f"{self.mark}: Moves = {moves+1}")
|
||||||
print(f"{self.mark}: Move = {move+1}, is_exp={is_exp}")
|
print(f"{self.mark}: Move = {move+1}, is_exp={is_exp}")
|
||||||
|
|
||||||
self.state_last = self.state
|
if self.state is not None:
|
||||||
self.state = self.state_from_move(state, move)
|
self.state_last = self.state.copy()
|
||||||
|
|
||||||
|
self.state = self.state_from_move(state.copy(), move)
|
||||||
|
|
||||||
# Learn
|
# Learn
|
||||||
if not is_exp:
|
if not is_exp and self.state_last is not None:
|
||||||
v0 = self.get_value(self.state_last)
|
v0 = self.get_value(self.state_last)
|
||||||
v1 = self.get_value(self.state)
|
v1 = self.get_value(self.state)
|
||||||
d = max(0, v1-v0)
|
d = v0 + self.alpha*(v1-v0)
|
||||||
if d > 0:
|
if d > 0:
|
||||||
self.set_value(self.state_last, self.alpha*d)
|
self.set_value(self.state_last, d)
|
||||||
if self.with_debug:
|
if self.with_debug:
|
||||||
print(f"{self.mark}: Learned {d:0.3f}")
|
print(f"{self.mark}: Learned {d:0.3f}")
|
||||||
|
|
||||||
return self.state, can_move
|
return self.state, can_move
|
||||||
|
|
||||||
def state_from_move(self, state, field):
|
def state_from_move(self, state: np.array, field):
|
||||||
return state[:field] + self.mark + state[field + 1:]
|
state.reshape(state.size)[field] = self.mark
|
||||||
|
return state
|
||||||
|
|
||||||
|
|
||||||
class PlayerProvider:
|
class PlayerProvider:
|
||||||
@@ -209,13 +119,16 @@ def choose_player(player_provider: PlayerProvider) -> list[APlayer]:
|
|||||||
else:
|
else:
|
||||||
result = [p2, p1]
|
result = [p2, p1]
|
||||||
|
|
||||||
|
p1.new_game()
|
||||||
|
p2.new_game()
|
||||||
|
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
def play(player_provider: PlayerProvider, k_max=10000, with_print=False):
|
def play(player_provider: PlayerProvider, k_max=10000, with_print=False):
|
||||||
for k in range(0, k_max):
|
for k in range(0, k_max):
|
||||||
players = choose_player(player_provider)
|
players = choose_player(player_provider)
|
||||||
state = "---------"
|
state = create_empty_state()
|
||||||
move = 1
|
move = 1
|
||||||
run = True
|
run = True
|
||||||
while run:
|
while run:
|
||||||
@@ -245,24 +158,15 @@ def play(player_provider: PlayerProvider, k_max=10000, with_print=False):
|
|||||||
|
|
||||||
move += 1
|
move += 1
|
||||||
|
|
||||||
|
|
||||||
px = MachinePlayer(mark='X', with_debug=False)
|
px = MachinePlayer(mark='X', with_debug=False)
|
||||||
po = MachinePlayer(mark='O', with_debug=False)
|
po = MachinePlayer(mark='O', with_debug=False)
|
||||||
|
|
||||||
do_training = 1
|
|
||||||
|
do_training = 0
|
||||||
if do_training:
|
if do_training:
|
||||||
players = PlayerProvider(px, po)
|
players = PlayerProvider(px, po)
|
||||||
play(players, 20000, False)
|
play(players, 4000, False)
|
||||||
|
|
||||||
players = PlayerProvider(MachinePlayer(mark='X', with_debug=True, values=px.values), MachinePlayer(mark='O', with_debug=True, values=po.values))
|
players = PlayerProvider(MachinePlayer(mark='X', with_debug=True, values=px.values), MachinePlayer(mark='O', with_debug=True, values=po.values))
|
||||||
play(players, 100, True)
|
play(players, 4000, True)
|
||||||
|
|
||||||
print("Testing")
|
|
||||||
test_sample([0.7, 0.1, 0.1, 0.1])
|
|
||||||
test_sample([0.1, 0.1, 0.2, 0.1])
|
|
||||||
test_sample([0.2, 0.8])
|
|
||||||
test_sample([0.1, 0.1])
|
|
||||||
test_sample([0.3, 0.0, 0.2, 0.0])
|
|
||||||
test_sample([0.0, 0.0, 0.0, 0.0])
|
|
||||||
test_sample([0.5, 0.0, 0.2, 0.3])
|
|
||||||
test_sample([0.3, 0.1, 0.4, 0.2])
|
|
||||||
|
|||||||
Reference in New Issue
Block a user