[ENH] Add the ability to convert a DAG to an MAG

doc/api.rst

@@ -43,6 +43,7 @@ causal graph operations.
 .. autosummary::
    :toctree: generated/
 
+   dag_to_mag
    valid_mag
    has_adc
    inducing_path

doc/whats_new/v0.2.rst

@@ -26,6 +26,7 @@ Version 0.2
 Changelog
 ---------
 - |Feature| Implement and test functions to validate an MAG and check the presence of almost directed cycles, by `Aryan Roy`_ (:pr:`91`)
+- |Feature| Implement and test functions to convert a DAG to MAG, by `Aryan Roy`_ (:pr:`96`)
 
 Code and Documentation Contributors
 -----------------------------------

pywhy_graphs/algorithms/generic.py

@@ -15,6 +15,7 @@
     "inducing_path",
     "has_adc",
     "valid_mag",
+    "dag_to_mag",
 ]
 
 
@@ -554,26 +555,29 @@
     .. footbibliography::
     """
     if L is None:
         L = set()
 
     if S is None:
         S = set()
 
     nodes = set(G.nodes)
 
     if node_x not in nodes or node_y not in nodes:
         raise ValueError("The provided nodes are not in the graph.")
 
     if node_x == node_y:
         raise ValueError("The source and destination nodes are the same.")
 
+    if (node_x in L) or (node_y in L) or (node_x in S) or (node_y in S):
+        return (False, [])
+
     edges = G.edges()
 
     # XXX: fix this when graphs are refactored to only check for directed/bidirected edge types
     for elem in edges.keys():
         if elem not in {"directed", "bidirected"}:
             if len(edges[elem]) != 0:
                 raise ValueError("Inducing Path is not defined for this graph.")
 
     path = []  # this will contain the path.
 
@@ -703,3 +707,77 @@
                 return False
 
     return True
+
+
+def dag_to_mag(G, L: Set = None, S: Set = None):
+    """Converts a DAG to a valid MAG.
+    The algorithm is defined in :footcite:`Zhang2008` on page 1877.
+
+    Parameters:
+    -----------
+    G : Graph
+        The graph.
+    L : Set
+        Nodes that are ignored on the path. Defaults to an empty set.
+    S : Set
+        Nodes that are always conditioned on. Defaults to an empty set.
+    """
+
+    if L is None:
+        L = set()
+
+    if S is None:
+        S = set()
+
+    # for each pair of nodes find if they have an inducing path between them.
+    # only then will they be adjacent in the MAG.
+
+    all_nodes = set(G.nodes)
+    adj_nodes = []
+
+    for source in all_nodes:
+        copy_all = all_nodes.copy()
+        copy_all.remove(source)
+        for dest in copy_all:
+            out = inducing_path(G, source, dest, L, S)
+            if out[0] is True and {source, dest} not in adj_nodes:
+                adj_nodes.append({source, dest})
+
+    # find the ancestors of B U S (ansB) and A U S (ansA) for each pair of adjacent nodes
+
+    mag = ADMG()
+
+    for elem in adj_nodes:
+
+        temp_list = list(elem)
+        a = set(temp_list[0])
+        b = set(temp_list[1])
+        aus = S.union(a)
+        bus = S.union(b)
+
+        ansA: Set[str] = set()
+        ansB: Set[str] = set()
+
+        for node in aus:
+            ansA = ansA.union(_directed_sub_graph_ancestors(G, node))
+
+        for node in bus:
+            ansB = ansB.union(_directed_sub_graph_ancestors(G, node))
+
+        if temp_list[0] in ansB and temp_list[1] not in ansA:
+            # if A is in ansB and B is not in ansA, A -> B
+            mag.add_edge(temp_list[0], temp_list[1], mag.directed_edge_name)
+
+        elif temp_list[0] not in ansB and temp_list[1] in ansA:
+            # if B is in ansA and A is not in ansB, A <- B
+            mag.add_edge(temp_list[1], temp_list[0], mag.directed_edge_name)
+
+        elif temp_list[0] not in ansB and temp_list[1] not in ansA:
+            # if A is not in ansB and B is not in ansA, A <-> B
+            mag.add_edge(temp_list[1], temp_list[0], mag.bidirected_edge_name)
+
+        elif temp_list[0] in ansB and temp_list[1] in ansA:
+            # if A is in ansB and B is in ansA, A - B
+            mag.add_edge(temp_list[1], temp_list[0], mag.undirected_edge_name)
+
+    return mag

pywhy_graphs/algorithms/tests/test_generic.py

@@ -348,3 +348,63 @@ def test_valid_mag():
     admg.add_edge("H", "J", admg.undirected_edge_name)
 
     assert not pywhy_graphs.valid_mag(admg)  # there is an undirected edge between H and J
+
+
+def test_dag_to_mag():
+
+    admg = ADMG()
+    admg.add_edge("A", "E", admg.directed_edge_name)
+    admg.add_edge("E", "S", admg.directed_edge_name)
+    admg.add_edge("H", "E", admg.directed_edge_name)
+    admg.add_edge("H", "R", admg.directed_edge_name)
+
+    S = {"S"}
+    L = {"H"}
+
+    out_edges = pywhy_graphs.dag_to_mag(admg, L, S).edges()
+
+    dir_edges = list(out_edges["directed"])
+    assert (
+        ("A", "R") in out_edges["directed"]
+        and ("E", "R") in out_edges["directed"]
+        and len(out_edges["directed"]) == 2
+    )
+    assert ("A", "E") in out_edges["undirected"]
+
+    admg = ADMG()
+    admg.add_edge("A", "E", admg.directed_edge_name)
+    admg.add_edge("H", "S", admg.directed_edge_name)
+    admg.add_edge("H", "E", admg.directed_edge_name)
+    admg.add_edge("H", "R", admg.directed_edge_name)
+
+    S = {"S"}
+    L = {"H"}
+
+    out_edges = pywhy_graphs.dag_to_mag(admg, L, S).edges()
+
+    dir_edges = list(out_edges["directed"])
+    assert ("A", "E") in out_edges["directed"] and len(out_edges["directed"]) == 1
+    assert ("E", "R") in out_edges["bidirected"]
+
+    admg = ADMG()
+    admg.add_edge("P", "S", admg.directed_edge_name)
+    admg.add_edge("S", "L", admg.directed_edge_name)
+    admg.add_edge("G", "S", admg.directed_edge_name)
+    admg.add_edge("G", "L", admg.directed_edge_name)
+    admg.add_edge("I", "S", admg.directed_edge_name)
+    admg.add_edge("J", "I", admg.directed_edge_name)
+    admg.add_edge("J", "S", admg.directed_edge_name)
+
+    S = set()
+    L = {"J"}
+
+    out_edges = pywhy_graphs.dag_to_mag(admg, L, S).edges()
+    dir_edges = list(out_edges["directed"])
+    assert (
+        ("G", "S") in dir_edges
+        and ("G", "L") in dir_edges
+        and ("S", "L") in dir_edges
+        and ("I", "S") in dir_edges
+        and ("P", "S") in dir_edges
+        and len(dir_edges) == 5
+    )