master-thesis/RecipeAnalysis/Recipe.py

1283 lines
44 KiB
Python

#!/usr/bin/env python3
# coding: utf-8
# # Recipe class
import sys
sys.path.append("../")
import settings
import pycrfsuite
import json
import db.db_settings as db_settings
from db.database_connection import DatabaseConnection
from Tagging.conllu_generator import ConlluGenerator
from Tagging.crf_data_generator import *
from difflib import SequenceMatcher
import numpy as np
import plotly.graph_objs as go
from plotly.offline import download_plotlyjs, init_notebook_mode, plot, iplot
from plotly.subplots import make_subplots
init_notebook_mode(connected=True)
from graphviz import Digraph
import itertools
import plotly.io as pio
pio.renderers.default = "jupyterlab"
from IPython.display import Markdown, HTML, display
# * sequence similarity matcher
def similar(a, b):
return SequenceMatcher(None, a, b).ratio()
def string_similarity(a,b):
"""
does the same like `similar` but also compares single words of multi word tokens
and returns the max similar value
"""
tokens_a = a.split()
tokens_b = b.split()
max_similarity = -1
max_a = None
max_b = None
for t_a in tokens_a:
for t_b in tokens_b:
s = similar(t_a, t_b)
if s > max_similarity:
max_similarity = s
max_a = t_a,
max_b = t_b,
return max_similarity, max_a, max_b
# * get vocabulary
import importlib.util
# loading ingredients:
spec = importlib.util.spec_from_file_location(
"ingredients", "../" + settings.ingredients_file)
ingredients = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ingredients)
# loading actions:
spec = importlib.util.spec_from_file_location(
"actions", "../" + settings.actions_file)
actions = importlib.util.module_from_spec(spec)
spec.loader.exec_module(actions)
# loading containers
spec = importlib.util.spec_from_file_location(
"containers", "../" + settings.container_file)
containers = importlib.util.module_from_spec(spec)
spec.loader.exec_module(containers)
# loading placeholders
spec = importlib.util.spec_from_file_location(
"placeholders", "../" + settings.placeholder_file)
placeholders = importlib.util.module_from_spec(spec)
spec.loader.exec_module(placeholders)
# helper function since the lemmatizer not always lemmatize in a meaningful way :shrug:
def check_ingredient(ing_token):
form = ing_token['form'].lower()
lemma = ing_token['lemma'].lower()
if form in ingredients.ingredients:
return True
if lemma in ingredients.ingredients_stemmed:
return True
if lemma.endswith('s'):
if lemma[:-1] in ingredients.ingredients_stemmed:
return True
else:
if lemma + 's' in ingredients.ingredients_stemmed:
return True
return False
tagger = pycrfsuite.Tagger()
tagger.open('../Tagging/test.crfsuite')
id_query = "select * from recipes where id like %s"
def escape_md_chars(s):
s = s.replace("*", "\*")
s = s.replace("(", "\(")
s = s.replace(")", "\)")
s = s.replace("[", "\[")
s = s.replace("]", "\]")
s = s.replace("_", "\_")
return s
import json
class Ingredient(object):
@staticmethod
def from_json(j):
d = json.loads(j)
ing = Ingredient(d['base'])
ing._action_set = set(d['actions'])
return ing
def __init__(self, base_ingredient, last_touched_instruction=0):
self._base_ingredient = base_ingredient
self._action_set = set()
self._last_touched_instruction = last_touched_instruction
self._is_mixed = False
def apply_action(self, action, instruction_number=0, touch=True):
if action in actions.mixing_cooking_verbs:
self.mark_for_mixing()
else:
self._action_set.add(action)
if touch:
self._last_touched_instruction = instruction_number
return self
def similarity(self, ingredient, use_actions=False, action_factor = 0.5):
sim,_,_ = string_similarity(self._base_ingredient, ingredient._base_ingredient)
if not use_actions:
return sim
return (1 - action_factor) + action_factor * similar(list(self._action_set), list(ingredient._action_set))
def mark_for_mixing(self):
self._is_mixed = True
def unmark_mixing(self):
self._is_mixed = False
def is_mixed(self):
return self._is_mixed
def most_similar_ingredient(self, ing_list, use_actions=False, action_factor=0.5):
best_index = -1
best_value = -1
for i, ing in enumerate(ing_list):
sim = self.similarity(ing, use_actions=use_actions, action_factor=action_factor)
if sim > best_value:
best_value = sim
best_index = i
return best_value, ing_list[best_index]
def copy(self):
result = Ingredient(self._base_ingredient, self._last_touched_instruction)
result._action_set = self._action_set.copy()
result._is_mixed = self._is_mixed
return result
def to_json(self):
result = {}
result['base'] = self._base_ingredient
result['actions'] = list(self._action_set)
return json.dumps(result)
def __repr__(self):
return f"{'|'.join(list(self._action_set))} 🠊 {self._base_ingredient} (last touched @ {self._last_touched_instruction})"
class RecipeState(object):
def __init__(self, initial_ingredients):
self._ingredients = initial_ingredients
self._seen_ingredients = set()
self._seen_actions = set()
self._mix_matrix = None
self._mix_labels = None
self._act_matrix = None
self._act_labels = None
self._ing_labels = None
self._mat_need_update = True
# set of (ing_a, ing_b) tuples
self._seen_mixes = set()
# set of (action, ing) tuples
self._seen_applied_actions = set()
for ing in self._ingredients:
self._seen_ingredients.add(ing.to_json())
def copy(self):
return RecipeState([ing.copy() for ing in self._ingredients])
def apply_action(self, action: str, ing: Ingredient, instruction_number=0, sim_threshold = 0.6, add_new_if_not_similar=True):
# find most similar ingredient to the given one and apply action on it
sim_val, best_ing = ing.most_similar_ingredient(self._ingredients)
# if sim_val is good enough, we apply the action to the best ingredient, otherwise
# we add a new ingredient to our set (and assume that it was not detected or listed in the
# ingredient set before)
self._mat_need_update = True
if sim_val > sim_threshold:
if action not in actions.stemmed_mixing_cooking_verbs:
self._seen_actions.add(action)
self._seen_applied_actions.add((action, best_ing.to_json()))
best_ing.apply_action(action, instruction_number)
self._seen_ingredients.add(best_ing.to_json())
elif add_new_if_not_similar:
self._ingredients.append(ing)
if action not in actions.stemmed_mixing_cooking_verbs:
self._seen_actions.add(action)
self._seen_ingredients.add(ing.to_json())
self._seen_applied_actions.add((action, ing.to_json()))
ing.apply_action(action, instruction_number)
self._seen_ingredients.add(ing.to_json())
def apply_action_on_all(self, action, instruction_number=0, exclude_instruction_number=None):
self._mat_need_update = True
for ing in self._ingredients:
if exclude_instruction_number is None or exclude_instruction_number != ing._last_touched_instruction:
if action not in actions.stemmed_mixing_cooking_verbs:
self._seen_actions.add(action)
self._seen_applied_actions.add((action, ing.to_json()))
ing.apply_action(action, instruction_number)
self._seen_ingredients.add(ing.to_json())
def apply_action_by_last_touched(action, last_touched_instruction, instruction_number=0):
self._mat_need_update = True
for ing in self.get_ingredients_touched_in_instruction(last_touched_instruction):
if action not in actions.stemmed_mixing_cooking_verbs:
self._seen_actions.add(action)
self._seen_applied_actions.add((action, ing.to_json()))
ing.apply_action(action, instruction_number)
self._seen_ingredients.add(ing.to_json())
def get_combined_ingredients(self):
combined = []
for ing in self._ingredients:
if ing.is_mixed():
combined.append(ing)
ing.unmark_mixing()
for x in combined:
for y in combined:
self._seen_mixes.add((x.to_json(), y.to_json()))
self._mat_need_update = True
return combined
def _update_matrices(self):
ing_list = list(self._seen_ingredients)
idx = {}
m = np.zeros((len(ing_list), len(ing_list)))
for i,ing in enumerate(ing_list):
idx[ing] = i
for x,y in self._seen_mixes:
m[idx[x], idx[y]] = 1
self._mix_matrix = m
self._mix_labels = [Ingredient.from_json(j) for j in ing_list]
ing_list = list(self._seen_ingredients)
idx_i = {}
act_list = list(self._seen_actions)
idx_a = {}
for i,ing in enumerate(ing_list):
idx_i[ing] = i
for i,act in enumerate(act_list):
idx_a[act] = i
m = np.zeros((len(act_list), len(ing_list)))
for act, ing in self._seen_applied_actions:
m[idx_a[act], idx_i[ing]] = 1
self._act_matrix = m
self._act_labels = act_list
self._ing_labels = [Ingredient.from_json(j) for j in ing_list]
self._mat_need_update = False
def get_mixing_matrix(self):
if self._mat_need_update:
self._update_matrices()
return self._mix_matrix, self._mix_labels
def get_action_matrix(self):
if self._mat_need_update:
self._update_matrices()
return self._act_matrix, self._act_labels, self._ing_labels
def get_ingredients_touched_in_instruction(self, instruction_number = 0):
ings = []
for ing in self._ingredients:
if ing._last_touched_instruction == instruction_number:
ings.append(ing)
return ings
def get_ingredients(self):
return self._ingredients
def __repr__(self):
s = ""
for ing in self._ingredients:
s += f"{str(ing)}\n"
return s
class Node(object):
def __init__(self, id, label, shape):
self.id = id
self.label = label
self.shape = shape
class GraphWrapper(object):
def __init__(self, comment="recipe graph"):
self._comment = comment
self._nodes = set()
self._nodes_by_id = {}
self._nodes_by_label = {}
self._edges = set()
self._to_node = {}
self._from_node = {}
def node(self, id, label, shape = None):
assert id not in self._nodes_by_id
n = Node(id, label, shape)
self._nodes.add(n)
self._nodes_by_id[id] = n
if label not in self._nodes_by_label:
self._nodes_by_label[label] = set()
self._nodes_by_label[label].add(n)
self._to_node[id] = set()
self._from_node[id] = set()
def edge(self, a, b):
assert a in self._nodes_by_id and b in self._nodes_by_id
self._edges.add((a,b))
self._from_node[a].add(b)
self._to_node[b].add(a)
def remove_edge(self, a, b):
self._edges.discard((a,b))
if a in self._from_node:
self._from_node[a].discard(b)
if b in self._to_node:
self._to_node[b].discard(a)
def remove_node(self, id, redirect_edges=False):
assert id in self._nodes_by_id
if redirect_edges:
f_set = self._from_node[id].copy()
t_set = self._to_node[id].copy()
self.remove_node(id)
for a in t_set:
for b in f_set:
self.edge(a,b)
return
# remove all edges
b_set = self._from_node[id].copy()
for b in b_set:
self.remove_edge(id, b)
a_set = self._to_node[id].copy()
for a in a_set:
self.remove_edge(a, id)
# remove node itself
n = self._nodes_by_id[id]
self._nodes_by_label[n.label].remove(n)
if len(self._nodes_by_label[n.label]) == 0:
del(self._nodes_by_label[n.label])
self._nodes.remove(n)
del(self._nodes_by_id[id])
del(self._from_node[id])
del(self._to_node[id])
def merge(self, a, b):
"""
merge a with b and return id of merged node
"""
assert a in self._nodes_by_id and b in self._nodes_by_id
if (a,b) in self._edges:
self.remove_edge(a,b)
if (b,a) in self._edges:
self.remove_edge(b,a)
to_merged = set()
from_merged = set()
if a in self._from_node:
from_merged = from_merged.union(self._from_node[a])
if b in self._from_node:
from_merged = from_merged.union(self._from_node[b])
if a in self._to_node:
to_merged = to_merged.union(self._to_node[a])
if b in self._to_node:
to_merged = to_merged.union(self._to_node[b])
from_merged.discard(a)
from_merged.discard(b)
to_merged.discard(a)
to_merged.discard(b)
merged_node = self._nodes_by_id[a]
self.remove_node(a)
self.remove_node(b)
self.node(merged_node.id, merged_node.label, merged_node.shape)
for x in to_merged:
self.edge(x, merged_node.id)
for x in from_merged:
self.edge(merged_node.id, x)
def insert_before(self, node_id, insert_id, insert_label, insert_shape):
assert insert_id not in self._nodes_by_id
assert node_id in self._nodes_by_id
to_node = self._to_node[node_id].copy()
for a in to_node:
self.remove_edge(a, node_id)
self.node(insert_id, insert_label, insert_shape)
for a in to_node:
self.edge(a, insert_id)
self.edge(insert_id, node_id)
def merge_adjacent_with_label(self, label):
"""
merge all adjacent nodes with given label
"""
assert label in self._nodes_by_label
node_set = self._nodes_by_label[label]
mix_set = set()
connected_clusters = {}
for x in node_set:
for y in node_set:
if (x.id, y.id) in self._edges:
# mark for merge
mix_set.add(x.id)
mix_set.add(y.id)
if x.id not in connected_clusters:
connected_clusters[x.id] = set()
if y.id not in connected_clusters:
connected_clusters[y.id] = set()
u = connected_clusters[x.id].union(connected_clusters[y.id])
u.add(x.id)
u.add(y.id)
for n in u:
connected_clusters[n] = u
clusters = []
while len(mix_set) > 0:
arbitrary_node = mix_set.pop()
# get cluster for node:
c = connected_clusters[arbitrary_node]
c_list = list(c)
# merge all nodes:
for i in range(len(c_list) - 1):
# note: order matters since 'merge' keeps the id of the first node!
self.merge(c_list[i + 1], c_list[i])
# subtract cluster set from mix_set
mix_set = mix_set.difference(c)
def merge_sisters(self):
sister_nodes = set()
sisters = {}
for label, node_set in self._nodes_by_label.items():
for x in node_set:
for y in node_set:
if x.id == y.id:
continue
if len(self._from_node[x.id].intersection(self._from_node[y.id])) > 0:
sister_nodes.add(x.id)
sister_nodes.add(y.id)
if x.id not in sisters:
sisters[x.id] = set()
if y.id not in sisters:
sisters[y.id] = set()
u = sisters[x.id].union(sisters[y.id])
u.add(x.id)
u.add(y.id)
for n in u:
sisters[n] = u
if len(sister_nodes) <= 1:
return False
while len(sister_nodes) > 0:
arbitrary_node = sister_nodes.pop()
# get cluster for node:
c = sisters[arbitrary_node]
c_list = list(c)
# merge all nodes:
for i in range(len(c_list) - 1):
# note: order matters since 'merge' keeps the id of the first node!
self.merge(c_list[i + 1], c_list[i])
i = 0
mix_id = "mix0"
while mix_id in self._nodes_by_id:
i += 1
mix_id = f"mix{i}"
self.insert_before(c_list[-1], mix_id, "mix", "diamond")
# subtract cluster set from mix_set
sister_nodes = sister_nodes.difference(c)
return True
def get_paths(self):
cluster = {}
nodes = set()
for a,b in self._edges:
if len(self._from_node[a]) == 1 and len(self._to_node[b]) == 1:
if a not in cluster:
cluster[a] = set()
if b not in cluster:
cluster[b] = set()
nodes.add(a)
nodes.add(b)
u = cluster[a].union(cluster[b])
u.add(a)
u.add(b)
for n in u:
cluster[n] = u
paths = []
while len(nodes) > 0:
arbitrary_node = nodes.pop()
# get cluster for node:
c = cluster[arbitrary_node]
paths.append(c)
nodes = nodes.difference(c)
return paths
def clean_paths(self):
for path in self.get_paths():
seen_labels = set()
for n in path:
l = self._nodes_by_id[n].label
if l == "mix" and len(self._to_node[n]) == 1:
self.remove_node(n, redirect_edges=True)
elif l in seen_labels:
self.remove_node(n, redirect_edges=True)
else:
seen_labels.add(l)
def simplify(self):
changed = True
while changed:
# merge all adjacent nodes with the same label
for key in self._nodes_by_label:
self.merge_adjacent_with_label(key)
# and now merge all sister nodes with the same label
# (just to make it more clean structured)
changed = self.merge_sisters()
self.clean_paths()
def compile_graph(self, simplify = False):
if simplify:
self.simplify()
dot = Digraph(self._comment)
for n in self._nodes:
dot.node(n.id, label=n.label, shape=n.shape)
for e in self._edges:
dot.edge(e[0], e[1])
return dot
class RecipeGraph(object):
def __init__(self, initial_ingreds=None):
self._base_ing_nodes = set()
self._dot = GraphWrapper(comment="recipe graph")
self._ing_state_mapping = {} # key: ingredient, value: state_id
self._seen_actions = set()
self._ings_connected_with_state = {} # key: state_id, value: set of ingreds
self._seen_actions_for_ingredient = {}
if initial_ingreds is not None:
for ing in initial_ingreds:
self.add_base_ingredient(ing)
def add_base_ingredient(self, ingredient):
if type(ingredient) == Ingredient:
self.add_base_ingredient(ingredient._base_ingredient)
return
self._base_ing_nodes.add(ingredient)
self._dot.node(ingredient, label=ingredient,shape="box")
self._ing_state_mapping[ingredient] = ingredient
self._ings_connected_with_state[ingredient] = set([ingredient])
self._seen_actions_for_ingredient[ingredient] = set()
def add_action(self, action, ingredient):
if type(ingredient) == Ingredient:
return self.add_action(ingredient._base_ingredient)
if ingredient not in self._seen_actions_for_ingredient:
self._seen_actions_for_ingredient[ingredient] = set()
if action in self._seen_actions_for_ingredient[ingredient]:
return False
self._seen_actions_for_ingredient[ingredient].add(action)
action_id = action + "0"
i = 0
while action_id in self._seen_actions:
i += 1
action_id = action + str(i)
self._seen_actions.add(action_id)
self._dot.node(action_id, action)
# get to the bottom of our tree (last known thing that happened to our ingredient)
last_node = self._ing_state_mapping[ingredient]
# update the reference of the last known state for all connected ingredients
# (and for ourselve)
connected_ingredients = self._ings_connected_with_state[last_node]
for ing_id in connected_ingredients:
self._ing_state_mapping[ing_id] = action_id
# set ingredient set for new node
self._ings_connected_with_state[action_id] = connected_ingredients.copy()
# connect nodes with an edge
self._dot.edge(last_node, action_id)
return True
def add_action_if_possible(self, action, ingredient):
# extract actions for ingredient
action_set = ingredient._action_set
if action_set.issubset(self._seen_actions_for_ingredient[ingredient._base_ingredient]):
return self.add_action(action, ingredient._base_ingredient)
return False
def mix_ingredients(self, ingredient_list):
assert len(ingredient_list) > 0
if type(ingredient_list[0]) == Ingredient:
self.mix_ingredients([ing._base_ingredient for ing in ingredient_list])
return
last_nodes = set([self._ing_state_mapping[ing] for ing in ingredient_list])
# create mixed ingredient set
ing_set = set()
for state in last_nodes:
ing_set = ing_set.union(self._ings_connected_with_state[state])
mix_action_id = "mix0"
i = 0
while mix_action_id in self._seen_actions:
i += 1
mix_action_id = f"mix{i}"
self._seen_actions.add(mix_action_id)
self._dot.node(mix_action_id, "mix", shape="diamond")
self._ings_connected_with_state[mix_action_id] = ing_set.copy()
for ing in ing_set:
self._ing_state_mapping[ing] = mix_action_id
for state in last_nodes:
self._dot.edge(state, mix_action_id)
def mix_if_possible(self, ingredient_list):
assert len(ingredient_list) > 0
assert type(ingredient_list[0]) == Ingredient
# check whether ingredients are mixed already
state_set = set(
[self._ing_state_mapping[ing._base_ingredient] for ing in ingredient_list]
)
if len(state_set) <= 1:
# all ingredients have the same last state → they're mixed already
return False
# check if action sets are matching the requirements
for ing in ingredient_list:
for act in ing._action_set:
if act not in self._seen_actions_for_ingredient[ing._base_ingredient]:
return False
# now we can mix the stuff:
self.mix_ingredients(ingredient_list)
return True
@staticmethod
def fromRecipeState(rec_state: RecipeState):
# get all ingredients
base_ingredients = set([ing._base_ingredient for ing in rec_state._ingredients])
mix_m, mix_label = rec_state.get_mixing_matrix()
act_m, act_a, act_i = rec_state.get_action_matrix()
graph = RecipeGraph(base_ingredients)
# create list of tuples: [action, ingredient]
seen_actions = np.array(list(itertools.product(act_a,act_i))).reshape((len(act_a), len(act_i), 2))
# create list of tuples [ingredient, ingredient]
seen_mixes = np.array(list(itertools.product(mix_label,mix_label))).reshape((len(mix_label), len(mix_label), 2))
seen_actions = seen_actions[act_m == 1]
seen_mixes = seen_mixes[mix_m == 1]
seen_actions = set([tuple(x) for x in seen_actions.tolist()])
seen_mixes = set([tuple(x) for x in seen_mixes.tolist()])
# for each ingredient get the list of unseen applied actions. (They were applied
# before the first instruction)
seen_actions_per_ingred = {}
for act, json_ing in rec_state._seen_applied_actions:
ing = Ingredient.from_json(json_ing)._base_ingredient
if ing not in seen_actions_per_ingred:
seen_actions_per_ingred[ing] = set()
seen_actions_per_ingred[ing].add(act)
unseen_actions_per_ingred = {}
for ing in rec_state._ingredients:
base = ing._base_ingredient
if base not in seen_actions_per_ingred:
unseen_actions_per_ingred[base] = ing._action_set.copy()
else:
unseen_actions_per_ingred[base] = ing._action_set.difference(seen_actions_per_ingred[base])
# for each ingredient: apply unseen actions first
for ing in rec_state._ingredients:
base = ing._base_ingredient
for act in unseen_actions_per_ingred[base]:
graph.add_action(act, base)
# iterate over all mixes and actions until the graph does not change anymore
# TODO: there are more efficient ways to do that!
changed = True
while changed:
changed = False
changed_ingreds = True
while changed_ingreds:
changed_ingreds = False
for mix in list(seen_mixes):
if graph.mix_if_possible([mix[0], mix[1]]):
changed = True
changed_ingreds = True
changed_acts = True
while changed_acts:
changed_acts = False
for act in list(seen_actions):
if graph.add_action_if_possible(act[0], act[1]):
changed = True
changed_acts = True
return graph
class Recipe(object):
def __init__(self, recipe_db_id = None):
self._sentences = None
self._title = None
self._part = None
self._ingredients = None
self._recipe_id = recipe_db_id
self._get_from_db()
self._extracted_ingredients = None # TODO
self.annotate_ingredients()
self.annotate_sentences()
def _get_from_db(self):
result = DatabaseConnection.global_single_query(id_query, (self._recipe_id))
assert len(result) > 0
result = result[0]
self._title = result['title']
self._part = result['part']
raw_sentences = json.loads(result['instructions'])
raw_ingredients = json.loads(result['ingredients'])
# throwing the raw data through our connlu generator to annotate them right
cg_sents = ConlluGenerator(["\n".join(raw_sentences)])
cg_ings = ConlluGenerator(["\n".join(raw_ingredients)])
cg_sents.tokenize()
cg_sents.pos_tagging_and_lemmatization()
cg_ings.tokenize()
cg_ings.pos_tagging_and_lemmatization()
# TODO
self._sentences = cg_sents.get_conllu_elements()[0]
self._ingredients = cg_ings.get_conllu_elements()[0]
#self._sentences = json.loads(result['instructions'])
#self._ingredients = json.loads(result['ingredients'])
def avg_sentence_length(self):
return sum([len(s) for s in self._sentences])/len(self._sentences)
def n_instructions(self):
return len(self._sentences)
def max_sentence_length(self):
return max([len(s) for s in self._sentences])
def keyword_ratio(self):
sentence_ratios = []
for sent in self._sentences:
# FIXME: only works if there are no other misc annotations!
sentence_ratios.append(sum([token['misc'] is not None for token in sent]))
return sum(sentence_ratios) / len(sentence_ratios)
def predict_labels(self):
features = [sent2features(sent) for sent in self._sentences]
labels = [tagger.tag(feat) for feat in features]
return labels
def predict_ingredient_labels(self):
features = [sent2features(sent) for sent in self._ingredients]
labels = [tagger.tag(feat) for feat in features]
return labels
def _annotate_sentences(self, sent_token_list, predictions):
# test whether we predicted an label or found it in our label list
for i, ing in enumerate(sent_token_list):
for j, token in enumerate(ing):
lemma = token['lemma']
# check for labels
if check_ingredient(token):
token.add_misc("food_type", "ingredient")
continue
if lemma in actions.stemmed_curated_cooking_verbs:
token.add_misc("food_type", "action")
continue
if predictions[i][j] == 'ingredient':
token.add_misc("food_type", "ingredient")
continue
#if predictions[i][j] == 'action':
# token.add_misc("food_type", "action")
# continue
if lemma in containers.stemmed_containers:
token.add_misc("food_type", "container")
continue
if predictions[i][j] == 'container':
token.add_misc("food_type", "container")
continue
if lemma in placeholders.stemmed_placeholders:
token.add_misc("food_type", "placeholder")
if predictions[i][j] == 'placeholder':
token.add_misc("food_type", "placeholder")
def annotate_ingredients(self):
self._annotate_sentences(self._ingredients, self.predict_ingredient_labels())
def annotate_sentences(self):
self._annotate_sentences(self._sentences, self.predict_labels())
def recipe_id(self):
return self._recipe_id
'''
# TODO: only conllu module compatible, and not with our own conllu classes
def serialize(self):
result = "# newdoc\n"
if self._recipe_id is not None:
result += f"# id: {self._recipe_id}\n"
for sent in self._sentences:
result += f"{sent.serialize()}"
return result + "\n"
'''
def display_recipe(self):
display(Markdown(f"## {self._title}\n({self._recipe_id})"))
display(Markdown(f"### Ingredients"))
display(Markdown("\n".join([f" * '{escape_md_chars(self.tokenlist2str(ing))}'" for ing in self._ingredients])))
display(Markdown(f"### Instructions"))
display(Markdown("\n".join([f" * {escape_md_chars(self.tokenlist2str(ins))}" for ins in self._sentences])))
def tokenlist2str(self, tokenlist):
return " ".join([token['form'] for token in tokenlist])
def tokenarray2str(self, tokenarray):
return "\n".join([self.tokenlist2str(tokenlist) for tokenlist in tokenarray])
def __repr__(self):
s = "recipe: " + (self._recipe_id if self._recipe_id else "") + "\n"
s += "instructions: \n"
for sent in self._sentences:
s += " ".join([token['form'] for token in sent]) + "\n"
s += "\nscores:\n"
s += f"avg_sent_length: {self.avg_sentence_length()}\n"
s += f"n_instructions: {self.n_instructions()}\n"
s += f"keyword_ratio: {self.keyword_ratio()}\n\n\n"
return s
# --------------------------------------------------------------------------
# functions for extracting ingredients
def extract_ingredients(self):
self._extracted_ingredients = []
for ing in self._ingredients:
entry_ing_tokens = []
entry_act_tokens = []
for token in ing:
t_misc = token['misc']
if t_misc is not None and "food_type" in t_misc:
ftype = t_misc['food_type']
if ftype == "ingredient":
entry_ing_tokens.append(token)
elif ftype == "action":
entry_act_tokens.append(token)
# find max cluster of ingredients and merge them
index_best = 0
best_size = 0
current_size = 0
for i, ing_token in enumerate(entry_ing_tokens):
if i == 0 or entry_ing_tokens[i - 1]['id'] + 1 == ing_token['id']:
current_size += 1
if current_size > best_size:
best_size = current_size
index_best = i - current_size + 1
if best_size == 0:
# unfortunately, no ingredient is found :(
continue
ingredient = Ingredient(" ".join([entry['lemma'] for entry in entry_ing_tokens[index_best:index_best + best_size]]))
# apply found actions:
for action in entry_act_tokens:
ingredient.apply_action(action['lemma'])
self._extracted_ingredients.append(ingredient)
return self._extracted_ingredients
def apply_instructions(self, confidence_threshold = 0.4, max_dist_last_token = 4, debug=False):
current_state = RecipeState(self._extracted_ingredients)
self._recipe_state = current_state
instruction_number = 0
for sent in self._sentences:
instruction_number += 1
if debug:
display(Markdown(f"----\n* **instruction {instruction_number}**:\n`" + escape_md_chars(self.tokenlist2str(sent)) + "`\n"))
instruction_ing_tokens = []
instruction_act_tokens = []
ing_dist_last_token = []
act_dist_last_token = []
last_token = -1
for i, token in enumerate(sent):
t_misc = token['misc']
if t_misc is not None and "food_type" in t_misc:
ftype = t_misc['food_type']
if ftype == "ingredient":
instruction_ing_tokens.append(token)
ing_dist_last_token.append(1000 if last_token < 0 else i - last_token)
last_token = i
elif ftype == "action":
instruction_act_tokens.append(token)
act_dist_last_token.append(1000 if last_token < 0 else i - last_token)
last_token = i
# cluster ingredient tokens together and apply actions on it:
clustered_ingredients = []
clustered_conllu_ids = []
clustered_last_tokens = []
i = 0
n = len(instruction_ing_tokens)
current_token_start = 0
while i < n:
current_token_start = i
clustered_conllu_ids.append(instruction_ing_tokens[i]['id'])
clustered_last_tokens.append(ing_dist_last_token[i])
ing_str = instruction_ing_tokens[i]['lemma']
while i+1 < n and instruction_ing_tokens[i+1]['id'] - instruction_ing_tokens[i]['id'] == 1:
ing_str += " " + instruction_ing_tokens[i+1]['lemma']
i += 1
clustered_ingredients.append(ing_str)
i += 1
def matching_action(ing_str, ing_id, action_token_list):
action = None
action_dists = [act['id'] - ing_id for act in action_token_list]
# so far: simple heuristic by matching to next action to the left
# (or first action to the right, if there is no one left to the ingredient)
for i in range(len(action_token_list)):
if action_dists[i] < 0:
action = action_token_list[i]
return action
ingredients_used = set()
actions_used = set()
if debug:
print("apply actions regular rule based:")
for i, ing_str in enumerate(clustered_ingredients):
ing = Ingredient(ing_str)
# get matching action:
action = matching_action(ing_str, clustered_conllu_ids[i], instruction_act_tokens)
if clustered_last_tokens[i] < max_dist_last_token:
if action is not None:
actions_used.add(action['lemma'])
ingredients_used.add(ing_str)
# apply action on state
current_state.apply_action(action['lemma'], ing, instruction_number=instruction_number, add_new_if_not_similar=False)
if debug:
print(f"\tapply {action['lemma']} on {ing}")
if debug:
print("try to match unused actions:")
# go throuh all actions. if we found an unused one, we assume it is applied either on the next right ingredient.
for act_token in instruction_act_tokens:
if act_token['lemma'] not in actions_used:
# fing next ingredient right to it
next_ing = None
for i, ing_str in enumerate(clustered_ingredients):
if clustered_conllu_ids[i] > act_token['id']:
actions_used.add(act_token['lemma'])
ingredients_used.add(ing_str)
ing = Ingredient(ing_str)
current_state.apply_action(act_token['lemma'], ing, instruction_number=instruction_number, add_new_if_not_similar=False)
if debug:
print(f"\tapply {act_token['lemma']} on {ing}")
break
actions_unused = []
ingredients_unused = []
for act_token in instruction_act_tokens:
if act_token['lemma'] in actions_used:
continue
actions_unused.append(act_token['lemma'])
for ing_str in clustered_ingredients:
if ing_str in ingredients_used:
continue
ingredients_unused.append(ing_str)
if debug:
print(f"\nunused actions: {actions_unused} \nunused ings: {ingredients_unused}\n")
if (instruction_number > 1):
if debug:
print("mixing ingredients based on mixing actions with last instruction:")
for ing in current_state.get_ingredients_touched_in_instruction(instruction_number -1):
ing.mark_for_mixing()
for ing in current_state.get_combined_ingredients():
if debug:
print(f"\t* {ing}")
if debug:
print("mixing all ingredients in this instruction:")
for ing_str in clustered_ingredients:
current_state.apply_action("mix", Ingredient(ing_str), instruction_number=instruction_number, add_new_if_not_similar=False)
for ing in current_state.get_combined_ingredients():
if debug:
print(f"\t* {ing}")
# if no ingredient is found, apply actions on all ingredients so far used
if len(clustered_ingredients) == 0 and len(actions_unused) > 0:
if debug:
print("\nno ingredients found. So apply actions on all ingredients that are touched so far:")
for action in actions_unused:
current_state.apply_action_on_all(action, instruction_number, exclude_instruction_number=0)
if debug:
print(f"\nstate after instruction {instruction_number}:")
print(current_state)
print("\n")
def plot_matrices(self):
if self._recipe_state is None:
print("Error: no recipe state found")
return
mixings, mix_labels = self._recipe_state.get_mixing_matrix()
x_labels = [f"{ing._base_ingredient} 🡸 ({' '.join([act for act in ing._action_set])})" for ing in mix_labels]
y_labels = [f"({' '.join([act for act in ing._action_set])}) 🢂 {ing._base_ingredient}" for ing in mix_labels]
fig = go.Figure(data=go.Heatmap(
z=mixings,
x=x_labels,
y=y_labels,
xgap = 1,
ygap = 1,))
fig.update_layout(
width=1024,
height=1024,
yaxis = dict(
scaleanchor = "x",
scaleratio = 1,
)
)
fig.show()
actions, act_labels, ing_labels = self._recipe_state.get_action_matrix()
fig = go.Figure(data=go.Heatmap(
z=actions,
x=[f"{ing._base_ingredient} 🡸 ({' '.join([act for act in ing._action_set])})" for ing in ing_labels],
y=[str(a) for a in act_labels],
xgap = 1,
ygap = 1,))
fig.update_layout(
width=1024,
height=1024,
yaxis = dict(
scaleanchor = "x",
scaleratio = 1,
)
)
fig.show()