"""Interface to the online high-score system"""
import json
import urllib
import serge.common
import serge.actor
from serge.blocks import concurrent
import serge.blocks.concurrent.futures
import serge.blocks.actors
[docs]class OnlineMethodCallFailed(Exception):
"""Calling the online method resulted in an error"""
[docs]class BaseHighScoreInterface(object):
"""Base class for interacting with the high score system"""
def __init__(self, app_url, secret_user=False):
"""Initialise the interface"""
self.app_url = app_url
self.secret_user = secret_user
def _getResult(self, method_url, parameters):
"""Returns the result of calling the web method url with the given parameters"""
parameter_list = list(parameters.iteritems())
if self.secret_user:
parameter_list.append(('secretUser', 1))
params = urllib.urlencode(parameter_list)
url = '%s/%s?%s' % (self.app_url, method_url, params)
url_file = urllib.urlopen(url)
result = url_file.read()
try:
return json.loads(result)
except ValueError, err:
raise OnlineMethodCallFailed('JSON result invalid (%s). URL "%s" returned "%s"' % (err, url, result))
[docs]class OnlineScoreTable(BaseHighScoreInterface):
"""Simple interface to the online high score table"""
def __init__(self, app_url, game):
"""Initialise the table connection"""
super(OnlineScoreTable, self).__init__(app_url)
#
self.game = game
[docs] def recordScore(self, player, category, score):
"""Record a score"""
result = self._getResult(
'record_score',
{
'game': self.game,
'player': player,
'category': category,
'score': score,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not record score: %s' % result['reason'])
[docs] def createPlayer(self, player):
"""Creates a new player"""
result = self._getResult(
'create_user',
{
'game': self.game,
'name': player,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not create user: %s' % result['reason'])
[docs] def getCategories(self):
"""Return the category names"""
result = self._getResult(
'get_game_categories',
{
'game': self.game,
}
)
if result['status'] == 'OK':
return result['categories']
else:
raise OnlineMethodCallFailed('Could not get categories: %s' % result['reason'])
[docs] def getScores(self, category, player, number):
"""Return the game scores for the given category and the given player (or * for all players)"""
result = self._getResult(
'get_scores',
{
'game': self.game,
'category': category,
'number': number,
'player': player,
}
)
if result['status'] == 'OK':
return result['scores']
else:
raise OnlineMethodCallFailed('Could not get scores: %s' % result['reason'])
[docs]class AsyncOnlineScoreTable(OnlineScoreTable):
"""Asynchronous version of the high score table
All the methods of the high score table return futures to
the results. If you want callbacks then you can add
a callback to the future once you get it back.
"""
def __init__(self, app_url, game, max_workers=1):
"""Initialise the table"""
super(AsyncOnlineScoreTable, self).__init__(app_url, game)
#
self._executor = concurrent.futures.ThreadPoolExecutor(max_workers)
def recordScore(self, player, category, score):
"""Record a score"""
return self._executor.submit(
super(AsyncOnlineScoreTable, self).recordScore, player, category, score)
[docs] def createPlayer(self, player):
"""Create a new player"""
return self._executor.submit(
super(AsyncOnlineScoreTable, self).createPlayer, player)
[docs] def getCategories(self):
"""Return the categories for a game"""
return self._executor.submit(
super(AsyncOnlineScoreTable, self).getCategories)
[docs] def getScores(self, category, player, number):
"""Return the scores for a game"""
return self._executor.submit(
super(AsyncOnlineScoreTable, self).getScores, category, player, number)
[docs] def recordScore(self, player, category, score):
"""Record the score for a game"""
return self._executor.submit(
super(AsyncOnlineScoreTable, self).recordScore, player, category, score)
[docs]class SimpleHSTableView(serge.actor.MountableActor):
"""A simple view of a high score table utilizing a layout
You can supply a theme and it will style using hs-font-colour,
hs-font-size etc
The format of the text should be a format string using the names
num, player and score.
"""
def __init__(self, tag, name, app_url, game_name, category_name,
player_name, max_scores, layout, theme,
fmt='%(num)d %(player)s %(score)s'):
"""Initialise the table"""
super(SimpleHSTableView, self).__init__(tag, name)
self.app_url = app_url
self.game_name = game_name
self.category_name = category_name
self.player_name = player_name
self.max_scores = max_scores
self.layout = layout
self.theme = theme
self.fmt = fmt
#
self.normal_colour = self.theme.getPropertyWithDefault('hs-font-colour', (255, 255, 255))
self.player_colour = self.theme.getPropertyWithDefault('hs-font-player-colour', (255, 255, 0))
#
# Create a table
self.table = AsyncOnlineScoreTable(self.app_url, self.game_name)
#
self.results = None
[docs] def addedToWorld(self, world):
"""Added the table to the world"""
super(SimpleHSTableView, self).addedToWorld(world)
#
self.mountActor(self.layout, (0, 0))
#
# Create text entries to show scores
L = self.theme.getPropertyWithDefault
self.entries = serge.actor.ActorCollection()
for idx in range(self.max_scores):
entry = self.layout.addActor(
serge.blocks.actors.FormattedText(
'hs-text', 'hs-text-%d' % idx, self.fmt,
colour=self.normal_colour,
font_name=L('hs-font-name', 'DEFAULT'),
font_size=L('hs-font-size', 20),
num=idx + 1, player='unknown', score=123,
)
)
entry.active = False
self.entries.append(entry)
#
# The callback used when we have updated our scores
self.done = None
[docs] def updateActor(self, interval, world):
"""Update the actor"""
super(SimpleHSTableView, self).updateActor(interval, world)
#
# Check if we got new scores
if self.results and self.results.done():
self.log.info('Got high score result')
try:
result = self.results.result()
except Exception, err:
self.log.error('Could not get high scores: %s' % err)
else:
self.entries.forEach().active = False
for idx, (player, score, category, date) in enumerate(result):
item = self.entries[idx]
item.setValue('player', player)
item.setValue('score', score)
item.active = True
item.visual.setColour(self.player_colour if player == self.player_name else self.normal_colour)
#
self.results = None
#
# Perform callback if we have one
if self.done:
self.done()
[docs] def updateScores(self, done=None):
"""Try to update the scores"""
self.log.info('Getting scores from remote site (%s)' %
('no callback' if not done else 'with callback',))
self.done = done
self.results = self.table.getScores(self.category_name, '*', self.max_scores)
[docs] def recordScore(self, score):
"""Record a new score"""
return self.table.recordScore(self.player_name, self.category_name, score)
[docs] def ensurePlayer(self):
"""Ensure that our player is defined"""
return self.table.createPlayer(self.player_name)
[docs]class HighScoreSystem(BaseHighScoreInterface):
"""Wrapper for useful functions to manage the high score system"""
[docs] def gameExists(self, name):
"""Return True if the named game exists"""
try:
self._getResult(
'get_scores',
{
'game': name
}
)
except OnlineMethodCallFailed:
return False
else:
return True
[docs] def createGame(self, name):
"""Create a game"""
result = self._getResult(
'create_game',
{
'game': name,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not create game: %s' % result['reason'])
[docs] def deleteGame(self, name):
"""Delete a game"""
result = self._getResult(
'delete_game',
{
'game': name,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not delete game: %s' % result['reason'])
[docs] def renameGame(self, name, new_name):
"""Rename a game"""
result = self._getResult(
'rename_game',
{
'game': name,
'new': new_name,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not rename game: %s' % result['reason'])
[docs] def categoryExists(self, game, category):
"""Return True if the named category exists for the named game"""
try:
result = self._getResult(
'get_scores',
{
'game': game,
'category': category,
'number': 1,
'player': '*',
}
)
except OnlineMethodCallFailed:
return False
else:
return result['status'] == 'OK'
[docs] def createGameCategory(self, name, category, value_name, sort_ascending, max_player_scores):
"""Rename a game"""
result = self._getResult(
'create_game_category',
{
'game': name,
'name': category,
'value_name': value_name,
'sort_ascending': int(sort_ascending),
'max_player_scores': max_player_scores,
}
)
if result['status'] == 'OK':
return result['key']
else:
raise OnlineMethodCallFailed('Could not create category: %s' % result['reason'])