fixes and cleanups to make acro vaguely functional again
Signed-off-by: Brian S. Stephan <bss@incorporeal.org>
This commit is contained in:
+247
-199
@@ -1,8 +1,14 @@
|
||||
"""An acromania style game for IRC."""
|
||||
"""An acromania style game for IRC.
|
||||
|
||||
SPDX-FileCopyrightText: © 2026 Brian S. Stephan <bss@incorporeal.org>
|
||||
SPDX-License-Identifier: GPL-3.0
|
||||
"""
|
||||
|
||||
import logging
|
||||
import random
|
||||
import threading
|
||||
import time
|
||||
from enum import Enum
|
||||
|
||||
from irc.client import NickMask, is_channel
|
||||
|
||||
@@ -11,142 +17,145 @@ from ircbot.lib import Plugin
|
||||
logger = logging.getLogger('acro.ircplugin')
|
||||
|
||||
|
||||
class Acro(Plugin):
|
||||
"""Play a game where users come up with silly definitions for randomly-generated acronyms."""
|
||||
class AcroGameState(Enum):
|
||||
"""Define states a game of Acromania can be in."""
|
||||
|
||||
class AcroGame(object):
|
||||
"""Track game details."""
|
||||
stopped = 0
|
||||
game_start = 1
|
||||
round_submission = 2
|
||||
round_voting = 3
|
||||
round_results = 4
|
||||
|
||||
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."""
|
||||
class AcroGame(object):
|
||||
"""Track game details."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize basic stuff."""
|
||||
self.acro = ""
|
||||
self.submissions = dict()
|
||||
self.sub_shuffle = []
|
||||
self.votes = dict()
|
||||
def __init__(self, bot, channel):
|
||||
"""Initialize basic stuff."""
|
||||
# running state
|
||||
self.state = AcroGameState.stopped
|
||||
self.should_quit = False
|
||||
self.scores = dict()
|
||||
self.rounds = []
|
||||
self.channel = channel
|
||||
|
||||
# default options
|
||||
self.seconds_to_submit = 60
|
||||
self.seconds_to_vote = 45
|
||||
self.seconds_to_submit_step = 10
|
||||
self.seconds_to_pause = 5
|
||||
# maintain a reference to the bot to generate messages back to IRC
|
||||
self.bot = bot
|
||||
|
||||
def __init__(self, bot, connection, event):
|
||||
"""Set up the game tracking and such."""
|
||||
# game state
|
||||
self.game = self.AcroGame()
|
||||
@property
|
||||
def current_round(self):
|
||||
"""Provide access to the current round of the game."""
|
||||
return self.rounds[-1]
|
||||
|
||||
super(Acro, self).__init__(bot, connection, event)
|
||||
def end_current_round(self):
|
||||
"""Clean up and output for ending the current round."""
|
||||
self.state = AcroGameState.round_results
|
||||
self.bot.reply(None, "voting's over! here are the scores for the round:", explicit_target=self.channel)
|
||||
self._print_round_scores()
|
||||
self._add_round_scores_to_game_scores()
|
||||
|
||||
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)
|
||||
# delay a bit
|
||||
time.sleep(self.current_round.seconds_to_pause)
|
||||
self._continue_or_quit()
|
||||
|
||||
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):
|
||||
def start_new_game(self, channel):
|
||||
"""Begin a new game, which will have multiple rounds."""
|
||||
self.game.state = 1
|
||||
self.game.channel = channel
|
||||
self.state = AcroGameState.game_start
|
||||
self.channel = channel
|
||||
self.bot.reply(None, "starting a new game of acro. it will run until you tell it to quit.",
|
||||
explicit_target=self.game.channel)
|
||||
explicit_target=self.channel)
|
||||
|
||||
self._start_new_round()
|
||||
self.start_new_round()
|
||||
|
||||
def _start_new_round(self):
|
||||
def start_new_round(self):
|
||||
"""Start a new round for play."""
|
||||
self.game.state = 2
|
||||
self.game.rounds.append(self.AcroRound())
|
||||
self.state = AcroGameState.round_submission
|
||||
self.rounds.append(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.current_round.acro = acro
|
||||
sleep_time = self.current_round.seconds_to_submit + (self.current_round.seconds_to_submit_step * (len(acro)-3))
|
||||
|
||||
self.bot.reply(None, "the round has started! your acronym is '{0:s}'. "
|
||||
"submit within {1:d} seconds via !acro submit [meaning]".format(acro, sleep_time),
|
||||
explicit_target=self.game.channel)
|
||||
explicit_target=self.channel)
|
||||
|
||||
t = threading.Thread(target=self.thread_do_process_submissions, args=(sleep_time,))
|
||||
t.daemon = True
|
||||
t.start()
|
||||
|
||||
def start_voting(self):
|
||||
"""Begin the voting period."""
|
||||
self.state = AcroGameState.round_voting
|
||||
self.current_round.shuffled_submissions = list(self.current_round.submissions.keys())
|
||||
random.SystemRandom().shuffle(self.current_round.shuffled_submissions)
|
||||
self.bot.reply(None, "here are the results. vote with !acro vote [number]", explicit_target=self.channel)
|
||||
self._print_round_acros()
|
||||
|
||||
t = threading.Thread(target=self.thread_do_process_votes, args=())
|
||||
t.daemon = True
|
||||
t.start()
|
||||
|
||||
def take_acro_submission(self, nick, submission):
|
||||
"""Take an acro submission and record it."""
|
||||
if self.state == AcroGameState.round_submission:
|
||||
sub_acro = self._turn_text_into_acro(submission)
|
||||
|
||||
if sub_acro == self.current_round.acro:
|
||||
self.current_round.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.current_round.acro, sub_acro)
|
||||
|
||||
def take_acro_vote(self, nick, vote):
|
||||
"""Take an acro vote and record it."""
|
||||
if self.state == AcroGameState.round_voting:
|
||||
acros = self.current_round.submissions
|
||||
if int(vote) > 0 and int(vote) <= len(acros):
|
||||
logger.debug("%s is %s", vote, self.current_round.shuffled_submissions[int(vote)-1])
|
||||
key = self.current_round.shuffled_submissions[int(vote)-1]
|
||||
|
||||
if key != nick:
|
||||
self.current_round.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 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.current_round.seconds_to_vote)
|
||||
self.end_current_round()
|
||||
|
||||
def _add_round_scores_to_game_scores(self):
|
||||
"""Apply the final round scores to the total scores for the game."""
|
||||
for s in list(self.current_round.votes.values()):
|
||||
votes = [x for x in list(self.current_round.votes.values()) if x == s]
|
||||
if s in list(self.scores.keys()):
|
||||
self.scores[s] += len(votes)
|
||||
else:
|
||||
self.scores[s] = len(votes)
|
||||
|
||||
def _continue_or_quit(self):
|
||||
"""Decide whether the game should continue or quit."""
|
||||
if self.state == AcroGameState.round_results:
|
||||
if self.should_quit:
|
||||
self._end_game()
|
||||
else:
|
||||
self.start_new_round()
|
||||
|
||||
def _end_game(self):
|
||||
"""Clean up the entire game."""
|
||||
self.state = AcroGameState.stopped
|
||||
self.should_quit = False
|
||||
self.bot.reply(None, "game's over! here are the final scores:", explicit_target=self.channel)
|
||||
self._print_game_scores()
|
||||
|
||||
@staticmethod
|
||||
def _generate_acro(): # noqa: C901
|
||||
"""Generate an acro to play with.
|
||||
@@ -156,8 +165,8 @@ class Acro(Plugin):
|
||||
"""
|
||||
acro = []
|
||||
# generate acro 3-8 characters long
|
||||
for i in range(1, random.SystemRandom.randint(4, 9)):
|
||||
letter = random.SystemRandom.randint(1, 182303)
|
||||
for i in range(1, random.SystemRandom().randint(4, 9)):
|
||||
letter = random.SystemRandom().randint(1, 182303)
|
||||
if letter <= 21912:
|
||||
acro.append('E')
|
||||
elif letter <= 38499:
|
||||
@@ -213,16 +222,28 @@ class Acro(Plugin):
|
||||
|
||||
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)
|
||||
def _print_game_scores(self):
|
||||
"""Print the final calculated scores."""
|
||||
for s in list(self.scores.keys()):
|
||||
self.bot.reply(None, " {0:s}: {1:d}".format(s, self.scores[s]), explicit_target=self.channel)
|
||||
|
||||
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)
|
||||
def _print_round_acros(self):
|
||||
"""Take the current round's acros and spit them to the channel."""
|
||||
i = 0
|
||||
for s in self.current_round.shuffled_submissions:
|
||||
logger.debug("%s is %s", str(i), s)
|
||||
self.bot.reply(None, " {0:d}: {1:s}".format(i+1, self.current_round.submissions[s]),
|
||||
explicit_target=self.channel)
|
||||
i += 1
|
||||
|
||||
def _print_round_scores(self):
|
||||
"""For the acros in the round, find the votes for them."""
|
||||
i = 0
|
||||
for s in list(self.current_round.submissions.keys()):
|
||||
votes = [x for x in list(self.current_round.votes.values()) if x == s]
|
||||
self.bot.reply(None, " {0:d} ({1:s}): {2:d}".format(i+1, s, len(votes)),
|
||||
explicit_target=self.channel)
|
||||
i += 1
|
||||
|
||||
@staticmethod
|
||||
def _turn_text_into_acro(text):
|
||||
@@ -233,91 +254,118 @@ class Acro(Plugin):
|
||||
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.SystemRandom.shuffle(self.game.rounds[-1].sub_shuffle)
|
||||
self.bot.reply(None, "here are the results. vote with !acro vote [number]", explicit_target=self.game.channel)
|
||||
self._print_round_acros()
|
||||
|
||||
t = threading.Thread(target=self.thread_do_process_votes, args=())
|
||||
t.daemon = True
|
||||
t.start()
|
||||
class AcroRound(object):
|
||||
"""Track a particular round of a game."""
|
||||
|
||||
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:
|
||||
logger.debug("%s is %s", str(i), s)
|
||||
self.bot.reply(None, " {0:d}: {1:s}".format(i+1, self.game.rounds[-1].submissions[s]),
|
||||
explicit_target=self.game.channel)
|
||||
i += 1
|
||||
def __init__(self):
|
||||
"""Initialize basic stuff."""
|
||||
self.acro = ""
|
||||
self.submissions = dict()
|
||||
self.shuffled_submissions = []
|
||||
self.votes = dict()
|
||||
|
||||
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):
|
||||
logger.debug("%s is %s", vote, self.game.rounds[-1].sub_shuffle[int(vote)-1])
|
||||
key = self.game.rounds[-1].sub_shuffle[int(vote)-1]
|
||||
# default options
|
||||
self.seconds_to_submit = 60
|
||||
self.seconds_to_vote = 45
|
||||
self.seconds_to_submit_step = 10
|
||||
self.seconds_to_pause = 5
|
||||
|
||||
if key != nick:
|
||||
self.game.rounds[-1].votes[nick] = key
|
||||
return "your vote has been recorded. it replaced any old vote for this round."
|
||||
|
||||
class Acro(Plugin):
|
||||
"""Play a game where users come up with silly definitions for randomly-generated acronyms."""
|
||||
|
||||
def __init__(self, bot, connection, event):
|
||||
"""Initialize the collection of games."""
|
||||
self.games = {}
|
||||
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 is_channel(event.target):
|
||||
try:
|
||||
game = self.games[event.target]
|
||||
if game.state == AcroGameState.stopped:
|
||||
return self.bot.reply(event, "there currently isn't a game running.")
|
||||
else:
|
||||
return "you can't vote for yourself!"
|
||||
else:
|
||||
return "you must vote for 1-{0:d}".format(len(acros))
|
||||
return self.bot.reply(event, "the game is running.")
|
||||
except KeyError:
|
||||
return self.bot.reply(event, "there is no game in progress for this channel.")
|
||||
|
||||
def _end_current_round(self):
|
||||
"""Clean up and output for ending the current round."""
|
||||
self.game.state = 4
|
||||
self.bot.reply(None, "voting's over! here are the scores for the round:", explicit_target=self.game.channel)
|
||||
self._print_round_scores()
|
||||
self._add_round_scores_to_game_scores()
|
||||
def handle_start(self, connection, event, match):
|
||||
"""Start the game, notify the channel, and begin timers."""
|
||||
if is_channel(event.target):
|
||||
try:
|
||||
game = self.games[event.target]
|
||||
except KeyError:
|
||||
game = AcroGame(self.bot, event.target)
|
||||
self.games[event.target] = game
|
||||
|
||||
# delay a bit
|
||||
time.sleep(self.game.rounds[-1].seconds_to_pause)
|
||||
self._continue_or_quit()
|
||||
if game.state != AcroGameState.stopped:
|
||||
return self.bot.reply(event, "the game is already running.")
|
||||
game.start_new_game(event.target)
|
||||
else:
|
||||
return self.bot.reply(event, "you must start the game from a channel.")
|
||||
|
||||
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.reply(None, " {0:d} ({1:s}): {2:d}".format(i+1, s, len(votes)),
|
||||
explicit_target=self.game.channel)
|
||||
i += 1
|
||||
def handle_submit(self, connection, event, match):
|
||||
"""Take a submission for an acro."""
|
||||
try:
|
||||
# HACK - need to implement a search by acro
|
||||
game = list(self.games.values())[0]
|
||||
nick = NickMask(event.source).nick
|
||||
submission = match.group(1)
|
||||
return self.bot.reply(event, game.take_acro_submission(nick, submission))
|
||||
except KeyError:
|
||||
return self.bot.reply(event, "there is no game in progress.")
|
||||
|
||||
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 handle_vote(self, connection, event, match):
|
||||
"""Take a vote for an acro."""
|
||||
try:
|
||||
# HACK - need to implement a search by acro
|
||||
game = list(self.games.values())[0]
|
||||
nick = NickMask(event.source).nick
|
||||
vote = match.group(1)
|
||||
return self.bot.reply(event, game.take_acro_vote(nick, vote))
|
||||
except KeyError:
|
||||
return self.bot.reply(event, "there is no game in progress.")
|
||||
|
||||
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 handle_quit(self, connection, event, match):
|
||||
"""Quit the game after the current round ends."""
|
||||
if is_channel(event.target):
|
||||
try:
|
||||
game = self.games[event.target]
|
||||
if game.state != AcroGameState.stopped:
|
||||
game.should_quit = True
|
||||
return self.bot.reply(event, "the game will end after the current round.")
|
||||
|
||||
def _end_game(self):
|
||||
"""Clean up the entire game."""
|
||||
self.game.state = 0
|
||||
self.game.quit = False
|
||||
self.bot.reply(None, "game's over! here are the final scores:", explicit_target=self.game.channel)
|
||||
self._print_game_scores()
|
||||
|
||||
def _print_game_scores(self):
|
||||
"""Print the final calculated scores."""
|
||||
for s in list(self.game.scores.keys()):
|
||||
self.bot.reply(None, " {0:s}: {1:d}".format(s, self.game.scores[s]), explicit_target=self.game.channel)
|
||||
return self.bot.reply(event, "the game is already stopped.")
|
||||
except KeyError:
|
||||
return self.bot.reply(event, "there is no game in progress for this channel.")
|
||||
|
||||
|
||||
plugin = Acro
|
||||
|
||||
Reference in New Issue
Block a user