A simple telnet based Taboo game in async Python
Marton Trencseni - Sun 06 October 2024 - Programming
Introduction
Our team at work is part remote and part in-office, so we frequently organize online gaming sessions as team building activities. We mix up the games, but one recurring favorite is Taboo: the objective of the game is for a player to have their partners guess the word on the player's card without using the word itself or five additional words listed on the card.
Over time, we have run into a number of issues with existing online Taboo servers:
- they seem like abandoned projects and we often run into show-stopper bugs
- the library of words/cards is low, so if we play for an hour, we often see the same card two to three times
- we cannot provide our own cards, eg. about Data Science terms
After our last gaming session encountered numerous repeated cards, I decided to build my own Taboo server in Python. To keep things simple and accessible, I opted for a Telnet-based interface, ensuring that everyone on our team could join the game without the need for specialized software. The playable game code is up on GitHub.
Architecture
Building a robust Taboo server involves handling multiple player connections, managing game states, and ensuring smooth gameplay even when players disconnect and reconnect. Leveraging Python's asyncio library, I structured the server to handle asynchronous tasks efficiently. Here's a breakdown of the key components and how they interact:
1. Handling Player Connections: The server listens for incoming connections and manages player sessions. Whether a player is joining for the first time or reconnecting, the server assigns them to a team and ensures they remain part of the game.
async def handle_player(self, reader, writer):
try:
writer.write(b"Enter your name: ")
await writer.drain()
data = await reader.read(1024)
name = data.decode().strip()
except Exception:
return
# Check if this player is reconnecting
for player in self.players:
if player["name"] == name:
player["reader"] = reader
player["writer"] = writer
player["connected"] = True
await self.send_to_player(player, "Welcome back! You have reconnected.")
await self.broadcast(f"{name} reconnected.")
return
player = {"name": name, "writer": writer, "reader": reader, "connected": True}
team = self.assign_team(player)
self.players.append(player)
await self.send_to_player(player, f"You are on team {team}")
self.send_to_admin(f"{name} joined team {team}")
await self.broadcast(self.teams_and_users())
self.send_to_admin("Type 'start' to begin the game.")
2. Assigning Teams: To ensure balanced gameplay, players are dynamically assigned to one of two teams based on the current team sizes.
def assign_team(self, player):
team = "A" if len(self.teams["A"]) <= len(self.teams["B"]) else "B"
self.teams[team].append(player)
return team
3. Game Flow Management: The core game logic is broken down into manageable functions: setting up each turn, waiting for player readiness, executing challenges, and ending each turn. This modular approach enhances readability and maintainability.
async def play_turn(self, team):
cluegiver, observer = await self.setup_turn(team)
await self.wait_for_cluegiver_to_start(cluegiver, team)
await self.execute_challenges(cluegiver, observer)
await self.end_turn(cluegiver, observer, team)
3a. Setting Up the Turn: Assigns roles and notifies players.
async def setup_turn(self, team):
cluegiver = self.teams[team][self.current_turn % len(self.teams[team])]
observer = random.choice(self.teams["A"] if team == "B" else self.teams["B"])
await self.broadcast(f"Round {self.current_turn} for team {team}: Cluegiver is {cluegiver['name']}, {observer['name']} from the other team is observing.")
await self.broadcast(f"Waiting for {cluegiver['name']} to start the turn...")
await self.send_to_player(observer, "You are the observer!")
await self.send_to_player(observer, "Make sure the cluegiver does not use forbidden words!")
await self.send_to_player(cluegiver, "You are the Cluegiver!")
await self.send_to_player(cluegiver, f"You have {self.turn_length_secs} seconds to go through as many challenges as you can.")
return cluegiver, observer
3b. Waiting for Cluegiver's Readiness: Ensures the cluegiver is ready before starting the turn.
async def wait_for_cluegiver_to_start(self, cluegiver, team):
await self.send_to_player(cluegiver, "Type 'start' if you're ready...")
while await self.read_from_player(cluegiver) != "start":
await self.send_to_player(cluegiver, "Type 'start' if you're ready...")
await self.broadcast(f"Round {self.current_turn} for team {team} has started!")
3c. Executing Challenges: Manages the challenge loop where the cluegiver gives clues without using forbidden words.
async def execute_challenges(self, cluegiver, observer):
start_time = time.time()
while time.time() - start_time < self.turn_length_secs:
challenge = self.challenges[self.challenge_index]
word, forbidden = challenge
await self.send_to_player(observer, f"Observer: The word is {word}. Forbidden: {', '.join(forbidden)}.")
await self.send_to_player(cluegiver, f"Cluegiver: The word is {word}. Forbidden: {', '.join(forbidden)}.")
await self.send_to_player(cluegiver, "Hit <enter> for next challenge.")
try:
data = await asyncio.wait_for(cluegiver["reader"].read(100), timeout=self.turn_length_secs - (time.time() - start_time))
response = data.decode().strip()
except asyncio.TimeoutError:
break
except Exception:
await self.on_disconnect(cluegiver)
finally:
self.challenge_index += 1
self.challenge_index %= len(self.challenges)
3d. Ending the Turn: Collects the number of correct guesses and updates the scores.
async def end_turn(self, cluegiver, observer, team):
await self.send_to_player(cluegiver, "Time's up!")
await self.send_to_player(observer, "Time's up!")
await self.broadcast(f"Waiting for {observer['name']} to input the number of correct guesses...")
correct_guesses = await self.get_valid_integer_from_player(observer, prompt="Please enter the number of correct guesses:")
await self.broadcast(f"In round {self.current_turn} team {team} guessed {correct_guesses} correctly.")
self.score[team] += correct_guesses
await self.broadcast(self.teams_and_users())
4. Broadcasting Messages: To keep all players informed, the server broadcasts messages to everyone, including updates and game status.
async def broadcast(self, message):
self.send_to_admin(message)
for player in self.players:
await self.send_to_player(player, message)
Getting started
Running your custom Taboo server is straightforward. Follow these steps to set it up and start playing:
1. Clone the Repository and Navigate to the Directory:
git clone https://github.com/mtrencseni/taboo.git
cd taboo
2. Prepare the challenges.csv File: Create a challenges.csv file containing Taboo words and their forbidden words in CSV format, or use mine. For example:
Python,programming,language,code,snake
Eiffel Tower,Paris,France,landmark,iron
3. Run the Server:
python3 taboo.py <port> <challenge_file> <turns> <turn_length_secs>
Replace <port>, <challenge_file>, <turns>, and <turn_length_secs> with your desired values. For example:
python3 taboo.py 12345 challenges.csv 10 60
- 12345: The port number the server will listen on.
- challenges.csv: The file containing game challenges.
- 10: The number of turns in the game.
- 60: The length of each turn in seconds.
4. Connect Players: Players can connect to the server using Telnet or any similar terminal-based network client:
telnet localhost 12345
5. Start the Game: Once all players are connected, the admin can start the game by typing start in the server console.
Conclusion
Whether you're a developer looking to enhance your skills or a team leader seeking a fun way to engage your team remotely, creating a multiplayer game server like this can be an exciting project. Feel free to explore the GitHub repository for the full source code and try it out yourself.
Happy coding and enjoy your game of Taboo!