kenkeiras
0fbb9238eb
A property dictionary is now only to be considered equal to a word when it shares at least one group, or neither has groups.
482 lines
16 KiB
Python
482 lines
16 KiB
Python
#!/usr/bin/env python
|
|
|
|
from . import knowledge_evaluation
|
|
|
|
from . import depth_meter
|
|
import logging
|
|
import re
|
|
import copy
|
|
|
|
from functools import reduce
|
|
from typing import List, Dict
|
|
from .modifiable_property import ModifiableProperty
|
|
from . import parameters
|
|
|
|
# TODO: more flexible tokenization
|
|
def to_tokens(text):
|
|
return re.findall(r'(\w+|[^\s])', text)
|
|
|
|
|
|
def make_template(knowledge_base, tokens, parsed):
|
|
matcher = list(tokens)
|
|
template = list(parsed)
|
|
logging.debug(" -- MK TEMPLATE --")
|
|
logging.debug("MATCHR: {}".format(matcher))
|
|
logging.debug("TEMPLT: {}".format(template))
|
|
for i in range(len(matcher)):
|
|
word = matcher[i]
|
|
if word in template:
|
|
template[template.index(word)] = i
|
|
matcher[i] = {
|
|
'groups': set(knowledge_base.knowledge[word]['groups'])
|
|
}
|
|
return tokens, matcher, template
|
|
|
|
|
|
def is_bottom_level(tree):
|
|
for element in tree:
|
|
if isinstance(element, list) or isinstance(element, tuple):
|
|
return False
|
|
return True
|
|
|
|
|
|
def get_lower_levels(parsed):
|
|
lower = []
|
|
def aux(subtree, path):
|
|
nonlocal lower
|
|
deeper = len(path) == 0
|
|
for i, element in enumerate(subtree):
|
|
if isinstance(element, list) or isinstance(element, tuple):
|
|
aux(element, path + (i,))
|
|
deeper = True
|
|
|
|
if not deeper:
|
|
lower.append((path, subtree))
|
|
|
|
aux(parsed, path=())
|
|
return lower
|
|
|
|
|
|
# TODO: probably optimize this, it creates lots of unnecessary tuples
|
|
def replace_position(tree, position, new_element):
|
|
logging.debug("REPLACE POSITIONS:")
|
|
logging.debug(" TREE : {}".format(tree))
|
|
logging.debug("POSITION: {}".format(position))
|
|
logging.debug("NEW ELEM: {}".format(new_element))
|
|
logging.debug("------------------")
|
|
|
|
def aux(current_tree, remaining_route):
|
|
if len(remaining_route) == 0:
|
|
return new_element
|
|
|
|
else:
|
|
step = remaining_route[0]
|
|
return (
|
|
tree[:step]
|
|
+ (aux(tree[step], remaining_route[1:]),)
|
|
+ tree[step + 2:]
|
|
)
|
|
|
|
result = aux(tree, position)
|
|
logging.debug("-RESULT: {}".format(result))
|
|
return result
|
|
|
|
|
|
def integrate_language(knowledge_base, example):
|
|
text = example["text"].lower()
|
|
parsed = example["parsed"]
|
|
|
|
resolved_parsed = copy.deepcopy(parsed)
|
|
tokens = to_tokens(text)
|
|
|
|
while True:
|
|
logging.debug("P: {}".format(resolved_parsed))
|
|
lower_levels = get_lower_levels(resolved_parsed)
|
|
logging.debug("Lower: {}".format(lower_levels))
|
|
if len(lower_levels) == 0:
|
|
break
|
|
|
|
for position, atom in lower_levels:
|
|
logging.debug("\x1b[1mSelecting\x1b[0m: {}".format(atom))
|
|
similar = get_similar_tree(knowledge_base, atom, tokens)
|
|
remix, (start_bounds, end_bounds) = build_remix_matrix(knowledge_base, tokens, atom, similar)
|
|
|
|
after_remix = apply_remix(tokens[len(start_bounds):-len(end_bounds)], remix)
|
|
logging.debug("--FIND MIX--")
|
|
logging.debug("-MIX- | {}".format(remix))
|
|
logging.debug("-FRM- | {}".format(tokens))
|
|
logging.debug("-AFT- | {}".format(after_remix))
|
|
|
|
print()
|
|
|
|
_, matcher, result = make_template(knowledge_base, after_remix, atom)
|
|
logging.debug("Tx: {}".format(after_remix))
|
|
logging.debug("Mx: {}".format(matcher))
|
|
logging.debug("Rx: {}".format(result))
|
|
logging.debug("Sx: {}".format(start_bounds))
|
|
logging.debug("Ex: {}".format(end_bounds))
|
|
|
|
|
|
assert(len(after_remix) + len(start_bounds) + len(end_bounds) == len(tokens))
|
|
logging.debug( " +-> {}".format(after_remix))
|
|
subquery_type = knowledge_evaluation.get_subquery_type(knowledge_base.knowledge, atom)
|
|
logging.debug(r" \-> <{}>".format(subquery_type))
|
|
|
|
# Clean remaining tokens
|
|
new_tokens = list(tokens)
|
|
offset = len(start_bounds)
|
|
for _ in range(len(remix)):
|
|
new_tokens.pop(offset)
|
|
|
|
# TODO: Get a specific types for... types
|
|
new_tokens.insert(offset, (subquery_type, remix))
|
|
tokens = new_tokens
|
|
|
|
resolved_parsed = replace_position(resolved_parsed, position, offset)
|
|
logging.debug("RP: {}".format(resolved_parsed))
|
|
logging.debug("AT: {}".format(atom))
|
|
logging.debug("#########")
|
|
|
|
|
|
tokens, matcher, result = make_template(knowledge_base, tokens, resolved_parsed)
|
|
logging.debug("T: {}".format(tokens))
|
|
logging.debug("M: {}".format(matcher))
|
|
logging.debug("R: {}".format(result))
|
|
logging.debug("---")
|
|
return tokens, matcher, result
|
|
|
|
|
|
def apply_remix(tokens, remix):
|
|
rebuilt = []
|
|
for i in remix:
|
|
if isinstance(i, int):
|
|
if i >= len(tokens):
|
|
return None
|
|
rebuilt.append(tokens[i])
|
|
else:
|
|
assert(isinstance(i, str))
|
|
rebuilt.append(i)
|
|
return rebuilt
|
|
|
|
|
|
def build_remix_matrix(knowledge_base, tokens, atom, similar):
|
|
tokens = list(tokens)
|
|
tokens, matcher, result = make_template(knowledge_base, tokens, atom)
|
|
similar_matcher, similar_result, similar_result_resolved, _, _ = similar
|
|
|
|
start_bounds, end_bounds = find_bounds(knowledge_base, matcher, similar_matcher)
|
|
|
|
for i, element in (end_bounds + start_bounds[::-1]):
|
|
matcher.pop(i)
|
|
tokens.pop(i)
|
|
|
|
possible_remixes = get_possible_remixes(knowledge_base, matcher, similar_matcher)
|
|
chosen_remix = possible_remixes[0]
|
|
|
|
return chosen_remix, (start_bounds, end_bounds)
|
|
|
|
|
|
def get_possible_remixes(knowledge_base, matcher, similar_matcher):
|
|
|
|
matrix = []
|
|
for element in matcher:
|
|
logging.debug("- {}".format(element))
|
|
logging.debug("+ {}".format(similar_matcher))
|
|
if element in similar_matcher or isinstance(element, dict):
|
|
if isinstance(element, dict):
|
|
indexes = all_matching_indexes(knowledge_base, similar_matcher, element)
|
|
else:
|
|
indexes = all_indexes(similar_matcher, element)
|
|
matrix.append(indexes)
|
|
else:
|
|
matrix.append([element])
|
|
|
|
# TODO: do some scoring to find the most "interesting combination"
|
|
return [list(x) for x in list(zip(*matrix))]
|
|
|
|
|
|
def all_indexes(collection, element):
|
|
indexes = []
|
|
base = 0
|
|
|
|
for _ in range(collection.count(element)):
|
|
i = collection.index(element, base)
|
|
base = i + 1
|
|
indexes.append(i)
|
|
|
|
return indexes
|
|
|
|
|
|
def all_matching_indexes(knowledge_base, collection, element):
|
|
indexes = []
|
|
|
|
assert("groups" in element)
|
|
element = element["groups"]
|
|
for i, instance in enumerate(collection):
|
|
if isinstance(instance, dict):
|
|
instance = instance["groups"]
|
|
elif instance in knowledge_base.knowledge:
|
|
instance = knowledge_base.knowledge[instance]["groups"]
|
|
|
|
intersection = set(instance) & set(element)
|
|
if (len(intersection) > 0 or (0 == len(instance) == len(element))):
|
|
indexes.append((i, intersection))
|
|
|
|
return [x[0] for x in sorted(indexes, key=lambda x: len(x[1]), reverse=True)]
|
|
|
|
|
|
def element_matches_groups(knowledge, element: Dict, groups):
|
|
if isinstance(groups, str) and groups in knowledge:
|
|
return len(knowledge[element].get("groups", set()) & element['groups']) > 0
|
|
elif isinstance(groups, dict):
|
|
return len(element.get("groups", set()) & element['groups']) > 0
|
|
return False
|
|
|
|
|
|
def find_bounds(knowledge, matcher, similar_matcher):
|
|
start_bounds = []
|
|
for i, element in enumerate(matcher):
|
|
if element in similar_matcher:
|
|
break
|
|
else:
|
|
start_bounds.append((i, element))
|
|
|
|
end_bounds = []
|
|
for i, element in enumerate(matcher[::-1]):
|
|
in_similar = False
|
|
if isinstance(element, str):
|
|
in_similar = element in similar_matcher
|
|
elif isinstance(element, dict):
|
|
in_similar = any(map(lambda groups: element_matches_groups(knowledge.knowledge,
|
|
element, groups),
|
|
similar_matcher))
|
|
|
|
if in_similar:
|
|
break
|
|
else:
|
|
end_bounds.append((len(matcher) - (i + 1), element))
|
|
|
|
return start_bounds, end_bounds
|
|
|
|
|
|
def get_similar_tree(knowledge_base, atom, tokens):
|
|
possibilities = []
|
|
|
|
# Find matching possibilities
|
|
for entry, tree in knowledge_base.trained:
|
|
if not is_bottom_level(tree):
|
|
continue
|
|
if tree[0] == atom[0]:
|
|
possibilities.append((entry, tree))
|
|
|
|
# Sort by more matching elements
|
|
sorted_possibilities = []
|
|
for (raw, possibility) in possibilities:
|
|
resolved = []
|
|
for element in atom:
|
|
if isinstance(element, str):
|
|
resolved.append(element)
|
|
else:
|
|
resolved.append(knowledge_evaluation.resolve(
|
|
knowledge_base.knowledge,
|
|
element,
|
|
raw))
|
|
|
|
# TODO: Probably should take into account the categories of the elements in the "intake" ([0]) element
|
|
atom_score = sum([resolved[i] == atom[i]
|
|
for i
|
|
in range(min(len(resolved),
|
|
len(atom)))])
|
|
token_score = sum([similar_token in tokens
|
|
for similar_token
|
|
in raw])
|
|
|
|
sorted_possibilities.append((raw, possibility, resolved, atom_score, token_score))
|
|
|
|
sorted_possibilities = sorted(sorted_possibilities, key=lambda p: p[3] * 100 + p[4], reverse=True)
|
|
if len(sorted_possibilities) < 1:
|
|
return None
|
|
|
|
for i, possibility in enumerate(sorted_possibilities):
|
|
logging.debug('---- POSSIBILITY #{} ----'.format(i))
|
|
similar_matcher, similar_result, similar_result_resolved, _, _ = possibility
|
|
logging.debug('AST: {}'.format(similar_result))
|
|
logging.debug('Based on: {}'.format(similar_matcher))
|
|
logging.debug('Results on: {}'.format(similar_result_resolved))
|
|
logging.debug('---------------------')
|
|
|
|
return sorted_possibilities[0]
|
|
|
|
|
|
# TODO: unroll this mess
|
|
def get_matching(sample, other):
|
|
l = len(sample[0])
|
|
other = list(filter(lambda x: len(x[0]) == l, other))
|
|
for i in range(l):
|
|
if len(other) == 0:
|
|
return []
|
|
|
|
if isinstance(sample[0][i], dict): # Dictionaries are compared by groups
|
|
other = list(filter(lambda x: isinstance(x[0][i], dict) and
|
|
len(x[0][i]['groups'] & sample[0][i]['groups']) > 0,
|
|
other))
|
|
|
|
elif isinstance(sample[0][i], tuple): # Tuples are compared by types [0]
|
|
other = list(filter(lambda x: isinstance(x[0][i], tuple) and
|
|
x[0][i][0] == sample[0][i][0],
|
|
other))
|
|
|
|
matching = []
|
|
for x in range(l): # Generate the combination of this and other(s) matcher
|
|
first_sample_data = sample[0][x]
|
|
if isinstance(first_sample_data, str):
|
|
matching.append(first_sample_data)
|
|
elif isinstance(first_sample_data, tuple):
|
|
matching.append(first_sample_data)
|
|
else:
|
|
this_groups = sample[0][x]['groups']
|
|
if len(other) > 0:
|
|
other_groups = reduce(lambda a, b: a & b,
|
|
map(lambda y: y[0][x]['groups'],
|
|
other))
|
|
this_groups = this_groups & other_groups
|
|
|
|
matching.append({'groups': this_groups})
|
|
return matching
|
|
|
|
|
|
def reprocess_language_knowledge(knowledge_base, examples):
|
|
examples = knowledge_base.examples + examples
|
|
|
|
pattern_examples = []
|
|
for i, sample in enumerate(examples):
|
|
other = examples[:i] + examples[i + 1:]
|
|
match = get_matching(sample, other)
|
|
if len(match) > 0:
|
|
sample = (match, sample[1],)
|
|
pattern_examples.append(sample)
|
|
|
|
return pattern_examples
|
|
|
|
|
|
def reverse_remix(tree_section, remix):
|
|
result_section = []
|
|
offset = 0
|
|
for origin in remix:
|
|
if isinstance(origin, int):
|
|
if (origin + offset) >= len(tree_section):
|
|
return None
|
|
|
|
result_section.append(copy.deepcopy(tree_section[origin + offset]))
|
|
else:
|
|
assert(isinstance(origin, str))
|
|
offset += 1
|
|
return result_section + tree_section[len(remix):]
|
|
|
|
|
|
def get_fit(knowledge, tokens, remaining_recursions=parameters.MAX_RECURSIONS):
|
|
results = []
|
|
for matcher, ast in knowledge.trained:
|
|
result = match_fit(knowledge, tokens, matcher, ast,
|
|
remaining_recursions)
|
|
|
|
if result is not None:
|
|
results.append(result)
|
|
print("XXX", result)
|
|
|
|
print(' - ' + '\n - '.join(map(str, results)))
|
|
if len(results) > 0:
|
|
return results[0]
|
|
|
|
|
|
def is_definite_minisegment(minisegment):
|
|
return isinstance(minisegment, str) or isinstance(minisegment, dict)
|
|
|
|
|
|
def match_token(knowledge, next_token, minisegment):
|
|
if isinstance(minisegment, dict):
|
|
return knowledge_evaluation.can_be_used_in_place(knowledge, next_token, minisegment)
|
|
elif isinstance(minisegment, str):
|
|
# TODO: check if the two elements can be used in each other place
|
|
return next_token == minisegment
|
|
|
|
return False
|
|
|
|
|
|
def resolve_fit(knowledge, fit, remaining_recursions):
|
|
fitted = []
|
|
for element in fit:
|
|
if is_definite_minisegment(element):
|
|
fitted.append(element)
|
|
else:
|
|
((result_type, remixer), tokens) = element
|
|
remixed_tokens = reverse_remix(tokens, remixer)
|
|
if remixed_tokens is None:
|
|
return None
|
|
|
|
# if len(tokens) == 3 and tokens[2] == 'electricity':
|
|
# logging.debug("--UNMIX--")
|
|
# logging.debug("-MIX- | {}".format(remixer))
|
|
# logging.debug("REMIX | {}".format(tokens))
|
|
# logging.debug(" T O | {}".format(remixed_tokens))
|
|
# if remixer != [0, 1, 2]:
|
|
# return None
|
|
|
|
minifit = get_fit(knowledge, remixed_tokens, remaining_recursions - 1)
|
|
if minifit is None:
|
|
return None
|
|
|
|
minitokens, miniast = minifit
|
|
logging.debug(" AST | {}".format(miniast))
|
|
subproperty = knowledge_evaluation.resolve(knowledge.knowledge, minitokens, miniast)
|
|
fitted.append(subproperty)
|
|
|
|
return fitted
|
|
|
|
|
|
def match_fit(knowledge, tokens, matcher, ast, remaining_recursions):
|
|
segment_possibilities = [([], tokens)] # Matched tokens, remaining tokens
|
|
indent = ' ' * (parameters.MAX_RECURSIONS - remaining_recursions)
|
|
for minisegment in matcher:
|
|
possibilities_after_round = []
|
|
for matched_tokens, remaining_tokens in segment_possibilities:
|
|
if len(remaining_tokens) < 1:
|
|
continue
|
|
|
|
if is_definite_minisegment(minisegment):
|
|
if match_token(knowledge, remaining_tokens[0], minisegment):
|
|
possibilities_after_round.append((
|
|
matched_tokens + [remaining_tokens[0]],
|
|
remaining_tokens[1:]
|
|
))
|
|
else:
|
|
# TODO: optimize this with a look ahead
|
|
for i in range(1, len(tokens)):
|
|
possibilities_after_round.append((
|
|
matched_tokens + [(minisegment, remaining_tokens[:i])],
|
|
remaining_tokens[i:]
|
|
))
|
|
else:
|
|
segment_possibilities = possibilities_after_round
|
|
|
|
fully_matched_segments = [(matched, remaining)
|
|
for (matched, remaining)
|
|
in segment_possibilities
|
|
if len(remaining) == 0]
|
|
|
|
resolved_fits = []
|
|
for fit, _ in fully_matched_segments:
|
|
print(indent + ":::", fit) # REMIXES HAVE TO BE APPLIED BEFORE!!!
|
|
|
|
print(indent + '*' * 20)
|
|
for fit, _ in fully_matched_segments:
|
|
print(indent + ":::", fit) # REMIXES HAVE TO BE APPLIED BEFORE!!!
|
|
resolved_fit = resolve_fit(knowledge, fit, remaining_recursions)
|
|
if resolved_fit is not None:
|
|
resolved_fits.append(resolved_fit)
|
|
|
|
if len(resolved_fits) == 0:
|
|
return None
|
|
|
|
return resolved_fits[0], ast
|