From 0c0ce7b0db4503c1bf3b4e604e807f4b77e4bf40 Mon Sep 17 00:00:00 2001 From: Stefano Pigozzi Date: Wed, 30 Dec 2020 17:05:25 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Create=20Command=20and=20PartialCom?= =?UTF-8?q?mand?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/source/autodoc/engineer.rst | 6 ++ royalnet/engineer/__init__.py | 1 + royalnet/engineer/command.py | 142 ++++++++++++++++++++++++++++++ royalnet/engineer/conversation.py | 42 ++++++--- 4 files changed, 181 insertions(+), 10 deletions(-) create mode 100644 royalnet/engineer/command.py diff --git a/docs/source/autodoc/engineer.rst b/docs/source/autodoc/engineer.rst index 188dc28e..341479bc 100644 --- a/docs/source/autodoc/engineer.rst +++ b/docs/source/autodoc/engineer.rst @@ -10,6 +10,12 @@ .. automodule:: royalnet.engineer.bullet +``command`` +----------- + +.. automodule:: royalnet.engineer.command + + ``conversation`` ----------- diff --git a/royalnet/engineer/__init__.py b/royalnet/engineer/__init__.py index 1ab3c5ee..80e05f81 100644 --- a/royalnet/engineer/__init__.py +++ b/royalnet/engineer/__init__.py @@ -5,6 +5,7 @@ All names are inspired by the `Engineer Class of Team Fortress 2 t.Optional[c.ConversationProtocol]: + log.debug(f"Awaiting a bullet...") + bullet: b.Bullet = await _sentry + + log.debug(f"Received: {bullet!r}") + + log.debug(f"Ensuring a message was received: {bullet!r}") + if not isinstance(bullet, b.Message): + log.debug(f"Returning: {bullet!r} is not a message") + return + + log.debug(f"Getting message text of: {bullet!r}") + if not (text := await bullet.text()): + log.debug(f"Returning: {bullet!r} has no text") + return + + log.debug(f"Searching for pattern: {text!r}") + if not (match := self.pattern.search(text)): + log.debug(f"Returning: Couldn't find pattern in {text!r}") + return + + log.debug(f"Match successful, getting capture groups of: {match!r}") + message_kwargs: t.Dict[str, str] = match.groupdict() + + log.debug(f"Passing args to function: {message_kwargs!r}") + return await super().run(_sentry=_sentry, **base_kwargs, **message_kwargs) + + def help(self) -> t.Optional[str]: + """ + Get help about this command. This defaults to returning the docstring of :field:`.f` . + + :return: The help :class:`str` for this command, or :data:`None` if the command has no docstring. + """ + return self.f.__doc__ + + def __repr__(self): + return f"<{self.__class__.__qualname__}: {self.name}>" + + +class PartialCommand: + """ + A PartialCommand is a :class:`.Command` having an unknown :attr:`~.Command.pattern` at the moment of creation. + + The pattern can specified later using :meth:`.complete`. + """ + def __init__(self, f: c.ConversationProtocol, name: str, syntax: str): + self.f: c.ConversationProtocol = f + """ + The function to pass to :attr:`.c.Conversation.f`. + """ + + if not self.name.isalnum(): + raise ValueError("Name should be alphanumeric") + if not self.name.islower(): + raise ValueError("Name should be lowercase") + + self.name: str = name + """ + The name of the command to pass to :attr`.Command.name`. Should be lowercase and containing only alphanumeric + characters. + """ + + self.syntax: str = syntax + """ + Part of the pattern from where the arguments should be captured. + """ + + @classmethod + def new(cls, *args, **kwargs): + """ + A decorator that instantiates a new :class:`Conversation` object using the decorated function. + + :return: The created :class:`Conversation` object. + It can still be called in the same way as the previous function! + """ + def decorator(f: c.ConversationProtocol): + partial_command = cls(f=f, *args, **kwargs) + log.debug(f"Created: {partial_command!r}") + return decorator + + def complete(self, pattern: str) -> Command: + """ + Complete the PartialCommand with a pattern, creating a :class:`Command` object. + + :param pattern: The pattern to add to the PartialCommand. It is first :meth:`str.format`\\ ted with the keyword + arguments ``name=self.name, syntax=self.syntax`` and later :func:`re.compile`\\ d. + :return: The complete :class:`Command`. + """ + pattern: re.Pattern = re.compile(pattern.format(name=self.name, syntax=self.syntax)) + return Command(f=self.f, name=self.name, pattern=pattern) + + def __repr__(self): + return f"<{self.__class__.__qualname__}: {self.name}>" + + +# Objects exported by this module +__all__ = ( + "Command", + "PartialCommand", +) diff --git a/royalnet/engineer/conversation.py b/royalnet/engineer/conversation.py index df89ffda..697ddae7 100644 --- a/royalnet/engineer/conversation.py +++ b/royalnet/engineer/conversation.py @@ -1,24 +1,30 @@ +# Module docstring """ Conversations are wrapper classes that can be applied to functions which await :class:`~royalnet.engineer.bullet.Bullet`\\ s from a :class:`~royalnet.engineer.sentry.Sentry`. """ +# Special imports from __future__ import annotations import royalnet.royaltyping as t +# External imports import logging -from . import sentry +# Internal imports +from . import sentry as s +# Special global objects log = logging.getLogger(__name__) +# Code class ConversationProtocol(t.Protocol): """ Typing annotation for Conversation functions. """ - def __call__(self, *, _sentry: sentry.Sentry, **kwargs) -> t.Awaitable[t.Optional[ConversationProtocol]]: + def __call__(self, *, _sentry: s.Sentry, **kwargs) -> t.Awaitable[t.Optional[ConversationProtocol]]: ... @@ -27,11 +33,18 @@ class Conversation: The base class for Conversations. It does nothing on its own except providing better debug information. """ - def __init__(self, f: ConversationProtocol): + def __init__(self, f: ConversationProtocol, *_, **__): self.f: ConversationProtocol = f + """ + The function that was wrapped by this conversation. + + It can be called by calling this object as it was a function:: + + my_conv(_sentry=_sentry, _msg=msg) + """ @classmethod - def new(cls): + def new(cls, *args, **kwargs): """ A decorator that instantiates a new :class:`Conversation` object using the decorated function. @@ -39,16 +52,25 @@ class Conversation: It can still be called in the same way as the previous function! """ def decorator(f: ConversationProtocol): - c = cls(f=f) - log.debug(f"Created: {repr(c)}") + c = cls(f=f, *args, **kwargs) + log.debug(f"Created: {c!r}") return decorator - def __call__(self, *, _sentry: sentry.Sentry, **kwargs) -> t.Awaitable[t.Optional[ConversationProtocol]]: - log.debug(f"Calling: {repr(self)}") - return self.f(_sentry=_sentry, **kwargs) + def __call__(self, *, _sentry: s.Sentry, **kwargs) -> t.Awaitable[t.Optional[ConversationProtocol]]: + log.debug(f"Calling: {self!r}") + return self.run(_sentry=_sentry, **kwargs) + + async def run(self, *, _sentry: s.Sentry, **kwargs) -> t.Optional[ConversationProtocol]: + """ + The coroutine function that generates the coroutines returned by :meth:`__call__` . + + It is a conversation itself. + """ + log.debug(f"Running: {self!r}") + return await self.f(_sentry=_sentry, **kwargs) def __repr__(self): - return f"" + return f"<{self.__class__.__qualname__} #{id(self)}>" __all__ = (