diff --git a/src/loveletter_cli/__main__.py b/src/loveletter_cli/__main__.py index 7768126..848e3d6 100644 --- a/src/loveletter_cli/__main__.py +++ b/src/loveletter_cli/__main__.py @@ -10,13 +10,15 @@ import time import traceback +from aioconsole import aprint + from loveletter_cli.data import HostVisibility, PlayMode, UserInfo from loveletter_cli.exceptions import Restart from loveletter_cli.session import ( GuestCLISession, HostCLISession, ) -from loveletter_cli.ui import ask_valid_input, print_exception, print_header +from loveletter_cli.ui import async_ask_valid_input, print_exception, print_header from loveletter_cli.utils import ( get_local_ip, get_public_ip, @@ -67,48 +69,55 @@ def main( file_path=(None if show_client_logs else pathlib.Path("./loveletter_cli.log")), ) - version = get_version() - - runners = { - PlayMode.JOIN: join_game, - PlayMode.HOST: functools.partial(host_game, show_server_logs=show_server_logs), - } - while True: - try: - print_header(f"Welcome to Love Letter (CLI)! [v{version}]", filler="~") - - user = ask_user() - print(f"Welcome, {user.username}!") - - mode = ask_play_mode() - print() - return runners[mode](user) - except Restart: - LOGGER.info("Restarting CLI") - continue - except Exception as e: - LOGGER.error("Unhandled exception in CLI", exc_info=e) - - traceback.print_exc() - time.sleep(0.2) - print("Unhandled exception:") - print_exception(e) - time.sleep(0.2) - - choice = ask_valid_input( - "What would you like to do?", - choices=UnhandledExceptionOptions, - default=UnhandledExceptionOptions.RESTART, - ) - if choice == UnhandledExceptionOptions.RESTART: + async def async_main(): + version = get_version() + + runners = { + PlayMode.JOIN: join_game, + PlayMode.HOST: functools.partial( + host_game, show_server_logs=show_server_logs + ), + } + while True: + try: + await print_header( + f"Welcome to Love Letter (CLI)! [v{version}]", filler="~" + ) + + user = await ask_user() + await aprint(f"Welcome, {user.username}!") + + mode = await ask_play_mode() + await aprint() + return await runners[mode](user) + except Restart: + LOGGER.info("Restarting CLI") continue - elif choice == UnhandledExceptionOptions.QUIT: - return - else: - assert False, f"Unhandled error option: {choice}" - - -def ask_user(): + except Exception as e: + LOGGER.error("Unhandled exception in CLI", exc_info=e) + + traceback.print_exc() + time.sleep(0.2) + await aprint("Unhandled exception:") + await print_exception(e) + time.sleep(0.2) + + choice = await async_ask_valid_input( + "What would you like to do?", + choices=UnhandledExceptionOptions, + default=UnhandledExceptionOptions.RESTART, + ) + if choice == UnhandledExceptionOptions.RESTART: + continue + elif choice == UnhandledExceptionOptions.QUIT: + return + else: + assert False, f"Unhandled error option: {choice}" + + asyncio.run(async_main()) + + +async def ask_user(): def parser(x: str) -> str: x = x.strip() x = " ".join(x.split()) # normalize spaces to 1 space @@ -122,15 +131,15 @@ def parser(x: str) -> str: ) return x - username = ask_valid_input("Enter your username: ", parser=parser) + username = await async_ask_valid_input("Enter your username: ", parser=parser) user = UserInfo(username) return user -def ask_play_mode() -> PlayMode: +async def ask_play_mode() -> PlayMode: prompt = f"Would you like to host a game or join an existing one? " error_message = "Not a valid mode: {choice!r}" - return ask_valid_input( + return await async_ask_valid_input( prompt=prompt, choices=PlayMode, error_message=error_message, @@ -138,9 +147,9 @@ def ask_play_mode() -> PlayMode: ) -def host_game(user: UserInfo, show_server_logs: bool): - print_header("Hosting a game") - mode = ask_valid_input( +async def host_game(user: UserInfo, show_server_logs: bool): + await print_header("Hosting a game") + mode = await async_ask_valid_input( "Choose the server_addresses's visibility:", choices=HostVisibility, default=HostVisibility.PUBLIC, @@ -154,25 +163,27 @@ def host_game(user: UserInfo, show_server_logs: bool): "127.0.0.1", str(addresses["local"]), ) # allow either localhost or local net. - print(f"Your address: {' | '.join(f'{v} ({k})' for k, v in addresses.items())}") - port = ask_port_for_hosting() + await aprint( + f"Your address: {' | '.join(f'{v} ({k})' for k, v in addresses.items())}" + ) + port = await ask_port_for_hosting() play_again = True while play_again: session = HostCLISession(user, hosts, port, show_server_logs=show_server_logs) - asyncio.run(session.manage()) + await session.manage() play_again = ask_play_again() -def ask_port_for_hosting() -> int: +async def ask_port_for_hosting() -> int: def parser(s: str) -> int: port = int(s) if not (socket.IPPORT_USERRESERVED <= port <= MAX_PORT): raise ValueError(port) return port - return ask_valid_input( + return await async_ask_valid_input( prompt=( f"Choose a port number >= {socket.IPPORT_USERRESERVED}, <= {MAX_PORT}:" ), @@ -182,19 +193,19 @@ def parser(s: str) -> int: ) -def join_game(user: UserInfo): - print_header("Joining game") - address = ask_address_for_joining() +async def join_game(user: UserInfo): + await print_header("Joining game") + address = await ask_address_for_joining() play_again = True while play_again: session = GuestCLISession(user, address) - asyncio.run(session.manage()) + await session.manage() play_again = ask_play_again() -def ask_address_for_joining() -> Address: +async def ask_address_for_joining() -> Address: def parser(s: str) -> Address: host, port = s.split(":") with valid8.validation("host", host, help_msg="Invalid host"): @@ -203,20 +214,20 @@ def parser(s: str) -> Address: valid8.validate("port", port, min_value=1, max_value=1 << 16, max_strict=True) return Address(host, port) - return ask_valid_input( + return await async_ask_valid_input( prompt='Enter the server\'s address: (format: ":")', parser=parser, ) -def ask_play_again() -> bool: +async def ask_play_again() -> bool: """ Ask whether to play again after a session has ended. :return: Whether the user wants to play again. :raises Restart: if the user wants to restart the CLI (main menu). """ - choice = ask_valid_input( + choice = await async_ask_valid_input( prompt="The game has ended, what would you like to do?", choices=GameEndOptions, ) diff --git a/src/loveletter_cli/session.py b/src/loveletter_cli/session.py index 3915ecc..b3eed55 100644 --- a/src/loveletter_cli/session.py +++ b/src/loveletter_cli/session.py @@ -9,7 +9,7 @@ import more_itertools as mitt import valid8 -from aioconsole import ainput +from aioconsole import ainput, aprint from multimethod import multimethod import loveletter.game @@ -72,7 +72,7 @@ async def play_game(self, game: RemoteGameShadowCopy): game_input = await handle(event) event = await generator.asend(game_input) except valid8.ValidationError as exc: - print(exc) + await aprint(exc) except StopAsyncIteration: break @@ -87,7 +87,7 @@ async def handle(e: gev.GameEvent) -> Optional[gev.GameInputRequest]: # ----------------------------- Game node stages ----------------------------- @handle.register async def handle(e: loveletter.game.PlayingRound) -> None: - print_header(f"ROUND {e.round_no}", filler="#") + await print_header(f"ROUND {e.round_no}", filler="#") @handle.register async def handle(e: rnd.Turn) -> None: @@ -97,18 +97,18 @@ async def handle(e: rnd.Turn) -> None: is_client = player is game.client_player possessive = "Your" if is_client else f"{player.username}'s" - print_header(f"{possessive} turn", filler="—") - draw_game(game) + await print_header(f"{possessive} turn", filler="—") + await draw_game(game) if is_client: - print(">>>>> It's your turn! <<<<<") + await aprint(">>>>> It's your turn! <<<<<") else: - print(f"It's {player.username}'s turn.") + await aprint(f"It's {player.username}'s turn.") @handle.register async def handle(e: rnd.PlayingCard) -> None: player, card = game.get_player(e.player), e.card is_client = player is game.client_player - print( + await aprint( f"{'You' if is_client else player.username} " f"{'have' if is_client else 'has'} chosen to play a {card.name}." ) @@ -118,14 +118,14 @@ async def handle(e: rnd.RoundEnd) -> None: get_username = lambda p: game.get_player(p).username # noqa await pause() # give a chance to see what's happened before the round end - print_header("Round end", filler="—") - draw_game(game, reveal=True) + await print_header("Round end", filler="—") + await draw_game(game, reveal=True) - print(">>>>> The round has ended! <<<<<") + await aprint(">>>>> The round has ended! <<<<<") if e.reason == rnd.RoundEnd.Reason.EMPTY_DECK: - print("There are no cards remaining in the deck.") + await aprint("There are no cards remaining in the deck.") if len(e.tie_contenders) == 1: - print( + await aprint( f"{get_username(e.winner)} wins with a {e.winner.hand.card}, " f"which is the highest card among those remaining." ) @@ -137,8 +137,8 @@ async def handle(e: rnd.RoundEnd) -> None: if len(contenders) == 2 else f"Each of {', '.join(contenders[:-1])} and {contenders[-1]}" ) - print(f"{contenders_str} have the highest card: a {card}.") - print( + await aprint(f"{contenders_str} have the highest card: a {card}.") + await aprint( f"But {get_username(e.winner)} has a higher sum of discarded" f" values, so they win." if len(e.winners) == 1 @@ -147,7 +147,7 @@ async def handle(e: rnd.RoundEnd) -> None: f" in a tie." ) elif e.reason == rnd.RoundEnd.Reason.ONE_PLAYER_STANDING: - print( + await aprint( f"{get_username(e.winner)} is the only player still alive, " f"so they win the round." ) @@ -155,18 +155,18 @@ async def handle(e: rnd.RoundEnd) -> None: @handle.register async def handle(e: loveletter.game.PointsUpdate) -> None: # print updates from last round - print("Points gained:") + await aprint("Points gained:") for player, delta in (+e.points_update).items(): - print(f" {player.username}: {delta:+}") - print() - print("Leaderboard:") + await aprint(f" {player.username}: {delta:+}") + await aprint() + await aprint("Leaderboard:") width = max(map(len, (p.username for p in game.players))) + 2 for i, (player, points) in enumerate(game.points.most_common(), start=1): - print( + await aprint( f"\t{i}. {player.username:{width}}" f"\t{points} {pluralize('token', points)} of affection" ) - print() + await aprint() await pause() # before going on to next round # ------------------------------ Remote events ------------------------------- @@ -176,7 +176,7 @@ async def handle(e: RemoteEvent) -> None: if isinstance(e.wrapped, mv.MoveStep): name = e.wrapped.__class__.__name__ message += f" ({camel_to_phrase(name)})" - print(message) + await aprint(message) # ----------------------------- Pre-move choices ----------------------------- @handle.register @@ -230,7 +230,7 @@ async def handle(e: mv.ChooseOneCard): names = [CardType(c).name.title() for c in e.options] options_members = {CardType(c).name: c for c in e.options} - print( + await aprint( f"You draw {num_drawn} {pluralize('card', num_drawn)}; " f"you now have these cards in your hand: {', '.join(names)}" ) @@ -247,8 +247,8 @@ async def handle(e: mv.ChooseOrderForDeckBottom): return e fmt = ", ".join(f"{i}: {CardType(c).name}" for i, c in enumerate(e.cards)) - print(f"Leftover cards: {fmt}") - print( + await aprint(f"Leftover cards: {fmt}") + await aprint( "You can choose which order to place these cards at the bottom of the " "deck. Use the numbers shown above to refer to each of the cards." ) @@ -285,7 +285,7 @@ async def handle(e: mv.CorrectCardGuess) -> None: is_client = player is game.client_player target_is_client = opponent is game.client_player possessive = "your" if target_is_client else f"{opponent.username}'s" - print( + await aprint( f"{'You' if is_client else player.username} correctly guessed " f"{possessive} {e.guess.name.title()}!" ) @@ -294,18 +294,18 @@ async def handle(e: mv.CorrectCardGuess) -> None: async def handle(e: mv.WrongCardGuess) -> None: player, opponent = map(game.get_player, (e.player, e.opponent)) if player is game.client_player: - print( + await aprint( f"You played a Guard against {opponent.username} and guessed a " f"{e.guess.name.title()}, but {opponent.username} doesn't have " f"that card." ) elif opponent is game.client_player: - print( + await aprint( f"{player.username} played a Guard against you and guessed a " f"{e.guess.name.title()}, but you don't have that card." ) else: - print( + await aprint( f"{player.username} played a Guard against {opponent.username} " f"and guessed a {e.guess.name.title()}, but {opponent.username} " f"doesn't have that card." @@ -321,38 +321,40 @@ async def handle(e: mv.PlayerEliminated) -> None: ) if not is_client: message += f" They had a {e.eliminated_card.name}." - print(message) + await aprint(message) @handle.register async def handle(e: mv.ShowOpponentCard) -> None: player, opponent = map(game.get_player, (e.player, e.opponent)) if player is game.client_player: - print( + await aprint( f"{opponent.username} shows their card to you, " f"revealing a {e.card_shown.name}." ) elif opponent is game.client_player: - print(f"You show your {e.card_shown.name} to {player.username}.") + await aprint(f"You show your {e.card_shown.name} to {player.username}.") else: - print(f"{opponent.username} shows their card to {player.username}.") + await aprint( + f"{opponent.username} shows their card to {player.username}." + ) @handle.register async def handle(e: mv.CardComparison) -> None: player, opponent = map(game.get_player, (e.player, e.opponent)) if player is game.client_player: - print( + await aprint( f"You and {opponent.username} compare your cards: " f"you have a {e.player_card.name}, " f"they have a {e.opponent_card.name}." ) elif opponent is game.client_player: - print( + await aprint( f"{player.username} compares their hand with yours: " f"they have a {e.player_card.name}, " f"you have a {e.opponent_card.name}." ) else: - print( + await aprint( f"{player.username} and {opponent.username} " f"compare their cards in secret." ) @@ -361,7 +363,7 @@ async def handle(e: mv.CardComparison) -> None: async def handle(e: mv.CardDiscarded) -> None: player = game.get_player(e.target) is_client = player is game.client_player - print( + await aprint( f"{'You' if is_client else player.username} " f"discard{'s' if not is_client else ''} a {e.discarded.name}." ) @@ -370,26 +372,26 @@ async def handle(e: mv.CardDiscarded) -> None: async def handle(e: mv.CardDealt) -> None: player = game.get_player(e.target) is_client = player is game.client_player - print( + await aprint( f"{'You' if is_client else player.username} " f"{'are' if is_client else 'is'} dealt another card from the deck." ) if is_client: - print(f"You get a {e.card_dealt.name}.") + await aprint(f"You get a {e.card_dealt.name}.") @handle.register async def handle(e: mv.CardChosen) -> None: player = game.get_player(e.player) if player is game.client_player: - print(f"You have chosen to keep the {e.choice.name}.") + await aprint(f"You have chosen to keep the {e.choice.name}.") else: - print(f"{player.username} has chosen a card to keep.") + await aprint(f"{player.username} has chosen a card to keep.") @handle.register async def handle(e: mv.CardsPlacedBottomOfDeck) -> None: player = game.get_player(e.player) is_client = player is game.client_player - print( + await aprint( f"{'You' if is_client else player.username} " f"{'have' if is_client else 'has'} placed back the other " f"{len(e.cards)} {pluralize('card', len(e.cards))} " @@ -400,7 +402,7 @@ async def handle(e: mv.CardsPlacedBottomOfDeck) -> None: async def handle(e: mv.ImmunityGranted) -> None: player = game.get_player(e.player) is_client = player is game.client_player - print( + await aprint( f"{'You' if is_client else player.username} " f"{'have' if is_client else 'has'} been granted immunity." ) @@ -410,11 +412,15 @@ async def handle(e: mv.CardsSwapped) -> None: king_player, target = map(game.get_player, (e.player, e.opponent)) if game.client_player in (king_player, target): opponent = target if game.client_player is king_player else king_player - print(f"You and {opponent.username} swap your cards.") - print(f"You give a {opponent.round_player.hand.card.name}.") - print(f"You get a {game.client_player.round_player.hand.card.name}.") + await aprint(f"You and {opponent.username} swap your cards.") + await aprint(f"You give a {opponent.round_player.hand.card.name}.") + await aprint( + f"You get a {game.client_player.round_player.hand.card.name}." + ) else: - print(f"{king_player.username} and {target.username} swap their cards.") + await aprint( + f"{king_player.username} and {target.username} swap their cards." + ) # --------------------------------- Helpers ---------------------------------- async def _player_choice( @@ -434,7 +440,9 @@ async def _player_choice( choice = await async_ask_valid_input(prompt, choices=choices) return choice.value else: - print("There are no valid targets, playing this card has no effect.") + await aprint( + "There are no valid targets, playing this card has no effect." + ) # TODO: allow cancel return mv.OpponentChoice.NO_TARGET @@ -448,12 +456,12 @@ async def handle(e: None): # special case for first "event" in loop below @staticmethod async def _show_game_end(game: RemoteGameShadowCopy): assert game.ended - print_header("GAME OVER", filler="#") + await print_header("GAME OVER", filler="#") end: loveletter.game.GameEnd = game.state # noqa try: winner = end.winner except ValueError: - print_centered("There were multiple winners!") + await print_centered("There were multiple winners!") winner_message = ( f"{', '.join(p.username for p in end.winners)} all won in a tie." ) @@ -463,8 +471,8 @@ async def _show_game_end(game: RemoteGameShadowCopy): else: winner_message = f"{winner.username} wins!" - print_centered(f"🏆🏆🏆 {winner_message} 🏆🏆🏆") - print() + await print_centered(f"🏆🏆🏆 {winner_message} 🏆🏆🏆") + await aprint() class HostCLISession(CommandLineSession): @@ -500,7 +508,7 @@ async def manage(self): await super().manage() self._host_has_joined_server = asyncio.Event() - print_header( + await print_header( f"Hosting game on {', '.join(f'{h}:{p}' for h, p in self.server_addresses)}" ) server_process = ServerProcess.new( @@ -510,7 +518,7 @@ async def manage(self): show_logs=self.show_server_logs, ) with server_process: - print("Joining the server...", end=" ") # see _player_joined() + await aprint("Joining the server...", end=" ") # see _player_joined() connection_task = await self._connect_localhost() await self._host_has_joined_server.wait() await watch_connection( @@ -537,7 +545,7 @@ async def _connect_localhost(self) -> asyncio.Task: raise error async def _ready_to_play(self) -> RemoteGameShadowCopy: - print("Waiting for other players to join the server.") + await aprint("Waiting for other players to join the server.") game = None while game is None: await ainput("Enter anything when ready to play...\n") @@ -545,24 +553,24 @@ async def _ready_to_play(self) -> RemoteGameShadowCopy: try: game = await self.client.wait_for_game() except RemoteValidationError as e: - print(e.help_message, end="\n\n") + await aprint(e.help_message, end="\n\n") except RemoteException as e: - print("Error in server while creating game:") - print_exception(e) + await aprint("Error in server while creating game:") + await print_exception(e) return game async def _player_joined(self, message: msg.PlayerJoined): if message.username == self.user.username: - print("Done.") # see manage() + await aprint("Done.") # see manage() self._host_has_joined_server.set() else: await self._host_has_joined_server.wait() # synchronize prints - print(f"{message.username} joined the server") + await aprint(f"{message.username} joined the server") @staticmethod async def _player_left(message: msg.PlayerDisconnected): - print(f"{message.username} left the server") + await aprint(f"{message.username} left the server") class GuestCLISession(CommandLineSession): @@ -578,7 +586,7 @@ def __init__(self, user: UserInfo, server_address: Address): async def manage(self): await super().manage() address = self.server_address - print_header(f"Joining game @ {address.host}:{address.port}") + await print_header(f"Joining game @ {address.host}:{address.port}") connection_task = await self._connect_to_server() await watch_connection( connection_task, main_task=self._manage_after_connection_established() @@ -599,8 +607,8 @@ class ConnectionErrorOptions(enum.Enum): try: connection = await self.client.connect(*self.server_address) except (ConnectionError, LogonError) as e: - print("Error while trying to connect to the server:") - print_exception(e) + await aprint("Error while trying to connect to the server:") + await print_exception(e) choice = await async_ask_valid_input( "What would you like to do? (" "RETRY: retry connecting to this server; " @@ -619,9 +627,9 @@ class ConnectionErrorOptions(enum.Enum): else: assert False - print("Successfully connected to the server.") + await aprint("Successfully connected to the server.") return connection async def _wait_for_game(self) -> RemoteGameShadowCopy: - print("Waiting for the host to start the game...") + await aprint("Waiting for the host to start the game...") return await self.client.wait_for_game() diff --git a/src/loveletter_cli/ui/board.py b/src/loveletter_cli/ui/board.py index eaabeec..4636f37 100644 --- a/src/loveletter_cli/ui/board.py +++ b/src/loveletter_cli/ui/board.py @@ -6,6 +6,7 @@ import more_itertools import numpy as np +from aioconsole import aprint from loveletter.cardpile import Deck, STANDARD_DECK_COUNTS from loveletter.cards import Card, CardType, Guard @@ -22,7 +23,7 @@ DEFAULT_CARD_WIDTH = DEFAULT_CARD_HEIGHT * CARD_ASPECT -def draw_game( +async def draw_game( game: RemoteGameShadowCopy, reveal: bool = False, player_card_size: int = 15, @@ -73,11 +74,11 @@ def username(p) -> str: else: return label - def print_blank_line(): - print(" " * board_cols) + async def print_blank_line(): + await aprint(" " * board_cols) - print_blank_line() - print_blank_line() + await print_blank_line() + await print_blank_line() # number of extra rows of players (one to the left one to the right) extra_player_rows = math.ceil((game_round.num_players - 2) / 2) @@ -89,11 +90,11 @@ def print_blank_line(): if reveal else [card_back_sprite(char="#")] * len(opposite.hand) ) - print_canvas(horizontal_join(sprites), align="^", width=board_cols) - print_blank_line() - print(format(username(opposite), center_fmt)) - print(format(cards_discarded_string(opposite), center_fmt)) - print_blank_line() + await print_canvas(horizontal_join(sprites), align="^", width=board_cols) + await print_blank_line() + await aprint(format(username(opposite), center_fmt)) + await aprint(format(cards_discarded_string(opposite), center_fmt)) + await print_blank_line() len_stack = len(game.current_round.deck.stack) num_set_aside = int(game.current_round.deck.set_aside is not None) @@ -107,9 +108,9 @@ def print_blank_line(): if game_round.num_players <= 2: assert extra_player_rows == 0 # print "economical" representation of deck to avoid increasing vertical length - print_blank_line() - print(format(deck_msg, center_fmt)) - print_blank_line() + await print_blank_line() + await aprint(format(deck_msg, center_fmt)) + await print_blank_line() else: # canvases for center strip: main -> the card sprites, footer -> the labels # make dummy sprite to make sure we get the dimensions right @@ -180,19 +181,19 @@ def print_blank_line(): center_block = underlay(base=center_block, layer=deck_layer) # print everything in this central strip: - print_canvas(center_block) + await print_canvas(center_block) - print_blank_line() + await print_blank_line() # this client's hand sprites = [card_sprite(c, size=player_card_size) for c in you.hand] - print_canvas(horizontal_join(sprites), align="^", width=board_cols) - print_blank_line() - print(format(username(you), center_fmt)) - print(format(cards_discarded_string(you), center_fmt)) + await print_canvas(horizontal_join(sprites), align="^", width=board_cols) + await print_blank_line() + await aprint(format(username(you), center_fmt)) + await aprint(format(cards_discarded_string(you), center_fmt)) - print_blank_line() - print_blank_line() + await print_blank_line() + await print_blank_line() # ---------------------------------- canvas creation ---------------------------------- @@ -511,7 +512,9 @@ def write_string( # ------------------------------------- utilities ------------------------------------- -def print_canvas(canvas: np.ndarray, align="", width: Optional[int] = None) -> None: +async def print_canvas( + canvas: np.ndarray, align="", width: Optional[int] = None +) -> None: """ Print the given canvas to stdout. @@ -525,4 +528,4 @@ def print_canvas(canvas: np.ndarray, align="", width: Optional[int] = None) -> N width = canvas.shape[1] fmt = f"{align}{width}" for row in canvas: - print(format(as_string(row), fmt)) + await aprint(format(as_string(row), fmt)) diff --git a/src/loveletter_cli/ui/input.py b/src/loveletter_cli/ui/input.py index b1cdcf3..55915bb 100644 --- a/src/loveletter_cli/ui/input.py +++ b/src/loveletter_cli/ui/input.py @@ -5,7 +5,7 @@ import more_itertools import valid8 -from aioconsole import ainput +from aioconsole import ainput, aprint from .misc import printable_width @@ -14,19 +14,6 @@ _DEFAULT = object() -def ask_valid_input(*args, **kwargs) -> _T: - error_message, parser, prompt, validation_errors = _ask_valid_input_parse_args( - *args, **kwargs - ) - - while True: - raw_input = input(prompt) - try: - return _parse_input(raw_input, parser, error_message, validation_errors) - except (valid8.ValidationError, *validation_errors): - continue - - async def async_ask_valid_input(*args, **kwargs): """Asynchronous version of :func:`ask_valid_input`.""" error_message, parser, prompt, validation_errors = _ask_valid_input_parse_args( @@ -36,7 +23,9 @@ async def async_ask_valid_input(*args, **kwargs): while True: raw_input = await ainput(prompt) try: - return _parse_input(raw_input, parser, error_message, validation_errors) + return await _parse_input( + raw_input, parser, error_message, validation_errors + ) except (valid8.ValidationError, *validation_errors): continue @@ -79,7 +68,6 @@ def parser(s: str) -> choices: s = s.casefold() else: case_fold = False - print("(interpreted case-*sensitively*)") normalized_members = ( { @@ -128,15 +116,15 @@ def parser(s: str, wrapped=parser) -> _T: return error_message, parser, prompt, validation_errors -def _parse_input(raw_input: str, parser, error_message, validation_errors) -> _T: +async def _parse_input(raw_input: str, parser, error_message, validation_errors) -> _T: raw_input = raw_input.strip() try: return parser(raw_input) except valid8.ValidationError as exc: - print(error_message.format(choice=raw_input, error=exc.get_help_msg())) + await aprint(error_message.format(choice=raw_input, error=exc.get_help_msg())) raise except validation_errors as exc: - print(error_message.format(choice=raw_input, error=exc)) + await aprint(error_message.format(choice=raw_input, error=exc)) raise @@ -148,7 +136,7 @@ def _decorate_prompt(prompt: str) -> str: return "\n".join(lines) -ask_valid_input.__doc__ = f""" +async_ask_valid_input.__doc__ = f""" Ask for user input until it satisfies a given validator. {_ask_valid_input_parse_args.__doc__} diff --git a/src/loveletter_cli/ui/misc.py b/src/loveletter_cli/ui/misc.py index 9ef91cc..ad49aef 100644 --- a/src/loveletter_cli/ui/misc.py +++ b/src/loveletter_cli/ui/misc.py @@ -3,6 +3,7 @@ import traceback import valid8 +from aioconsole import aprint from multimethod import multimethod from loveletter_multiplayer import RemoteException @@ -15,32 +16,32 @@ def printable_width() -> int: @valid8.validate_arg("filler", valid8.validation_lib.length_between(1, 1)) -def print_header(text: str, filler: str = "-"): - print() +async def print_header(text: str, filler: str = "-"): + await aprint() width = printable_width() - print(format(f" {text} ", f"{filler}^{width - 1}"), end="\n\n") + await aprint(format(f" {text} ", f"{filler}^{width - 1}"), end="\n\n") @valid8.validate_arg("line", lambda s: "\n" not in s) -def print_centered(line: str): - print(format(line, f"^{printable_width()}")) +async def print_centered(line: str): + await aprint(format(line, f"^{printable_width()}")) @multimethod -def print_exception(exception: BaseException): +async def print_exception(exception: BaseException): text = "\n".join(traceback.format_exception_only(type(exception), exception)) - return _gcd_print_exception(text) + return await _gcd_print_exception(text) @print_exception.register -def print_exception(exception: RemoteException): +async def print_exception(exception: RemoteException): text = f"{exception.exc_type.__name__}: {exception.exc_message}" - return _gcd_print_exception(text) + return await _gcd_print_exception(text) -def _gcd_print_exception(text: str): +async def _gcd_print_exception(text: str): text = textwrap.indent(text, prefix=" " * 4 + "!!! ") - print(text, end="\n\n") + await aprint(text, end="\n\n") def pluralize(word: str, count: int) -> str: