[ENH] Add the ability to find Proper Possibly Directed Paths

pywhy_graphs/algorithms/generic.py

@@ -19,6 +19,7 @@
     "dag_to_mag",
     "is_maximal",
     "all_vstructures",
+    "possibly_directed_path",
 ]
 
 
@@ -855,3 +856,99 @@ def all_vstructures(G: nx.DiGraph, as_edges: bool = False):
                 else:
                     vstructs.add((p1, node, p2))  # type: ignore
     return vstructs
+
+
+def check_back_arrow(G: ADMG, X, Y: set):
+
+    out = set()
+
+    for elem in Y:
+        if not (
+            G.has_edge(X, elem, G.bidirected_edge_name) or G.has_edge(elem, X, G.directed_edge_name)
+        ):
+            out.update(elem)
+
+    return out
+
+
+def get_X_neighbors(G, X: set):
+
+    out = []
+
+    for elem in X:
+        elem_neighbors = set(G.neighbors(elem))
+        elem_possible_neighbors = check_back_arrow(G, elem, elem_neighbors)
+        to_remove = X.intersection(elem_possible_neighbors)
+        elem_neighbors = elem_possible_neighbors - to_remove
+
+        if len(elem_neighbors) != 0:
+            for nbh in elem_neighbors:
+                temp = dict()
+                temp[0] = elem
+                temp[1] = nbh
+                out.append(temp)
+    return out
+
+
+def recursively_find_pd_paths(G, X, paths, Y):
+
+    counter = 0
+    new_paths = []
+
+    for i in range(len(paths)):
+        cur_elem = paths[i][list(paths[i].keys())[-1]]
+
+        if cur_elem in Y:
+            new_paths.append(paths[i])
+            continue
+
+        nbr_temp = G.neighbors(cur_elem)
+        nbr_possible = check_back_arrow(G, cur_elem, nbr_temp)
+
+        if len(nbr_possible) == 0:
+            new_paths.append(paths[i].copy())
+
+        possible_end = nbr_possible.intersection(Y)
+
+        if len(possible_end) != 0:
+            for elem in possible_end:
+                temp_path = paths[i].copy()
+                temp_path[len(temp_path)] = elem
+                new_paths.append(temp_path)
+
+        remaining_nodes = nbr_possible - possible_end
+        remaining_nodes = (
+            remaining_nodes
+            - remaining_nodes.intersection(paths[i].values())
+            - remaining_nodes.intersection(X)
+        )
+
+        temp_arr = []
+        for elem in remaining_nodes:
+            temp_paths = paths[i].copy()
+            temp_paths[len(temp_paths)] = elem
+            temp_arr.append(temp_paths)
+
+        new_paths.extend(recursively_find_pd_paths(G, X, temp_arr, Y))
+
+    return new_paths
+
+
+def possibly_directed_path(G, X: Optional[Set] = None, Y: Optional[Set] = None):
+
+    if isinstance(X, set):
+        x_neighbors = get_X_neighbors(G, X)
+    else:
+        nbr_temp = G.neighbors(X)
+        nbr_possible = check_back_arrow(nbr_temp)
+        x_neighbors = []
+
+        for elem in nbr_possible:
+            temp = dict()
+            temp[0] = X
+            temp[1] = elem
+            x_neighbors.append(temp)
+
+    path_list = recursively_find_pd_paths(G, X, x_neighbors, Y)
+
+    return path_list

pywhy_graphs/algorithms/tests/test_generic.py

@@ -2,7 +2,7 @@
 import pytest
 
 import pywhy_graphs
-from pywhy_graphs import ADMG
+from pywhy_graphs import ADMG, PAG
 from pywhy_graphs.algorithms import all_vstructures
 
 
@@ -496,3 +496,207 @@ def test_all_vstructures():
     # Assert that the returned values are as expected
     assert len(v_structs_edges) == 0
     assert len(v_structs_tuples) == 0
+
+
+def test_possibly_directed():
+    # X <- Y <-> Z <-> H; Z -> X
+
+    admg = ADMG()
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "H", admg.directed_edge_name)
+
+    Y = {"H"}
+    X = {"Y"}
+
+    correct = [{0: "Y", 1: "X", 2: "Z", 3: "H"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+
+    admg = ADMG()
+    admg.add_edge("A", "X", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "H", admg.directed_edge_name)
+
+    Y = {"H"}
+    X = {"Y", "A"}
+
+    correct = [{0: "A", 1: "X", 2: "Z", 3: "H"}, {0: "Y", 1: "X", 2: "Z", 3: "H"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+
+    admg = ADMG()
+    admg.add_edge("X", "A", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "H", admg.directed_edge_name)
+
+    Y = {"H"}
+    X = {"Y", "A"}
+
+    correct = [{0: "Y", 1: "X", 2: "Z", 3: "H"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+
+    admg = ADMG()
+    admg.add_edge("X", "A", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "H", admg.directed_edge_name)
+    admg.add_edge("K", "Z", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [{0: "Y", 1: "X", 2: "Z", 3: "H"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+
+    admg = ADMG()
+    admg.add_edge("A", "X", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "H", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [
+        {0: "A", 1: "X", 2: "Z", 3: "H"},
+        {0: "A", 1: "X", 2: "Z", 3: "K"},
+        {0: "Y", 1: "X", 2: "Z", 3: "H"},
+        {0: "Y", 1: "X", 2: "Z", 3: "K"},
+    ]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+    assert correct[2] == out[2]
+    assert correct[3] == out[3]
+
+    admg = ADMG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("G", "C", admg.directed_edge_name)
+    admg.add_edge("C", "H", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [{0: "A", 1: "G", 2: "C", 3: "H"}, {0: "Y", 1: "X", 2: "Z", 3: "K"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+
+    admg = ADMG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("G", "C", admg.directed_edge_name)
+    admg.add_edge("C", "H", admg.directed_edge_name)
+    admg.add_edge("Z", "C", admg.directed_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [
+        {0: "A", 1: "G", 2: "C", 3: "H"},
+        {0: "Y", 1: "X", 2: "Z", 3: "K"},
+        {0: "Y", 1: "X", 2: "Z", 3: "C", 4: "H"},
+    ]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+    assert correct[2] == out[2]
+
+    admg = ADMG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("A", "H", admg.directed_edge_name)
+    admg.add_edge("K", "G", admg.directed_edge_name)
+    admg.add_edge("K", "H", admg.directed_edge_name)
+
+    Y = {"G", "H"}
+    X = {"A", "K"}
+
+    correct = [{0: "K", 1: "G"}, {0: "K", 1: "H"}, {0: "A", 1: "G"}, {0: "A", 1: "H"}]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+    assert correct[2] == out[2]
+    assert correct[3] == out[3]
+
+    admg = ADMG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("G", "C", admg.directed_edge_name)
+    admg.add_edge("C", "H", admg.directed_edge_name)
+    admg.add_edge("Z", "C", admg.bidirected_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [
+        {0: "A", 1: "G", 2: "C", 3: "H"},
+        {0: "Y", 1: "X", 2: "Z", 3: "K"},
+    ]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+
+    admg = ADMG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("G", "C", admg.directed_edge_name)
+    admg.add_edge("C", "H", admg.directed_edge_name)
+    admg.add_edge("Z", "C", admg.bidirected_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [
+        {0: "A", 1: "G", 2: "C", 3: "H"},
+        {0: "Y", 1: "X", 2: "Z", 3: "K"},
+        {0: "Y", 1: "X", 2: "Z", 3: "C", 4: "H"},
+        {0: "A", 1: "G", 2: "C", 3: "Z", 4: "K"},
+    ]
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+    assert correct[2] == out[2]
+
+    admg = PAG()
+    admg.add_edge("A", "G", admg.directed_edge_name)
+    admg.add_edge("G", "C", admg.directed_edge_name)
+    admg.add_edge("C", "H", admg.directed_edge_name)
+    admg.add_edge("Z", "C", admg.circle_edge_name)
+    admg.add_edge("C", "Z", admg.circle_edge_name)
+    admg.add_edge("Y", "X", admg.directed_edge_name)
+    admg.add_edge("X", "Z", admg.directed_edge_name)
+    admg.add_edge("Z", "K", admg.directed_edge_name)
+
+    Y = {"H", "K"}
+    X = {"Y", "A"}
+
+    correct = [
+        [
+            {0: "Y", 1: "X", 2: "Z", 3: "K"},
+            {0: "Y", 1: "X", 2: "Z", 3: "C", 4: "H"},
+            {0: "A", 1: "G", 2: "C", 3: "H"},
+            {0: "A", 1: "G", 2: "C", 3: "Z", 4: "K"},
+        ]
+    ]
+
+    out = pywhy_graphs.possibly_directed_path(admg, X, Y)
+    assert correct[0] == out[0]
+    assert correct[1] == out[1]
+    assert correct[2] == out[2]
+    assert correct[3] == out[3]