self.move_pattern = None
self.coverage = None
-
+ self.possible_moves = {}
def init_from_copy(self, c):
# Draw the two possible types underneath the current_type image
for i in range(len(self.types)):
if always_reveal_states == True or self.types[i][0] != '?':
- img = small_images[self.colour][self.types[i]]
+ if self.types[i][0] == '?':
+ img = small_images[self.colour][self.types[i][1:]]
+ else:
+ img = small_images[self.colour][self.types[i]]
else:
img = small_images[self.colour]["unknown"] # If the type hasn't been revealed, show a placeholder
# Collapses the wave function!
def select(self):
- if self.current_type == "unknown":
+ if self.current_type == "unknown" or not self.choice in [0,1]:
self.choice = random.randint(0,1)
if self.types[self.choice][0] == '?':
self.types[self.choice] = self.types[self.choice][1:]
self.king = {"white" : None, "black" : None} # We need to keep track of the king, because he is important
self.max_moves = None
self.moves = 0
+ self.move_stack = []
for c in ["black", "white"]:
del self.unrevealed_types[c]["unknown"]
# Update the board when a piece has been selected
# "type" is apparently reserved, so I'll use "state"
- def update_select(self, x, y, type_index, state):
+ def update_select(self, x, y, type_index, state, sanity=True):
piece = self.grid[x][y]
if piece.types[type_index] == "unknown":
- if not state in self.unrevealed_types[piece.colour].keys():
+ if not state in self.unrevealed_types[piece.colour].keys() and sanity == True:
raise Exception("SANITY: Too many " + piece.colour + " " + state + "s")
self.unrevealed_types[piece.colour][state] -= 1
if self.unrevealed_types[piece.colour][state] <= 0:
if len(self.possible_moves(piece)) <= 0:
piece.deselect() # Piece can't move; deselect it
+
+ # Piece needs to recalculate moves
+ piece.possible_moves = None
# Update the board when a piece has been moved
- def update_move(self, x, y, x2, y2):
+ def update_move(self, x, y, x2, y2, sanity=True):
+
piece = self.grid[x][y]
+ #print "Moving " + str(x) + "," + str(y) + " to " + str(x2) + "," + str(y2) + "; possible_moves are " + str(self.possible_moves(piece))
+
+ if not [x2,y2] in self.possible_moves(piece) and sanity == True:
+ raise Exception("ILLEGAL move " + str(x2)+","+str(y2))
+
self.grid[x][y] = None
taken = self.grid[x2][y2]
if taken != None:
piece.deselect() # Uncollapse (?) the wavefunction!
self.moves += 1
+
+ # All other pieces need to recalculate moves
+ for p in self.pieces["white"] + self.pieces["black"]:
+ p.possible_moves = None
+
#self.verify()
# Update the board from a string
# Guesses what to do based on the format of the string
- def update(self, result):
+ def update(self, result, sanity=True):
#print "Update called with \"" + str(result) + "\""
# String always starts with 'x y'
try:
raise Exception("GIBBERISH \""+ str(result) + "\"") # Raise expectations
piece = self.grid[x][y]
- if piece == None:
+ if piece == None and sanity == True:
raise Exception("EMPTY")
# If a piece is being moved, the third token is '->'
raise Exception("GIBBERISH \"" + str(result) + "\"") # Raise the alarm
# Move the piece (take opponent if possible)
- self.update_move(x, y, x2, y2)
+ self.update_move(x, y, x2, y2, sanity)
else:
# Otherwise we will just assume a piece has been selected
except:
raise Exception("GIBBERISH \"" + result + "\"") # Throw a hissy fit
+
# Select the piece
- self.update_select(x, y, type_index, state)
+ self.update_select(x, y, type_index, state, sanity)
return result
if prob > 0:
result.update({p : prob})
- self.verify()
+ #self.verify()
return result
for i in range(len(p.types)):
t = p.types[i]
- prob = 0.5
+ prob = 1.0 / float(len(p.types))
if t == "unknown" or p.types[i][0] == '?':
total_types = 0
for t2 in self.unrevealed_types[p.colour].keys():
for t2 in self.unrevealed_types[p.colour].keys():
prob2 = float(self.unrevealed_types[p.colour][t2]) / float(total_types)
- p.current_type = t2
- for point in self.possible_moves(p, reject_allied):
+ #p.current_type = t2
+ for point in self.possible_moves(p, reject_allied, state=t2):
result[point[0]][point[1]] += prob2 * prob
else:
- p.current_type = t
- for point in self.possible_moves(p, reject_allied):
- result[point[0]][point[1]] += prob
+ #p.current_type = t
+ for point in self.possible_moves(p, reject_allied, state=t):
+ result[point[0]][point[1]] += prob
- self.verify()
- p.current_type = "unknown"
+ #self.verify()
+ #p.current_type = "unknown"
return result
def prob_is_type(self, p, state):
# This is probably inefficient, but I looked at some sample chess games and they seem to actually do things this way
# reject_allied indicates whether squares occupied by allied pieces will be removed
# (set to false to check for defense)
- def possible_moves(self, p, reject_allied = True):
- result = []
+ def possible_moves(self, p, reject_allied = True, state=None):
if p == None:
+ raise Exception("SANITY: No piece")
+
+
+
+ if state != None and state != p.current_type:
+ old_type = p.current_type
+ p.current_type = state
+ result = self.possible_moves(p, reject_allied, state=None)
+ p.current_type = old_type
return result
+
+
+
+
+ result = []
+
if p.current_type == "unknown":
if g != None and (g.colour == p.colour and reject_allied == True):
result.remove(point) # Remove allied pieces
- self.verify()
+ #self.verify()
+
+ p.possible_moves = result
return result
# I typed the full statement about 30 times before writing this function...
def on_board(self, x, y):
return (x >= 0 and x < w) and (y >= 0 and y < h)
+
+ # Pushes a move temporarily
+ def push_move(self, piece, x, y):
+ target = self.grid[x][y]
+ self.move_stack.append([piece, target, piece.x, piece.y, x, y])
+ [piece.x, piece.y] = [x, y]
+ self.grid[x][y] = piece
+ self.grid[piece.x][piece.y] = None
+
+ for p in self.pieces["white"] + self.pieces["black"]:
+ p.possible_moves = None
+
+ # Restore move
+ def pop_move(self):
+ #print str(self.move_stack)
+ [piece, target, x1, y1, x2, y2] = self.move_stack[len(self.move_stack)-1]
+ self.move_stack = self.move_stack[:-1]
+ piece.x = x1
+ piece.y = y1
+ self.grid[x1][y1] = piece
+ if target != None:
+ target.x = x2
+ target.y = y2
+ self.grid[x2][y2] = target
+
+ for p in self.pieces["white"] + self.pieces["black"]:
+ p.possible_moves = None
+
# --- board.py --- #
import subprocess
import select
import platform
-
+import re
agent_timeout = -1.0 # Timeout in seconds for AI players to make moves
# WARNING: Won't work for windows based operating systems
self.colour = colour
def update(self, result):
- pass
+ return result
def reset_board(self, s):
pass
if self.p.stdout in ready:
#sys.stderr.write("Reading from " + str(self.p) + " 's stdout...\n")
try:
- result = self.p.stdout.readline().strip("\r\n")
+ result = self.p.stdout.readline().strip(" \t\r\n")
#sys.stderr.write("Read \'" + result + "\' from " + str(self.p) + "\n")
return result
except: # Exception, e:
line = self.get_response()
try:
- result = map(int, line.split(" "))
+ m = re.match("\s*(\d+)\s+(\d+)\s*", line)
+ result = map(int, [m.group(1), m.group(2)])
except:
raise Exception("GIBBERISH \"" + str(line) + "\"")
return result
def update(self, result):
#print "Update " + str(result) + " called for AgentPlayer"
self.send_message(result)
-
+ return result
def get_move(self):
line = self.get_response()
try:
- result = map(int, line.split(" "))
+ m = re.match("\s*(\d+)\s+(\d+)\s*", line)
+ result = map(int, [m.group(1), m.group(2)])
+
except:
raise Exception("GIBBERISH \"" + str(line) + "\"")
return result
pass
else:
sys.stdout.write(result + "\n")
+ return result
# Default internal player (makes random moves)
def update(self, result):
self.board.update(result)
- self.board.verify()
+ #self.board.verify()
+ return result
def reset_board(self, s):
self.board.reset_board(s)
# Get total probability that the move is protected
- [xx,yy] = [piece.x, piece.y]
- [piece.x, piece.y] = [x, y]
- self.board.grid[x][y] = piece
- self.board.grid[xx][yy] = None
+ self.board.push_move(piece, x, y)
+
+
defenders = self.board.coverage(x, y, piece.colour, reject_allied = False)
d_prob = 0.0
if (a_prob > 1.0):
a_prob = 1.0
- self.board.grid[x][y] = target
- self.board.grid[xx][yy] = piece
- [piece.x, piece.y] = [xx, yy]
+ self.board.pop_move()
+
# Score of the move
class Network():
def __init__(self, colour, address = None):
self.socket = socket.socket()
+ self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#self.socket.setblocking(0)
+
+ self.server = (address == None)
if colour == "white":
self.port = 4562
# print str(self) + " listens on port " + str(self.port)
if address == None:
- self.host = socket.gethostname()
+ self.host = "0.0.0.0" #socket.gethostname() # Breaks things???
self.socket.bind((self.host, self.port))
self.socket.listen(5)
self.src.send("ok\n")
if self.get_response() == "QUIT":
self.src.close()
+ self.address = (address, self.port)
def get_response(self):
# Timeout the start of the message (first character)
game.stop()
return True
+
class NetworkSender(Player,Network):
self.address = address
def connect(self):
- Network.__init__(self, self.base_player.colour, self.address)
-
+ nAttempts=3
+ for i in range(nAttempts):
+ try:
+ Network.__init__(self, self.colour, self.address)
+ debug(str(self) +" connected to " + str(self.address))
+ return
+ except Exception, e:
+ debug(str(self) +" attempt " + str(i) + ": " + str(e.message))
+
+ raise Exception("NETWORK - Can't connect to " + str(self.address))
def select(self):
[x,y] = self.base_player.select()
choice = self.board.grid[x][y]
s = str(x) + " " + str(y)
- #print str(self) + ".select sends " + s
+ #debug(str(self) + " sends: " + str(s))
self.send_message(s)
return [x,y]
def get_move(self):
[x,y] = self.base_player.get_move()
s = str(x) + " " + str(y)
- #print str(self) + ".get_move sends " + s
+ #debug(str(self) + " sends: " + str(s))
self.send_message(s)
return [x,y]
def update(self, s):
+
self.base_player.update(s)
+ if self.server == True:
+ #debug(str(self) + " sends: " + str(s))
+ self.send_message(s)
+ return s
+
s = s.split(" ")
[x,y] = map(int, s[0:2])
selected = self.board.grid[x][y]
if selected != None and selected.colour == self.colour and len(s) > 2 and not "->" in s:
s = " ".join(s[0:3])
for i in range(2):
- if selected.types_revealed[i] == True:
+ if selected.types[i][0] != '?':
s += " " + str(selected.types[i])
else:
s += " unknown"
- #print str(self) + ".update sends " + s
+ #debug(str(self) +" sending: " + str(s))
self.send_message(s)
self.base_player.quit(final_result)
#self.src.send("QUIT " + str(final_result) + "\n")
self.src.close()
+
+ def __str__(self):
+ s = "NetworkSender:"
+ if self.server:
+ s += "server"
+ else:
+ s += "client"
+ s += ":"+str(self.address)
+ return s
+
class NetworkReceiver(Player,Network):
def __init__(self, colour, address=None):
- Player.__init__(self, address, colour)
+ s = "@network"
+ if address != None:
+ s += ":"+str(address)
+ Player.__init__(self, s, colour)
self.address = address
self.board = None
+
def connect(self):
- Network.__init__(self, self.colour, self.address)
+ nAttempts=3
+ for i in range(nAttempts):
+ try:
+ Network.__init__(self, self.colour, self.address)
+ debug(str(self) +" connected to " + str(self.address))
+ return
+ except Exception, e:
+ debug(str(self) +" attempt " + str(i) + ": " + str(e.message))
+
+ raise Exception("NETWORK - Can't connect to " + str(self.address))
+
def select(self):
s = self.get_response()
- #print str(self) + ".select gets " + s
+ #debug(str(self) +".select reads: " + str(s))
[x,y] = map(int,s.split(" "))
if x == -1 and y == -1:
#print str(self) + ".select quits the game"
return [x,y]
def get_move(self):
s = self.get_response()
- #print str(self) + ".get_move gets " + s
+ #debug(str(self) +".get_move reads: " + str(s))
[x,y] = map(int,s.split(" "))
if x == -1 and y == -1:
#print str(self) + ".get_move quits the game"
return [x,y]
def update(self, result):
-
- result = result.split(" ")
- [x,y] = map(int, result[0:2])
- selected = self.board.grid[x][y]
- if selected != None and selected.colour == self.colour and len(result) > 2 and not "->" in result:
- s = self.get_response()
- #print str(self) + ".update - receives " + str(s)
- s = s.split(" ")
- selected.choice = int(s[2])
- for i in range(2):
- selected.types[i] = str(s[3+i])
- if s[3+i] == "unknown":
- selected.types_revealed[i] = False
- else:
- selected.types_revealed[i] = True
- selected.current_type = selected.types[selected.choice]
- else:
- pass
- #print str(self) + ".update - ignore result " + str(result)
+ if self.server == True:
+ return result
+ s = self.get_response()
+ #debug(str(self) + ".update reads: " + str(s))
+ if not "->" in s.split(" "):
+ self.board.update(s, sanity=False)
+ return s
def quit(self, final_result):
self.src.close()
+
+ def __str__(self):
+ s = "NetworkReceiver:"
+ if self.server:
+ s += "server"
+ else:
+ s += "client"
+ s += ":"+str(self.address)
+ return s
# --- network.py --- #
import threading
def setup(self, board, players):
for p in players:
- self.log.write("# " + p.colour + " : " + p.name + "\n")
+ self.log.write("# " + str(p.colour) + " : " + str(p.name) + "\n")
self.log.write("# Initial board\n")
for x in range(0, w):
def close(self):
self.getter.stop()
+
+class FileReplay():
+ def __init__(self, filename):
+ self.f = open(filename, "r", 0)
+ self.filename = filename
+ self.mod = os.path.getmtime(filename)
+ self.count = 0
+
+ def readline(self):
+ line = self.f.readline()
+
+ while line == "":
+ mod2 = os.path.getmtime(self.filename)
+ if mod2 > self.mod:
+ #sys.stderr.write("File changed!\n")
+ self.mod = mod2
+ self.f.close()
+ self.f = open(self.filename, "r", 0)
+
+ new_line = self.f.readline()
+
+ if " ".join(new_line.split(" ")[0:3]) != "# Short log":
+ for i in range(self.count):
+ new_line = self.f.readline()
+ #sys.stderr.write("Read back " + str(i) + ": " + str(new_line) + "\n")
+ new_line = self.f.readline()
+ else:
+ self.count = 0
+
+ line = new_line
+
+ self.count += 1
+ return line
+
+ def close(self):
+ self.f.close()
+
def log(s):
for l in log_files:
l.write(s)
+def debug(s):
+ sys.stderr.write("# DEBUG: " + s + "\n")
+
def log_init(board, players):
for l in log_files:
[x,y] = p.select() # Player selects a square
if self.stopped():
break
-
+
+ if not (isinstance(p, Network) and p.server == False):
+ result = self.board.select(x, y, colour = p.colour)
+ else:
+ #debug(str(self) + " don't update local board")
+ result = ""
-
-
- result = self.board.select(x, y, colour = p.colour)
+ result = p.update(result)
for p2 in self.players:
- p2.update(result) # Inform players of what happened
+ if p2 != p:
+ result = p2.update(result) # Inform players of what happened
log(result)
if self.stopped():
break
+ if len(line) <= 0:
+ continue
if line[0] == '#':
self.board.update_select(x, y, int(tokens[2]), tokens[len(tokens)-1])
if isinstance(graphics, GraphicsThread):
with graphics.lock:
- graphics.state["moves"] = self.board.possible_moves(target)
+ if target.current_type != "unknown":
+ graphics.state["moves"] = self.board.possible_moves(target)
+ else:
+ graphics.state["moves"] = None
time.sleep(turn_delay)
else:
self.board.update_move(x, y, x2, y2)
small_images[c].update({p : pygame.image.load(os.path.join(image_dir, c + "_" + p + "_small.png"))})
# --- images.py --- #
graphics_enabled = True
+
try:
import pygame
+ os.environ["SDL_VIDEO_ALLOW_SCREENSAVER"] = "1"
except:
graphics_enabled = False
+import time
self.error = 0
self.lock = threading.RLock()
self.cond = threading.Condition()
+ self.sleep_timeout = None
+ self.last_event = time.time()
+ self.blackout = False
#print "Test font"
pygame.font.Font(os.path.join(os.path.curdir, "data", "DejaVuSans.ttf"), 32).render("Hello", True,(0,0,0))
while not self.stopped():
- #print "Display grid"
- self.board.display_grid(window = self.window, grid_sz = self.grid_sz) # Draw the board
+ if self.sleep_timeout == None or (time.time() - self.last_event) < self.sleep_timeout:
+
+ #print "Display grid"
+ self.board.display_grid(window = self.window, grid_sz = self.grid_sz) # Draw the board
- #print "Display overlay"
- self.overlay()
+ #print "Display overlay"
+ self.overlay()
- #print "Display pieces"
- self.board.display_pieces(window = self.window, grid_sz = self.grid_sz) # Draw the board
+ #print "Display pieces"
+ self.board.display_pieces(window = self.window, grid_sz = self.grid_sz) # Draw the board
+ self.blackout = False
+
+ elif pygame.mouse.get_focused() and not self.blackout:
+ os.system("xset dpms force off")
+ self.blackout = True
+ self.window.fill((0,0,0))
pygame.display.flip()
for event in pygame.event.get():
- if event.type == pygame.QUIT:
+ self.last_event = time.time()
+ if event.type == pygame.QUIT or (event.type == pygame.KEYDOWN and event.key == pygame.K_q):
if isinstance(game, GameThread):
with game.lock:
game.final_result = ""
break
elif event.type == pygame.MOUSEBUTTONDOWN:
self.mouse_down(event)
+
elif event.type == pygame.MOUSEBUTTONUP:
- self.mouse_up(event)
+ self.mouse_up(event)
+
+
-
+
import time
turn_delay = 0.5
+sleep_timeout = None
[game, graphics] = [None, None]
def make_player(name, colour):
global src_file
global graphics_enabled
global always_reveal_states
+ global sleep_timeout
max_moves = None
src_file = None
elif arg[1] == '-' and arg[2:] == "reveal":
always_reveal_states = True
elif (arg[1] == '-' and arg[2:] == "graphics"):
- graphics_enabled = not graphics_enabled
+ graphics_enabled = True
+ elif (arg[1] == '-' and arg[2:] == "no-graphics"):
+ graphics_enabled = False
elif (arg[1] == '-' and arg[2:].split("=")[0] == "file"):
# Load game from file
if len(arg[2:].split("=")) == 1:
if f[0:7] == "http://":
src_file = HttpReplay(f)
else:
- src_file = open(f.split(":")[0], "r", 0)
+ src_file = FileReplay(f.split(":")[0])
if len(f.split(":")) == 2:
max_moves = int(f.split(":")[1])
agent_timeout = -1
else:
agent_timeout = float(arg[2:].split("=")[1])
+ elif (arg[1] == '-' and arg[2:].split("=")[0] == "blackout"):
+ # Screen saver delay
+ if len(arg[2:].split("=")) == 1:
+ sleep_timeout = -1
+ else:
+ sleep_timeout = float(arg[2:].split("=")[1])
elif (arg[1] == '-' and arg[2:] == "help"):
# Help
if graphics_enabled == True:
try:
graphics = GraphicsThread(game.board, grid_sz = [64,64]) # Construct a GraphicsThread!
+
+ graphics.sleep_timeout = sleep_timeout
except Exception,e:
graphics = None
if graphics != None:
graphics.board.display_grid(graphics.window, graphics.grid_sz)
graphics.message("Connecting to " + p.colour + " player...")
+
+ # Handle race condition by having clients wait longer than servers to connect
+ if p.address != None:
+ time.sleep(0.2)
p.connect()
if src_file != None and src_file != sys.stdin:
src_file.close()
+ sys.stdout.write(game.final_result + "\n")
+
return error
# This is how python does a main() function...
sys.exit(102)
# --- main.py --- #
-# EOF - created from make on Thu Jan 31 13:37:15 WST 2013
+# EOF - created from make on Tue Apr 2 15:05:07 WST 2013
git.ucc.asn.au / progcomp2013.git / blobdiff
UCC git Repository :: git.ucc.asn.au