Opsdroid-RasaSR/__init__.py

90 lines
4.1 KiB
Python
Raw Normal View History

2021-11-03 11:57:55 +00:00
from opsdroid.skill import Skill
from opsdroid.matchers import match_rasanlu, match_parse
import random
2021-11-12 12:28:05 +00:00
import importlib
2021-11-03 11:57:55 +00:00
class Respond(Skill):
2021-11-12 11:59:33 +00:00
# Constructor gets configs for the skill and applies the decorator
# to match using rasa.
2021-11-03 11:57:55 +00:00
def __init__(self, opsdroid, config):
super().__init__(opsdroid, config)
2021-11-12 11:59:33 +00:00
self._get_configs()
2021-11-03 11:57:55 +00:00
self.respond = match_rasanlu(self.intent)(self.respond)
2021-11-12 11:59:33 +00:00
# This utility function will grab the relevent config variables
# for this skill to function.
def _get_configs(self):
# The intent to match.
self.intent = self.config.get("intent")
# Possible responses given a match.
self.response_options = self.config.get("response_options")
# A bool to check if we want to use entities in our response.
self.use_entities = self.config.get("use_entities", False)
# Fallback responses if we do not get the required entities
# from Rasa.
self.fallback_response_options = self.config.get("fallback_response_options",
["Sorry, Rasa did not extract sufficient entities to fulfil your request"])
2021-11-12 12:28:05 +00:00
# An optional callback used for more advanced responses. Such
# functions should simply return the final response string and
# will be given the message and extracted entities (if enabled).
self.callback_file = self.config.get("callback_file", None)
self.callback_name = self.config.get("callback_name", None)
2021-11-12 11:59:33 +00:00
# Extract the entities from the rasanlu response
def _get_entities(self, message):
# This includes data such as the extractor used, which we
# don't need.
raw_entities = message.rasanlu["entities"]
entities = {}
for entity in raw_entities:
entity_type = entity["entity"]
unique_entity_type = self._get_unique_key(entity_type, entities)
entity_value = entity["value"]
entities[unique_entity_type] = entity_value
return entities
# Appends the smallest integer to the name of the key in order to
# create a new key that is not already in the dictionary.
def _get_unique_key(self, base_key, dictionary):
counter = 0
while True:
unique_key_attempt = base_key + str(counter)
if unique_key_attempt in dictionary:
counter += 1
continue
return unique_key_attempt
# Takes the response and attempts to replace all entities in it
# with the extracted entities. Failing that, we will fall back to a
# fallback response.
def _substitute_entities(self, response, entities):
try:
response = response.format(**entities)
except KeyError as e:
chosen_fallback = random.choice(self.fallback_response_options)
return chosen_fallback
return response
2021-11-03 11:57:55 +00:00
# This line is to add properties to respond on declaration
# as doing this in __init__ would cause an exception to be
# raised. If you don't believe me, remove it!
@match_parse("hgftgyhjknbvhgftyuihjkvhgftyuijkbjhgtyyuijhkjghfdtrytyihujkbvncgfxdtrytuyiuhkbjvbhcfdytryuhjkbvhcfdyrtuyiuhkbhj")
async def respond(self, message):
2021-11-12 11:59:33 +00:00
chosen_response = random.choice(self.response_options)
2021-11-12 12:28:05 +00:00
entities = None # In case we want callbacks without entities
2021-11-12 11:59:33 +00:00
if self.use_entities:
entities = self._get_entities(message)
chosen_response = self._substitute_entities(chosen_response, entities)
2021-11-12 12:28:05 +00:00
if self.callback_file != None and self.callback_name != None:
2021-11-22 18:03:48 +00:00
# The following three lines allow us to import the module using an absolute path dynamically
spec = importlib.util.spec_from_file_location("callback", self.callback_file)
callback_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(callback_module)
# Now we can get the response from the callback.
2021-11-12 12:28:05 +00:00
chosen_response = eval("callback_module." + self.callback_name + "(message, entities)")
await message.respond(chosen_response)