<commit_before>import networkx as nx

from pybbn.graph.edge import Edge, EdgeType
from pybbn.graph.graph import Graph
from pybbn.graph.node import BbnNode
from pybbn.graph.variable import Variable


class Dag(Graph):
    """
    Directed acyclic graph.
    """

    def __init__(self):
        """
        Ctor.
        """
        Graph.__init__(self)

    def get_parents(self, id):
        """
        Gets the parent IDs of the specified node.
        :param id: Node id.
        :return: Array of parent ids.
        """
        return [x for x in self.map if id in self.map[x]]

    def get_children(self, node_id):
        """
        Gets the children IDs of the specified node.
        :param node_id: Node id.
        :return: Array of children ids.
        """
        return [x for x in self.map[node_id]]

    def __shouldadd__(self, edge):
        """
        Checks if the specified directed edge should be added.
        :param edge: Directed edge.
        :return: A boolean indicating if the edge should be added.
        """
        if EdgeType.DIRECTED != edge.type:
            return False

        parent = edge.i
        child = edge.j

        if parent.id == child.id:
            return False

        if child.id not in self.map[parent.id] and parent.id not in self.map[child.id]:
            if not PathDetector(self, child.id, parent.id).exists():
                return True

        return False

    def edge_exists(self, id1, id2):
        """
        Checks if a directed edge exists between the specified id. e.g. id1 -> id2
        :param id1: Node id.
        :param id2: Node id.
        :return: A boolean indicating if a directed edge id1 -> id2 exists.
        """
        if id2 in self.map[id1] and id1 not in self.map[id2]:
            return True
        return False

    def to_nx_graph(self):
        """
        Converts this DAG to a NX DiGraph for visualization.
        :return: A tuple, where the first item is the NX DiGraph and the second items are the node labels.
        """
        g = nx.DiGraph()
        labels = []

        for k, node in self.nodes.items():
            g.add_node(node.id)
            t = (node.id, node.variable.name)
            labels.append(t)

        for k, edge in self.edges.items():
            pa = edge.i.id
            ch = edge.j.id
            g.add_edges_from([(pa, ch, {})])

        return g, dict(labels)


class Bbn(Dag):
    """
    BBN.
    """

    def __init__(self):
        """Ctor."""
        Dag.__init__(self)
        self.parents = {}

    def get_parents_ordered(self, id):
        return sorted(self.parents[id]) if id in self.parents else []

    def __edge_added__(self, edge):
        if edge.j.id not in self.parents:
            self.parents[edge.j.id] = []

        if edge.i.id not in self.parents[edge.j.id]:
            self.parents[edge.j.id].append(edge.i.id)

    def __shouldadd__(self, edge):
        """
        Checks if the specified directed edge should be added.
        :param edge: Directed edge.
        :return: A boolean indicating if the directed edge should be added.
        """
        if isinstance(edge.i, BbnNode) and isinstance(edge.j, BbnNode):
            return True
        return Dag.__shouldadd__(edge)

    @staticmethod
    def to_dict(bbn):
        """
        Gets a JSON serializable dictionary representation.
        :param bbn: BBN.
        :return: Dictionary.
        """
        return {
            'nodes': {n.id: n.to_dict() for n in bbn.get_nodes()},
            'edges': [{'pa': edge.i.id, 'ch': edge.j.id} for _, edge in bbn.edges.items()]
        }

    @staticmethod
    def from_dict(d):
        """
        Creates a BBN from a dictionary (deserialized JSON).
        :param d: Dictionary.
        :return: BBN.
        """
        def get_variable(d):
            return Variable(d['id'], d['name'], d['values'])

        def get_bbn_node(d):
            bbn_node = BbnNode(get_variable(d['variable']), d['probs'])
            bbn_node.metadata = d['metadata']
            return bbn_node

        nodes = {k: get_bbn_node(n) for k, n in d['nodes'].items()}
        edges = d['edges']

        bbn = Bbn()

        for k, n in nodes.items():
            bbn.add_node(n)

        for e in edges:
            pa_id = e['pa']
            ch_id = e['ch']

            pa = nodes[pa_id] if pa_id in nodes else nodes[str(pa_id)]
            ch = nodes[ch_id] if ch_id in nodes else nodes[str(ch_id)]

            bbn.add_edge(Edge(pa, ch, EdgeType.DIRECTED))

        return bbn


class PathDetector(object):
    """<commit_msg>revert; don't serde meta (for internal use)<commit_after>import networkx as nx

from pybbn.graph.edge import Edge, EdgeType
from pybbn.graph.graph import Graph
from pybbn.graph.node import BbnNode
from pybbn.graph.variable import Variable


class Dag(Graph):
    """
    Directed acyclic graph.
    """

    def __init__(self):
        """
        Ctor.
        """
        Graph.__init__(self)

    def get_parents(self, id):
        """
        Gets the parent IDs of the specified node.
        :param id: Node id.
        :return: Array of parent ids.
        """
        return [x for x in self.map if id in self.map[x]]

    def get_children(self, node_id):
        """
        Gets the children IDs of the specified node.
        :param node_id: Node id.
        :return: Array of children ids.
        """
        return [x for x in self.map[node_id]]

    def __shouldadd__(self, edge):
        """
        Checks if the specified directed edge should be added.
        :param edge: Directed edge.
        :return: A boolean indicating if the edge should be added.
        """
        if EdgeType.DIRECTED != edge.type:
            return False

        parent = edge.i
        child = edge.j

        if parent.id == child.id:
            return False

        if child.id not in self.map[parent.id] and parent.id not in self.map[child.id]:
            if not PathDetector(self, child.id, parent.id).exists():
                return True

        return False

    def edge_exists(self, id1, id2):
        """
        Checks if a directed edge exists between the specified id. e.g. id1 -> id2
        :param id1: Node id.
        :param id2: Node id.
        :return: A boolean indicating if a directed edge id1 -> id2 exists.
        """
        if id2 in self.map[id1] and id1 not in self.map[id2]:
            return True
        return False

    def to_nx_graph(self):
        """
        Converts this DAG to a NX DiGraph for visualization.
        :return: A tuple, where the first item is the NX DiGraph and the second items are the node labels.
        """
        g = nx.DiGraph()
        labels = []

        for k, node in self.nodes.items():
            g.add_node(node.id)
            t = (node.id, node.variable.name)
            labels.append(t)

        for k, edge in self.edges.items():
            pa = edge.i.id
            ch = edge.j.id
            g.add_edges_from([(pa, ch, {})])

        return g, dict(labels)


class Bbn(Dag):
    """
    BBN.
    """

    def __init__(self):
        """Ctor."""
        Dag.__init__(self)
        self.parents = {}

    def get_parents_ordered(self, id):
        return sorted(self.parents[id]) if id in self.parents else []

    def __edge_added__(self, edge):
        if edge.j.id not in self.parents:
            self.parents[edge.j.id] = []

        if edge.i.id not in self.parents[edge.j.id]:
            self.parents[edge.j.id].append(edge.i.id)

    def __shouldadd__(self, edge):
        """
        Checks if the specified directed edge should be added.
        :param edge: Directed edge.
        :return: A boolean indicating if the directed edge should be added.
        """
        if isinstance(edge.i, BbnNode) and isinstance(edge.j, BbnNode):
            return True
        return Dag.__shouldadd__(edge)

    @staticmethod
    def to_dict(bbn):
        """
        Gets a JSON serializable dictionary representation.
        :param bbn: BBN.
        :return: Dictionary.
        """
        return {
            'nodes': {n.id: n.to_dict() for n in bbn.get_nodes()},
            'edges': [{'pa': edge.i.id, 'ch': edge.j.id} for _, edge in bbn.edges.items()]
        }

    @staticmethod
    def from_dict(d):
        """
        Creates a BBN from a dictionary (deserialized JSON).
        :param d: Dictionary.
        :return: BBN.
        """
        def get_variable(d):
            return Variable(d['id'], d['name'], d['values'])

        def get_bbn_node(d):
            return BbnNode(get_variable(d['variable']), d['probs'])

        nodes = {k: get_bbn_node(n) for k, n in d['nodes'].items()}
        edges = d['edges']

        bbn = Bbn()

        for k, n in nodes.items():
            bbn.add_node(n)

        for e in edges:
            pa_id = e['pa']
            ch_id = e['ch']

            pa = nodes[pa_id] if pa_id in nodes else nodes[str(pa_id)]
            ch = nodes[ch_id] if ch_id in nodes else nodes[str(ch_id)]

            bbn.add_edge(Edge(pa, ch, EdgeType.DIRECTED))

        return bbn


class PathDetector(object):
    """
{'fraction_non_alphanumeric': 0.11365127980748195, 'fraction_numerical': 0.0026252461168234523, 'mean_word_length': 1.7892007321537522, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1830535', 'n_tokens_mistral': 2844, 'n_tokens_neox': 2779, 'n_words': 866}