Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_model_accuracies_are_similar_before_and_after_caching(kwik_e_mart_app_path):
# clear model cache
model_cache_path = MODEL_CACHE_PATH.format(app_path=kwik_e_mart_app_path)
try:
shutil.rmtree(MODEL_CACHE_PATH.format(app_path=kwik_e_mart_app_path))
except FileNotFoundError:
pass
# Make sure no cache exists
assert os.path.exists(model_cache_path) is False
nlp = NaturalLanguageProcessor(kwik_e_mart_app_path)
nlp.build(incremental=True)
nlp.dump()
intent_eval = nlp.domains["store_info"].intent_classifier.evaluate()
entity_eval = (
nlp.domains["store_info"]
.intents["get_store_hours"]
.entity_recognizer.evaluate()
)
intent_accuracy_no_cache = intent_eval.get_accuracy()
.intents["change_alarm"]
.entity_recognizer.evaluate()
)
role_eval = (
nlp.domains["times_and_dates"]
.intents["change_alarm"]
.entities["sys_time"]
.role_classifier.evaluate()
)
entity_accuracy_no_cache = entity_eval.get_accuracy()
role_accuracy_no_cache = role_eval.get_accuracy()
example_cache = os.listdir(
MODEL_CACHE_PATH.format(app_path=home_assistant_app_path)
)[0]
nlp = NaturalLanguageProcessor(home_assistant_app_path)
nlp.load(example_cache)
# make sure cache exists
assert os.path.exists(model_cache_path) is True
entity_eval = (
nlp.domains["times_and_dates"]
.intents["change_alarm"]
.entity_recognizer.evaluate()
)
role_eval = (
nlp.domains["times_and_dates"]
.intents["change_alarm"]
def test_model_accuracies_are_similar_before_and_after_caching(kwik_e_mart_app_path):
# clear model cache
model_cache_path = MODEL_CACHE_PATH.format(app_path=kwik_e_mart_app_path)
try:
shutil.rmtree(MODEL_CACHE_PATH.format(app_path=kwik_e_mart_app_path))
except FileNotFoundError:
pass
# Make sure no cache exists
assert os.path.exists(model_cache_path) is False
nlp = NaturalLanguageProcessor(kwik_e_mart_app_path)
nlp.build(incremental=True)
nlp.dump()
intent_eval = nlp.domains["store_info"].intent_classifier.evaluate()
entity_eval = (
nlp.domains["store_info"]
.intents["get_store_hours"]
.entity_recognizer.evaluate()
def test_all_classifier_are_unique_for_incremental_builds(home_assistant_app_path):
nlp = NaturalLanguageProcessor(home_assistant_app_path)
nlp.build(incremental=True)
example_cache = os.listdir(
MODEL_CACHE_PATH.format(app_path=home_assistant_app_path)
)[0]
unique_hashs = set()
for domain in nlp.domains:
for intent in nlp.domains[domain].intents:
_, cached_path = get_entity_model_paths(
home_assistant_app_path, domain, intent, timestamp=example_cache
)
hash_val = open(cached_path + ".hash", "r").read()
assert hash_val not in unique_hashs
unique_hashs.add(hash_val)
for entity in (
nlp.domains[domain].intents[intent].entity_recognizer.entity_types
):
_, cached_path = get_role_model_paths(
# Make sure no cache exists
assert os.path.exists(model_cache_path) is False
nlp = NaturalLanguageProcessor(kwik_e_mart_app_path)
nlp.build(incremental=True)
nlp.dump()
intent_eval = nlp.domains["store_info"].intent_classifier.evaluate()
entity_eval = (
nlp.domains["store_info"]
.intents["get_store_hours"]
.entity_recognizer.evaluate()
)
intent_accuracy_no_cache = intent_eval.get_accuracy()
entity_accuracy_no_cache = entity_eval.get_accuracy()
example_cache = os.listdir(MODEL_CACHE_PATH.format(app_path=kwik_e_mart_app_path))[
0
]
nlp = NaturalLanguageProcessor(kwik_e_mart_app_path)
nlp.load(example_cache)
# make sure cache exists
assert os.path.exists(model_cache_path) is True
intent_eval = nlp.domains["store_info"].intent_classifier.evaluate()
entity_eval = (
nlp.domains["store_info"]
.intents["get_store_hours"]
.entity_recognizer.evaluate()
)
intent_accuracy_cached = intent_eval.get_accuracy()
entity_accuracy_cached = entity_eval.get_accuracy()
def test_model_accuracies_are_similar_before_and_after_caching(home_assistant_app_path):
# clear model cache
model_cache_path = MODEL_CACHE_PATH.format(app_path=home_assistant_app_path)
try:
shutil.rmtree(MODEL_CACHE_PATH.format(app_path=home_assistant_app_path))
except FileNotFoundError:
pass
# Make sure no cache exists
assert os.path.exists(model_cache_path) is False
nlp = NaturalLanguageProcessor(home_assistant_app_path)
nlp.build(incremental=True)
nlp.dump()
entity_eval = (
nlp.domains["times_and_dates"]
.intents["change_alarm"]
.entity_recognizer.evaluate()
)
role_eval = (
def _load_cached_models(self):
if not self._hash_to_model_path:
self._hash_to_model_path = {}
cache_path = MODEL_CACHE_PATH.format(app_path=self.app_path)
for dir_path, _, file_names in os.walk(cache_path):
for filename in [f for f in file_names if f.endswith(".hash")]:
file_path = os.path.join(dir_path, filename)
hash_val = open(file_path, "r").read()
classifier_file_path = file_path.split(".hash")[0]
if not os.path.exists(classifier_file_path):
logger.warning("Could not find the serialized model")
continue
self._hash_to_model_path[hash_val] = classifier_file_path
main_cache_location = QUERY_CACHE_PATH.format(app_path=app.app_path)
tmp_cache_location = QUERY_CACHE_TMP_PATH.format(app_path=app.app_path)
if os.path.exists(main_cache_location):
os.remove(main_cache_location)
if os.path.exists(tmp_cache_location):
os.remove(tmp_cache_location)
logger.info("Query cache deleted")
except FileNotFoundError:
logger.info("No query cache to delete")
return
if model_cache:
model_cache_path = MODEL_CACHE_PATH.format(app_path=app.app_path)
if not os.path.exists(model_cache_path):
logger.warning("Model cache directory doesn't exist")
return
if days:
for ts_folder in os.listdir(model_cache_path):
full_path = os.path.join(model_cache_path, ts_folder)
if not os.path.isdir(full_path):
logger.warning(
"Expected timestamped folder. Ignoring the file %s.", full_path
)
continue
try: