/usr/share/pyshared/pychess/ic/ICGameModel.py is in pychess 0.10.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | from pychess.System.Log import log
from pychess.Utils.GameModel import GameModel
from pychess.Utils.Offer import Offer
from pychess.Utils.const import *
from pychess.Players.Human import Human
from pychess.ic import GAME_TYPES
class ICGameModel (GameModel):
def __init__ (self, connection, ficsgame, timemodel):
assert ficsgame.game_type in GAME_TYPES.values()
GameModel.__init__(self, timemodel, ficsgame.game_type.variant)
self.connection = connection
self.ficsgame = ficsgame
connections = self.connections
connections[connection.bm].append(connection.bm.connect("boardUpdate", self.onBoardUpdate))
connections[connection.bm].append(connection.bm.connect("obsGameEnded", self.onGameEnded))
connections[connection.bm].append(connection.bm.connect("curGameEnded", self.onGameEnded))
connections[connection.bm].append(connection.bm.connect("gamePaused", self.onGamePaused))
connections[connection.om].append(connection.om.connect("onActionError", self.onActionError))
connections[connection].append(connection.connect("disconnected", self.onDisconnected))
rated = "rated" if ficsgame.rated else "unrated"
# This is in the format that ficsgame.com writes these PGN headers
self.tags["Event"] = "FICS %s %s game" % (rated, ficsgame.game_type.fics_name)
self.tags["Site"] = "FICS"
def __repr__ (self):
s = GameModel.__repr__(self)
s = s.replace("<GameModel", "<ICGameModel")
s = s.replace(", players=", ", ficsgame=%s, players=" % self.ficsgame)
return s
@property
def display_text (self):
t = "[ "
if self.timemodel:
t += self.timemodel.display_text + " "
t += self.ficsgame.display_rated.lower() + " "
if self.ficsgame.game_type.display_text:
t += self.ficsgame.game_type.display_text + " "
return t + "]"
def __disconnect (self):
if self.connections is None: return
for obj in self.connections:
# Humans need to stay connected post-game so that "GUI > Actions" works
if isinstance(obj, Human):
continue
for handler_id in self.connections[obj]:
if obj.handler_is_connected(handler_id):
log.debug("ICGameModel.__disconnect: object=%s handler_id=%s\n" % \
(repr(obj), repr(handler_id)))
obj.disconnect(handler_id)
self.connections = None
def hasGuestPlayers (self):
for player in (self.ficsgame.wplayer, self.ficsgame.bplayer):
if player.isGuest():
return True
return False
def onBoardUpdate (self, bm, gameno, ply, curcol, lastmove, fen, wname, bname, wms, bms):
log.debug(("ICGameModel.onBoardUpdate: id=%s self.ply=%s self.players=%s gameno=%s " + \
"wname=%s bname=%s ply=%s curcol=%s lastmove=%s fen=%s wms=%s bms=%s\n") % \
(str(id(self)), str(self.ply), repr(self.players), str(gameno), str(wname), str(bname), \
str(ply), str(curcol), str(lastmove), str(fen), str(wms), str(bms)))
if gameno != self.ficsgame.gameno or len(self.players) < 2 or wname != self.players[0].ichandle \
or bname != self.players[1].ichandle:
return
log.debug("ICGameModel.onBoardUpdate: id=%d, self.players=%s: updating time and/or ply\n" % \
(id(self), str(self.players)))
if self.timemodel:
log.debug("ICGameModel.onBoardUpdate: id=%d self.players=%s: updating timemodel\n" % \
(id(self), str(self.players)))
self.timemodel.updatePlayer (WHITE, wms/1000.)
self.timemodel.updatePlayer (BLACK, bms/1000.)
if ply < self.ply:
log.debug("ICGameModel.onBoardUpdate: id=%d self.players=%s self.ply=%d ply=%d: TAKEBACK\n" % \
(id(self), str(self.players), self.ply, ply))
offers = self.offers.keys()
for offer in offers:
if offer.type == TAKEBACK_OFFER:
# There can only be 1 outstanding takeback offer for both players on FICS,
# (a counter-offer by the offeree for a takeback for a different number of
# moves replaces the initial offer) so we can safely remove all of them
del self.offers[offer]
self.undoMoves(self.ply-ply)
def onGameEnded (self, bm, ficsgame):
if ficsgame == self.ficsgame and len(self.players) >= 2:
log.debug(
"ICGameModel.onGameEnded: self.players=%s ficsgame=%s\n" % \
(repr(self.players), repr(ficsgame)))
self.end(ficsgame.result, ficsgame.reason)
def setPlayers (self, players):
GameModel.setPlayers(self, players)
if self.players[WHITE].icrating:
self.tags["WhiteElo"] = self.players[WHITE].icrating
if self.players[BLACK].icrating:
self.tags["BlackElo"] = self.players[BLACK].icrating
def onGamePaused (self, bm, gameno, paused):
if paused:
self.pause()
else: self.resume()
# we have to do this here rather than in acceptRecieved(), because
# sometimes FICS pauses/unpauses a game clock without telling us that the
# original offer was "accepted"/"received", such as when one player offers
# "pause" and the other player responds not with "accept" but "pause"
for offer in self.offers.keys():
if offer.type in (PAUSE_OFFER, RESUME_OFFER):
del self.offers[offer]
def onDisconnected (self, connection):
if self.status in (WAITING_TO_START, PAUSED, RUNNING):
self.end (KILLED, DISCONNECTED)
############################################################################
# Offer management #
############################################################################
def offerRecieved (self, player, offer):
log.debug("ICGameModel.offerRecieved: offerer=%s %s\n" % (repr(player), offer))
if player == self.players[WHITE]:
opPlayer = self.players[BLACK]
else: opPlayer = self.players[WHITE]
if offer.type == CHAT_ACTION:
opPlayer.putMessage(offer.param)
elif offer.type in (RESIGNATION, FLAG_CALL):
self.connection.om.offer(offer, self.ply)
elif offer.type in OFFERS:
if offer not in self.offers:
log.debug("ICGameModel.offerRecieved: %s.offer(%s)\n" % (repr(opPlayer), offer))
self.offers[offer] = player
opPlayer.offer(offer)
# If the offer was an update to an old one, like a new takebackvalue
# we want to remove the old one from self.offers
for offer_ in self.offers.keys():
if offer.type == offer_.type and offer != offer_:
del self.offers[offer_]
def acceptRecieved (self, player, offer):
log.debug("ICGameModel.acceptRecieved: accepter=%s %s\n" % (repr(player), offer))
if player.__type__ == LOCAL:
if offer not in self.offers or self.offers[offer] == player:
player.offerError(offer, ACTION_ERROR_NONE_TO_ACCEPT)
else:
log.debug("ICGameModel.acceptRecieved: connection.om.accept(%s)\n" % offer)
self.connection.om.accept(offer)
del self.offers[offer]
# We don't handle any ServerPlayer calls here, as the fics server will
# know automatically if he/she accepts an offer, and will simply send
# us the result.
def checkStatus (self):
pass
def onActionError (self, om, offer, error):
self.emit("action_error", offer, error)
#
# End
#
def end (self, status, reason):
if self.status in UNFINISHED_STATES:
self.__disconnect()
if self.isObservationGame():
self.connection.bm.unobserve(self.ficsgame)
else:
self.connection.om.offer(Offer(ABORT_OFFER), -1)
self.connection.om.offer(Offer(RESIGNATION), -1)
GameModel.end(self, status, reason)
|