Configurable Generation of Synthetic Schemas and Knowledge Graphs at Your Fingertips
# Configuration file for the Sphinx documentation builder.

from datetime import date
import sphinx_rtd_theme

# -- Project information

project = "PyGraft"
copyright = f"2023-{date.today().year}, Nicolas Hubert"
author = "Nicolas Hubert"

release = "0.1.0"

# -- General configuration

extensions = [
    "sphinx.ext.duration",
    "sphinx.ext.doctest",
    "sphinx.ext.autodoc",
    'sphinxcontrib.bibtex',
    "sphinxemoji.sphinxemoji",
    "sphinx.ext.autosummary",
    "sphinx.ext.mathjax",
    "sphinx.ext.viewcode",
    "sphinx.ext.intersphinx",
]

bibtex_bibfiles = ['bibliography.bib']
sphinxemoji_style = "twemoji"

# -- Extension configuration

# -- Options for intersphinx extension

# Example configuration for intersphinx: refer to the Python standard library.
intersphinx_mapping = {
    "python": ("https://docs.python.org/3", None),
    "numpy": ("https://numpy.org/doc/stable", None),
    "sphinx": ("https://www.sphinx-doc.org/en/master/", None)
}

# generate autosummary pages
autosummary_generate = True

# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:
source_suffix = ['.rst', '.md']
# The master toctree document.
master_doc = "index"
language = "en"
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = "sphinx"

intersphinx_disabled_domains = ["std"]

templates_path = ["_templates"]

# General information about the project.
project = 'PyGraft'
copyright = 'PyGraft is licensed under the MIT License'
author = 'Nicolas Hubert'

# -- Options for HTML output

html_theme = "sphinx_rtd_theme"
html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
html_theme_options = {
    'style_nav_header_background': '#FFFFFF',
}

# html_static_path = ["_static"]
html_css_files = [
    'style.css'
]
html_logo = "logo.svg"

# -- Options for HTMLHelp output

# Output file base name for HTML help builder.
htmlhelp_basename = "PyGraftdoc"

# -- Options for EPUB output
epub_show_urls = "footnote"

docs/source/conf.py
nicolas-hbt/pygraft
from .pygraft import create_template, create_json_template, create_yaml_template, generate_schema, generate_kg, generate

__all__ = [
    "create_template",
    "create_json_template",
    "create_yaml_template",
    "generate_schema",
    "generate_kg",
    "generate",
]

pygraft/__init__.py
nicolas-hbt/pygraft
import random
import copy
import numpy as np
from collections import defaultdict
from pygraft.utils_schema import *
from tabulate import tabulate


class ClassGenerator:
    def __init__(self, **kwargs):
        self.init_params(**kwargs)
        self.init_class_structures()

    def init_params(self, **kwargs):
        """
        Initializes the parameters for the class generator.

        Args:
            self (object): The instance of the ClassGenerator.
            kwargs (dict): The keyword arguments.

        Returns:
            None
        """
        self.num_classes = kwargs.get("num_classes")
        self.max_hierarchy_depth = kwargs.get("max_hierarchy_depth")
        self.avg_class_depth = kwargs.get("avg_class_depth")
        self.class_inheritance_ratio = kwargs.get("class_inheritance_ratio")
        self.avg_disjointness = kwargs.get("avg_disjointness")
        self.verbose = kwargs.get("verbose")

    def init_class_structures(self):
        """
        Initializes the class structures for the class generator.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:    
            None
        """
        self.classes = None
        self.class2superclass_direct = {}
        self.class2superclasses_transitive = defaultdict(list)
        self.class2subclasses_direct = defaultdict(list)
        self.class2subclasses_transitive = defaultdict(list)
        self.layer2classes = defaultdict(list)

    def generate_class_schema(self):
        """
        Generates a class schema based on the given parameters.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            class_info (dict): A dictionary containing various information about the class.
        """
        self.generate_class_hierarchy()
        self.generate_class_disjointness()

        if self.verbose:
            self.print_schema()

        class_info = self.assemble_class_info()

        return class_info

    def assemble_class_info(self):
        """
        Assembles and returns information about the current class schema.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            class_info (dict): A dictionary containing various information about the class.
        """
        class_info = {
            "num_classes": len(self.classes),
            "classes": self.classes,
            "hierarchy_depth": get_max_depth(self.layer2classes),
            "avg_class_depth": round(calculate_average_depth(self.layer2classes), 2),
            "class_inheritance_ratio": round(
                calculate_inheritance_ratio(self.class2superclass_direct, self.class2subclasses_direct), 2
            ),
            "direct_class2subclasses": self.class2subclasses_direct,
            "direct_class2superclass": self.class2superclass_direct,
            "transitive_class2subclasses": self.class2subclasses_transitive,
            "transitive_class2superclasses": self.class2superclasses_transitive,
            "avg_class_disjointness": round(calculate_class_disjointness(self.disjointwith, len(self.classes)), 2),
            "class2disjoints": self.disjointwith,
            "class2disjoints_symmetric": self.mutual_disjointness,
            "class2disjoints_extended": {c: list(set(classes)) for c, classes in self.disjointwith_extended.items()},
            "layer2classes": {int(k): v for k, v in self.layer2classes.items()},
            "class2layer": generate_class2layer(self.layer2classes),
        }

        return class_info

    def generate_classes(self):
        """
        Generates the classes for the given self instance based on the
        given number of classes.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            None
        """
        self.classes = [f"C{i}" for i in range(1, self.num_classes + 1)]

    def generate_class_hierarchy(self):
        """
        Generate the class hierarchy for the given self instance.
        This function creates a hierarchical structure of classes based on the
        existing classes in the self instance. It assigns each class to a layer
        and establishes the parent-child relationships between them.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            None
        """
        self.generate_classes()

        unconnected_classes = copy.deepcopy(self.classes)

        c = unconnected_classes.pop()
        self.link_child2parent(c, "owl:Thing", layer=1)

        for layer in range(1, self.max_hierarchy_depth):
            c2 = unconnected_classes.pop()
            self.link_child2parent(c2, c, layer=layer + 1)
            c = c2

        current_avg_depth = calculate_average_depth(self.layer2classes)
        current_inheritance_ratio = calculate_inheritance_ratio(
            self.class2superclass_direct, self.class2subclasses_direct
        )

        stochastic_noise_until = int(len(unconnected_classes) * 0.5)

        while unconnected_classes:
            c = unconnected_classes.pop()

            if (
                random.random() < 0.35
                and len(unconnected_classes) >= stochastic_noise_until
                and self.max_hierarchy_depth > 3
            ):
                self.noisy_placing(c, current_avg_depth, current_inheritance_ratio)
            else:
                self.smart_placing(c, current_avg_depth, current_inheritance_ratio)

            current_avg_depth = calculate_average_depth(self.layer2classes)
            current_inheritance_ratio = calculate_inheritance_ratio(
                self.class2superclass_direct, self.class2subclasses_direct
            )

    def smart_placing(self, c, current_avg_depth, current_inheritance_ratio):
        """
        Determines the appropriate action to take based on the current average depth 
        and current inheritance ratio.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): A given class.
            current_avg_depth (float): The current average depth.
            current_inheritance_ratio (float): The current inheritance ratio.

        Returns:
            None
        """

        if current_avg_depth <= self.avg_class_depth and current_inheritance_ratio <= self.class_inheritance_ratio:
            self.create_deep_leaf_realistic(c)
        elif current_avg_depth <= self.avg_class_depth and current_inheritance_ratio > self.class_inheritance_ratio:
            self.create_deep_child_realistic(c)
        elif current_avg_depth > self.avg_class_depth and current_inheritance_ratio <= self.class_inheritance_ratio:
            self.create_shallow_leaf(c)
        elif current_avg_depth > self.avg_class_depth and current_inheritance_ratio > self.class_inheritance_ratio:
            self.create_shallow_leaf_root(c)

    def noisy_placing(self, c, current_avg_depth, current_inheritance_ratio):
        """
        Adds noise to the class hierarchy tree to create more diverse shapes.
        This function adds randomness to the placement of classes in the hierarchy tree.
        Without this optional noise, trees tend to be vertical with only a few parents having most children.
        By putting more weight on the intermediate layers, which tend to be underpopulated otherwise,
        the hierarchy becomes more realistic.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be placed in the hierarchy tree.
            current_avg_depth (float): The current average depth of the hierarchy tree.
            current_inheritance_ratio (float): The current inheritance ratio of the hierarchy tree.

        Returns:
            None
        """

        focus_layers = [key - 1 for key, value in self.layer2classes.items() if value]
        # the parent will not be in 1st or penultimate layer:
        focus_layers = [l for l in focus_layers if l not in [0, 1, self.max_hierarchy_depth - 1]]

        if focus_layers:
            layer = random.choice(focus_layers)
            parent = random.choice(self.layer2classes[layer])
            self.link_child2parent(c, parent, layer=layer + 1)
        else:
            self.smart_placing(c, current_avg_depth, current_inheritance_ratio)

    def create_deep_leaf_realistic(self, c):
        """
        Create a child to an already existing parent which is deep in the class hierarchy.

        => |S| += 1; |C| += 1; |L| += 1
        => Inheritance ratio increases.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be placed.

        Returns:
            None
        """
        deep_layers = [key - 1 for key, value in self.layer2classes.items() if value and key >= self.avg_class_depth]
        layer = random.choice(deep_layers)

        while True:
            current_parents = [c for c in self.layer2classes[layer] if c in self.class2subclasses_direct.keys()]

            if current_parents:
                parent = random.choice(current_parents)
                self.link_child2parent(c, parent, layer=layer + 1)
                break
            else:
                layer -= 1

    def create_deep_leaf_deterministic(self, c):
        """
        Creates a child to an already existing parent which is deep in the class hierarchy.

        => |S| += 1; |C| += 1; |L| += 1
        => Inheritance ratio increases.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be placed.

        Returns:
            None
        """
        found = False
        layer = max((key for key, value in self.layer2classes.items() if value), default=None) - 1

        while not found:
            current_parents = [c for c in self.layer2classes[layer] if c in self.class2subclasses_direct.keys()]

            if current_parents:
                found = True
                parent = random.choice(current_parents)
                self.link_child2parent(c, parent, layer=layer + 1)
            else:
                layer -= 1

    def create_deep_child_realistic(self, c):
        """
        Creates a child to a leaf which is deep in the class hierarchy.

        => |S| += 1; |C| += 1; |L| = |L| (unchanged)
        => Inheritance ratio decreases.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be placed.

        Returns:
            None
        """
        found = False
        deep_layers = [key - 1 for key, value in self.layer2classes.items() if value and key >= self.avg_class_depth]
        layer = random.choice(deep_layers)

        while not found:
            current_leaves = [c for c in self.layer2classes[layer] if c not in self.class2subclasses_direct.keys()]

            if current_leaves:
                found = True
                parent = random.choice(current_leaves)
                self.link_child2parent(c, parent, layer=layer + 1)
            else:
                layer -= 1

    def create_deep_child_deterministic(self, c):
        """
        Creates a child to a leaf which is deep in the class hierarchy.

        => |S| += 1; |C| += 1; |L| = |L| (unchanged)
        => Inheritance ratio decreases.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be placed.

        Returns:
            None
        """
        found = False
        layer = max((key for key, value in self.layer2classes.items() if value), default=None) - 1

        while not found:
            current_leaves = [c for c in self.layer2classes[layer] if c not in self.class2subclasses_direct.keys()]

            if current_leaves:
                found = True
                parent = random.choice(current_leaves)
                self.link_child2parent(c, parent, layer=layer + 1)
            else:
                layer -= 1

    def create_shallow_leaf(self, c):
        """
        Creates a shallow leaf by adding a new class to the class hierarchy.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The class to be added as a shallow leaf.

        Returns:
            None
        """
        layer = 1
        parent = random.choice(self.layer2classes[layer])
        self.link_child2parent(c, parent, layer=layer + 1)

    def create_shallow_leaf_root(self, c):
        """
        Creates a leaf which is placed just under the root of the class hierarchy.

        Args:
            self (object): The instance of the ClassGenerator.
            c (str): The name of the class.

        Returns:
            None
        """
        self.link_child2parent(c, "owl:Thing", layer=1)

    def generate_class_disjointness(self):
        """
        Generates class disjointness by randomly selecting two classes and making them incompatible.
        Updates the class mappings and extend the incompatibilities to subclasses.
        Calculates the current class disjointness and stops when average disjointness threshold is reached.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            None
        """
        current_class_disjointness = 0.0
        self.class2superclasses_transitive, self.class2subclasses_transitive = extend_class_mappings(
            self.class2superclass_direct
        )
        self.disjointwith, self.disjointwith_extended, self.mutual_disjointness = {}, {}, set()

        while current_class_disjointness < self.avg_disjointness:
            found = False
            while not found:
                # pick one class A randomly
                A = random.choice(self.classes)
                # pick another class B randomly that is neither a transitive parent nor child of A
                B = random.choice(self.classes)
                while (
                    B == A
                    or (A in self.class2superclasses_transitive and B in self.class2superclasses_transitive[A])
                    or (A in self.class2subclasses_transitive and B in self.class2subclasses_transitive[A])
                ):
                    B = random.choice(self.classes)
                found = True

            # make A and B incompatible
            self.disjointwith.setdefault(A, []).append(B)
            self.disjointwith.setdefault(B, []).append(A)
            self.disjointwith_extended.setdefault(A, []).append(B)
            self.disjointwith_extended.setdefault(B, []).append(A)
            mutual_disj_key = f"{A}-{B}" if int(A[1:]) < int(B[1:]) else f"{B}-{A}"
            self.mutual_disjointness.add(mutual_disj_key)
            # iterate through subclasses of A and B, and make them incompatible
            self.extend_incompatibilities(A, B)
            # update current class disjointness
            current_class_disjointness = calculate_class_disjointness(self.disjointwith, self.num_classes)

    def extend_incompatibilities(self, class_A, class_B):
        """
        Extends the incompatibilities between two classes.

        Args:
            self (object): The instance of the ClassGenerator.
            class_A (str): The first class.
            class_B (str): The second class.

        Returns:
            None
        """
        children_A = self.class2subclasses_transitive.get(class_A, [])
        children_B = self.class2subclasses_transitive.get(class_B, [])

        # Iterate over the transitive children of class A
        for child_A in children_A:
            # Iterate over the transitive children of class B
            for child_B in children_B:
                self.disjointwith_extended.setdefault(child_A, []).append(child_B)
                self.disjointwith_extended.setdefault(child_B, []).append(child_A)
                mutual_disj_key = (
                    f"{child_A}-{child_B}" if int(child_A[1:]) < int(child_B[1:]) else f"{child_B}-{child_A}"
                )
                self.mutual_disjointness.add(mutual_disj_key)

        # Handle the case where class A has no children
        if not children_A:
            for child_B in children_B:
                self.disjointwith_extended.setdefault(class_A, []).append(child_B)
                self.disjointwith_extended.setdefault(child_B, []).append(class_A)
                mutual_disj_key = (
                    f"{class_A}-{child_B}" if int(class_A[1:]) < int(child_B[1:]) else f"{child_B}-{class_A}"
                )
                self.mutual_disjointness.add(mutual_disj_key)

        # Handle the case where class B has no children
        if not children_B:
            for child_A in children_A:
                self.disjointwith_extended.setdefault(child_A, []).append(class_B)
                self.disjointwith_extended.setdefault(class_B, []).append(child_A)
                mutual_disj_key = (
                    f"{class_B}-{child_A}" if int(class_B[1:]) < int(child_A[1:]) else f"{child_A}-{class_B}"
                )
                self.mutual_disjointness.add(mutual_disj_key)

    def link_child2parent(self, child, parent, layer):
        self.class2subclasses_direct[parent] = [child]
        self.class2superclass_direct[child] = parent
        self.layer2classes[layer].append(child)

    def print_schema(self):
        """
        Prints the generated class schema.

        Args:
            self (object): The instance of the ClassGenerator.

        Returns:
            None
        """
        print("Ontology Generated.")
        print("===================")
        print("Ontology Parameters:")
        print("===================")
        print("\n")

        table = [
            ["Number of Classes", len(self.classes), self.num_classes],
            ["Maximum Hierarchy Depth", get_max_depth(self.layer2classes), self.max_hierarchy_depth],
            ["Average Class Depth", round(calculate_average_depth(self.layer2classes), 2), self.avg_class_depth],
            [
                "Class Inheritance Ratio",
                round(calculate_inheritance_ratio(self.class2superclass_direct, self.class2subclasses_direct), 2),
                self.class_inheritance_ratio,
            ],
            [
                "Average Disjointness",
                round(calculate_class_disjointness(self.disjointwith, len(self.classes)), 2),
                self.avg_disjointness,
            ],
        ]

        headers = ["Class Metric", "Value", "Specified Value"]
        table_str = tabulate(table, headers, tablefmt="pretty")
        print(table_str)

pygraft/class_generator.py
nicolas-hbt/pygraft
import copy
import time
import numpy as np
from collections import Counter
import json
import itertools
from owlready2 import *
from rdflib import Graph as RDFGraph, Namespace, URIRef, RDF, OWL
from tqdm.auto import tqdm
from pygraft.utils_kg import *
from pygraft.utils import reasoner


class InstanceGenerator:
    def __init__(self, **kwargs):
        self.init_params(**kwargs)
        self.init_utils(**kwargs)

    def init_params(self, **kwargs):
        """
        Initializes general KG information with user-specified parameters.

        Args:
            self (object): The instance of the InstanceGenerator.
            kwargs (dict): Dictionary of parameter names and values.

        Returns:
            None
        """
        self.schema = kwargs.get("schema")
        self.num_entities = kwargs.get("num_entities")
        self.num_triples = kwargs.get("num_triples")
        self.relation_balance_ratio = kwargs.get("relation_balance_ratio")
        self.prop_untyped_entities = kwargs.get("prop_untyped_entities")
        self.avg_depth_specific_class = kwargs.get("avg_depth_specific_class")
        self.multityping = kwargs.get("multityping")
        self.avg_multityping = kwargs.get("avg_multityping")
        self.multityping = False if self.avg_multityping == 0.0 else self.multityping

    def init_utils(self, **kwargs):
        """
        Initializes auxiliary information.

        Args:
            self (object): The instance of the InstanceGenerator.
            kwargs (dict): Dictionary of parameter names and values.

        Returns:
            None
        """
        self.directory = f"output/{self.schema}/"
        self.format = kwargs.get("format")
        self.fast_gen = kwargs.get("fast_gen")
        self.oversample = kwargs.get("oversample")
        self.fast_ratio = get_fast_ratio(self.num_entities) if self.fast_gen else 1
        self.oversample_every = int(self.num_triples / self.fast_ratio)
        self.load_schema_info()

    def load_schema_info(self):
        """
        Loads schema information from class_info and relation_info json files.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """

        with open(f"{self.directory}class_info.json", "r") as file:
            self.class_info = json.load(file)
        with open(f"{self.directory}relation_info.json", "r") as file:
            self.relation_info = json.load(file)

        if self.avg_depth_specific_class > (self.class_info["hierarchy_depth"] + 1):
            self.avg_depth_specific_class = self.class_info["hierarchy_depth"] - 1

    def assemble_instance_info(self):
        """
        Assembles the KG information and returns a dictionary
        containing statistics and user parameters.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            kg_info (dict): A dictionary containing information about the KG.
        """
        observed_entities = {entity for tup in self.kg for entity in tup[::2]}
        typed_observed = {entity for entity in observed_entities if entity in self.ent2classes_specific}
        observed_relations = {tup[1] for tup in self.kg}
        kg_info = {
            "user_parameters": {
                "schema": self.schema,
                "num_entities": self.num_entities,
                "num_triples": self.num_triples,
                "relation_balance_ratio": self.relation_balance_ratio,
                "prop_untyped_entities": self.prop_untyped_entities,
                "avg_depth_specific_class": self.avg_depth_specific_class,
                "multityping": self.multityping,
                "avg_multityping": self.avg_multityping,
            },
            "statistics": {
                "num_entities": len(observed_entities),
                "num_instantiated_relations": len(observed_relations),
                "num_triples": len(self.kg),
                "prop_untyped_entities": round(1 - (len(typed_observed) / len(observed_entities)), 2),
                "avg_depth_specific_class": self.current_avg_depth_specific_class,
                "avg_multityping": round(self.calculate_avg_multityping(), 2) if len(self.is_typed) > 0 else 0.0,
            },
        }

        with open(self.directory + "kg_info.json", "w") as file:
            json.dump(kg_info, file, indent=4)

        return kg_info

    def write_kg(self):
        """
        Writes the KG to a file.
        Initializes a new RDFGraph object and parses the schema.
        Each triple in the KG is added to the full graph.
        The full graph is then serialized to a file and checked for consistency.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        self.graph = RDFGraph()
        self.graph.parse(f"{self.directory}schema.rdf") if self.format == "xml" else self.graph.parse(
            f"{self.directory}schema.{self.format}"
        )

        schema = Namespace("http://pygraf.t/")
        self.graph.bind("sc", schema)

        for h, r, t in tqdm(self.kg, desc="Writing instance triples", unit="triples", colour="red"):
            self.graph.add((URIRef(schema + h), URIRef(schema + r), URIRef(schema + t)))

            if h in self.ent2classes_specific:
                for c in self.ent2classes_specific[h]:
                    self.graph.add((URIRef(schema + h), RDF.type, URIRef(schema + c)))

            if t in self.ent2classes_specific:
                for c in self.ent2classes_specific[t]:
                    self.graph.add((URIRef(schema + t), RDF.type, URIRef(schema + c)))

        self.graph.serialize(
            f"{self.directory}full_graph.rdf", format="xml"
        ) if self.format == "xml" else self.graph.serialize(
            f"{self.directory}full_graph.{self.format}", format=self.format
        )

        kg_file = (
            f"{self.directory}full_graph.rdf" if self.format == "xml" else f"{self.directory}full_graph.{self.format}"
        )
        reasoner(resource_file=kg_file, resource="KG")

    def generate_kg(self):
        self.pipeline()
        self.check_asymmetries()
        self.check_inverseof_asymmetry()
        self.check_dom_range()
        self.procedure_1()
        self.procedure_2()
        kg_info = self.assemble_instance_info()
        self.write_kg()

    def assign_most_specific(self):
        """
        Assigns the most specific class to each entity based on the hierarchy depth.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        hierarchy_depth = self.class_info["hierarchy_depth"] + 1
        shape = hierarchy_depth / (hierarchy_depth - 1)
        numbers = np.random.power(shape, size=len(self.is_typed))
        scaled_numbers = numbers / np.mean(numbers) * self.avg_depth_specific_class
        generated_numbers = np.clip(np.floor(scaled_numbers), 1, hierarchy_depth).astype(int)
        generated_numbers = [n if n < hierarchy_depth else hierarchy_depth for n in generated_numbers]
        self.current_avg_depth_specific_class = np.mean(generated_numbers)
        self.ent2layer_specific = {e: l for e, l in zip(self.is_typed, generated_numbers)}
        self.ent2classes_specific = {
            e: [np.random.choice(self.layer2classes[l])] for e, l in self.ent2layer_specific.items()
        }

    def complete_typing(self):
        """
        Completes the typing for the current entity (if multityping is enabled).

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        current_avg_multityping = 1.0
        entity_list = list(copy.deepcopy(self.is_typed))
        cpt = 0

        if entity_list:
            while current_avg_multityping < self.avg_multityping and cpt < 10:
                ent = np.random.choice(entity_list)
                most_specific_classes = self.ent2classes_specific[ent]
                specific_layer = self.ent2layer_specific[ent]
                compatible_classes = self.find_compatible_classes(most_specific_classes)
                specific_compatible_classes = list(
                    set(self.layer2classes[specific_layer]).intersection(set(compatible_classes))
                )
                specific_compatible_classes = [
                    cl for cl in specific_compatible_classes if cl not in most_specific_classes
                ]

                if specific_compatible_classes:
                    other_specific_class = np.random.choice(specific_compatible_classes)
                    self.ent2classes_specific[ent].append(other_specific_class)
                    current_avg_multityping = self.calculate_avg_multityping()
                    cpt = 0
                else:
                    cpt += 1

    def check_multityping(self):
        """
        Checks the multityping of entities and updates the badly typed entities.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        self.badly_typed = {}

        for e, classes in self.ent2classes_transitive.items():
            for c in classes:
                disj = self.class2disjoints_extended.get(c, [])
                if set(disj).intersection(classes):
                    self.badly_typed[e] = {"all_classes": classes, "problematic_class": c, "disjointwith": disj}
                    # keep only one of its most_specific classes and update its transitive classes
                    self.ent2classes_specific[e] = np.random.choice(self.ent2classes_specific[e])
                    self.ent2classes_transitive[e] = self.class_info["transitive_class2superclasses"][
                        self.ent2classes_specific[e]
                    ]
                    break

    def extend_superclasses(self):
        """
        Extends the superclasses of entities.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        self.ent2classes_transitive = {k: set(v) for k, v in self.ent2classes_specific.items()}

        for ent, specific_cls in self.ent2classes_specific.items():
            # Extend superclasses recursively
            for specific_cl in specific_cls:
                self.ent2classes_transitive[ent].update(
                    set(self.class_info["transitive_class2superclasses"][specific_cl])
                )
            self.ent2classes_transitive[ent] = list(set(self.ent2classes_transitive[ent]))

    def calculate_avg_multityping(self):
        """
        Calculates the average value of the multityping in the KG.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            float: The average value of multityping.
        """
        specific_cl_instanciations = len(list(itertools.chain(*self.ent2classes_specific.values())))
        return specific_cl_instanciations / len(self.is_typed)

    def find_compatible_classes(self, class_list):
        """
        Finds the classes that are compatible with the given class list.

        Args:
            self (object): The instance of the InstanceGenerator.
            class_list (list): A list of classes.

        Returns:
            set: A set of compatible classes.
        """
        disjoint_cls = set()

        for c in class_list:
            disjoint_cls.update(self.class2disjoints_extended.get(c, []))

        # Find all the classes that are not disjoint with any of the specific classes
        compatible_classes = [
            c
            for c in self.class2disjoints_extended.keys()
            if all(c not in self.class2disjoints_extended.get(specific_c, []) for specific_c in class_list)
        ]

        return set(compatible_classes) - set(class_list) | set(self.non_disjoint_classes)

    def pipeline(self):
        """
        Pipeline for processing entities and subsequently generating triples.

        Args:
            self (object): The instance of the InstanceGenerator.

        Return:
            None
        """

        if self.fast_gen:
            self.entities = [f"E{i}" for i in range(1, int(self.num_entities / self.fast_ratio) + 1)]
        else:
            self.entities = [f"E{i}" for i in range(1, self.num_entities + 1)]

        entities = copy.deepcopy(self.entities)
        np.random.shuffle(entities)

        threshold = int(len(self.entities) * (1 - self.prop_untyped_entities))
        self.is_typed = set(entities[:threshold])

        self.layer2classes = {int(k): v for k, v in self.class_info["layer2classes"].items()}
        self.class2layer = self.class_info["class2layer"]
        self.class2disjoints_extended = self.class_info["class2disjoints_extended"]
        self.classes = self.class_info["classes"]
        self.non_disjoint_classes = set(self.classes) - set(self.class2disjoints_extended.keys())

        self.assign_most_specific()

        if self.multityping:
            self.complete_typing()

        self.extend_superclasses()
        self.check_multityping()

        if self.fast_gen:
            ent2classes_spec_values = list(self.ent2classes_specific.values())
            ent2classes_trans_values = list(self.ent2classes_transitive.values())
            last_ent = len(self.entities)

            for _ in range(1, self.fast_ratio):
                entity_batch = [
                    f"E{i}" for i in range(last_ent + 1, last_ent + int(self.num_entities / self.fast_ratio) + 1)
                ]
                np.random.shuffle(entity_batch)
                threshold = int(len(entity_batch) * (1 - self.prop_untyped_entities))
                typed_entities = entity_batch[:threshold]
                self.is_typed.update(typed_entities)
                ent2classes_specific = {e: ent2classes_spec_values[idx] for idx, e in enumerate(typed_entities)}
                ent2classes_transitive = {e: ent2classes_trans_values[idx] for idx, e in enumerate(typed_entities)}
                self.ent2classes_specific.update(ent2classes_specific)
                self.ent2classes_transitive.update(ent2classes_transitive)
                self.entities += entity_batch
                last_ent = len(self.entities)

        self.generate_triples()

    def distribute_relations(self):
        """
        Distributes relations based on the number of triples and the relation balance ratio.

        Args:
            self (object): The instance of the InstanceGenerator.

        Return:
            None

        """
        self.num_relations = len(self.relation_info["relations"])

        if self.num_triples < self.num_relations:
            self.triples_per_rel = {f"R{i}": 1 if i < self.num_triples else 0 for i in range(self.num_relations)}
        else:
            mean = int(self.num_triples / len(self.relation_info["relations"]))
            spread_coeff = (1 - self.relation_balance_ratio) * mean
            self.relation_weights = generate_random_numbers(mean, spread_coeff, self.num_relations)
            self.triples_per_rel = {
                r: np.ceil(tpr)
                for r, tpr in zip(self.relation_info["relations"], np.array(self.relation_weights) * self.num_triples)
            }

    def generate_triples(self):
        """
        Generates triples for the KG.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        self.class2entities = {}

        for e, classes in self.ent2classes_transitive.items():
            for c in classes:
                self.class2entities.setdefault(c, []).append(e)

        self.class2unseen = copy.deepcopy(self.class2entities)
        self.flattened_unseen = list(set((itertools.chain(*self.class2entities.values()))))

        self.untyped_entities_priority = set(self.entities) - set(self.is_typed)
        self.untyped_entities = list(copy.deepcopy(self.untyped_entities_priority))

        self.rel2dom = self.relation_info["rel2dom"]
        self.rel2range = self.relation_info["rel2range"]
        self.rel2patterns = self.relation_info["rel2patterns"]

        self.kg = set()

        self.distribute_relations()

        self.last_oversample = 0

        attempt = 0
        while len(self.kg) < self.num_triples:
            rnd_r = np.random.choice(self.relation_info["relations"], p=self.relation_weights)
            new_triple = self.generate_one_triple(rnd_r)
            attempt += 1
            is_consistent = self.check_consistency(new_triple) if None not in new_triple else False
            if is_consistent:
                self.kg.add(new_triple)
                attempt = 0
            if attempt > 10:
                break

    def generate_one_triple(self, r):
        """
        Generates a single triple based on the given relation.

        Args:
            self (object): The instance of the InstanceGenerator.
            r (str): The relation for which to generate the triple.

        Returns:
            tuple: A tuple containing the head entity, relation, and tail entity of the generated triple.
        """
        r2dom = self.rel2dom.get(r)
        r2range = self.rel2range.get(r)
        token_dom, token_range = False, False

        if r2dom:
            h_list = self.class2unseen.get(r2dom, [])
            h = np.random.choice(h_list) if h_list else None
            if h is not None and self.check_class_disjointness(h, r2dom):
                token_dom = True
            else:
                is_valid = False
                attempt = 0
                while not is_valid and attempt < 10:
                    attempt += 1
                    class2entities = self.class2entities.get(r2dom, [])
                    if class2entities:
                        h = np.random.choice(class2entities)
                        is_valid = self.check_class_disjointness(h, r2dom)
                    else:
                        h = None
                if not is_valid:
                    h = None

        else:
            if len(self.untyped_entities) > 0:
                h = (
                    self.untyped_entities_priority.pop()
                    if self.untyped_entities_priority
                    else np.random.choice(self.untyped_entities)
                )
            else:
                h = np.random.choice(self.flattened_unseen)

        if r2range:
            t_list = self.class2unseen.get(r2range, [])
            t = np.random.choice(t_list) if t_list else None
            if t is not None and self.check_class_disjointness(t, r2range) and h is not None:
                self.class2unseen[r2range].remove(t)
                if token_dom and h in self.class2unseen[r2dom]:
                    self.class2unseen[r2dom].remove(h)
            else:
                is_valid = False
                attempt = 0
                while not is_valid and attempt < 10:
                    attempt += 1
                    class2entities = self.class2entities.get(r2range, [])
                    if class2entities:
                        t = np.random.choice(class2entities)
                        is_valid = self.check_class_disjointness(t, r2range)
                    else:
                        t = None
                if not is_valid:
                    t = None

        else:
            if len(self.untyped_entities) > 0:
                t = (
                    self.untyped_entities_priority.pop()
                    if self.untyped_entities_priority
                    else np.random.choice(self.untyped_entities)
                )
            else:
                t = np.random.choice(self.flattened_unseen)

        return (h, r, t)

    def check_consistency(self, triple):
        """
        Checks the consistency of a triple before adding it to the KG.

        Args:
            self (object): The instance of the InstanceGenerator.
            triple (tuple): A tuple representing a candidate triple (h, r, t).

        Returns:
            bool: True if the triple is consistent, False otherwise.
        """
        h, r, t = triple[0], triple[1], triple[2]

        if not h or not t:
            return False

        if r in self.relation_info["irreflexive_relations"] and h == t:
            return False

        if r in self.relation_info["asymmetric_relations"]:
            if h == t or (t, r, h) in self.kg:
                return False

        if r in self.relation_info["functional_relations"]:
            selected_triples = [triple for triple in self.kg if triple[:2] == (h, r)]
            if selected_triples:
                return False

        if r in self.relation_info["inversefunctional_relations"]:
            selected_triples = [triple for triple in self.kg if triple[1:] == (r, t)]
            if selected_triples:
                return False

        return True

    def check_inverseof_asymmetry(self):
        """
        Checks if the inverse-of-asymmetry condition holds for the given relations.

        This method checks if the inverse-of-asymmetry condition holds for each pair of relations (R1, R2)
        in the rel2inverse dictionary. The inverse-of-asymmetry condition states that if R1 is the inverse of R2,
        and either R1 or R2 is asymmetric, then the same (h, t) pair cannot be observed with both R1 and R2.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        rel2inverse = self.generate_rel2inverse()

        for r1, r2 in rel2inverse.items():
            if r1 in self.relation_info["asymmetric_relations"] or r2 in self.relation_info["asymmetric_relations"]:
                subset_kg = list(filter(lambda triple: triple[1] in (r1, r2), self.kg))

                if len(set(subset_kg)) < len(subset_kg):
                    counter = Counter(subset_kg)
                    duplicates = [h_t for h_t, count in counter.items() if count > 1]

                    for duplicate in duplicates:
                        self.kg.discard((duplicate[0], r1, duplicates[1]))

    def check_dom_range(self):
        """
        Checks the domain and range of triples in the KG and removes inconsistent triples.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        to_remove = set()

        for triple in self.kg:
            h, r, t = triple[0], triple[1], triple[2]
            r2dom, r2range = self.rel2dom.get(r), self.rel2range.get(r)
            if r2dom and h in self.ent2classes_transitive:
                is_valid = self.check_class_disjointness(h, r2dom)
                if not is_valid:
                    to_remove.add(triple)
            if r2range and t in self.ent2classes_transitive:
                is_valid = self.check_class_disjointness(t, r2range)
                if not is_valid:
                    to_remove.add(triple)

        self.kg -= to_remove

    def generate_rel2inverse(self):
        """
        Generates a dictionary containing pairs of inverse relations.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            rel2inverse (dict): A dictionary containing the inverse of the relation.
        """
        rel2inverse = self.relation_info["rel2inverse"]
        keys = list(rel2inverse.keys())
        values = list(rel2inverse.values())
        is_matching = all(keys[i] == values[i + 1] for i in range(0, len(keys), 2) if i + 1 < len(values))
        assert is_matching
        # because of the symmetric key-value pairs:
        rel2inverse = {k: v for i, (k, v) in enumerate(self.relation_info["rel2inverse"].items()) if i % 2 == 0}

        return rel2inverse

    def check_asymmetries(self):
        """
        Checks for asymmetries in the KG.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """

        for r in self.relation_info["asymmetric_relations"]:
            subset_kg = list(filter(lambda triple: triple[1] == r, self.kg))
            symmetric_dict = {}

            for triple in subset_kg:
                symmetric_triple = (triple[2], triple[1], triple[0])
                if symmetric_triple in subset_kg and symmetric_triple not in symmetric_dict.keys():
                    symmetric_dict[triple] = symmetric_triple

            to_remove = set(symmetric_dict.values())
            self.kg -= to_remove

    def check_class_disjointness(self, ent, expected_class):
        """
        Checks for class disjointness (owl:disjointWith) between domain/range of a relation
        and the classes to which belong the randomly sampled entity.

        Args:
            self (object): The instance of the InstanceGenerator.
            ent (str): The entity to check.
            expected_class (str): The expected class as domain or range of a relation.

        Returns:
            bool: True if the entity classes and expected class are disjoint, False otherwise.
        """
        classes_entity_side = self.ent2classes_transitive[ent]
        classes_relation_side = self.class_info["transitive_class2superclasses"][expected_class]

        for c in classes_relation_side:
            disj = self.class2disjoints_extended.get(c, [])
            if set(disj).intersection(set(classes_entity_side)):
                return False

        return True

    def oversample_triples_inference(self):
        """
        Infers new triples to be added to the KG based on logical deductions.
        Allows reaching user-specified number of triples without increasing the number of entities.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        used_relations = set()
        id2pattern = {
            1: self.relation_info["inverseof_relations"],
            2: self.relation_info["symmetric_relations"],
            3: self.relation_info["subrelations"],
        }
        attempt = 0

        while len(self.kg) < self.num_triples:
            attempt += 1
            chosen_id = np.random.randint(1, len(id2pattern) + 1)
            pattern2rels = id2pattern[chosen_id]
            np.random.shuffle(pattern2rels)

            if pattern2rels:
                rel = pattern2rels[0]

                if rel not in used_relations:
                    attempt = 0
                    used_relations.add(rel)
                    subset_kg = set([triple for triple in self.kg if triple[1] == rel])

                    if chosen_id == 1:
                        inv_rel = self.relation_info["rel2inverse"][rel]
                        inferred_triples = inverse_inference(subset_kg, inv_rel)

                    elif chosen_id == 2:
                        inferred_triples = symmetric_inference(subset_kg)

                    elif chosen_id == 3:
                        super_rel = self.relation_info["rel2superrel"][rel]
                        inferred_triples = subproperty_inference(subset_kg, super_rel)

                    # inferred_triples = inferred_triples[: 0.5 * int(len(inferred_triples))]
                    self.kg = self.kg | set(inferred_triples)

                    if len(self.kg) >= self.num_triples:
                        return

            if attempt > 1000:
                break

    def procedure_1(self):
        """
        Checks that domains and ranges are compatible with ent2classes_transitive of instantiated triples.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """

        for rel in self.rel2dom:
            if self.rel2dom[rel] in self.class2disjoints_extended:
                subset_kg = set([triple for triple in self.kg if triple[1] == rel])
                disjoint_with_dom = self.class2disjoints_extended[self.rel2dom[rel]]
                wrong_heads = set()
                for h, _, _ in subset_kg:
                    if h in self.ent2classes_transitive:
                        intersection = set(self.ent2classes_transitive[h]).intersection(disjoint_with_dom)
                        if intersection:
                            wrong_heads.add(h)

                problematic_triples = {
                    (head, relation, tail) for head, relation, tail in subset_kg if head in wrong_heads
                }
                self.kg -= problematic_triples

        for rel in self.rel2range:
            if self.rel2range[rel] in self.class2disjoints_extended:
                subset_kg = set([triple for triple in self.kg if triple[1] == rel])
                disjoint_with_range = self.class2disjoints_extended[self.rel2range[rel]]
                wrong_tails = set()
                for _, _, t in subset_kg:
                    if t in self.ent2classes_transitive:
                        intersection = set(self.ent2classes_transitive[t]).intersection(disjoint_with_range)
                        if intersection:
                            wrong_tails.add(t)

                problematic_triples = {
                    (head, relation, tail) for head, relation, tail in subset_kg if tail in wrong_tails
                }
                self.kg -= problematic_triples

    def procedure_2(self):
        """
        Checks if the inverse relationship between r1 and r2 satisfies certain conditions.

        Args:
            self (object): The instance of the InstanceGenerator.

        Returns:
            None
        """
        rel2inverse = self.generate_rel2inverse()
        for r1 in rel2inverse:
            r2 = rel2inverse[r1]
            subset_kg = set([triple for triple in self.kg if triple[1] == r1])
            if r2 in self.rel2range and self.rel2range[r2] in self.class2disjoints_extended:
                range_r2 = self.rel2range[r2]
                disjoint_with = self.class2disjoints_extended[range_r2]
                wrong_heads = set()
                for h, _, _ in subset_kg:
                    if h in self.ent2classes_transitive:
                        intersection = set(self.ent2classes_transitive[h]).intersection(disjoint_with)
                        if intersection:
                            wrong_heads.add(h)

                problematic_triples = {
                    (head, relation, tail) for head, relation, tail in subset_kg if tail in wrong_heads
                }
                self.kg -= problematic_triples

            if r2 in self.rel2dom and self.rel2dom[r2] in self.class2disjoints_extended:
                dom_r2 = self.rel2dom[r2]
                disjoint_with = self.class2disjoints_extended[dom_r2]
                wrong_tails = set()
                for _, _, t in subset_kg:
                    if t in self.ent2classes_transitive:
                        intersection = set(self.ent2classes_transitive[t]).intersection(disjoint_with)
                        if intersection:
                            wrong_tails.add(t)

                problematic_triples = {
                    (head, relation, tail) for head, relation, tail in subset_kg if tail in wrong_tails
                }
                self.kg -= problematic_triples

pygraft/kg_generator.py
nicolas-hbt/pygraft
import argparse
import os
from pygraft import create_json_template, create_yaml_template, generate_schema, generate_kg, generate


def parse_arguments():
    parser = argparse.ArgumentParser(description="================== Schema & KG Generator ==================")
    parser.add_argument(
        "-t",
        "--template",
        action="store_true",
        default=None,
        help="Create a config template in the current working directory.",
    )
    parser.add_argument(
        "-e", "--extension", type=str, default=None, help="Config file extension. Options: json | yaml | yml"
    )
    parser.add_argument(
        "-jt",
        "--json_template",
        action="store_true",
        default=None,
        help="Create a json config template in the current working directory.",
    )
    parser.add_argument(
        "-yt",
        "--yaml_template",
        action="store_true",
        default=None,
        help="Create a yml config template in the current working directory.",
    )
    parser.add_argument("-conf", "--config", type=str, default=None, help="Load a given config file.")
    parser.add_argument(
        "-g",
        "--gen",
        type=str,
        default=None,
        help="Which function to call. Options: generate_schema | generate_kg | generate",
    )

    args = parser.parse_args()
    return args


def main():
    args = parse_arguments()

    if args.template:
        if args.extension == "json":
            create_json_template()
        else:
            create_yaml_template()

    if args.json_template:
        create_json_template()

    if args.yaml_template:
        create_yaml_template()

    if args.config:
        _, file_extension = os.path.splitext(args.config)

        if file_extension.lower() not in [".yml", ".yaml", ".json"]:
            raise ValueError(
                "Please, specify the --extension parameter with json, yaml, or yml. Alternatively, append the file format of your file for the --config parameter."
            )

    if args.gen == "generate_schema":
        generate_schema(args.config)
    if args.gen == "generate_kg":
        generate_kg(args.config)
    if args.gen == "generate":
        generate(args.config)

pygraft/main.py
nicolas-hbt/pygraft
from .schema_constructor import SchemaBuilder
from .utils import (
    get_most_recent_subfolder,
    load_config,
    initialize_folder,
    check_schema_arguments,
    check_kg_arguments,
    print_ascii_header,
    load_json_template,
    load_yaml_template,
)
from .class_generator import ClassGenerator
from .relation_generator import RelationGenerator
from .kg_generator import InstanceGenerator


def create_template(extension="yml"):
    """
    Creates a template file for the user to fill in.

    Args:
        extension (str, optional): File extension of the template file. Defaults to "yml".

    Raises:
        ValueError: If the extension is not one of the following: json, yaml, yml

    Returns:
        None
    """
    if extension == "json":
        load_json_template()
    elif extension in {"yaml", "yml"}:
        load_yaml_template()
    else:
        raise ValueError(
            f"Unknown extension file format: {extension}. Please enter one of the following: json, yaml, yml"
        )


def create_json_template():
    """
    Creates a json template file for the user to fill in.

    Args:
        None
    
    Returns:
        None
    """
    load_json_template()


def create_yaml_template():
    """
    Creates a yaml template file for the user to fill in.

    Args:
        None

    Returns:
        None
    """
    load_yaml_template()


def generate_schema(path):
    """
    Generates a schema based on the user's configuration file.
    
    Args:
        path (str): Path to the user's configuration file.
        
    Returns:
        None
    """
    config = load_config(path)
    check_schema_arguments(config)
    config["schema_name"] = initialize_folder(config["schema_name"])

    print_ascii_header()

    class_generator = ClassGenerator(
        num_classes=config["num_classes"],
        max_hierarchy_depth=config["max_hierarchy_depth"],
        avg_class_depth=config["avg_class_depth"],
        class_inheritance_ratio=config["class_inheritance_ratio"],
        avg_disjointness=config["avg_disjointness"],
        verbose=config["verbose"],
    )
    class_info = class_generator.generate_class_schema()

    relation_generator = RelationGenerator(
        class_info=class_info,
        num_relations=config["num_relations"],
        relation_specificity=config["relation_specificity"],
        prop_profiled_relations=config["prop_profiled_relations"],
        profile_side=config["profile_side"],
        prop_symmetric_relations=config["prop_symmetric_relations"],
        prop_inverse_relations=config["prop_inverse_relations"],
        prop_functional_relations=config["prop_functional_relations"],
        prop_transitive_relations=config["prop_transitive_relations"],
        prop_subproperties=config["prop_subproperties"],
        prop_reflexive_relations=config["prop_reflexive_relations"],
        prop_irreflexive_relations=config["prop_irreflexive_relations"],
        prop_asymmetric_relations=config["prop_asymmetric_relations"],
        prop_inverse_functional_relations=config["prop_inverse_functional_relations"],
        verbose=config["verbose"],
    )
    relation_info = relation_generator.generate_relation_schema()

    schema_builder = SchemaBuilder(class_info, relation_info, config["schema_name"], config["format"])
    schema_builder.building_pipeline()


def generate_kg(path):
    """
    Generates a knowledge graph based on the user's configuration file.

    Args:
        path (str): Path to the user's configuration file.

    Returns:
        None
    """
    config = load_config(path)
    check_kg_arguments(config)
    if config["schema_name"] is None:
        most_recent_subfolder_name = get_most_recent_subfolder("output")
        config["schema_name"] = most_recent_subfolder_name

    print_ascii_header()

    instance_generator = InstanceGenerator(
        schema=config["schema_name"],
        num_entities=config["num_entities"],
        num_triples=config["num_triples"],
        relation_balance_ratio=config["relation_balance_ratio"],
        fast_gen=config["fast_gen"],
        oversample=config["oversample"],
        prop_untyped_entities=config["prop_untyped_entities"],
        avg_depth_specific_class=config["avg_depth_specific_class"],
        multityping=config["multityping"],
        avg_multityping=config["avg_multityping"],
        format=config["format"],
    )
    instance_generator.generate_kg()


def generate(path):
    """
    Generates a schema and knowledge graph based on the user's configuration file.

    Args:
        path (str): Path to the user's configuration file.
    
    Returns:
        None
    """
    config = load_config(path)
    check_schema_arguments(config)
    check_kg_arguments(config)
    config["schema_name"] = initialize_folder(config["schema_name"])

    print_ascii_header()

    class_generator = ClassGenerator(
        num_classes=config["num_classes"],
        max_hierarchy_depth=config["max_hierarchy_depth"],
        avg_class_depth=config["avg_class_depth"],
        class_inheritance_ratio=config["class_inheritance_ratio"],
        avg_disjointness=config["avg_disjointness"],
        verbose=config["verbose"],
    )
    class_info = class_generator.generate_class_schema()

    relation_generator = RelationGenerator(
        class_info=class_info,
        num_relations=config["num_relations"],
        relation_specificity=config["relation_specificity"],
        prop_profiled_relations=config["prop_profiled_relations"],
        profile_side=config["profile_side"],
        prop_symmetric_relations=config["prop_symmetric_relations"],
        prop_inverse_relations=config["prop_inverse_relations"],
        prop_functional_relations=config["prop_functional_relations"],
        prop_transitive_relations=config["prop_transitive_relations"],
        prop_subproperties=config["prop_subproperties"],
        prop_reflexive_relations=config["prop_reflexive_relations"],
        prop_irreflexive_relations=config["prop_irreflexive_relations"],
        prop_asymmetric_relations=config["prop_asymmetric_relations"],
        prop_inverse_functional_relations=config["prop_inverse_functional_relations"],
        verbose=config["verbose"],
    )
    relation_info = relation_generator.generate_relation_schema()

    schema_builder = SchemaBuilder(class_info, relation_info, config["schema_name"], config["format"])
    schema_builder.building_pipeline()

    instance_generator = InstanceGenerator(
        schema=config["schema_name"],
        num_entities=config["num_entities"],
        num_triples=config["num_triples"],
        relation_balance_ratio=config["relation_balance_ratio"],
        fast_gen=config["fast_gen"],
        oversample=config["oversample"],
        prop_untyped_entities=config["prop_untyped_entities"],
        avg_depth_specific_class=config["avg_depth_specific_class"],
        multityping=config["multityping"],
        avg_multityping=config["avg_multityping"],
        format=config["format"],
    )
    instance_generator.generate_kg()

pygraft/pygraft.py
nicolas-hbt/pygraft
import numpy as np
import random
from collections import defaultdict
import json
import itertools
import pkg_resources
from tabulate import tabulate


class RelationGenerator:
    def __init__(self, **kwargs):
        """
        Initializes the RelationGenerator.

        Args:
            self (object): The instance of the RelationGenerator.
            **kwargs: Additional keyword arguments.

        Keyword Args:
            class_info (dict): A dictionary containing information about the classes.
            num_relations (int): The number of relations to generate.
            relation_specificity (float): The desired specificity of the relations.
            prop_profiled_relations (float): The desired proportion of profiled relations.
            profile_side (str): The side to profile relations on.
            verbose (bool): Whether to print the relation schema.
            prop_symmetric_relations (float): The desired proportion of symmetric relations.
            prop_inverse_relations (float): The desired proportion of inverse relations.
            prop_functional_relations (float): The desired proportion of functional relations.
            prop_inverse_functional_relations (float): The desired proportion of inverse functional relations.
            prop_transitive_relations (float): The desired proportion of transitive relations.
            prop_subproperties (float): The desired proportion of subproperties.
            prop_reflexive_relations (float): The desired proportion of reflexive relations.
            prop_irreflexive_relations (float): The desired proportion of irreflexive relations.
            prop_asymmetric_relations (float): The desired proportion of asymmetric relations.

        Returns:
            None
        """
        self.init_params(**kwargs)
        self.get_one_rel_compatibilities()
        self.get_inverseof_compatibilities()

    def init_params(self, **kwargs):
        """
        Initializes general relation information with user-specified parameters.

        Args:
            self (object): The instance of the RelationGenerator.
            **kwargs: Additional keyword arguments.

        Keyword Args:
            class_info (dict): A dictionary containing information about the classes.
            num_relations (int): The number of relations to generate.
            relation_specificity (float): The desired specificity of the relations.
            prop_profiled_relations (float): The desired proportion of profiled relations.
            profile_side (str): The side to profile relations on.
            verbose (bool): Whether to print the relation schema.
            prop_symmetric_relations (float): The desired proportion of symmetric relations.
            prop_inverse_relations (float): The desired proportion of inverse relations.
            prop_functional_relations (float): The desired proportion of functional relations.
            prop_inverse_functional_relations (float): The desired proportion of inverse functional relations.
            prop_transitive_relations (float): The desired proportion of transitive relations.
            prop_subproperties (float): The desired proportion of subproperties.
            prop_reflexive_relations (float): The desired proportion of reflexive relations.
            prop_irreflexive_relations (float): The desired proportion of irreflexive relations.
            prop_asymmetric_relations (float): The desired proportion of asymmetric relations.

        Returns:
            None
        """
        self.class_info = kwargs.get("class_info")
        self.num_relations = kwargs.get("num_relations")
        self.relation_specificity = kwargs.get("relation_specificity")
        self.prop_profiled_relations = kwargs.get("prop_profiled_relations")
        self.profile_side = kwargs.get("profile_side")
        self.verbose = kwargs.get("verbose")
        self.init_property_props(**kwargs)

    def init_property_props(self, **kwargs):
        """
        Initializes proportions of relation properties with user-specified parameters.

        Args:
            self (object): The instance of the RelationGenerator.
            **kwargs: Additional keyword arguments.

        Keyword Args:
            class_info (dict): A dictionary containing information about the classes.
            num_relations (int): The number of relations to generate.
            relation_specificity (float): The desired specificity of the relations.
            prop_profiled_relations (float): The desired proportion of profiled relations.
            profile_side (str): The side to profile relations on.
            verbose (bool): Whether to print the relation schema.
            prop_symmetric_relations (float): The desired proportion of symmetric relations.
            prop_inverse_relations (float): The desired proportion of inverse relations.
            prop_functional_relations (float): The desired proportion of functional relations.
            prop_inverse_functional_relations (float): The desired proportion of inverse functional relations.
            prop_transitive_relations (float): The desired proportion of transitive relations.
            prop_subproperties (float): The desired proportion of subproperties.
            prop_reflexive_relations (float): The desired proportion of reflexive relations.
            prop_irreflexive_relations (float): The desired proportion of irreflexive relations.
            prop_asymmetric_relations (float): The desired proportion of asymmetric relations. 

        Returns:
            None
        """
        self.prop_symmetric_relations = kwargs.get("prop_symmetric_relations")
        self.prop_inverse_relations = kwargs.get("prop_inverse_relations")
        self.prop_functional_relations = kwargs.get("prop_functional_relations")
        self.prop_inverse_functional_relations = kwargs.get("prop_inverse_functional_relations")
        self.prop_transitive_relations = kwargs.get("prop_transitive_relations")
        self.prop_subproperties = kwargs.get("prop_subproperties")
        self.prop_reflexive_relations = kwargs.get("prop_reflexive_relations")
        self.prop_irreflexive_relations = kwargs.get("prop_irreflexive_relations")
        self.prop_asymmetric_relations = kwargs.get("prop_asymmetric_relations")

    def generate_relation_schema(self):
        """
        Generates the relation schema.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            relation_info (dict): The assembled relation information.
        """
        self.generate_relations()

        if self.verbose:
            self.print_schema()

        relation_info = self.assemble_relation_info()

        return relation_info

    def assemble_relation_info(self):
        """
        Assembles and returns information about the relations.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            relation_info (dict): A dictionary containing statistics about the relations.
        """
        num_relations = len(self.relations)
        prop_reflexive = round(len(self.reflexive_relations) / num_relations, 2)
        prop_irreflexive = round(len(self.irreflexive_relations) / num_relations, 2)
        prop_functional = round(len(self.functional_relations) / num_relations, 2)
        prop_inversefunctional = round(len(self.inversefunctional_relations) / num_relations, 2)
        prop_symmetric = round(len(self.symmetric_relations) / num_relations, 2)
        prop_asymmetric = round(len(self.asymmetric_relations) / num_relations, 2)
        prop_transitive = round(len(self.transitive_relations) / num_relations, 2)
        prop_inverseof = round(len(self.inverseof_relations) / num_relations, 2)
        prop_subpropertyof = round(2 * len(self.prop2superprop) / num_relations, 2)
        prop_profiled_relations = round(self.current_profile_ratio, 2)
        relation_specificity = round(self.calculate_relation_specificity(), 2)

        relation_info = {
            "statistics": {
                "num_relations": num_relations,
                "prop_reflexive": prop_reflexive,
                "prop_irreflexive": prop_irreflexive,
                "prop_functional": prop_functional,
                "prop_inversefunctional": prop_inversefunctional,
                "prop_symmetric": prop_symmetric,
                "prop_asymmetric": prop_asymmetric,
                "prop_transitive": prop_transitive,
                "prop_inverseof": prop_inverseof,
                "prop_subpropertyof": prop_subpropertyof,
                "prop_profiled_relations": prop_profiled_relations,
                "relation_specificity": relation_specificity,
            },
            "relations": self.relations,
            "rel2patterns": self.rel2patterns,
            "pattern2rels": self.pattern2rels,
            "reflexive_relations": self.reflexive_relations,
            "irreflexive_relations": self.irreflexive_relations,
            "symmetric_relations": self.symmetric_relations,
            "asymmetric_relations": self.asymmetric_relations,
            "functional_relations": self.functional_relations,
            "inversefunctional_relations": self.inversefunctional_relations,
            "transitive_relations": self.transitive_relations,
            "inverseof_relations": self.inverseof_relations,
            "subrelations": self.subproperties,
            "rel2inverse": self.rel2inverse,
            "rel2dom": self.rel2dom,
            "rel2range": self.rel2range,
            "rel2superrel": self.prop2superprop,
        }

        return relation_info

    def generate_relations(self):
        """
        Generates relations and adds various properties to them.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        self.relations = [f"R{i}" for i in range(1, self.num_relations + 1)]
        self.rel2patterns = {
            r: set() for r in self.relations
        }  # contains current ObjectProperties for each generated relation
        self.pattern2rels = {tuple(sorted(p)): set() for p in self.one_rel_compatibilities}

        self.reflexive_relations = random.sample(
            self.relations, k=int(len(self.relations) * self.prop_reflexive_relations)
        )
        self.update_rel2patterns("owl:Reflexive")
        self.irreflexive_relations = random.sample(
            list(set(self.relations) - set(self.reflexive_relations)),
            k=int(len(self.relations) * self.prop_irreflexive_relations),
        )
        self.update_rel2patterns("owl:Irreflexive")

        self.add_property("owl:Symmetric")
        self.add_property("owl:Asymmetric")
        self.add_property("owl:Transitive")
        self.add_property("owl:Functional")
        self.add_property("owl:InverseFunctional")

        self.inverseof_relations = []
        self.rel2inverse = {}
        self.add_inverseof()

        self.class2disjoints_extended = self.class_info["class2disjoints_extended"]
        self.layer2classes = self.class_info["layer2classes"]
        self.class2layer = self.class_info["class2layer"]
        self.unprofiled_relations = {}

        # Reflexivity is incompatible with domain and range assertions
        self.unprofiled_relations["both"] = [r for r, p in self.rel2patterns.items() if "owl:Reflexive" not in p]
        self.unprofiled_relations["dom"] = [r for r, p in self.rel2patterns.items() if "owl:Reflexive" not in p]
        self.unprofiled_relations["range"] = [r for r, p in self.rel2patterns.items() if "owl:Reflexive" not in p]
        self.num_relations_wo_reflexive = len(self.unprofiled_relations["both"])

        self.rel2dom = {}
        self.rel2range = {}
        self.current_profile_ratio = 0.0

        while self.current_profile_ratio < self.prop_profiled_relations:
            self.add_one_relation_profile()
            self.current_profile_ratio = (len(self.rel2dom) + len(self.rel2range)) / (
                2 * self.num_relations_wo_reflexive
            )

        self.add_property("rdfs:subPropertyOf")

    def calculate_profile_ratio(self):
        """
        Calculates the profile ratio.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            float: The calculated profile ratio.
        """
        return (len(self.rel2dom) + len(self.rel2range)) / (2 * len(self.relations))

    def add_one_relation_profile(self):
        """
        Adds one relation profile based on the value of `profile_side`.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        if self.profile_side == "both":
            self.add_complete_relation_profile()
        elif self.profile_side == "partial":
            self.add_partial_relation_profile()

    def add_partial_relation_profile(self):
        """
        Generates a partial relation profile by assigning a domain/range to a relation.
        This function selects a relation (either from the "domain" or "range" category)
        that has not been profiled yet and assigns a randomly sampled class to it. The
        function also updates the relation-specificity based on the chosen class. If the
        selected relation is transitive or symmetric, the function also assigns the
        sampled class to the corresponding "range" or "domain" relation. If the selected
        relation has an inverse relation, the function assigns the inverse relation the
        same class as the selected relation, as long as it is not reflexive.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        current_rel_specificity = self.calculate_relation_specificity()
        sampled_class = self.sample_class(current_rel_specificity)
        domain_or_range = "domain" if random.random() < 0.5 else "range"

        if domain_or_range == "domain":
            rel = self.unprofiled_relations["dom"].pop(0)
            rel2patterns = self.rel2patterns[rel]
            self.rel2dom[rel] = sampled_class

            if "owl:Transitive" in rel2patterns or "owl:Symmetric" in rel2patterns:
                self.rel2range[rel] = sampled_class
                if rel in self.unprofiled_relations["range"]:
                    self.unprofiled_relations["range"].remove(rel)
                    if rel in self.rel2inverse:
                        inverse_rel = self.rel2inverse[rel]
                        self.unprofiled_relations["dom"].remove(
                            inverse_rel
                        ) if inverse_rel in self.unprofiled_relations["dom"] else None
                        self.rel2dom[inverse_rel] = self.rel2range[rel]

            if rel in self.rel2inverse:
                inverse_rel = self.rel2inverse[rel]
                # if inverse_rel in self.unprofiled_relations["range"] and "owl:Reflexive" not in self.rel2patterns[inverse_rel]:
                self.unprofiled_relations["range"].remove(inverse_rel) if inverse_rel in self.unprofiled_relations[
                    "range"
                ] else None
                self.rel2range[inverse_rel] = self.rel2dom[rel]

        else:
            rel = self.unprofiled_relations["range"].pop(0)
            rel2patterns = self.rel2patterns[rel]
            self.rel2range[rel] = sampled_class

            if "owl:Transitive" in rel2patterns or "owl:Symmetric" in rel2patterns:
                self.rel2dom[rel] = sampled_class
                if rel in self.unprofiled_relations["dom"]:
                    self.unprofiled_relations["dom"].remove(rel)
                    if rel in self.rel2inverse:
                        inverse_rel = self.rel2inverse[rel]
                        self.unprofiled_relations["range"].remove(
                            inverse_rel
                        ) if inverse_rel in self.unprofiled_relations["range"] else None
                        self.rel2range[inverse_rel] = self.rel2dom[rel]

            if rel in self.rel2inverse:
                inverse_rel = self.rel2inverse[rel]
                # if inverse_rel in self.unprofiled_relations["dom"] and "owl:Reflexive" not in self.rel2patterns[inverse_rel]:
                self.unprofiled_relations["dom"].remove(inverse_rel) if inverse_rel in self.unprofiled_relations[
                    "dom"
                ] else None
                self.rel2dom[inverse_rel] = self.rel2range[rel]

    def add_complete_relation_profile(self):
        """
        Generates a complete relation profile by assigning a domain and a range to a relation.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        current_rel_specificity = self.calculate_relation_specificity()
        sampled_domain = self.sample_class(current_rel_specificity)
        rel = self.unprofiled_relations["both"].pop(0)
        rel2patterns = self.rel2patterns[rel]

        if "owl:Reflexive" not in rel2patterns:
            self.rel2dom[rel] = sampled_domain
            current_rel_specificity = self.calculate_relation_specificity()
            if "owl:Transitive" in rel2patterns or "owl:Symmetric" in rel2patterns:
                self.rel2range[rel] = sampled_domain
            else:
                sampled_range = self.sample_class_constrained(current_rel_specificity, sampled_domain)
                self.rel2range[rel] = sampled_range

            if rel in self.rel2inverse:
                inverse_rel = self.rel2inverse[rel]

                if inverse_rel in self.unprofiled_relations["both"]:
                    self.unprofiled_relations["both"].remove(inverse_rel)

                self.rel2dom[inverse_rel] = self.rel2range[rel]
                self.rel2range[inverse_rel] = self.rel2dom[rel]

    def calculate_relation_specificity(self):
        """
        Calculates the specificity of relations.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        domains = list(self.rel2dom.values())
        ranges = list(self.rel2range.values())
        both = domains + ranges

        return np.mean([self.class2layer[c] for c in both])

    def sample_class(self, current_rel_specificity):
        """
        Takes in a current relative specificity value
        and returns a random class from a list of filtered classes
        such that the current relative specificity value converges towards user-specified value.

        Args:
            self (object): The instance of the RelationGenerator.
            current_rel_specificity (float): The current relative specificity value.

        Returns:
            str: A random class from a list of filtered classes.
        """
        potential_classes = self.filter_classes(current_rel_specificity)

        return random.choice(potential_classes)

    def sample_class_constrained(self, current_rel_specificity, other_class):
        """
        Returns a compatible class based on the current relational specificity and other class.

        Args:
            self (object): The instance of the RelationGenerator.
            current_rel_specificity (float): The current relational specificity.
            other_class (str): The other class.

        Returns:
            str: A compatible class.
        """
        compatible_classes = []

        while not compatible_classes:
            potential_classes = self.filter_classes(current_rel_specificity)
            compatible_classes = list(set(potential_classes) - set(self.class2disjoints_extended.get(other_class, [])))

        return random.choice(compatible_classes)

    def filter_classes(self, current_rel_specificity):
        """
        Filters classes based on the current relational specificity.

        Args:
            self (object): The instance of the RelationGenerator.
            current_rel_specificity (float): The current relational specificity.

        Returns:
            list: A list of filtered classes.
        """
        if current_rel_specificity < self.relation_specificity:
            filtered_classes = [
                cl for layer, cl in self.layer2classes.items() if layer > int(self.relation_specificity)
            ]
            if random.random() < 0.1:  # add some noise
                filtered_classes = [
                    cl for layer, cl in self.layer2classes.items() if layer <= int(self.relation_specificity)
                ]
        else:
            filtered_classes = [
                cl for layer, cl in self.layer2classes.items() if layer <= int(self.relation_specificity)
            ]
            if random.random() < 0.1:  # add some noise
                filtered_classes = [
                    cl for layer, cl in self.layer2classes.items() if layer > int(self.relation_specificity)
                ]

        return list(itertools.chain.from_iterable(filtered_classes))

    def get_one_rel_compatibilities(self):
        """
        Gets all valid combinations of relation properties.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        file_path = pkg_resources.resource_filename("pygraft", "property_checks/combinations.json")

        with open(file_path, "r") as file:
            data = json.load(file)

        compatibilities = [key for key, value in data.items() if value == "True"]
        self.one_rel_compatibilities = [op.split(",") for op in compatibilities]

    def get_inverseof_compatibilities(self):
        """
        Gets all valid combinations of inverse relations.

        Args:
            self (object): The instance of the RelationGenerator.
        
        Returns:
            None
        """
        self.compat_inverseof = {}
        file_path = pkg_resources.resource_filename("pygraft", "property_checks/compat_p1p2_inverseof.txt")

        with open(file_path, "r") as file:
            for line in file:
                line = line.strip()
                if "True" in line:
                    line = line.split(" ")
                    line[0] = line[0][:-1]
                    line = line[0].split("|")
                    v1 = frozenset(line[0].strip().split(","))
                    v2 = frozenset(line[1].strip().split(","))
                    if v1 not in self.compat_inverseof:
                        self.compat_inverseof[v1] = [v2]
                    else:
                        self.compat_inverseof[v1].append(v2)

    def update_rel2patterns(self, property):
        """
        Updates self.rel2patterns dictionary based on the given property.

        Args:
            self (object): The instance of the RelationGenerator.
            property (str): The property to add to the self.rel2patterns dictionary.

        Returns:
            None
        """
        property_mappings = {
            "owl:Reflexive": getattr(self, "reflexive_relations", set()),
            "owl:Irreflexive": getattr(self, "irreflexive_relations", set()),
            "owl:Symmetric": getattr(self, "symmetric_relations", set()),
            "owl:Asymmetric": getattr(self, "asymmetric_relations", set()),
            "owl:Transitive": getattr(self, "transitive_relations", set()),
        }

        if property in property_mappings:
            self.rel2patterns = {
                rel: property_set | {property} if rel in property_mappings[property] else property_set
                for rel, property_set in self.rel2patterns.items()
            }

        self.update_pattern2rels()

    def update_pattern2rels(self):
        """
        Updates self.pattern2rels dictionary.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        self.pattern2rels = defaultdict(set)
        _ = [self.pattern2rels[frozenset(value)].add(key) for key, value in self.rel2patterns.items()]

    def add_property(self, property):
        """
        Adds properties to relations.

        Args:
            self (object): The instance of the RelationGenerator.
            property (str): The property to add.

        Returns:
            None
        """
        # get all valid combinations featuring the desired property
        combinations_with_property = list(filter(lambda combi: property in combi, self.one_rel_compatibilities))
        # remove the desired property so that later on we ensure to sample relation s.t. adding a property is legit
        combinations_without_property = [
            list(filter(lambda item: item != property, combi)) for combi in combinations_with_property
        ]
        # get all relations having one of the valid combinations
        relation_pool = set().union(
            *[self.pattern2rels.get(frozenset(combi), set()) for combi in combinations_without_property]
        )

        if property == "owl:Functional":
            self.functional_relations = []
            potential_relations = [key for key, values in self.rel2patterns.items() if values == set()]
            while (
                len(self.functional_relations) < self.prop_functional_relations * self.num_relations
                and potential_relations
            ):
                new_functional_relation = potential_relations.pop()
                self.functional_relations.append(new_functional_relation)
                self.rel2patterns[new_functional_relation] = set(self.rel2patterns[new_functional_relation]) | {
                    "owl:Functional"
                }

        if property == "owl:InverseFunctional":
            self.inversefunctional_relations = []
            X = random.uniform(0.25, 0.75)
            potential_relations = [key for key, values in self.rel2patterns.items() if not values]
            while (
                len(self.inversefunctional_relations) < X * self.prop_inverse_functional_relations * self.num_relations
                and potential_relations
            ):
                new_inversefunctional_relation = potential_relations.pop()
                self.inversefunctional_relations.append(new_inversefunctional_relation)
                self.rel2patterns[new_inversefunctional_relation] = set(
                    self.rel2patterns[new_inversefunctional_relation]
                ) | {"owl:InverseFunctional"}

            potential_relations = [key for key, values in self.rel2patterns.items() if not values - {"owl:Functional"}]
            potential_relations = [r for r in potential_relations if r not in self.inversefunctional_relations]
            while (
                len(self.inversefunctional_relations) < self.prop_inverse_functional_relations * self.num_relations
                and potential_relations
            ):
                new_inversefunctional_relation = potential_relations.pop()
                self.inversefunctional_relations.append(new_inversefunctional_relation)
                self.rel2patterns[new_inversefunctional_relation] = set(
                    self.rel2patterns[new_inversefunctional_relation]
                ) | {"owl:InverseFunctional"}

        if property == "rdfs:subPropertyOf":
            self.prop2superprop = {}
            self.subproperties = []
            rels = self.relations[:]
            for r1 in rels:
                for r2 in rels:
                    if (
                        r1 != r2
                        and r1 not in self.prop2superprop
                        and r2 not in self.prop2superprop
                        and self.rel2inverse.get(r1) != r2
                        and self.rel2inverse.get(r2) != r1
                        and self.rel2patterns[r1] == self.rel2patterns[r2]
                    ):
                        if (
                            r1 not in self.rel2dom
                            and r1 not in self.rel2range
                            and r2 not in self.rel2dom
                            and r1 not in self.rel2range
                        ):
                            self.prop2superprop[r1] = r2
                            self.subproperties.append(r1)
                            break
                        if r1 in self.rel2dom and r1 in self.rel2range and r2 in self.rel2dom and r1 in self.rel2range:
                            # 1
                            if self.rel2dom.get(r1) == self.rel2dom.get(r2) and self.rel2range.get(
                                r1
                            ) == self.rel2range.get(r2):
                                self.prop2superprop[r1] = r2
                                self.subproperties.append(r1)
                                break
                            # 2
                            elif (
                                self.rel2dom.get(r1) == self.rel2dom.get(r2)
                                and self.rel2range.get(r2)
                                in self.class_info["transitive_class2superclasses"][self.rel2range.get(r1)]
                            ):
                                self.prop2superprop[r1] = r2
                                self.subproperties.append(r1)
                                break
                            # 3
                            elif (
                                self.rel2range.get(r1) == self.rel2range.get(r2)
                                and self.rel2dom.get(r2)
                                in self.class_info["transitive_class2superclasses"][self.rel2dom.get(r1)]
                            ):
                                self.prop2superprop[r1] = r2
                                self.subproperties.append(r1)
                                break
                            # 4
                            elif (
                                self.rel2dom.get(r2)
                                in self.class_info["transitive_class2superclasses"][self.rel2dom.get(r1)]
                                and self.rel2range.get(r2)
                                in self.class_info["transitive_class2superclasses"][self.rel2range.get(r1)]
                            ):
                                self.prop2superprop[r1] = r2
                                self.subproperties.append(r1)
                                break

                if 2 * len(self.prop2superprop) >= self.prop_subproperties * self.num_relations:
                    return

        if property == "owl:Symmetric":
            sample_size = int(len(self.relations) * self.prop_symmetric_relations)
            if sample_size > len(relation_pool):
                sample_size = len(relation_pool)
            self.symmetric_relations = random.sample(list(relation_pool), k=sample_size)

        if property == "owl:Asymmetric":
            sample_size = int(len(self.relations) * self.prop_asymmetric_relations)
            if sample_size > len(relation_pool):
                sample_size = len(relation_pool)
            self.asymmetric_relations = random.sample(list(relation_pool), k=sample_size)

        if property == "owl:Transitive":
            sample_size = int(len(self.relations) * self.prop_transitive_relations)
            if sample_size > len(relation_pool):
                sample_size = len(relation_pool)
            self.transitive_relations = random.sample(list(relation_pool), k=sample_size)

        # update rel2patterns
        self.update_rel2patterns(property)

    def add_inverseof(self):
        """
        Determines and adds inverse relations based on observed patterns and compatibility.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        observed_patterns = [frozenset(op) for op in self.rel2patterns.values()]
        running_inverseof_prop = 0.0
        attempt = 0
        warning_msg = 0

        # first with relations without pattern
        unpatterned_relations = [r for r in self.relations if not self.rel2patterns[r]]
        unpatterned_relations = (
            unpatterned_relations[:-1] if len(unpatterned_relations) % 2 == 1 else unpatterned_relations
        )

        while running_inverseof_prop < self.prop_inverse_relations and len(unpatterned_relations) >= 2:
            first_rel = unpatterned_relations.pop()
            second_rel = unpatterned_relations.pop()
            self.pair_inverseof(first_rel, second_rel)
            running_inverseof_prop = self.calculate_inverseof()

        while running_inverseof_prop < self.prop_inverse_relations:
            attempt += 1
            try:
                first_pattern = random.choice(observed_patterns)
                compatible_patterns = set(self.compat_inverseof[frozenset(first_pattern)])
                possible_pattern = set(observed_patterns).intersection(compatible_patterns)
                second_pattern = random.choice(list(possible_pattern))
                first_rel = random.choice(list(self.pattern2rels[frozenset(first_pattern)]))
                second_rel = random.choice(list(self.pattern2rels[frozenset(second_pattern)]))

                if (
                    first_rel != second_rel
                    and first_rel not in self.rel2inverse
                    and second_rel not in self.rel2inverse
                    and "owl:Reflexive" not in self.rel2patterns[first_rel]
                    and "owl:Reflexive" not in self.rel2patterns[second_rel]
                    and "owl:Irreflexive" not in self.rel2patterns[first_rel]
                    and "owl:Irreflexive" not in self.rel2patterns[second_rel]
                    and "owl:Symmetric" not in self.rel2patterns[first_rel]
                    and "owl:Symmetric" not in self.rel2patterns[second_rel]
                    and not (
                        "owl:Asymmetric" in self.rel2patterns[first_rel]
                        and "owl:Asymmetric" in self.rel2patterns[second_rel]
                    )
                ):
                    self.pair_inverseof(first_rel, second_rel)
                    running_inverseof_prop = self.calculate_inverseof()
                    attempt = 0
                else:
                    attempt += 1
                    if attempt > 1000:
                        self.prop_inverse_relations -= 0.005
                        attempt = 0
                        warning_msg = 1
            except:
                continue

        if warning_msg:
            print("Proportion of inverse relations reduced due to incompatibilities with other properties.")

    def pair_inverseof(self, rel, inv_rel):
        """
        Pairs relations as inverse relations.

        Args:
            self (object): The instance of the RelationGenerator.
            rel (str): The first relation.
            inv_rel (str): The second relation.

        Returns:
            None
        """
        self.inverseof_relations.append(rel)
        self.inverseof_relations.append(inv_rel)
        self.rel2inverse[rel] = inv_rel
        self.rel2inverse[inv_rel] = rel

    def calculate_inverseof(self):
        """
        Calculates the proportion of inverse relations.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            float: The proportion of inverse relations.
        """
        return len(self.inverseof_relations) / len(self.relations)

    def print_schema(self):
        """
        Prints the relation schema and
        displays various metrics and values related to the relations.

        Args:
            self (object): The instance of the RelationGenerator.

        Returns:
            None
        """
        print("\n")

        table = [
            ["Number of Relations", len(self.relations), self.num_relations],
            [
                "SubProperty Proportion",
                round(2 * len(self.prop2superprop) / len(self.relations), 2),
                self.prop_subproperties,
            ],
            [
                "Reflexive Relations",
                round(len(self.reflexive_relations) / len(self.relations), 2),
                self.prop_reflexive_relations,
            ],
            [
                "Irreflexive Relations",
                round(len(self.irreflexive_relations) / len(self.relations), 2),
                self.prop_irreflexive_relations,
            ],
            [
                "Functional Relations",
                round(len(self.functional_relations) / len(self.relations), 2),
                self.prop_functional_relations,
            ],
            [
                "InverseFunctional Relations",
                round(len(self.inversefunctional_relations) / len(self.relations), 2),
                self.prop_inverse_functional_relations,
            ],
            [
                "Symmetric Relations",
                round(len(self.symmetric_relations) / len(self.relations), 2),
                self.prop_symmetric_relations,
            ],
            [
                "Asymmetric Relations",
                round(len(self.asymmetric_relations) / len(self.relations), 2),
                self.prop_asymmetric_relations,
            ],
            [
                "Transitive Relations",
                round(len(self.transitive_relations) / len(self.relations), 2),
                self.prop_transitive_relations,
            ],
            [
                "InverseOf Relations",
                round(len(self.inverseof_relations) / len(self.relations), 2),
                round(self.prop_inverse_relations, 2),
            ],
            ["Profiled Relations", round(self.current_profile_ratio, 2), self.prop_profiled_relations],
            ["Relation Specificity", round(self.calculate_relation_specificity(), 2), self.relation_specificity],
        ]

        headers = ["Relation Metric", "Value", "Specified Value"]
        table_str = tabulate(table, headers, tablefmt="pretty")
        print(table_str)
        print("\n")

pygraft/relation_generator.py
nicolas-hbt/pygraft
import warnings

warnings.filterwarnings("ignore")

import json
from owlready2 import *
from rdflib import Graph, Namespace, RDF, RDFS, OWL, URIRef
from tqdm.auto import tqdm
import os
from datetime import datetime
from pygraft.utils import reasoner


class SchemaBuilder:
    def __init__(self, class_info, relation_info, folder_name, format):
        """
        Initializes the SchemaBuilder class.

        Args:
            self (object): The instance of the SchemaBuilder.
            class_info (dict): A dictionary containing class information.
            relation_info (dict): A dictionary containing relation information.
            folder_name (str): The name of the folder to be created. If None, a folder with the current date and time will be created.
            format (str): The format of the output file. Can be either "xml" or "ttl".

        Returns:
            None
        """
        self.class_info = class_info
        self.relation_info = relation_info
        self.format = format
        self.initialize_folder(folder_name)
        self.save_dict()

    def initialize_folder(self, folder_name):
        """
        Initializes a folder for output files.

        Args:
            self (object): The instance of the SchemaBuilder.
            folder_name (str): The name of the folder to be created. If None, a folder with the current date and time will be created.

        Returns:
            None
        """
        output_folder = "output/"
        if folder_name is None:
            output_folder += datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        else:
            output_folder += folder_name

        self.directory = f"{output_folder}/"
        if not os.path.exists(self.directory):
            os.makedirs(self.directory)

    def save_dict(self):
        """
        Saves the dictionary containing relation information and class information to JSON files.

        Args:
            self (object): The instance of the SchemaBuilder.

        Returns:
            None
        """
        rel_info = {}

        for k, v in self.relation_info.items():
            if k == "rel2patterns":
                rel_info["rel2patterns"] = {r: list(p) for r, p in self.relation_info[k].items()}
            elif k != "pattern2rels":
                rel_info[k] = v

        with open(f"{self.directory}relation_info.json", "w") as file:
            json.dump(rel_info, file, indent=4)

        class_dict = {k: list(v) if isinstance(v, set) else v for k, v in self.class_info.items()}

        with open(f"{self.directory}class_info.json", "w") as file:
            json.dump(class_dict, file, indent=4)

    def building_pipeline(self):
        """
        Initializes and builds the pipeline for creating the graph.

        This function initializes a new graph object and sets up the necessary namespaces for ontology, RDF, and RDFS.
        It binds the namespaces to the graph and adds the OWL ontology to the graph.
        It also adds the CC0 license URI to the ontology.
        After setting up the namespaces and ontology, it calls helper functions to add classes, relations, and test the schema.
        Finally, it prints a message indicating that the schema has been created.

        Args:
            self (object): The instance of the SchemaBuilder.

        Returns:
            None
        """
        self.graph = Graph()

        owl = Namespace("http://www.w3.org/2002/07/owl")
        rdf = Namespace("http://www.w3.org/1999/02/22-rdf-syntax-ns")
        rdfs = Namespace("http://www.w3.org/2000/01/rdf-schema")
        self.schema = Namespace("http://pygraf.t/")

        self.graph.bind("owl", owl)
        self.graph.bind("rdf", rdf)
        self.graph.bind("rdfs", rdfs)
        self.graph.bind("sc", self.schema)

        ontology_uri = URIRef(self.schema)
        self.graph.add((ontology_uri, RDF.type, OWL.Ontology))
        cc0_license_uri = URIRef("https://creativecommons.org/publicdomain/zero/1.0/")
        self.graph.add((URIRef(ontology_uri), URIRef("http://purl.org/dc/terms/license"), cc0_license_uri))

        self.add_classes()
        self.add_relations()

        print(f"\nSchema created.")

        ontology_file = (
            f"{self.directory}schema.rdf" if self.format == "xml" else f"{self.directory}schema.{self.format}"
        )
        reasoner(resource_file=ontology_file, resource="schema")

    def add_classes(self):
        """
        Adds classes to the graph based on the given class info.

        Args:
            self (object): The instance of the SchemaBuilder.

        Returns:
            None
        """
        classes = self.class_info["classes"]
        class2superclass = self.class_info["direct_class2superclass"]
        class2disjoints = self.class_info["class2disjoints"]

        for c in tqdm(classes, desc="Writing classes", unit="classes", colour="red"):
            class_URI = URIRef(self.schema + str(c))

            self.graph.add((class_URI, RDF.type, OWL.Class))

            if c in class2superclass:
                sp = class2superclass[c]

                if sp == "owl:Thing":
                    self.graph.add((class_URI, RDFS.subClassOf, OWL.Thing))
                else:
                    self.graph.add((class_URI, RDFS.subClassOf, URIRef(self.schema + str(sp))))

            if c in class2disjoints:
                for c2 in class2disjoints[c]:
                    self.graph.add((class_URI, OWL.disjointWith, URIRef(self.schema + str(c2))))

        print("\n")

    def add_relations(self):
        """
        Adds relations to the graph based on the provided relation information.

        This function iterates through each relation and performs the following steps:
        1. Creates a relation URI using the schema and the relation ID.
        2. Adds the rdf:type property to the relation URI with a value of OWL.ObjectProperty.
        3. If the relation has associated patterns, it adds the corresponding OWL property types to the relation URI.
        4. If the relation is not reflexive, it adds the domain and range assertions to the relation URI.
        5. If the relation has an inverse, it adds the inverseOf property to the relation URI.
        6. If the relation has a superrelation, it adds the subPropertyOf property to the relation URI.
        7. Serializes the graph to the specified output file.

        Args:
            self (object): The instance of the SchemaBuilder.

        Returns:
            None
        """
        relations = self.relation_info["relations"]
        rel2patterns = self.relation_info["rel2patterns"]
        rel2dom = self.relation_info["rel2dom"]
        rel2range = self.relation_info["rel2range"]
        rel2inverse = self.relation_info["rel2inverse"]
        rel2superrel = self.relation_info["rel2superrel"]

        for r in tqdm(relations, desc="Writing relations", unit="relations", colour="red"):
            relation_URI = URIRef(self.schema + str(r))

            # rdf:type
            self.graph.add((relation_URI, RDF.type, OWL.ObjectProperty))

            if r in rel2patterns:
                for object_property in rel2patterns[r]:
                    if object_property == "owl:Symmetric":
                        self.graph.add((relation_URI, RDF.type, OWL.SymmetricProperty))
                    if object_property == "owl:Asymmetric":
                        self.graph.add((relation_URI, RDF.type, OWL.AsymmetricProperty))
                    if object_property == "owl:Reflexive":
                        if r in rel2dom and r in rel2range:
                            if rel2dom[r] == rel2range[r]:
                                continue
                            else:
                                self.graph.add((relation_URI, RDF.type, OWL.ReflexiveProperty))
                        else:
                            self.graph.add((relation_URI, RDF.type, OWL.ReflexiveProperty))
                    if object_property == "owl:Irreflexive":
                        self.graph.add((relation_URI, RDF.type, OWL.IrreflexiveProperty))
                    if object_property == "owl:Transitive":
                        self.graph.add((relation_URI, RDF.type, OWL.TransitiveProperty))
                    if object_property == "owl:Functional":
                        self.graph.add((relation_URI, RDF.type, OWL.FunctionalProperty))
                    if object_property == "owl:InverseFunctional":
                        self.graph.add((relation_URI, RDF.type, OWL.InverseFunctionalProperty))

            if r in rel2dom and "owl:Reflexive" not in rel2patterns[r]:
                # https://oborel.github.io/obo-relations/reflexivity/: "Reflexivity is incompatible with domain and range assertions."
                domain_URI = URIRef(self.schema + str(rel2dom[r]))
                self.graph.add((relation_URI, RDFS.domain, domain_URI))

            if r in rel2range and "owl:Reflexive" not in rel2patterns[r]:
                # https://oborel.github.io/obo-relations/reflexivity/: "Reflexivity is incompatible with domain and range assertions."
                range_URI = URIRef(self.schema + str(rel2range[r]))
                self.graph.add((relation_URI, RDFS.range, range_URI))

            if r in rel2inverse:
                r_inv = URIRef(self.schema + str(rel2inverse[r]))
                self.graph.add((relation_URI, OWL.inverseOf, r_inv))

            if r in rel2superrel:
                superrel = URIRef(self.schema + str(rel2superrel[r]))
                self.graph.add((relation_URI, RDFS.subPropertyOf, superrel))

        print("\n")

        output_file = f"{self.directory}schema.rdf" if self.format == "xml" else f"{self.directory}schema.{self.format}"
        self.graph.serialize(output_file, format=self.format)

    def test_schema(self):
        """
        Tests the schema by loading the ontology file and running a reasoner.

        Args:
            self (object): The instance of the SchemaBuilder.

        Returns:
            None
        """
        ontology_file = (
            f"{self.directory}schema.rdf" if self.format == "xml" else f"{self.directory}schema.{self.format}"
        )
        ontology = get_ontology(ontology_file)

        try:
            with ontology.load():
                sync_reasoner_hermit(infer_property_values=False, debug=True)
        except OwlReadyInconsistentOntologyError:
            print("Inconsistent ontology.")

pygraft/schema_constructor.py
nicolas-hbt/pygraft
import pickle
import json
import pathlib
from owlready2 import *
from art import *
import random
import os
import yaml
from datetime import datetime
import pkg_resources
import shutil

font_styles = ["dancingfont", "rounded", "varsity", "wetletter", "chunky"]


def print_ascii_header():
    """
    
    """
    header = text2art("PyGraft", font=random.choice(font_styles))
    print("\n")
    print(header)
    print("\n")


def initialize_folder(folder_name):
    """
    Initializes a folder for output files.

    Args:
        self (object): The instance of the SchemaBuilder.
        folder_name (str): The name of the folder to be created. If None, a folder with the current date and time will be created.

    Returns:
        None
    """
    output_folder = "output/"
    if folder_name is None or folder_name == "None":
        folder_name = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

    output_folder += folder_name

    directory = f"{output_folder}/"
    if not os.path.exists(directory):
        os.makedirs(directory)

    return folder_name


def load_config(path):
    """
    Loads a configuration from a JSON or YAML file.
    
    Args:
        path (str): The path to the configuration file.

    Raises:
        ValueError: If the configuration file format is not supported.
        
    Returns:
        dict: The configuration dictionary.
    """
    path = pathlib.Path(path)

    if path.suffix == ".json":
        with path.open() as file:
            return json.load(file)

    if path.suffix in {".yaml", ".yml"}:
        with path.open() as file:
            return yaml.safe_load(file)

    raise ValueError(f"Unknown configuration file format: {path.suffix}. Valid formats: .json, .yaml, .yml")


def get_most_recent_subfolder(folder_path):
    """
    Gets the most recent subfolder in the given folder path.

    Args:
        folder_path (str): The path to the folder.

    Returns:
        str or None: The name of the most recent subfolder or None if there are no subfolders.
    """
    subfolders = [f.path for f in os.scandir(folder_path) if f.is_dir()]
    if not subfolders:
        return None
    most_recent_subfolder = max(subfolders, key=os.path.getctime)

    return os.path.basename(most_recent_subfolder)


def check_schema_arguments(config):
    """
    Checks the validity of the schema arguments.

    Args:
        config (dict): The configuration dictionary.

    Raises:
        AssertionError: If the proportions of owl:Asymmetric and owl:Symmetric relations sum to more than 1,
                        or if the proportions of owl:Irreflexive and owl:Reflexive relations sum to more than 1.
        AssertionError: If the current PyGraft version does not handle rdfs:subPropertyOf, owl:FunctionalProperty,
                        and owl:InverseFunctionalProperty at the same time.

    Returns:
        None
    """
    assert (
        config["prop_symmetric_relations"] + config["prop_asymmetric_relations"] <= 1.0
    ), "Proportions of owl:Asymmetric and owl:Symmetric relations cannot sum to more than 1."
    assert (
        config["prop_reflexive_relations"] + config["prop_irreflexive_relations"] <= 1.0
    ), "Proportions of owl:Irreflexive and owl:Reflexive relations cannot sum to more than 1."
    assert (
        config["avg_class_depth"] < config["max_hierarchy_depth"]
    ), "The average class depth value cannot be set higher than the class hierarchy depth."

    assert (
        config["prop_subproperties"] == 0.0
        and (config["prop_functional_relations"] >= 0.0 or config["prop_inverse_functional_relations"] >= 0.0)
    ) or (
        config["prop_subproperties"] >= 0.0
        and (config["prop_functional_relations"] == 0.0 or config["prop_inverse_functional_relations"] == 0.0)
    ), """
    The current PyGraft version does not handle rdfs:subPropertyOf, owl:FunctionalProperty, and owl:InverseFunctionalProperty **at the same time**.
    Retry choosing either:
    (1) -psub 0.0     -pfr value1  -pifr value2
    (2) -psub value3  -pfr 0.0     -pifr 0.0
    """


def check_kg_arguments(config):
    """
    Checks the validity of the knowledge graph arguments.

    Args:
        config (dict): The configuration dictionary.
    
    Returns:
        None
    """
    if config["multityping"] == False:
        config["avg_multityping"] = 1.0


def reasoner(resource_file=None, infer_property_values=False, debug=False, keep_tmp_file=False, resource="schema"):
    """
    Runs the HermiT reasoner on the given OWL file.

    Args:
        resource_file (str): The path to the OWL file.
        infer_property_values (bool): Whether to infer property values.
        debug (bool): Whether to print the debug information.
        keep_tmp_file (bool): Whether to keep the temporary file.
        resource (str): The name of the resource.

    Returns:
        None
    """
    graph = get_ontology(resource_file).load()
    try:
        sync_reasoner_hermit(
            graph, infer_property_values=infer_property_values, debug=debug, keep_tmp_file=keep_tmp_file
        )
        print(f"\nConsistent {resource}.\n")
        graph.destroy()
    except OwlReadyInconsistentOntologyError:
        print(f"\nInconsistent {resource}.\n")
        graph.destroy()


def save_dict_to_text(data_dict, file_path):
    """
    Saves a dictionary to a text file.
    
    Args:
        data_dict (dict): The dictionary to be saved.
        file_path (str): The path to the file.
        
    Returns:
        None
    """
    with open(file_path, "w") as file:
        for k, v in data_dict.items():
            if isinstance(v, list):
                for v2 in v:
                    file.write(str(k) + "\t" + str(v2) + "\n")
            else:
                file.write(str(k) + "\t" + str(v) + "\n")


def save_dict_to_pickle(data_dict, file_path):
    """
    Saves a dictionary to a pickle file.

    Args:
        data_dict (dict): The dictionary to be saved.
        file_path (str): The path to the file.
    
    Returns:
        None
    """
    with open(file_path, "wb") as file:
        pickle.dump(data_dict, file)


def save_set_uris_to_text(set_uris, file_path):
    """
    Saves a set of triples to a text file.
    
    Args:
        set_uris (set): The set of triples to be saved.
        file_path (str): The path to the file.
        
    Returns:
        None
    """
    with open(file_path, "w") as file:
        for t in set_uris:
            file.write(f"""<{t[0]}> <{t[1]}> <{t[2]}> .\n""")


def save_set_ids_to_text(set_ids, file_path):
    """
    Saves a set of triples to a text file.

    Args:
        set_ids (set): The set of triples to be saved.
        file_path (str): The path to the file.

    Returns:
        None
    """
    with open(file_path, "w") as file:
        for t in set_ids:
            file.write(f"""{t[0]}\t{t[1]}\t{t[2]}\n""")


def load_json(file_path):
    """
    Loads a JSON file.

    Args:
        file_path (str): The path to the file.

    Returns:
        dict: The loaded JSON file.
    """
    path = pathlib.Path(file_path)
    with path.open() as file:
        return json.load(file)


def load_json_template():
    """
    Loads a JSON file.
    
    Args:
        file_path (str): The path to the file.
        
    Returns:
        dict: The loaded JSON file.
    """
    json_file_path = pkg_resources.resource_filename("pygraft", "examples/template.json")
    destination_directory = os.getcwd()
    # Use the 'cp' command to copy the file
    # subprocess.run(["cp", json_file_path, destination_directory])
    shutil.copy(json_file_path, destination_directory)


def load_yaml_template():
    """
    Loads a YAML file.

    Args:
        file_path (str): The path to the file.

    Returns:
        dict: The loaded YAML file.
    """
    yaml_file_path = pkg_resources.resource_filename("pygraft", "examples/template.yml")
    destination_directory = os.getcwd()
    # Use the 'cp' command to copy the file
    # subprocess.run(["cp", yaml_file_path, destination_directory])
    shutil.copy(yaml_file_path, destination_directory)

pygraft/utils.py
nicolas-hbt/pygraft
import numpy as np


def generate_weight_vector(size, spread):
    """
    Generates a weight vector of size `size` with random values between 0 and 1.

    Args:
        size (int): The size of the weight vector.
        spread (float): The spread of the weight vector.

    Returns:
        A weight vector of size `size` with random values between 0 and 1.
    """
    if not 0 <= spread <= 1:
        raise ValueError("Spread parameter must be between 0 and 1.")

    weights = [np.random.rand() for _ in range(size)]
    total_weight = sum(weights)
    weights = [weight / total_weight for weight in weights]
    balanced_weights = [weight * (1 - spread) + (spread / size) for weight in weights]

    return balanced_weights


def generate_random_numbers(mean, std_dev, size):
    """
    Generates random numbers from a normal distribution with mean `mean` and standard deviation `std_dev`.

    Args:
        mean (float): The mean of the normal distribution.
        std_dev (float): The standard deviation of the normal distribution.
        size (int): The size of the output array.
    
    Returns:
        An array of random numbers from a normal distribution with mean `mean` and standard deviation `std_dev`.
    """
    # Generate random numbers from a normal distribution
    numbers = np.random.normal(mean, std_dev, size)
    # Normalize the numbers so their sum is equal to 1
    normalized_numbers = numbers / np.sum(numbers)
    # Clip the numbers between a small positive value and 1
    clipped_numbers = np.clip(normalized_numbers, np.finfo(float).eps, 1)
    # Normalize again to ensure the sum is exactly 1
    clipped_numbers = clipped_numbers / np.sum(clipped_numbers)

    return clipped_numbers


def get_fast_ratio(num_entities):
    """
    Makes the KG generation process faster by lowering the diversity in entities' profiles.
    
    Args:
        num_entities (int): The number of entities.
        
    Returns:
        The fast ratio for the given number of entities.
    """
    if num_entities >= 1000000:
        return 15
    if num_entities >= 500000:
        return 10
    elif num_entities >= 50000:
        return 5
    elif num_entities >= 30000:
        return 3
    else:
        return 1


def transitive_inference(triples, original_triples):
    """
    Infers new triples to be added using transitive inference.

    Args:
        triples (set): The set of triples.
        original_triples (set): The set of original triples.

    Returns:
        set: The set of inferred triples.
    """
    inferred_triples = set()
    print(list(triples))
    rel = list(triples)[0][1]
    for triple1 in original_triples:
        for triple2 in triples:
            if triple1[2] == triple2[0]:
                inferred_triple = (triple1[0], rel, triple2[2])

                if inferred_triple not in triples:
                    inferred_triples.add(inferred_triple)

    if inferred_triples:
        triples.update(inferred_triples)  # for next run
        inferred_triples.update(transitive_inference(triples, original_triples))

    return inferred_triples


def inverse_inference(kg, inv_rel):
    """
    Infers new triples to be added using inverse inference.

    Args:
        kg (set): The set of triples.
        inv_rel (str): The inverse relation.

    Returns:
        set: The set of inferred triples.
    """
    return set((triple[2], inv_rel, triple[0]) for triple in kg)


def symmetric_inference(kg):
    """
    Infers new triples to be added using symmetric inference.
    
    Args:
        kg (set): The set of triples.
        
    Returns:
        set: The set of inferred triples.
    """
    return set((triple[2], triple[1], triple[0]) for triple in kg)


def reflexive_inference(kg):
    """
    Infers new triples to be added using reflexive inference.

    Args:
        kg (set): The set of triples.

    Returns:
        set: The set of inferred triples.
    """
    return set((triple[0], triple[1], triple[0]) for triple in kg) | set(
        (triple[2], triple[1], triple[2]) for triple in kg
    )


def subproperty_inference(kg, super_rel):
    """
    Infers new triples to be added using subproperty inference.

    Args:
        kg (set): The set of triples.
        super_rel (str): The superproperty relation.

    Returns:
        set: The set of inferred triples.
    """
    return set((triple[0], super_rel, triple[2]) for triple in kg)


def filter_symmetric(arr):
    """
    Filters out symmetric triples from an array.

    Args:
        arr (np.ndarray): The array of triples.

    Returns:
        np.ndarray: The filtered array of triples.
    """
    result = []
    for row in arr:
        if row.tolist() not in result and row[::-1].tolist() not in result:
            result.append(row.tolist())
    return np.array(result)

pygraft/utils_kg.py
nicolas-hbt/pygraft
from collections import defaultdict


def non_trivial_children(class2superclass_direct):
    """
    Returns a list of classes that have at least one non-trivial parent.
    
    Args:
        class2superclass_direct (dict): A dictionary mapping classes to their direct superclasses.
        
    Returns:
        list: A list of classes that have at least one non-trivial parent.
    """
    return [c for c in class2superclass_direct.keys() if class2superclass_direct[c] != "owl:Thing"]


def get_subclassof_count(class2layer):
    """
    Returns the number of classes that have at least one subclass.
    
    Args:
        class2layer (dict): A dictionary mapping classes to their layers.
        
    Returns:
        int: The number of classes that have at least one non-trivial parent.
    """
    return len([key for key, value in class2layer.items() if value > 1])


def get_leaves(class2superclass_direct, class2subclasses_direct):
    """
    Returns a list of classes that have no subclasses, i.e. leaves.

    Args:
        class2superclass_direct (dict): A dictionary mapping classes to their direct superclasses.
        class2subclasses_direct (dict): A dictionary mapping classes to their direct subclasses.

    Returns:
        list: A list of classes that have no subclasses.
    """
    return set(class2superclass_direct.keys()) - set(class2subclasses_direct.keys())


def get_max_depth(layer2classes):
    """
    Returns the maximum depth of the schema.

    Args:
        layer2classes (dict): A dictionary mapping layers to classes.

    Returns:
        int: The maximum depth of the schema.
    """
    return max((key for key, value in layer2classes.items() if value), default=None)


def calculate_inheritance_ratio(class2superclass_direct, class2subclasses_direct):
    """
    Calculates the inheritance ratio of the schema.

    Args:
        class2superclass_direct (dict): A dictionary mapping classes to their direct superclasses.
        class2subclasses_direct (dict): A dictionary mapping classes to their direct subclasses.

    Returns:
        float: The inheritance ratio of the schema.
    """
    n_classes = len(class2superclass_direct.keys())
    n_leaves = len(get_leaves(class2superclass_direct, class2subclasses_direct))
    n_non_trivial_children = len(non_trivial_children(class2superclass_direct))

    return n_non_trivial_children / (n_classes - n_leaves)


def calculate_average_depth(layer2classes):
    """
    Calculates the average depth of the schema.

    Args:
        layer2classes (dict): A dictionary mapping layers to classes.

    Returns:
        float: The average depth of the schema.
    """
    denominator = sum(map(len, layer2classes.values()))
    numerator = 0.0

    for key, value in layer2classes.items():
        numerator += key * len(value)

    return numerator / denominator


def calculate_class_disjointness(class2disjoint, num_classes):
    """
    Calculates the class disjointness of the schema.

    Args:
        class2disjoint (dict): A dictionary mapping classes to their disjoint classes.
        num_classes (int): The number of classes.

    Returns:
        float: The class disjointness of the schema.
    """
    return len(class2disjoint) / (2 * num_classes)


def get_all_superclasses(class_name, direct_class2superclass):
    """
    Returns a list of all superclasses of a given class.
    
    Args:
        class_name (str): The name of the class.
        direct_class2superclass (dict): A dictionary mapping classes to their direct superclasses.
        
    Returns:
        list: A list of all superclasses of the given class.
    """
    superclasses = []

    if class_name in direct_class2superclass:
        superclass = direct_class2superclass[class_name]
        superclasses.append(superclass)
        superclasses.extend(get_all_superclasses(superclass, direct_class2superclass))

    return superclasses


def get_all_subclasses(transitive_class2superclass):
    """
    Returns a dictionary mapping classes to their transitive subclasses.

    Args:
        transitive_class2superclass (dict): A dictionary mapping classes to their transitive superclasses.

    Returns:
        dict: A dictionary mapping classes to their subclasses.
    """
    class2subclasses = defaultdict(list)

    for subclass, superclasses in transitive_class2superclass.items():
        for superclass in superclasses:
            class2subclasses[superclass].append(subclass)

    return dict(class2subclasses)


def extend_class_mappings(direct_class2superclass):
    """
    Extends the class mappings to include transitive superclasses and subclasses.

    Args:
        direct_class2superclass (dict): A dictionary mapping classes to their direct superclasses.

    Returns:
        tuple: A tuple containing the extended class mappings.
    """
    transitive_class2superclass = {}
    transitive_class2subclasses = {}

    for class_name in direct_class2superclass:
        # Extend superclasses recursively
        transitive_superclasses = get_all_superclasses(class_name, direct_class2superclass)
        transitive_class2superclass[class_name] = transitive_superclasses

    transitive_class2subclasses = get_all_subclasses(transitive_class2superclass)

    return transitive_class2superclass, transitive_class2subclasses


def generate_class2layer(layer2classes):
    """
    Generates a dictionary mapping classes to their layers.

    Args:
        layer2classes (dict): A dictionary mapping layers to classes.

    Returns:
        dict: A dictionary mapping classes to their layers.
    """
    class2layer = {}

    for layer, classes in layer2classes.items():
        for c in classes:
            class2layer[c] = layer

    return class2layer

pygraft/utils_schema.py
nicolas-hbt/pygraft
from setuptools import setup, find_packages

with open("README.md", "r", encoding="utf8") as fh:
    long_description = fh.read()

setup(
  name = 'pygraft',
  version = '0.0.3',
  packages = find_packages(exclude=('docs')),
  include_package_data=True,
  zip_safe=False,
  package_data={
        'pygraft': ['examples/template.json', 'examples/template.yml', 'property_checks/combinations.json', 'property_checks/compat_p1p2_inverseof.txt']
  },
    entry_points={
    'console_scripts': [
        'pygraft = pygraft.main:main',
    ],
  },
  license='MIT',
  description = 'PyGraft: Configurable Generation of Schemas and Knowledge Graphs at Your Fingertips',
  author = 'Nicolas Hubert',
  author_email = 'nicolas.hubert@univ-lorraine.fr',
  url = 'https://github.com/nicolas-hbt',
  long_description=long_description,
  long_description_content_type="text/markdown",
  keywords = ['Knowledge Graph',
              'Ontology', 
              'Schema',
              'Semantic Web',
              'Synthetic Data Generator'],
  install_requires=[ 
      'numpy>=1.24.0',         
      'matplotlib>=3.7.0',
      'Owlready2>=0.41',
      'rdflib>=6.2.0',
      'pyyaml',
      'tabulate',
      'art',
      'tqdm'
      ],
  python_requires='>=3.7',
  classifiers=[
    'Development Status :: 4 - Beta',
    'Intended Audience :: Science/Research',
    'Topic :: Software Development :: Build Tools',
    'License :: OSI Approved :: MIT License', 
    'Programming Language :: Python :: 3 :: Only',
    'Programming Language :: Python :: 3.7',
    'Programming Language :: Python :: 3.8',
    'Programming Language :: Python :: 3.9',
    'Programming Language :: Python :: 3.10',
    'Programming Language :: Python :: 3.11',
    'Topic :: Scientific/Engineering :: Artificial Intelligence',
    'Topic :: Scientific/Engineering :: Information Analysis'
  ],
)

setup.py
nicolas-hbt/pygraft