Skip to content

from pax25.applications.parsers import no_arguments

Command Router

The command router is built to allow for easy routing of commands. In most applications, the user has the option to enter one of several commands. The command router allows for specification of these commands, basic parsing, and execution of the commands.

When paired with the help system, it allows for easy programming of applications.

Initializing a Command Router

To build a command router, import the CommandRouter class and invoke it. You will probably want to do this in the setup class, saving it on self:

from pax25 import Application
from pax25.applications.router import CommandRouter


class MyApplication(Application):
    ...
    def setup(self):
        ...
        self.router = CommandRouter()

Adding commands

Once you have a command router, you can add commands to it. Adding commands to a router requires building a CommandSpec and using the .add method:

from pax25 import Application
from pax25.applications.router import CommandRouter, CommandSpec
from pax25.applications.utils import send_message


class MyApplication(Application):
    ...
    def setup(self):
        ...
        self.router = CommandRouter()
        # Create the command
        command = CommandSpec(
            command='Hello',
            function=self.say_hello,
            # See the article on the help system for more information on this feature.
            help="Greets the user.",
            aliases=('Hi', 'hola'),
        )
        # Add it to the router.
        self.router.add(command)

    def say_hello(self, connection, context):
        # This will say "Hi! You used the hello command!"
        send_message(connection, f"Hi! You used the {context.spec.command} command!")

Routing commands

Once commands are added to the router, you can send messages from the user to the command router, and it will execute the command. Commands can have their own parsers to interpret their arguments as needed.

from pax25 import Application
from pax25.applications.router import CommandRouter, CommandSpec
from pax25.applications.utils import send_message
from pax25.applications.parsers import no_arguments, ParseError

def positive_number_parser(args):
    """
    A simple parser that converts the arguments into a number.
    """
    try:
        number = int(args)
    except ValueError:
        raise ParseError("That's not a number!")
    if number <= 0:
        raise ParseError("The number must be a positive integer!")
    return number

class MyApplication(Application):
    ...

    def setup(self):
        ...
        self.router = CommandRouter()
        # Create the command
        hello_command = CommandSpec(
            command='Hello',
            function=self.say_hello,
            # See the article on the help system for more information on this feature.
            help="Greets the user.",
            aliases=('Hi', 'hola'),
            # Parsers for commands are pluggable. This one indicates that no
            # arguments are used. These functions take the raw arguments as a string,
            # and then their return value is set as the .args value on the
            # context.
            parser=no_arguments,
        )
        countdown_command = CommandSpec(
            command='countdown',
            function=self.countdown,
            help="Counts down from a number.",
            aliases=('blastoff',),
            # At the bottom of this example file, you'll find the implementation for
            # number_parser.
            parser=positive_number_parser,
        )
        # Add these to the router.
        self.router.add(hello_command)
        self.router.add(countdown_command)

    def say_hello(self, connection, context):
        # This will say "Hi! You used the hello command!"
        send_message(connection, f"Hi! You used the {context.spec.command} command!")

    def countdown(self, connection, context):
        # This will count down from the number supplied.
        # context.args is an integer here, since that's the return value of
        # positive_number_parser.
        current = context.args
        while current > 0:
            send_message(connection, f"{current}!")
            current -= 1
        send_message(connection, "Blast off!")

    def on_message(self, connection, message: str):
        # Will lookup the command (if it exists) and run it, or else send a useful error.
        self.router.route(connection, message)

pax25.applications.router.CommandRouter

Router object that allows us to quickly route to command functions based on a command string sent by a user.

Source code in pax25/applications/router.py
class CommandRouter:
    """
    Router object that allows us to quickly route to command functions based on a
    command string sent by a user.
    """

    def __init__(
        self,
        *,
        default: CommandSpec[Any] = default_command,
        post_command_func: Callable[[Connection], None] = default_post_command_func,
    ) -> None:
        """
        Creates an autocompleting command map that handles input and runs any matching
        command.
        """
        self.command_store: AutocompleteDict[CommandSpec[Any]] = AutocompleteDict()
        # Canonical listing of all command names, used for checking conflicts.
        self.command_set: set[str] = set()
        # Canonical listing of all aliases, used for checking conflicts.
        self.alias_set: set[str] = set()
        self.default = default
        # A 'post command' that runs after a command is complete.
        self.post_command = post_command_func

    @property
    def help_available(self) -> bool:
        """
        Returns a boolean indicating if there's a help command installed.
        """
        try:
            results = self.command_store["HELP"]
            # Help command is ambiguous if there's more than 1.
            # This could still be wrong if there is an entry named something like
            # 'helpmeplz' and it's not a help command. But we're going to discount that
            # possibility here. If it ever becomes a real problem we'll refactor this
            # to something more robust.
            return len(results) == 1
        except KeyError:
            return False

    def add(self, *args: CommandSpec[Any]) -> None:
        """
        Add commands to the command router.
        """
        for arg in args:
            command = arg.command.upper()
            aliases = set(alias.upper() for alias in arg.aliases)
            to_check = (command, *aliases)
            for entry in to_check:
                if entry in self.command_set or entry in self.alias_set:
                    existing_spec = self.command_store[entry]
                    raise ConfigurationError(
                        f"Found preexisting entry with conflicting name or "
                        f"aliases when adding spec {repr(arg)}. Conflicting "
                        f"entry was: {repr(existing_spec)}"
                    )
            for entry in to_check:
                self.command_store[entry] = arg
            self.command_set |= {command}
            self.alias_set |= aliases

    def remove(self, *args: CommandSpec[Any]) -> None:
        """
        Remove commands from the command router.
        """
        for arg in args:
            command = arg.command.upper()
            if command not in self.command_store.store:
                raise KeyError(f"Command does not exist, {repr(arg.command)}")
            if self.command_store.store[command] != arg:
                raise KeyError(
                    f"Command {repr(arg.command)} exists, but is for a different spec!"
                )
            del self.command_store[command]
            self.command_set -= {command}
            for alias in arg.aliases:
                del self.command_store[alias.upper()]
                self.alias_set -= {alias.upper()}

    def route(self, connection: Connection, raw_command: str) -> None:
        """
        Routes a user to a command function based on their selection, or gives them
        a hint otherwise.
        """
        segments = raw_command.split(maxsplit=1)
        try:
            first_segment = segments[0]
            command = first_segment.upper()
        except IndexError:
            # No command specified-- it was an empty string. Run the default.
            context = CommandContext(
                command="",
                args="",
                spec=self.default,
                raw_input="",
            )
            self.default.function(connection, context)
            self.post_command(connection)
            return
        args = ""
        if len(segments) == 2:
            args = segments[1]
        try:
            candidates = self.command_store[command]
        except KeyError:
            context = CommandContext(
                command=first_segment,
                args=args,
                spec=self.default,
                raw_input=raw_command,
            )
            self.default.function(connection, context)
            self.post_command(connection)
            return
        if len(candidates) > 1:
            possibilities = sorted(entry.command for entry in candidates)
            send_message(
                connection,
                "Ambiguous command. "
                f"Did you mean one of these?: {', '.join(possibilities)}",
            )
            return
        [spec] = candidates
        try:
            parser_spec = ParserSpec(
                raw_input=raw_command,
                command=command,
                args=args,
                connection=connection,
                command_spec=spec,
            )
            parsed_args = spec.parser(parser_spec)
            context = CommandContext(
                command=first_segment,
                args=parsed_args,
                spec=spec,
                raw_input=raw_command,
            )
        except ParseError as err:
            error_string = str(err)
            if self.help_available:
                error_string += f"\rTry: help {spec.command}"
            send_message(connection, error_string)
            self.post_command(connection)
            return
        spec.function(connection, context)
        self.post_command(connection)

help_available: bool property

Returns a boolean indicating if there's a help command installed.

__init__(*, default: CommandSpec[Any] = default_command, post_command_func: Callable[[Connection], None] = default_post_command_func) -> None

Creates an autocompleting command map that handles input and runs any matching command.

Source code in pax25/applications/router.py
def __init__(
    self,
    *,
    default: CommandSpec[Any] = default_command,
    post_command_func: Callable[[Connection], None] = default_post_command_func,
) -> None:
    """
    Creates an autocompleting command map that handles input and runs any matching
    command.
    """
    self.command_store: AutocompleteDict[CommandSpec[Any]] = AutocompleteDict()
    # Canonical listing of all command names, used for checking conflicts.
    self.command_set: set[str] = set()
    # Canonical listing of all aliases, used for checking conflicts.
    self.alias_set: set[str] = set()
    self.default = default
    # A 'post command' that runs after a command is complete.
    self.post_command = post_command_func

add(*args: CommandSpec[Any]) -> None

Add commands to the command router.

Source code in pax25/applications/router.py
def add(self, *args: CommandSpec[Any]) -> None:
    """
    Add commands to the command router.
    """
    for arg in args:
        command = arg.command.upper()
        aliases = set(alias.upper() for alias in arg.aliases)
        to_check = (command, *aliases)
        for entry in to_check:
            if entry in self.command_set or entry in self.alias_set:
                existing_spec = self.command_store[entry]
                raise ConfigurationError(
                    f"Found preexisting entry with conflicting name or "
                    f"aliases when adding spec {repr(arg)}. Conflicting "
                    f"entry was: {repr(existing_spec)}"
                )
        for entry in to_check:
            self.command_store[entry] = arg
        self.command_set |= {command}
        self.alias_set |= aliases

remove(*args: CommandSpec[Any]) -> None

Remove commands from the command router.

Source code in pax25/applications/router.py
def remove(self, *args: CommandSpec[Any]) -> None:
    """
    Remove commands from the command router.
    """
    for arg in args:
        command = arg.command.upper()
        if command not in self.command_store.store:
            raise KeyError(f"Command does not exist, {repr(arg.command)}")
        if self.command_store.store[command] != arg:
            raise KeyError(
                f"Command {repr(arg.command)} exists, but is for a different spec!"
            )
        del self.command_store[command]
        self.command_set -= {command}
        for alias in arg.aliases:
            del self.command_store[alias.upper()]
            self.alias_set -= {alias.upper()}

route(connection: Connection, raw_command: str) -> None

Routes a user to a command function based on their selection, or gives them a hint otherwise.

Source code in pax25/applications/router.py
def route(self, connection: Connection, raw_command: str) -> None:
    """
    Routes a user to a command function based on their selection, or gives them
    a hint otherwise.
    """
    segments = raw_command.split(maxsplit=1)
    try:
        first_segment = segments[0]
        command = first_segment.upper()
    except IndexError:
        # No command specified-- it was an empty string. Run the default.
        context = CommandContext(
            command="",
            args="",
            spec=self.default,
            raw_input="",
        )
        self.default.function(connection, context)
        self.post_command(connection)
        return
    args = ""
    if len(segments) == 2:
        args = segments[1]
    try:
        candidates = self.command_store[command]
    except KeyError:
        context = CommandContext(
            command=first_segment,
            args=args,
            spec=self.default,
            raw_input=raw_command,
        )
        self.default.function(connection, context)
        self.post_command(connection)
        return
    if len(candidates) > 1:
        possibilities = sorted(entry.command for entry in candidates)
        send_message(
            connection,
            "Ambiguous command. "
            f"Did you mean one of these?: {', '.join(possibilities)}",
        )
        return
    [spec] = candidates
    try:
        parser_spec = ParserSpec(
            raw_input=raw_command,
            command=command,
            args=args,
            connection=connection,
            command_spec=spec,
        )
        parsed_args = spec.parser(parser_spec)
        context = CommandContext(
            command=first_segment,
            args=parsed_args,
            spec=spec,
            raw_input=raw_command,
        )
    except ParseError as err:
        error_string = str(err)
        if self.help_available:
            error_string += f"\rTry: help {spec.command}"
        send_message(connection, error_string)
        self.post_command(connection)
        return
    spec.function(connection, context)
    self.post_command(connection)