Command Interface

This is the code reference for the Command system interface.

open_mafia_engine.commands.raw

MafiaBadCommand

Bad command was passed.

RawCommand dataclass

Raw parsed command

open_mafia_engine.commands.lobby

AbstractLobby

A lobby, with mappings to user-defined objects.

Attributes

admins : dict Mapping of admin names to user objects. Override this. players : dict Mapping of player names to user objects. Override this.

admins: Dict[str, TUser] property readonly

Return mapping of str to admins.

players: Dict[str, TUser] property readonly

Return mapping of str to players.

add_admin(self, name: str, user: TUser)

Add the admin to the game.

Source code in commands/lobby.py

@abstractmethod
def add_admin(self, name: str, user: TUser):
    """Add the admin to the game."""

add_player(self, name: str, user: TUser)

Add the player to the game.

Source code in commands/lobby.py

@abstractmethod
def add_player(self, name: str, user: TUser):
    """Add the player to the game."""

remove_admin(self, name: str, user: TUser)

Remove the admin from the game.

Source code in commands/lobby.py

@abstractmethod
def remove_admin(self, name: str, user: TUser):
    """Remove the admin from the game."""

remove_player(self, name: str, user: TUser)

Remove the player from the game.

Source code in commands/lobby.py

@abstractmethod
def remove_player(self, name: str, user: TUser):
    """Remove the player from the game."""

AutoAddStrLobby

String-based lobby that adds any users automatically.

admins: Dict[str, str] property readonly

Return mapping of str to admins.

players: Dict[str, str] property readonly

Return mapping of str to players.

add_admin(self, name: str, user: str)

Add the admin to the game.

Source code in commands/lobby.py

def add_admin(self, name: str, user: str):
    assert name == user
    self._admin_names.add(name)

add_player(self, name: str, user: str)

Add the player to the game.

Source code in commands/lobby.py

def add_player(self, name: str, user: str):
    assert name == user
    self._player_names.add(name)

remove_admin(self, name: str, user: str)

Remove the admin from the game.

Source code in commands/lobby.py

def remove_admin(self, name: str, user: str):
    self._admin_names.discard(name)

remove_player(self, name: str, user: str)

Remove the player from the game.

Source code in commands/lobby.py

def remove_player(self, name: str, user: str):
    self._player_names.discard(name)

SimpleDictLobby

Simple lobby implementation.

admins: Dict[str, TUser] property readonly

Return mapping of str to admins.

players: Dict[str, TUser] property readonly

Return mapping of str to players.

add_admin(self, name: str, user: TUser)

Add the admin to the game.

Source code in commands/lobby.py

def add_admin(self, name: str, user: TUser):
    self._admins[name] = user

add_player(self, name: str, user: TUser)

Add the player to the game.

Source code in commands/lobby.py

def add_player(self, name: str, user: TUser):
    self._players[name] = user

remove_admin(self, name: str, user: TUser)

Remove the admin from the game.

Source code in commands/lobby.py

def remove_admin(self, name: str, user: TUser):
    if name in self._admins:
        del self._admins[name]

remove_player(self, name: str, user: TUser)

Remove the player from the game.

Source code in commands/lobby.py

def remove_player(self, name: str, user: TUser):
    if name in self._players:
        del self._players[name]

open_mafia_engine.commands.parser

AbstractCommandParser

Base for command parsers.

parse(self, source: str, obj: str) -> List[RawCommand]

Parses obj into zero, one or more RawCommand objects.

Source code in commands/parser.py

@abstractmethod
def parse(self, source: str, obj: str) -> List[RawCommand]:
    """Parses `obj` into zero, one or more `RawCommand` objects."""
    return []

ShellCommandParser

Implementation for shell-like command parsing.

parse(self, source: str, obj: str) -> List[RawCommand]

Parses obj into zero, one or more RawCommand objects.

Source code in commands/parser.py

def parse(self, source: str, obj: str) -> List[RawCommand]:
    if not isinstance(obj, str):
        # raise TypeError(f"Expected str, got {obj!r}")
        logger.warning(f"Cannot parse object, ignoring: {obj!r}")
        return []

    text = obj
    # TODO: keyword and flag arguments? Not just positional :)
    raw = shlex.split(text, posix=True)
    res = RawCommand(source, raw[0], args=tuple(raw[1:]))
    return [res]

open_mafia_engine.commands.runner

CommandHandler

Descriptor, wrapper for command functions.

Parameters

func : Callable The wrapped function. game : bool If True, allows use during the game. Default is True. lobby : bool If True, allows use in the lobby. Default is False. admin : bool If True, this is an admin-only command. name : None or str Command name. If None (default), uses function's name.

CommandRunner

Interprets and runs commands on the Mafia engine.

This handles both pre-game commands (such as joining or starting the game) and commands during the game. Basic Mafia commands are already added.

To add or override a command handler, use the command decorator:

class MyRunner(CommandRunner[MyUserType]):
    @command("foo", game=True)
    def my_foo(self, source: str, *args, **kwargs):
        return

The command decorator defaults to non-admin, in-game actions. Specifying lobby=True will, by default, allow only lobby use. If you want it always on, use @command("name", game=True, lobby=True)

Commands that don't match any given command will call default_command(). The default implementation raises a MafiaBadCommand exception, but you may override it (e.g. to ignore it).

Attributes

parser : AbstractCommandParser Your own parser implementation, or ShellCommandParser() by default. lobby : AbstractLobby Your own lobby implementation, or SimpleDictLobbyTUser by default. game : None or Game The game state. Defaults to None. score_cutoff : int The score cutoff for matching command names, from 0 to 100. Default is 80 (relatively strict).

in_game: bool property readonly

True if the game is active.

in_lobby: bool property readonly

True if the game is not active.

all_available_commands() -> List[str] classmethod

Returns all commands that are registered.

Source code in commands/runner.py

@classmethod
def all_available_commands(cls) -> List[str]:
    """Returns all commands that are registered."""
    return list(cls.registered_commands.keys())

change_phase(source: str, new_phase: Optional[str] = None)

Changes the phase (admin action).

create_game(source: str, builder_name: str, *args, **kwargs)

Creates the game.

    TODO: Need to standardize builder options.
    For example, 'game name' would be pretty important!

currently_available_commands(self, source: str = None) -> List[str]

Returns all commands that are currently available.

Source code in commands/runner.py

def currently_available_commands(self, source: str = None) -> List[str]:
    """Returns all commands that are currently available."""
    all_cmds = self.all_available_commands()

    def chk(cmd: str) -> bool:
        # FIXME
        try:
            self.pre_dispatch_check(cmd, source=source)
        except Exception:
            return False
        return True

    return [cmd for cmd in all_cmds if chk(cmd)]

default_command(self, source: str, name: str, *args, **kwargs) -> NoReturn

Default command definition - override.

Source code in commands/runner.py

def default_command(self, source: str, name: str, *args, **kwargs) -> NoReturn:
    """Default command definition - override."""
    rc = RawCommand(source, name, args, kwargs)
    raise MafiaBadCommand(rc)

destroy_game(source: str)

Stops the game immediately.

    Do we even need this? I assume you'd just make a new runner... :)

    TODO: Maybe a bit more elegantly?...
    TODO: Auto-stop the game when EGameEnded occurs? Not sure how that would work.

dispatch(self, rc: RawCommand) -> Any

Calls the relevant command given by rc.

Source code in commands/runner.py

def dispatch(self, rc: RawCommand) -> Any:
    """Calls the relevant command given by `rc`."""
    try:
        ch: CommandHandler = self.pre_dispatch_check(rc.name, source=rc.source)
    except KeyError:
        # Default command
        return self.default_command(rc.source, rc.name, *rc.args, **rc.kwargs)
    except Exception:
        raise

    # TODO: Possibly add debug output here?
    return ch.func(self, rc.source, *rc.args, **rc.kwargs)

do(source: str, abil_name: str, *args, **kwargs)

Activates an ability.

find_command(self, name: str) -> CommandHandler

Finds the closes command handler.

Source code in commands/runner.py

def find_command(self, name: str) -> CommandHandler:
    """Finds the closes command handler."""
    matcher = FuzzyMatcher(
        choices=self.registered_commands,
        score_cutoff=self.score_cutoff,
    )
    return matcher[name]

force_join(source: str, *targets: List[str])

Force one or more users to join the lobby.

join(source: str)

Joins the game as a player.

kick(source: str, *args, **kwargs)

Kick a player from the lobby or game.

    TODO: Implement!
    Will probably need change in AbstractLobby.
    Will definitely need change in Game to handle kicking/modkilling...

leave(source: str)

Leaves the lobby.

parse_and_run(self, source: str, obj: str) -> List[Tuple[RawCommand, Any]]

Parses commands and runs them.

Source code in commands/runner.py

def parse_and_run(self, source: str, obj: str) -> List[Tuple[RawCommand, Any]]:
    """Parses commands and runs them."""
    rcs = self.parser.parse(source, obj)
    res = []
    for rc in rcs:
        self.pre_check_command(rc)
        out_i = self.dispatch(rc)
        res.append((rcs, out_i))
    return res

pre_check_command(self, rc: RawCommand)

Override for additional command checking

Source code in commands/runner.py

def pre_check_command(self, rc: RawCommand):
    """Override for additional command checking"""

pre_dispatch_check(self, cmd: str, source: str = None) -> CommandHandler

Gets the command handler for a given command name, with checked access.

Checks whether the command is available for a certain source. If not - raises an exception.

Source code in commands/runner.py

def pre_dispatch_check(self, cmd: str, source: str = None) -> CommandHandler:
    """Gets the command handler for a given command name, with checked access.

    Checks whether the command is available for a certain source.
    If not - raises an exception.
    """
    try:
        ch: CommandHandler = self.find_command(cmd)
    except KeyError:
        # TODO: Instead return a default command? ...
        raise KeyError(f"No such command: {cmd!r}.")

    if ch.admin:
        if source not in self.lobby.admin_names:
            raise RuntimeError(f"Command {ch.name!r} requires admin privileges.")
    if self.in_lobby and not ch.lobby:
        raise RuntimeError(f"Command {ch.name!r} not available in lobby.")
    if self.in_game and not ch.game:
        raise RuntimeError(f"Command {ch.name!r} not available during game.")
    return ch

unvote(source: str, *args, **kwargs)

Unvotes from all previous votes.

user_for(self, name: str) -> Optional[TUser]

Gets the user by name from the lobby.

Source code in commands/runner.py

def user_for(self, name: str) -> Optional[TUser]:
    """Gets the user by name from the lobby."""
    return self.lobby.get(name, None)

vote(source: str, *args, **kwargs)

Votes for the target(s).

command(name_or_func: Union[str, Callable], *, game: bool = None, lobby: bool = None, admin: bool = False) -> Union[Callable]

Decorator to register methods as commands.

By default, "game" is True, "lobby" is not game, and "admin" is False.

Source code in commands/runner.py

def command(
    name_or_func: Union[str, Callable],
    *,
    game: bool = None,
    lobby: bool = None,
    admin: bool = False,
) -> Union[Callable]:
    """Decorator to register methods as commands.

    By default, "game" is True, "lobby" is `not game`, and "admin" is False.
    """

    if game is None:
        if lobby is None:
            game = True
            lobby = False
        else:
            game = not lobby
    else:
        if lobby is None:
            lobby = not game
    game = bool(game)
    lobby = bool(lobby)
    admin = bool(admin)

    if isinstance(name_or_func, str):

        def inner(func: Callable) -> CommandHandler:
            return CommandHandler(
                func, name=name_or_func, game=game, lobby=lobby, admin=admin
            )

        return inner
    elif callable(name_or_func):
        func = name_or_func
        return CommandHandler(func, game=game, lobby=lobby, admin=admin)
    else:
        raise TypeError(f"Argument to `command` is neither str, nor callable.")