import logging
from collections import namedtuple
import typing
from typing import Text, List, Dict, Any
from rasa_core.channels import OutputChannel
from rasa_core.nlg import NaturalLanguageGenerator
logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
from rasa_core.trackers import DialogueStateTracker
class Element(dict):
__acceptable_keys = ['title', 'item_url', 'image_url',
'subtitle', 'buttons']
def __init__(self, *args, **kwargs):
kwargs = {key: value
for key, value in kwargs.items()
if key in self.__acceptable_keys}
super(Element, self).__init__(*args, **kwargs)
# Makes a named tuple with entries text and data
BotMessage = namedtuple("BotMessage", "text data")
class Button(dict):
pass
[docs]class Dispatcher(object):
"""Send messages back to user"""
def __init__(self,
sender_id: Text,
output_channel: OutputChannel,
nlg: NaturalLanguageGenerator) -> None:
self.sender_id = sender_id
self.output_channel = output_channel
self.nlg = nlg
self.latest_bot_messages = []
[docs] def utter_response(self, message: Dict[Text, Any]) -> None:
"""Send a message to the client."""
bot_message = BotMessage(text=message.get("text"),
data={"elements": message.get("elements"),
"buttons": message.get("buttons"),
"attachment": message.get("image")})
self.latest_bot_messages.append(bot_message)
self.output_channel.send_response(self.sender_id, message)
[docs] def utter_message(self, text: Text) -> None:
""""Send a text to the output channel"""
# Adding the text to the latest bot messages (with no data)
bot_message = BotMessage(text=text,
data=None)
self.latest_bot_messages.append(bot_message)
self.output_channel.send_text_message(self.sender_id, text)
[docs] def utter_custom_message(self, *elements: Dict[Text, Any]) -> None:
"""Sends a message with custom elements to the output channel."""
bot_message = BotMessage(text=None,
data={"elements": elements})
self.latest_bot_messages.append(bot_message)
self.output_channel.send_custom_message(self.sender_id, elements)
[docs] def utter_attachment(self, attachment: Text) -> None:
"""Send a message to the client with attachments."""
bot_message = BotMessage(text=None,
data={"attachment": attachment})
self.latest_bot_messages.append(bot_message)
self.output_channel.send_image_url(self.sender_id, attachment)
# TODO: deprecate this function
[docs] def utter_template(self,
template: Text,
tracker: 'DialogueStateTracker',
silent_fail: bool = False,
**kwargs: Any
) -> None:
""""Send a message to the client based on a template."""
message = self._generate_response(template,
tracker,
silent_fail,
**kwargs)
if not message:
return
self.utter_response(message)
def _generate_response(self,
template: Text,
tracker: 'DialogueStateTracker',
silent_fail: bool = False,
**kwargs: Any
) -> Dict[Text, Any]:
""""Generate a response."""
message = self.nlg.generate(template, tracker,
self.output_channel.name(),
**kwargs)
if message is None and not silent_fail:
logger.error("Couldn't create message for template '{}'."
"".format(template))
return message