mirror of
https://github.com/JarvyJ/HomeIntent.git
synced 2022-02-11 01:01:05 +03:00
Add unit tests and refactor intents.py (#53)
* starting unit tests (there's a lot more to do!) * refactored intents class closes #17 and #39
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
from pydantic import AnyHttpUrl, BaseModel
|
||||
from typing import Set
|
||||
|
||||
from pydantic import AnyHttpUrl, BaseModel
|
||||
|
||||
from . import cover, fan, group, light, lock, remote, shopping_list, switch
|
||||
from .api import HomeAssistantAPI
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from home_intent import Intents
|
||||
from enum import IntFlag, auto
|
||||
|
||||
from home_intent import Intents
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from enum import IntFlag, auto
|
||||
|
||||
from home_intent import Intents
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
from home_intent import Intents, get_file
|
||||
from typing import Dict
|
||||
|
||||
from pydantic import BaseModel, conint
|
||||
import yaml
|
||||
from typing import Dict
|
||||
|
||||
from home_intent import Intents, get_file
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from home_intent import Intents
|
||||
from enum import IntFlag, auto
|
||||
|
||||
from home_intent import Intents
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import ciso8601
|
||||
|
||||
from home_intent import Intents
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
@@ -11,8 +11,8 @@ from requests.exceptions import Timeout
|
||||
from audio_config import AudioConfig
|
||||
from intent_handler import IntentHandler
|
||||
from intents import Intents, Sentence
|
||||
from rhasspy_api import RhasspyAPI, RhasspyError
|
||||
from path_finder import get_file
|
||||
from rhasspy_api import RhasspyAPI, RhasspyError
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -1,351 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from functools import wraps, partial, partialmethod
|
||||
import inspect
|
||||
import logging
|
||||
from pathlib import PosixPath
|
||||
import re
|
||||
from typing import Callable, Dict, List, Optional, Union
|
||||
|
||||
from pydantic import BaseModel, Extra
|
||||
import yaml
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
# we'll likely have to get more sophisticated than regexes eventually
|
||||
SLOT_REGEX = re.compile(r"\(\$([a-z_]*)")
|
||||
TAG_REGEX = re.compile(r"""{([a-z_]*)}""")
|
||||
|
||||
|
||||
class IntentException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class SlotCustomization(BaseModel, extra=Extra.forbid):
|
||||
add: Optional[List[Union[str, Dict[str, str]]]]
|
||||
remove: Optional[List[str]]
|
||||
|
||||
|
||||
class SentenceModification(BaseModel, extra=Extra.forbid):
|
||||
add: Optional[List[str]]
|
||||
remove: Optional[List[str]]
|
||||
|
||||
|
||||
class SentenceAlias(BaseModel, extra=Extra.forbid):
|
||||
sentences: List[str]
|
||||
slots: Optional[Dict[str, str]]
|
||||
|
||||
|
||||
class SentenceCustomization(BaseModel, extra=Extra.forbid):
|
||||
sentences: Optional[SentenceModification]
|
||||
alias: Optional[List[SentenceAlias]]
|
||||
enable: Optional[bool] = None
|
||||
|
||||
|
||||
class Customization(BaseModel, extra=Extra.forbid):
|
||||
slots: Optional[Dict[str, SlotCustomization]]
|
||||
intents: Optional[Dict[str, SentenceCustomization]]
|
||||
enable_all: Optional[bool] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Sentence:
|
||||
sentences: List[str]
|
||||
func: Callable
|
||||
slots: List[str]
|
||||
disabled: bool = False
|
||||
disabled_reason: str = None
|
||||
beta: bool = False
|
||||
|
||||
|
||||
class Intents:
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
self.all_slots = {}
|
||||
self.all_sentences = {}
|
||||
self.slot_modifications = {}
|
||||
self.events = {"register_sentences": []}
|
||||
|
||||
def dictionary_slots(self, func):
|
||||
LOGGER.debug(f"Registering {func.__name__}")
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
slot_dictionary = func(*arg, **kwargs)
|
||||
reverse_slot_dictionary = {v: k for (k, v) in slot_dictionary.items()}
|
||||
non_dictionary_additions = []
|
||||
if func.__name__ in self.slot_modifications:
|
||||
if self.slot_modifications[func.__name__].remove:
|
||||
for slots_to_remove in self.slot_modifications[func.__name__].remove:
|
||||
if slots_to_remove in slot_dictionary:
|
||||
del slot_dictionary[slots_to_remove]
|
||||
elif slots_to_remove in reverse_slot_dictionary:
|
||||
del slot_dictionary[reverse_slot_dictionary[slots_to_remove]]
|
||||
else:
|
||||
LOGGER.warning(
|
||||
f"'{slots_to_remove}' not in slot list for {func.__name__}"
|
||||
)
|
||||
|
||||
if self.slot_modifications[func.__name__].add:
|
||||
for slot_addition in self.slot_modifications[func.__name__].add:
|
||||
|
||||
if isinstance(slot_addition, str):
|
||||
if slot_addition in slot_dictionary:
|
||||
del slot_dictionary[slot_addition]
|
||||
non_dictionary_additions.append(
|
||||
f"{_sanitize_slot(slot_addition)}{{{func.__name__}}}"
|
||||
)
|
||||
|
||||
elif isinstance(slot_addition, dict):
|
||||
synonyms, value = next(iter(slot_addition.items()))
|
||||
if synonyms in slot_dictionary:
|
||||
del slot_dictionary[synonyms]
|
||||
elif value in reverse_slot_dictionary:
|
||||
del slot_dictionary[reverse_slot_dictionary[value]]
|
||||
slot_dictionary[synonyms] = value
|
||||
|
||||
slot_list = [
|
||||
f"{_sanitize_slot(x)}{{{func.__name__}:{slot_dictionary[x]}}}"
|
||||
for x in slot_dictionary
|
||||
]
|
||||
slot_list.extend(non_dictionary_additions)
|
||||
return slot_list
|
||||
|
||||
self.all_slots[func.__name__] = wrapper
|
||||
return wrapper
|
||||
|
||||
def slots(self, func):
|
||||
LOGGER.debug(f"Registering {func.__name__}")
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
slot_values = set(func(*arg, **kwargs))
|
||||
synonmym_values = []
|
||||
if func.__name__ in self.slot_modifications:
|
||||
if self.slot_modifications[func.__name__].remove:
|
||||
for slots_to_remove in self.slot_modifications[func.__name__].remove:
|
||||
if slots_to_remove in slot_values:
|
||||
slot_values.remove(slots_to_remove)
|
||||
else:
|
||||
LOGGER.warning(
|
||||
f"'{slots_to_remove}' not in slot list for {func.__name__}"
|
||||
)
|
||||
|
||||
if self.slot_modifications[func.__name__].add:
|
||||
for slot_addition in self.slot_modifications[func.__name__].add:
|
||||
|
||||
if isinstance(slot_addition, str):
|
||||
if slot_addition in slot_values:
|
||||
slot_values.remove(slot_addition)
|
||||
slot_values.add(slot_addition)
|
||||
|
||||
elif isinstance(slot_addition, dict):
|
||||
synonyms, value = next(iter(slot_addition.items()))
|
||||
if synonyms in slot_values:
|
||||
slot_values.remove(synonyms)
|
||||
synonmym_values.append(
|
||||
f"{_sanitize_slot(synonyms)}{{{func.__name__}:{value}}}"
|
||||
)
|
||||
|
||||
slot_list = [f"{_sanitize_slot(x)}{{{func.__name__}}}" for x in slot_values]
|
||||
slot_list.extend(synonmym_values)
|
||||
|
||||
return slot_list
|
||||
|
||||
self.all_slots[func.__name__] = wrapper
|
||||
return wrapper
|
||||
|
||||
def sentences(self, sentences: List[str]):
|
||||
def inner(func):
|
||||
if not isinstance(sentences, list):
|
||||
raise IntentException(f"The sentences decorator expects a list for {func}")
|
||||
sentence_slots = _check_if_args_in_sentence_slots(sentences, func)
|
||||
self.all_sentences[func.__name__] = Sentence(sentences, func, sentence_slots)
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
LOGGER.info(f"Running function {func.__name__}")
|
||||
func(*arg, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def beta(self, func):
|
||||
self.all_sentences[func.__name__].disabled = True
|
||||
self.all_sentences[func.__name__].beta = True
|
||||
self.all_sentences[func.__name__].disabled_reason = "BETA"
|
||||
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def default_disable(self, reason: str):
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
self.all_sentences[func.__name__].disabled_reason = reason
|
||||
self.all_sentences[func.__name__].disabled = True
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def on_event(self, event):
|
||||
if event != "register_sentences":
|
||||
raise IntentException(
|
||||
"Currently you can only register events during 'register_sentences'"
|
||||
)
|
||||
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
self.events[event].append(func)
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def disable_intent(self, sentence_func: Union[Callable, str]):
|
||||
if isinstance(sentence_func, str):
|
||||
self.all_sentences[sentence_func].disabled = True
|
||||
else:
|
||||
self.all_sentences[sentence_func.__name__].disabled = True
|
||||
|
||||
def enable_intent(self, sentence_func: Union[Callable, str]):
|
||||
if isinstance(sentence_func, str):
|
||||
self.all_sentences[sentence_func].disabled = False
|
||||
else:
|
||||
self.all_sentences[sentence_func.__name__].disabled = False
|
||||
|
||||
def disable_all(self):
|
||||
self.all_sentences = {}
|
||||
|
||||
def enable_all(self):
|
||||
for _, sentence in self.all_sentences:
|
||||
sentence.disabled = False
|
||||
|
||||
def handle_customization(self, customization_file: PosixPath, class_instance):
|
||||
LOGGER.info(f"Loading customization file {customization_file}")
|
||||
customization_yaml = yaml.load(
|
||||
customization_file.read_text("utf-8"), Loader=yaml.SafeLoader
|
||||
)
|
||||
component_customization = Customization(**customization_yaml)
|
||||
if component_customization.enable_all is not None:
|
||||
if component_customization.enable_all is True:
|
||||
self.enable_all()
|
||||
elif component_customization.enable_all is False:
|
||||
self.disable_all()
|
||||
|
||||
if component_customization.intents:
|
||||
for intent, customization in component_customization.intents.items():
|
||||
if intent in self.all_sentences:
|
||||
self._customize_intents(intent, customization, class_instance)
|
||||
else:
|
||||
raise IntentException(
|
||||
f"'{intent}'' not in intent sentences: {self.all_sentences.keys()}"
|
||||
)
|
||||
|
||||
if component_customization.slots:
|
||||
for slot, customization in component_customization.slots.items():
|
||||
if slot in self.all_slots:
|
||||
self.slot_modifications[slot] = customization
|
||||
else:
|
||||
raise IntentException(f"'{slot}' not associated with {self.name}")
|
||||
|
||||
def _customize_intents(self, intent: str, customization: SentenceCustomization, class_instance):
|
||||
if customization.enable is not None:
|
||||
if customization.enable is True:
|
||||
self.enable_intent(intent)
|
||||
elif customization.enable is False:
|
||||
self.disable_intent(intent)
|
||||
|
||||
if customization.sentences:
|
||||
if customization.sentences.add:
|
||||
self.all_sentences[intent].sentences.extend(customization.sentences.add)
|
||||
|
||||
if customization.sentences.remove:
|
||||
for sentence_to_remove in customization.sentences.remove:
|
||||
if sentence_to_remove in self.all_sentences[intent].sentences:
|
||||
self.all_sentences[intent].sentences.remove(sentence_to_remove)
|
||||
else:
|
||||
LOGGER.warning(f"'{sentence_to_remove}' not in {intent} ")
|
||||
|
||||
if customization.alias:
|
||||
for count, alias in enumerate(customization.alias):
|
||||
sentences = alias.sentences
|
||||
funcname = f"{intent}.alias_function.{count}"
|
||||
|
||||
# alright, this is some insanity. We get the alias' intent func dynamically
|
||||
# and then populate its arguments, then re-inject it into the class with a new name
|
||||
if alias.slots:
|
||||
alias_function = partial(
|
||||
getattr(class_instance, intent).__func__, **alias.slots
|
||||
)
|
||||
else:
|
||||
alias_function = getattr(class_instance, intent).__func__
|
||||
setattr(class_instance, funcname, alias_function)
|
||||
sentence_slots = _get_slots_from_sentences(sentences)
|
||||
self.all_sentences[funcname] = Sentence(sentences, alias_function, sentence_slots)
|
||||
|
||||
|
||||
def _sanitize_slot(slot_name: str):
|
||||
# okay, maybe a regex would be better at this point...
|
||||
return "".join(
|
||||
x if x.isalnum() or x in ("(", ")", "|", "[", "]", ".", ":") else " " for x in slot_name
|
||||
)
|
||||
|
||||
|
||||
def _get_slots_from_sentences(sentences: List[str]):
|
||||
sentence_slots = set()
|
||||
for sentence in sentences:
|
||||
sentence_slots.update((SLOT_REGEX.findall(sentence)))
|
||||
|
||||
return sentence_slots
|
||||
|
||||
|
||||
def _get_tags_from_sentences(sentences: List[str]):
|
||||
sentence_tags = set()
|
||||
for sentence in sentences:
|
||||
sentence_tags.update(TAG_REGEX.findall(sentence))
|
||||
|
||||
return sentence_tags
|
||||
|
||||
|
||||
def _check_if_args_in_sentence_slots(sentences, func):
|
||||
sentence_slots = _get_slots_from_sentences(sentences)
|
||||
sentence_tags = _get_tags_from_sentences(sentences)
|
||||
|
||||
argument_spec = inspect.getfullargspec(func)
|
||||
|
||||
# first arg is 'self', the last args are optionals with defaults set
|
||||
required_args = (
|
||||
argument_spec.args[1 : -len(argument_spec.defaults)]
|
||||
if argument_spec.defaults
|
||||
else argument_spec.args[1:]
|
||||
)
|
||||
|
||||
# check if arg in sentence slot
|
||||
for arg in required_args:
|
||||
valid_argument = arg in sentence_slots or arg in sentence_tags
|
||||
if not valid_argument:
|
||||
if arg not in sentence_slots:
|
||||
raise IntentException(
|
||||
f"The argument '{arg}' is not associated in the sentence for {func}. "
|
||||
f"Make sure the sentence decorator includes a (${arg}) or "
|
||||
"remove it as an argument."
|
||||
)
|
||||
|
||||
if arg not in sentence_tags:
|
||||
raise IntentException(
|
||||
f"The argument '{arg}' is not associated in the sentence for {func}. "
|
||||
f"Make sure the sentence decorator includes a {{{arg}}} or "
|
||||
"remove it as an argument."
|
||||
)
|
||||
|
||||
return sentence_slots
|
||||
2
home_intent/intents/__init__.py
Normal file
2
home_intent/intents/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
from .intents import Intents
|
||||
from .util import IntentException, Sentence
|
||||
123
home_intent/intents/customization_mixin.py
Normal file
123
home_intent/intents/customization_mixin.py
Normal file
@@ -0,0 +1,123 @@
|
||||
from functools import partial
|
||||
import logging
|
||||
from pathlib import PosixPath
|
||||
from typing import Callable, Dict, List, Optional, Union
|
||||
|
||||
from pydantic import BaseModel, Extra
|
||||
import yaml
|
||||
|
||||
from .util import IntentException, Sentence, _get_slots_from_sentences
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SlotCustomization(BaseModel, extra=Extra.forbid):
|
||||
add: Optional[List[Union[str, Dict[str, str]]]]
|
||||
remove: Optional[List[str]]
|
||||
|
||||
|
||||
class SentenceModification(BaseModel, extra=Extra.forbid):
|
||||
add: Optional[List[str]]
|
||||
remove: Optional[List[str]]
|
||||
|
||||
|
||||
class SentenceAlias(BaseModel, extra=Extra.forbid):
|
||||
sentences: List[str]
|
||||
slots: Optional[Dict[str, str]]
|
||||
|
||||
|
||||
class SentenceCustomization(BaseModel, extra=Extra.forbid):
|
||||
sentences: Optional[SentenceModification]
|
||||
alias: Optional[List[SentenceAlias]]
|
||||
enable: Optional[bool] = None
|
||||
|
||||
|
||||
class Customization(BaseModel, extra=Extra.forbid):
|
||||
slots: Optional[Dict[str, SlotCustomization]]
|
||||
intents: Optional[Dict[str, SentenceCustomization]]
|
||||
enable_all: Optional[bool] = None
|
||||
|
||||
|
||||
class IntentCustomizationMixin:
|
||||
def handle_customization(self, customization_file: PosixPath, class_instance):
|
||||
LOGGER.info(f"Loading customization file {customization_file}")
|
||||
customization_yaml = yaml.load(
|
||||
customization_file.read_text("utf-8"), Loader=yaml.SafeLoader
|
||||
)
|
||||
component_customization = Customization(**customization_yaml)
|
||||
if component_customization.enable_all is not None:
|
||||
if component_customization.enable_all is True:
|
||||
self._enable_all()
|
||||
elif component_customization.enable_all is False:
|
||||
self._disable_all()
|
||||
|
||||
if component_customization.intents:
|
||||
for intent, customization in component_customization.intents.items():
|
||||
if intent in self.all_sentences:
|
||||
self._customize_intents(intent, customization, class_instance)
|
||||
else:
|
||||
raise IntentException(
|
||||
f"'{intent}'' not in intent sentences: {self.all_sentences.keys()}"
|
||||
)
|
||||
|
||||
if component_customization.slots:
|
||||
for slot, customization in component_customization.slots.items():
|
||||
if slot in self.all_slots:
|
||||
self.slot_modifications[slot] = customization
|
||||
else:
|
||||
raise IntentException(f"'{slot}' not associated with {self.name}")
|
||||
|
||||
def _enable_all(self):
|
||||
for _, sentence in self.all_sentences.items():
|
||||
sentence.disabled = False
|
||||
|
||||
def _disable_all(self):
|
||||
for _, sentence in self.all_sentences.items():
|
||||
sentence.disabled = True
|
||||
|
||||
def _customize_intents(self, intent: str, customization: SentenceCustomization, class_instance):
|
||||
if customization.enable is not None:
|
||||
if customization.enable is True:
|
||||
self._enable_intent(intent)
|
||||
elif customization.enable is False:
|
||||
self._disable_intent(intent)
|
||||
|
||||
if customization.sentences:
|
||||
if customization.sentences.add:
|
||||
self.all_sentences[intent].sentences.extend(customization.sentences.add)
|
||||
|
||||
if customization.sentences.remove:
|
||||
for sentence_to_remove in customization.sentences.remove:
|
||||
if sentence_to_remove in self.all_sentences[intent].sentences:
|
||||
self.all_sentences[intent].sentences.remove(sentence_to_remove)
|
||||
else:
|
||||
LOGGER.warning(f"'{sentence_to_remove}' not in {intent} ")
|
||||
|
||||
if customization.alias:
|
||||
for count, alias in enumerate(customization.alias):
|
||||
sentences = alias.sentences
|
||||
funcname = f"{intent}.alias_function.{count}"
|
||||
|
||||
# alright, this is some insanity. We get the alias' intent func dynamically
|
||||
# and then populate its arguments, then re-inject it into the class with a new name
|
||||
if alias.slots:
|
||||
alias_function = partial(
|
||||
getattr(class_instance, intent).__func__, **alias.slots
|
||||
)
|
||||
else:
|
||||
alias_function = getattr(class_instance, intent).__func__
|
||||
setattr(class_instance, funcname, alias_function)
|
||||
sentence_slots = _get_slots_from_sentences(sentences)
|
||||
self.all_sentences[funcname] = Sentence(sentences, alias_function, sentence_slots)
|
||||
|
||||
def _enable_intent(self, sentence_func: Union[Callable, str]):
|
||||
if isinstance(sentence_func, str):
|
||||
self.all_sentences[sentence_func].disabled = False
|
||||
else:
|
||||
self.all_sentences[sentence_func.__name__].disabled = False
|
||||
|
||||
def _disable_intent(self, sentence_func: Union[Callable, str]):
|
||||
if isinstance(sentence_func, str):
|
||||
self.all_sentences[sentence_func].disabled = True
|
||||
else:
|
||||
self.all_sentences[sentence_func.__name__].disabled = True
|
||||
182
home_intent/intents/intents.py
Normal file
182
home_intent/intents/intents.py
Normal file
@@ -0,0 +1,182 @@
|
||||
from functools import wraps
|
||||
import logging
|
||||
from typing import Callable, Dict, List
|
||||
|
||||
from .customization_mixin import IntentCustomizationMixin, SlotCustomization
|
||||
from .util import (IntentException, Sentence, _check_if_args_in_sentence_slots,
|
||||
_sanitize_slot)
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Intents(IntentCustomizationMixin):
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
self.all_slots: Dict[str, Callable] = {}
|
||||
self.all_sentences: Dict[str, Sentence] = {}
|
||||
self.slot_modifications: Dict[str, SlotCustomization] = {}
|
||||
self.events = {"register_sentences": []}
|
||||
|
||||
def dictionary_slots(self, func):
|
||||
LOGGER.debug(f"Registering {func.__name__}")
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
slot_dictionary = func(*arg, **kwargs)
|
||||
non_dictionary_additions = []
|
||||
if func.__name__ in self.slot_modifications:
|
||||
non_dictionary_additions = self._handle_dictionary_slot_modification(
|
||||
func.__name__, slot_dictionary
|
||||
)
|
||||
|
||||
slot_list = [
|
||||
f"{_sanitize_slot(x)}{{{func.__name__}:{slot_dictionary[x]}}}"
|
||||
for x in slot_dictionary
|
||||
]
|
||||
slot_list.extend(non_dictionary_additions)
|
||||
return slot_list
|
||||
|
||||
self.all_slots[func.__name__] = wrapper
|
||||
return wrapper
|
||||
|
||||
def slots(self, func):
|
||||
LOGGER.debug(f"Registering {func.__name__}")
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
slot_values = set(func(*arg, **kwargs))
|
||||
synonym_values = []
|
||||
if func.__name__ in self.slot_modifications:
|
||||
synonym_values = self._handle_slot_modification(func.__name__, slot_values)
|
||||
|
||||
slot_list = [f"{_sanitize_slot(x)}{{{func.__name__}}}" for x in slot_values]
|
||||
slot_list.extend(synonym_values)
|
||||
|
||||
return slot_list
|
||||
|
||||
self.all_slots[func.__name__] = wrapper
|
||||
return wrapper
|
||||
|
||||
def sentences(self, sentences: List[str]):
|
||||
def inner(func):
|
||||
if not isinstance(sentences, list):
|
||||
raise IntentException(f"The sentences decorator expects a list for {func}")
|
||||
sentence_slots = _check_if_args_in_sentence_slots(sentences, func)
|
||||
self.all_sentences[func.__name__] = Sentence(sentences, func, sentence_slots)
|
||||
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
LOGGER.info(f"Running function {func.__name__}")
|
||||
func(*arg, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def beta(self, func):
|
||||
if func.__name__ not in self.all_sentences:
|
||||
raise IntentException("Put the beta decorator above the sentences decorator")
|
||||
self.all_sentences[func.__name__].disabled = True
|
||||
self.all_sentences[func.__name__].beta = True
|
||||
self.all_sentences[func.__name__].disabled_reason = "BETA"
|
||||
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def default_disable(self, reason: str):
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
if func.__name__ not in self.all_sentences:
|
||||
raise IntentException(
|
||||
"Put the default_disable decorator above the sentences decorator"
|
||||
)
|
||||
|
||||
self.all_sentences[func.__name__].disabled_reason = reason
|
||||
self.all_sentences[func.__name__].disabled = True
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def on_event(self, event):
|
||||
if event != "register_sentences":
|
||||
raise IntentException(
|
||||
"Currently you can only register events during 'register_sentences'"
|
||||
)
|
||||
|
||||
def inner(func):
|
||||
@wraps(func)
|
||||
def wrapper(*arg, **kwargs):
|
||||
return func(*arg, **kwargs)
|
||||
|
||||
self.events[event].append(func)
|
||||
return wrapper
|
||||
|
||||
return inner
|
||||
|
||||
def _handle_dictionary_slot_modification(self, func_name, slot_dictionary) -> List[str]:
|
||||
non_dictionary_additions = []
|
||||
reverse_slot_dictionary = {v: k for (k, v) in slot_dictionary.items()}
|
||||
|
||||
if self.slot_modifications[func_name].remove:
|
||||
for slots_to_remove in self.slot_modifications[func_name].remove:
|
||||
if slots_to_remove in slot_dictionary:
|
||||
del slot_dictionary[slots_to_remove]
|
||||
elif slots_to_remove in reverse_slot_dictionary:
|
||||
del slot_dictionary[reverse_slot_dictionary[slots_to_remove]]
|
||||
else:
|
||||
LOGGER.warning(f"'{slots_to_remove}' not in slot list for {func_name}")
|
||||
|
||||
if self.slot_modifications[func_name].add:
|
||||
for slot_addition in self.slot_modifications[func_name].add:
|
||||
|
||||
if isinstance(slot_addition, str):
|
||||
if slot_addition in slot_dictionary:
|
||||
del slot_dictionary[slot_addition]
|
||||
non_dictionary_additions.append(
|
||||
f"{_sanitize_slot(slot_addition)}{{{func_name}}}"
|
||||
)
|
||||
|
||||
elif isinstance(slot_addition, dict):
|
||||
synonyms, value = next(iter(slot_addition.items()))
|
||||
if synonyms in slot_dictionary:
|
||||
del slot_dictionary[synonyms]
|
||||
elif value in reverse_slot_dictionary:
|
||||
del slot_dictionary[reverse_slot_dictionary[value]]
|
||||
slot_dictionary[synonyms] = value
|
||||
|
||||
return non_dictionary_additions
|
||||
|
||||
def _handle_slot_modification(self, func_name, slot_values):
|
||||
synonym_values = []
|
||||
|
||||
if self.slot_modifications[func_name].remove:
|
||||
for slots_to_remove in self.slot_modifications[func_name].remove:
|
||||
if slots_to_remove in slot_values:
|
||||
slot_values.remove(slots_to_remove)
|
||||
else:
|
||||
LOGGER.warning(f"'{slots_to_remove}' not in slot list for {func_name}")
|
||||
|
||||
if self.slot_modifications[func_name].add:
|
||||
for slot_addition in self.slot_modifications[func_name].add:
|
||||
|
||||
if isinstance(slot_addition, str):
|
||||
if slot_addition in slot_values:
|
||||
slot_values.remove(slot_addition)
|
||||
slot_values.add(slot_addition)
|
||||
|
||||
elif isinstance(slot_addition, dict):
|
||||
synonyms, value = next(iter(slot_addition.items()))
|
||||
if synonyms in slot_values:
|
||||
slot_values.remove(synonyms)
|
||||
synonym_values.append(f"{_sanitize_slot(synonyms)}{{{func_name}:{value}}}")
|
||||
|
||||
return synonym_values
|
||||
79
home_intent/intents/util.py
Normal file
79
home_intent/intents/util.py
Normal file
@@ -0,0 +1,79 @@
|
||||
from dataclasses import dataclass
|
||||
import inspect
|
||||
import re
|
||||
from typing import Callable, List
|
||||
|
||||
# we'll likely have to get more sophisticated than regexes eventually
|
||||
SLOT_REGEX = re.compile(r"\(\$([a-z_]*)")
|
||||
TAG_REGEX = re.compile(r"""{([a-z_]*)}""")
|
||||
|
||||
|
||||
class IntentException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Sentence:
|
||||
sentences: List[str]
|
||||
func: Callable
|
||||
slots: List[str]
|
||||
disabled: bool = False
|
||||
disabled_reason: str = ""
|
||||
beta: bool = False
|
||||
|
||||
|
||||
def _sanitize_slot(slot_name: str):
|
||||
# okay, maybe a regex would be better at this point...
|
||||
return "".join(
|
||||
x if x.isalnum() or x in ("(", ")", "|", "[", "]", ".", ":") else " " for x in slot_name
|
||||
)
|
||||
|
||||
|
||||
def _get_slots_from_sentences(sentences: List[str]):
|
||||
sentence_slots = set()
|
||||
for sentence in sentences:
|
||||
sentence_slots.update((SLOT_REGEX.findall(sentence)))
|
||||
|
||||
return sentence_slots
|
||||
|
||||
|
||||
def _get_tags_from_sentences(sentences: List[str]):
|
||||
sentence_tags = set()
|
||||
for sentence in sentences:
|
||||
sentence_tags.update(TAG_REGEX.findall(sentence))
|
||||
|
||||
return sentence_tags
|
||||
|
||||
|
||||
def _check_if_args_in_sentence_slots(sentences, func):
|
||||
sentence_slots = _get_slots_from_sentences(sentences)
|
||||
sentence_tags = _get_tags_from_sentences(sentences)
|
||||
|
||||
argument_spec = inspect.getfullargspec(func)
|
||||
|
||||
# first arg is 'self', the last args are optionals with defaults set
|
||||
required_args = (
|
||||
argument_spec.args[1 : -len(argument_spec.defaults)]
|
||||
if argument_spec.defaults
|
||||
else argument_spec.args[1:]
|
||||
)
|
||||
|
||||
# check if arg in sentence slot
|
||||
for arg in required_args:
|
||||
valid_argument = arg in sentence_slots or arg in sentence_tags
|
||||
if not valid_argument:
|
||||
if arg not in sentence_slots:
|
||||
raise IntentException(
|
||||
f"The argument '{arg}' is not associated in the sentence for {func}. "
|
||||
f"Make sure the sentence decorator includes a (${arg}) or "
|
||||
"remove it as an argument."
|
||||
)
|
||||
|
||||
if arg not in sentence_tags:
|
||||
raise IntentException(
|
||||
f"The argument '{arg}' is not associated in the sentence for {func}. "
|
||||
f"Make sure the sentence decorator includes a {{{arg}}} or "
|
||||
"remove it as an argument."
|
||||
)
|
||||
|
||||
return sentence_slots
|
||||
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Home Intent Tests"""
|
||||
1
tests/disable_all.yaml
Normal file
1
tests/disable_all.yaml
Normal file
@@ -0,0 +1 @@
|
||||
enable_all: false
|
||||
41
tests/sample_light.py
Normal file
41
tests/sample_light.py
Normal file
@@ -0,0 +1,41 @@
|
||||
from home_intent import Intents, get_file
|
||||
from typing import Dict
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
class SampleLight:
|
||||
def __init__(self):
|
||||
self.ha = {}
|
||||
self.color_temp_to_name = {}
|
||||
|
||||
@intents.dictionary_slots
|
||||
def color(self):
|
||||
colors = ["red", "yellow", "blue", "light goldenrod", "light blue"]
|
||||
return {color: color.replace(" ", "") for color in colors}
|
||||
|
||||
@intents.slots
|
||||
def light(self):
|
||||
return ["bedroom", "kitchen", "attic"]
|
||||
|
||||
@intents.sentences(["toggle the ($light) [light]", "turn (on|off) the ($light) [light]"])
|
||||
def toggle_light(self, light):
|
||||
return "Turning light"
|
||||
|
||||
@intents.sentences(["turn on the ($light) [light]"])
|
||||
def turn_on(self, light):
|
||||
return "Turning light"
|
||||
|
||||
@intents.sentences(["turn off the ($light) [light]"])
|
||||
def turn_off(self, light):
|
||||
return "Turning light"
|
||||
|
||||
@intents.beta
|
||||
@intents.sentences(["(set | change | make) the ($light) [light] [to] ($color_temperature)"])
|
||||
def change_color_temperature(self, light, color_temperature):
|
||||
return "Setting to "
|
||||
|
||||
@intents.default_disable("uhhh, no good for some reason!")
|
||||
@intents.sentences(["(set | change | make) the ($light) [light] [to] ($color_temperature)"])
|
||||
def change_color_temperature2(self, light, color_temperature):
|
||||
return "Setting to "
|
||||
19
tests/sample_light_broken_arg.py
Normal file
19
tests/sample_light_broken_arg.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from home_intent import Intents, get_file
|
||||
from typing import Dict
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
class SampleLight:
|
||||
def __init__(self):
|
||||
self.ha = {}
|
||||
self.color_temp_to_name = {}
|
||||
|
||||
@intents.dictionary_slots
|
||||
def color(self):
|
||||
colors = ["red", "yellow", "blue", "light goldenrod"]
|
||||
return {color: color.replace(" ", "") for color in colors}
|
||||
|
||||
@intents.sentences(["toggle the ($light) [light]", "turn (on|off) the ($light) [light]"])
|
||||
def toggle_light(self, lights):
|
||||
return "Turning light"
|
||||
19
tests/sample_light_broken_sentence.py
Normal file
19
tests/sample_light_broken_sentence.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from home_intent import Intents, get_file
|
||||
from typing import Dict
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
class SampleLight:
|
||||
def __init__(self):
|
||||
self.ha = {}
|
||||
self.color_temp_to_name = {}
|
||||
|
||||
@intents.dictionary_slots
|
||||
def color(self):
|
||||
colors = ["red", "yellow", "blue", "light goldenrod"]
|
||||
return {color: color.replace(" ", "") for color in colors}
|
||||
|
||||
@intents.sentences(["toggle the ($lights) [light]", "turn (on|off) the ($lights) [light]"])
|
||||
def toggle_light(self, light):
|
||||
return "Turning light"
|
||||
19
tests/sample_light_broken_sentence2.py
Normal file
19
tests/sample_light_broken_sentence2.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from home_intent import Intents, get_file
|
||||
from typing import Dict
|
||||
|
||||
intents = Intents(__name__)
|
||||
|
||||
|
||||
class SampleLight:
|
||||
def __init__(self):
|
||||
self.ha = {}
|
||||
self.color_temp_to_name = {}
|
||||
|
||||
@intents.dictionary_slots
|
||||
def color(self):
|
||||
colors = ["red", "yellow", "blue", "light goldenrod"]
|
||||
return {color: color.replace(" ", "") for color in colors}
|
||||
|
||||
@intents.sentences(["toggle the ($lights) [light]", "turn (on|off) the ($light) [light]"])
|
||||
def toggle_light(self, light):
|
||||
return "Turning light"
|
||||
17
tests/test_bad_imports.py
Normal file
17
tests/test_bad_imports.py
Normal file
@@ -0,0 +1,17 @@
|
||||
import pytest
|
||||
from home_intent.intents import IntentException
|
||||
|
||||
|
||||
def test_broken_arg_fails():
|
||||
with pytest.raises(IntentException) as exception:
|
||||
from .sample_light_broken_arg import SampleLight, intents
|
||||
|
||||
|
||||
def test_broken_sentence_fails():
|
||||
with pytest.raises(IntentException) as exception:
|
||||
from .sample_light_broken_sentence import SampleLight, intents
|
||||
|
||||
|
||||
# def test_broken_sentence_fails2():
|
||||
# with pytest.raises(IntentException) as exception:
|
||||
# from .sample_light_broken_sentence2 import SampleLight, intents
|
||||
14
tests/test_base_cases.py
Normal file
14
tests/test_base_cases.py
Normal file
@@ -0,0 +1,14 @@
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
|
||||
def test_registered_slots():
|
||||
assert len(intents.all_slots) == 2
|
||||
|
||||
|
||||
def test_registered_sentences():
|
||||
assert len(intents.all_sentences) == 5
|
||||
|
||||
|
||||
def test_disabled_sentences():
|
||||
print(intents.all_sentences)
|
||||
assert len({k: v for (k, v) in intents.all_sentences.items() if v.disabled is True}) == 2
|
||||
21
tests/test_various_sentence_customizations.py
Normal file
21
tests/test_various_sentence_customizations.py
Normal file
@@ -0,0 +1,21 @@
|
||||
import pytest
|
||||
from pathlib import PosixPath
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def intents():
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
sample_light = SampleLight()
|
||||
intents.handle_customization(
|
||||
PosixPath("./tests/various_sentence_customizations.yaml"), sample_light
|
||||
)
|
||||
|
||||
return intents
|
||||
|
||||
|
||||
def test_registered_sentences(intents):
|
||||
assert len(intents.all_sentences) == 5
|
||||
assert len({k: v for (k, v) in intents.all_sentences.items() if v.disabled is True}) == 2
|
||||
assert intents.all_sentences["toggle_light"].disabled == True
|
||||
assert intents.all_sentences["change_color_temperature"].disabled == False
|
||||
88
tests/test_various_slot_customizations.py
Normal file
88
tests/test_various_slot_customizations.py
Normal file
@@ -0,0 +1,88 @@
|
||||
import pytest
|
||||
from pathlib import PosixPath
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def colors():
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
# load up the light/customization
|
||||
sample_light = SampleLight()
|
||||
intents.handle_customization(
|
||||
PosixPath("./tests/various_slot_customizations.yaml"), sample_light
|
||||
)
|
||||
|
||||
return sample_light.color()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lights():
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
sample_light = SampleLight()
|
||||
intents.handle_customization(
|
||||
PosixPath("./tests/various_slot_customizations.yaml"), sample_light
|
||||
)
|
||||
|
||||
return sample_light.light()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def intents():
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
sample_light = SampleLight()
|
||||
intents.handle_customization(
|
||||
PosixPath("./tests/various_slot_customizations.yaml"), sample_light
|
||||
)
|
||||
|
||||
return intents
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_intents():
|
||||
from .sample_light import intents, SampleLight
|
||||
|
||||
sample_light = SampleLight()
|
||||
intents.handle_customization(PosixPath("./tests/disable_all.yaml"), sample_light)
|
||||
|
||||
return intents
|
||||
|
||||
|
||||
def test_slot(lights):
|
||||
assert "kitchen{light}" in lights
|
||||
assert "attic{light}" in lights
|
||||
|
||||
|
||||
def test_slot_addition(lights):
|
||||
assert "bathroom{light}" in lights
|
||||
assert "laundry{light:basement}" in lights
|
||||
|
||||
|
||||
def test_slot_removal(lights):
|
||||
assert "bedroom{light}" not in lights
|
||||
|
||||
|
||||
def test_dictionary_slot(colors):
|
||||
assert "yellow{color:yellow}" in colors
|
||||
assert "red{color:red}" in colors
|
||||
|
||||
|
||||
def test_dictionary_slot_addition(colors):
|
||||
assert "green{color}" in colors
|
||||
assert "lime green{color:limegreen}" in colors
|
||||
|
||||
|
||||
def test_dictionary_slot_removal(colors):
|
||||
assert "blue{color:blue}" not in colors
|
||||
assert "light blue{color:blue}" not in colors
|
||||
|
||||
|
||||
def test_registered_sentences(intents):
|
||||
assert len(intents.all_sentences) == 5
|
||||
assert len({k: v for (k, v) in intents.all_sentences.items() if v.disabled is True}) == 0
|
||||
|
||||
|
||||
def test_disable_all(no_intents):
|
||||
assert len(no_intents.all_sentences) == 5
|
||||
assert len({k: v for (k, v) in no_intents.all_sentences.items() if v.disabled is True}) == 5
|
||||
20
tests/various_sentence_customizations.yaml
Normal file
20
tests/various_sentence_customizations.yaml
Normal file
@@ -0,0 +1,20 @@
|
||||
intents:
|
||||
change_color_temperature:
|
||||
enable: true
|
||||
|
||||
toggle_light:
|
||||
enable: false
|
||||
|
||||
turn_off:
|
||||
sentences:
|
||||
add:
|
||||
- "power off my ($light)"
|
||||
remove:
|
||||
- "turn off the ($light) [light]"
|
||||
|
||||
# turn_on:
|
||||
# alias:
|
||||
# - sentences:
|
||||
# - "Turn on fitzs light"
|
||||
# slots:
|
||||
# light: "bedroom"
|
||||
20
tests/various_slot_customizations.yaml
Normal file
20
tests/various_slot_customizations.yaml
Normal file
@@ -0,0 +1,20 @@
|
||||
enable_all: true
|
||||
|
||||
slots:
|
||||
light:
|
||||
add:
|
||||
- "bathroom"
|
||||
- "laundry": "basement"
|
||||
|
||||
remove:
|
||||
- "bedroom"
|
||||
|
||||
color:
|
||||
add:
|
||||
- "green"
|
||||
- "lime green": "limegreen"
|
||||
|
||||
remove:
|
||||
- "blue"
|
||||
- "lightblue"
|
||||
- "nonreal color"
|
||||
Reference in New Issue
Block a user