from modifiable_property import ModifiableProperty def resolve(knowledge_base, elements, value): if isinstance(value, int): return elements[value] elif isinstance(value, tuple) or isinstance(value, list): return integrate_information(knowledge_base, { "elements": elements, "parsed": value, }) return value # TODO: improve typing def infer_type(result): if isinstance(result, bool): return "bool" elif isinstance(result, int): return "int" else: raise Exception("Unknown type for value: {}".format(result)) def get_subquery_type(knowledge_base, atom): subquery_result = integrate_information(knowledge_base, { "parsed": atom, "elements": [], }) assert (subquery_result is not None) result = subquery_result.getter() result_type = infer_type(result) return result_type def property_for_value(knowledge_base, value): # logging.debug(value) # logging.debug(knowledge_base[value]) return knowledge_base[value]['as_property'] def modifiable_property_from_property(prop, path, value): def getter(): nonlocal prop, path, value return (path in prop) and prop[path] == value def setter(): nonlocal prop, path, value prop[path] = value return ModifiableProperty( getter=getter, setter=setter, ) def exists_property_with_value(knowledge_base, elements, subj, value): subj = resolve(knowledge_base, elements, subj) value = resolve(knowledge_base, elements, value) if subj not in knowledge_base: knowledge_base[subj] = {} return modifiable_property_from_property( prop=knowledge_base[subj], path=property_for_value(knowledge_base, value), value=value ) def modifiable_element_for_existance_in_set(container, set_name, element): def getter(): nonlocal container, set_name, element return (set_name in container) and (element in container[set_name]) def setter(): nonlocal container, set_name, element return container[set_name].add(element) return ModifiableProperty( getter=getter, setter=setter, ) def pertenence_to_group(knowledge_base, elements, subj, group): subj = resolve(knowledge_base, elements, subj) group = resolve(knowledge_base, elements, group) if subj not in knowledge_base: knowledge_base[subj] = {} if "groups" not in knowledge_base[subj]: knowledge_base[subj]["groups"] = set() return modifiable_element_for_existance_in_set( container=knowledge_base[subj], set_name="groups", element=group ) def has_capacity(knowledge_base, elements, subj, capacity): subj = resolve(knowledge_base, elements, subj) capacity = resolve(knowledge_base, elements, capacity) if subj not in knowledge_base: knowledge_base[subj] = {} if "capacities" not in knowledge_base[subj]: knowledge_base[subj]["capacities"] = set() return modifiable_element_for_existance_in_set( container=knowledge_base[subj], set_name="capacities", element=capacity ) def question(knowledge_base, elements, subj): subj = resolve(knowledge_base, elements, subj) if isinstance(subj, ModifiableProperty): return subj.getter() return subj knowledge_ingestion = { "exists-property-with-value": exists_property_with_value, "pertenence-to-group": pertenence_to_group, "has-capacity": has_capacity, "question": question, } def integrate_information(knowledge_base, example): method = example['parsed'][0] args = example['parsed'][1:] elements = example.get('elements', None) return knowledge_ingestion[method](knowledge_base, elements, *args)