"""An acromania style game for IRC.""" import logging import random import threading import time from irc.client import NickMask, is_channel from ircbot.lib import Plugin log = logging.getLogger('acro.ircplugin') class Acro(Plugin): """Play a game where users come up with silly definitions for randomly-generated acronyms.""" class AcroGame(object): """Track game details.""" def __init__(self): """Initialize basic stuff.""" # running state self.state = 0 self.quit = False self.scores = dict() self.rounds = [] self.channel = '' class AcroRound(object): """Track a particular round of a game.""" def __init__(self): """Initialize basic stuff.""" self.acro = "" self.submissions = dict() self.sub_shuffle = [] self.votes = dict() # default options self.seconds_to_submit = 60 self.seconds_to_vote = 45 self.seconds_to_submit_step = 10 self.seconds_to_pause = 5 def __init__(self, bot, connection, event): """Set up the game tracking and such.""" # game state self.game = self.AcroGame() super(Acro, self).__init__(bot, connection, event) def start(self): """Set up handlers.""" self.connection.reactor.add_global_regex_handler(['pubmsg', 'privmsg'], r'^!acro\s+status$', self.handle_status, -20) self.connection.reactor.add_global_regex_handler(['pubmsg', 'privmsg'], r'^!acro\s+start$', self.handle_start, -20) self.connection.reactor.add_global_regex_handler(['pubmsg', 'privmsg'], r'^!acro\s+submit\s+(.*)$', self.handle_submit, -20) self.connection.reactor.add_global_regex_handler(['pubmsg', 'privmsg'], r'^!acro\s+vote\s+(\d+)$', self.handle_vote, -20) self.connection.reactor.add_global_regex_handler(['pubmsg', 'privmsg'], r'^!acro\s+quit$', self.handle_quit, -20) super(Acro, self).start() def stop(self): """Tear down handlers.""" self.connection.reactor.remove_global_regex_handler(['pubmsg', 'privmsg'], self.handle_status) self.connection.reactor.remove_global_regex_handler(['pubmsg', 'privmsg'], self.handle_start) self.connection.reactor.remove_global_regex_handler(['pubmsg', 'privmsg'], self.handle_submit) self.connection.reactor.remove_global_regex_handler(['pubmsg', 'privmsg'], self.handle_vote) self.connection.reactor.remove_global_regex_handler(['pubmsg', 'privmsg'], self.handle_quit) super(Acro, self).stop() def handle_status(self, connection, event, match): """Return the status of the currently-running game, if there is one.""" if self.game.state == 0: return self.bot.reply(event, "there currently isn't a game running.") else: return self.bot.reply(event, "the game is running.") def handle_start(self, connection, event, match): """Start the game, notify the channel, and begin timers.""" if self.game.state != 0: return self.bot.reply(event, "the game is already running.") else: if is_channel(event.target): self._start_new_game(event.target) else: return self.bot.reply("you must start the game from a channel.") def handle_submit(self, connection, event, match): """Take a submission for an acro.""" nick = NickMask(event.source).nick submission = match.group(1) return self.bot.reply(event, self._take_acro_submission(nick, submission)) def handle_vote(self, connection, event, match): """Take a vote for an acro.""" nick = NickMask(event.source).nick vote = match.group(1) return self.bot.reply(event, self._take_acro_vote(nick, vote)) def handle_quit(self, connection, event, match): """Quit the game after the current round ends.""" if self.game.state != 0: self.game.quit = True return self.bot.reply(event, "the game will end after the current round.") def thread_do_process_submissions(self, sleep_time): """Wait for players to provide acro submissions, and then kick off voting.""" time.sleep(sleep_time) self._start_voting() def thread_do_process_votes(self): """Wait for players to provide votes, and then continue or quit.""" time.sleep(self.game.rounds[-1].seconds_to_vote) self._end_current_round() def _start_new_game(self, channel): """Begin a new game, which will have multiple rounds.""" self.game.state = 1 self.game.channel = channel self.bot.privmsg(self.game.channel, "starting a new game of acro. it will run until you tell it to quit.") self._start_new_round() def _start_new_round(self): """Start a new round for play.""" self.game.state = 2 self.game.rounds.append(self.AcroRound()) acro = self._generate_acro() self.game.rounds[-1].acro = acro sleep_time = self.game.rounds[-1].seconds_to_submit + (self.game.rounds[-1].seconds_to_submit_step * (len(acro)-3)) self.bot.privmsg(self.game.channel, "the round has started! your acronym is '{0:s}'. " "submit within {1:d} seconds via !acro submit [meaning]".format(acro, sleep_time)) t = threading.Thread(target=self.thread_do_process_submissions, args=(sleep_time,)) t.daemon = True t.start() @staticmethod def _generate_acro(): """Generate an acro to play with. Letter frequencies pinched from http://www.math.cornell.edu/~mec/2003-2004/cryptography/subs/frequencies.html """ acro = [] # generate acro 3-8 characters long for i in range(1, random.randint(4, 9)): letter = random.randint(1, 182303) if letter <= 21912: acro.append('E') elif letter <= 38499: acro.append('T') elif letter <= 53309: acro.append('A') elif letter <= 67312: acro.append('O') elif letter <= 80630: acro.append('I') elif letter <= 93296: acro.append('N') elif letter <= 104746: acro.append('S') elif letter <= 115723: acro.append('R') elif letter <= 126518: acro.append('H') elif letter <= 134392: acro.append('D') elif letter <= 141645: acro.append('L') elif letter <= 146891: acro.append('U') elif letter <= 151834: acro.append('C') elif letter <= 156595: acro.append('M') elif letter <= 160795: acro.append('F') elif letter <= 164648: acro.append('Y') elif letter <= 168467: acro.append('W') elif letter <= 172160: acro.append('G') elif letter <= 175476: acro.append('P') elif letter <= 178191: acro.append('B') elif letter <= 180210: acro.append('V') elif letter <= 181467: acro.append('K') elif letter <= 181782: acro.append('X') elif letter <= 181987: acro.append('Q') elif letter <= 182175: acro.append('J') elif letter <= 182303: acro.append('Z') return "".join(acro) def _take_acro_submission(self, nick, submission): """Take an acro submission and record it.""" if self.game.state == 2: sub_acro = self._turn_text_into_acro(submission) if sub_acro == self.game.rounds[-1].acro: self.game.rounds[-1].submissions[nick] = submission return "your submission has been recorded. it replaced any old submission for this round." else: return "the current acro is '{0:s}', not '{1:s}'".format(self.game.rounds[-1].acro, sub_acro) @staticmethod def _turn_text_into_acro(text): """Turn text into an acronym.""" words = text.split() acro = [] for w in words: acro.append(w[0].upper()) return "".join(acro) def _start_voting(self): """Begin the voting period.""" self.game.state = 3 self.game.rounds[-1].sub_shuffle = list(self.game.rounds[-1].submissions.keys()) random.shuffle(self.game.rounds[-1].sub_shuffle) self.bot.privmsg(self.game.channel, "here are the results. vote with !acro vote [number]") self._print_round_acros() t = threading.Thread(target=self.thread_do_process_votes, args=()) t.daemon = True t.start() def _print_round_acros(self): """Take the current round's acros and spit them to the channel.""" i = 0 for s in self.game.rounds[-1].sub_shuffle: log.debug("%s is %s", str(i), s) self.bot.privmsg(self.game.channel, " {0:d}: {1:s}".format(i+1, self.game.rounds[-1].submissions[s])) i += 1 def _take_acro_vote(self, nick, vote): """Take an acro vote and record it.""" if self.game.state == 3: acros = self.game.rounds[-1].submissions if int(vote) > 0 and int(vote) <= len(acros): log.debug("%s is %s", vote, self.game.rounds[-1].sub_shuffle[int(vote)-1]) key = self.game.rounds[-1].sub_shuffle[int(vote)-1] if key != nick: self.game.rounds[-1].votes[nick] = key return "your vote has been recorded. it replaced any old vote for this round." else: return "you can't vote for yourself!" else: return "you must vote for 1-{0:d}".format(len(acros)) def _end_current_round(self): """Clean up and output for ending the current round.""" self.game.state = 4 self.bot.privmsg(self.game.channel, "voting's over! here are the scores for the round:") self._print_round_scores() self._add_round_scores_to_game_scores() # delay a bit time.sleep(self.game.rounds[-1].seconds_to_pause) self._continue_or_quit() def _print_round_scores(self): """For the acros in the round, find the votes for them.""" i = 0 for s in list(self.game.rounds[-1].submissions.keys()): votes = [x for x in list(self.game.rounds[-1].votes.values()) if x == s] self.bot.privmsg(self.game.channel, " {0:d} ({1:s}): {2:d}".format(i+1, s, len(votes))) i += 1 def _add_round_scores_to_game_scores(self): """Apply the final round scores to the totall scores for the game.""" for s in list(self.game.rounds[-1].votes.values()): votes = [x for x in list(self.game.rounds[-1].votes.values()) if x == s] if s in list(self.game.scores.keys()): self.game.scores[s] += len(votes) else: self.game.scores[s] = len(votes) def _continue_or_quit(self): """Decide whether the game should continue or quit.""" if self.game.state == 4: if self.game.quit: self._end_game() else: self._start_new_round() def _end_game(self): """Clean up the entire game.""" self.game.state = 0 self.game.quit = False self.bot.privmsg(self.game.channel, "game's over! here are the final scores:") self._print_game_scores() def _print_game_scores(self): """Print the final calculated scores.""" for s in list(self.game.scores.keys()): self.bot.privmsg(self.game.channel, " {0:s}: {1:d}".format(s, self.game.scores[s])) plugin = Acro