Merge pull request #63 from causy-dev/d-separation

add edge cases in check for d-separations
causy-dev · Nov 6, 2024 · 2930320 · 2930320
2 parents b4ff852 + 2bae91f
commit 2930320
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 1 deletion.
diff --git a/causy/graph.py b/causy/graph.py
@@ -328,9 +328,11 @@ def are_nodes_d_separated(
 
         # check whether there is an open path on which all colliders are in the conditioning set and all non-colliders are not in the conditioning set
         list_of_results_for_paths = []
+        if list(self.all_paths_on_underlying_undirected_graph(u, v)) == []:
+            return True
         for path in self.all_paths_on_underlying_undirected_graph(u, v):
             if len(path) == 2:
-                list_of_results_for_paths.append(False)
+                is_path_blocked = False
 
             for i in range(1, len(path) - 1):
                 is_path_blocked = False

diff --git a/causy/sensitivity_analysis/internal_consistency.py b/causy/sensitivity_analysis/internal_consistency.py
diff --git a/tests/test_internal_consistency.py b/tests/test_internal_consistency.py
diff --git a/tests/test_path_classification.py b/tests/test_path_classification.py
@@ -231,3 +231,29 @@ def test_are_nodes_d_separated_3(self):
         graph.add_directed_edge(node3, node2, {"test": "test"})
         graph.add_directed_edge(node1, node3, {"test": "test"})
         self.assertFalse(graph.are_nodes_d_separated(node1, node3, []))
+
+    def test_are_nodes_d_separated_5(self):
+        new_graph_manager = GraphManager
+        new_graph_manager.__bases__ = (
+            GraphBaseAccessMixin,
+            DirectedEdge.GraphAccessMixin,
+        )
+        graph = new_graph_manager()
+        node1 = graph.add_node("test1", [1, 2, 3])
+        node2 = graph.add_node("test2", [1, 2, 3])
+        node3 = graph.add_node("test3", [1, 2, 3])
+        graph.add_directed_edge(node1, node2, {"test": "test"})
+        self.assertTrue(graph.are_nodes_d_separated(node1, node3, []))
+
+    def test_are_nodes_d_separated_6(self):
+        new_graph_manager = GraphManager
+        new_graph_manager.__bases__ = (
+            GraphBaseAccessMixin,
+            DirectedEdge.GraphAccessMixin,
+        )
+        graph = new_graph_manager()
+        node1 = graph.add_node("test1", [1, 2, 3])
+        node2 = graph.add_node("test2", [1, 2, 3])
+        node3 = graph.add_node("test3", [1, 2, 3])
+        graph.add_directed_edge(node1, node2, {"test": "test"})
+        self.assertTrue(graph.are_nodes_d_separated(node1, node3, [node2]))