Merge pull request #15 from bradendubois/develop

Develop

API.py

@@ -107,7 +107,7 @@ def p(self, y: set, x: set) -> float:
 
         return result
 
-    def joint_distribution_table(self) -> list:
+    def joint_distribution_table(self) -> ConditionalProbabilityTable:
         """
         Compute a joint distribution table across the entire model loaded.
         @return: A list of tuples, (Outcomes, P), where Outcomes is a unique set of Outcome objects for the model, and
@@ -117,12 +117,8 @@ def joint_distribution_table(self) -> list:
 
         if self._print_result:
             keys = sorted(self._cg.variables.keys())
-            rows = [[",".join(map(str, outcomes)), [], p] for outcomes, p in result]
-            rows.append(["Total:", [], sum(map(lambda r: r[1], result))])
-            cpt = ConditionalProbabilityTable(Variable(",".join(keys), [], []), [], rows)
-
             self._output.result(f"Joint Distribution Table for: {','.join(keys)}")
-            self._output.result(f"{cpt}")
+            self._output.result(f"{result}")
 
         return result
 

src/api/backdoor_paths.py

@@ -11,15 +11,15 @@ def api_backdoor_paths_parse(query: str) -> (set, set):
         of the arrow, and the third as all vertices are the right of the bar, respectively.
     """
     def clean(x):
-        return set(map(lambda y: y.strip(), x.strip().split(" ")))
+        return set(map(lambda y: y.strip(), x.strip().split(",")))
 
     l, r = query.split("->")
 
     if "|" in r:
         s = r.split("|")
         r, dcf = clean(s[0]), clean(s[1])
     else:
-        r, dcf = clean(r), {}
+        r, dcf = clean(r), set()
 
     return {
         "src": clean(l),

src/api/joint_distribution_table.py

@@ -1,9 +1,10 @@
 from itertools import product
 from src.probability.structures.CausalGraph import CausalGraph
-from src.probability.structures.VariableStructures import Outcome
+from src.probability.structures.ConditionalProbabilityTable import ConditionalProbabilityTable
+from src.probability.structures.VariableStructures import Outcome, Variable
 
 
-def api_joint_distribution_table(cg: CausalGraph) -> list:
+def api_joint_distribution_table(cg: CausalGraph) -> ConditionalProbabilityTable:
     """
     Compute and return a joint distribution table for the given model.
     @param cg: A CausalGraph to compute the JDT for.
@@ -17,4 +18,9 @@ def api_joint_distribution_table(cg: CausalGraph) -> list:
         outcomes = {Outcome(x, cross[i]) for i, x in enumerate(sorted_keys)}
         results.append((outcomes, cg.probability_query(outcomes, set())))
 
-    return results
+    keys = sorted(cg.variables.keys())
+    rows = [[",".join(map(str, outcomes)), [], p] for outcomes, p in results]
+    rows.append(["Total:", [], sum(map(lambda r: r[1], results))])
+    cpt = ConditionalProbabilityTable(Variable(",".join(keys), [], []), [], rows)
+
+    return cpt

src/graphs/full/test.json

@@ -0,0 +1,25 @@
+{
+  "name": "Two-Variable Test",
+  "model": {
+
+    "Y": {
+      "latent": true,
+      "outcomes": ["y", "~y"],
+      "table": [
+        ["y", 0.6],
+        ["~y", 0.4]
+      ]
+    },
+
+    "X":{
+      "outcomes": ["x", "~x"],
+      "parents": ["Y"],
+      "table": [
+        ["x", "y", 0.7],
+        ["x", "~y", 0.2],
+        ["~x", "y", 0.3],
+        ["~x", "~y", 0.8]
+      ]
+    }
+  }
+}

src/probability/structures/BackdoorController.py

@@ -111,7 +111,7 @@ def get_backdoor_paths(cur: str, path: list, path_list: list, previous="up") ->
         backdoor_paths = get_backdoor_paths(s, [], [])
 
         # Filter out the paths that don't "enter" x; see the definition of a backdoor path
-        return list(filter(lambda l: l[0] in self.graph.children(l[1]), backdoor_paths))
+        return list(filter(lambda l: l[0] in self.graph.children(l[1]) and l[1] != t, backdoor_paths))
 
     def all_dcf_sets(self, src: set, dst: set) -> list:
         """

src/probability/structures/Probability_Engine.py

@@ -165,7 +165,7 @@ def _compute(self, head: list, body: list, depth=0) -> float:
                 result_1 = self._compute(child, head + new_body, depth+1)
                 result_2 = self._compute(head, new_body, depth+1)
                 result_3 = self._compute(child, new_body, depth+1)
-                if result_3 == 0:       # Avoid dividing by 0!
+                if result_3 == 0:       # Avoid dividing by 0! coverage: skip
                     self.output.detail(f"{str_3} = 0, therefore the result is 0.", x=depth)
                     return 0
 
@@ -190,33 +190,32 @@ def _compute(self, head: list, body: list, depth=0) -> float:
         if missing_parents:
             self.output.detail("Attempting application of Jeffrey's Rule", x=depth)
 
-            # Try an approach beginning with each missing parent
-            for missing_parent in missing_parents:
+        for missing_parent in missing_parents:
 
-                try:
-                    # Add one parent back in and recurse
-                    parent_outcomes = self.outcomes[missing_parent]
+            try:
+                # Add one parent back in and recurse
+                parent_outcomes = self.outcomes[missing_parent]
 
-                    # Consider the missing parent and sum every probability involving it
-                    total = 0.0
-                    for parent_outcome in parent_outcomes:
+                # Consider the missing parent and sum every probability involving it
+                total = 0.0
+                for parent_outcome in parent_outcomes:
 
-                        as_outcome = Outcome(missing_parent, parent_outcome)
+                    as_outcome = Outcome(missing_parent, parent_outcome)
 
-                        self.output.detail(p_str(head, [as_outcome] + body), "*", p_str([as_outcome], body), x=depth)
+                    self.output.detail(p_str(head, [as_outcome] + body), "*", p_str([as_outcome], body), x=depth)
 
-                        result_1 = self._compute(head, [as_outcome] + body, depth+1)
-                        result_2 = self._compute([as_outcome], body, depth+1)
-                        outcome_result = result_1 * result_2
+                    result_1 = self._compute(head, [as_outcome] + body, depth+1)
+                    result_2 = self._compute([as_outcome], body, depth+1)
+                    outcome_result = result_1 * result_2
 
-                        total += outcome_result
+                    total += outcome_result
 
-                    self.output.detail(rep, "=", total, x=depth)
-                    self._store_computation(rep, total)
-                    return total
+                self.output.detail(rep, "=", total, x=depth)
+                self._store_computation(rep, total)
+                return total
 
-                except ProbabilityException:    # coverage: skip
-                    self.output.detail("Failed to resolve by Jeffrey's Rule", x=depth)
+            except ProbabilityException:    # coverage: skip
+                self.output.detail("Failed to resolve by Jeffrey's Rule", x=depth)
 
         ###############################################
         #            Interventions / do(X)            #

src/util/helpers.py

@@ -1,6 +1,8 @@
 from itertools import chain, combinations
 from typing import Iterator
 
+from src.config.settings import Settings
+
 
 def power_set(variable_list: list or set, allow_empty_set=True) -> Iterator[any]:
     """
@@ -49,3 +51,13 @@ def p_str(lhs: list, rhs: list) -> str:
         return f'P({", ".join(map(str, lhs))})'
 
     return f'P({", ".join(map(str, lhs))} | {", ".join(map(str, rhs))})'
+
+
+def within_precision(a: float, b: float) -> bool:
+    """
+    Check whether two values differ by an amount less than some number of digits of precision
+    @param a: The first value
+    @param b: The second value
+    @return: True if the values are within the margin of error acceptable, False otherwise
+    """
+    return abs(a - b) < 1 / (10 ** Settings.regression_levels_of_precision)

src/validation/backdoors/backdoor_path_tests.py

@@ -87,7 +87,7 @@ def backdoor_tests(graph_location: Path) -> (bool, str):
         success, msg = model_backdoor_validation(bc, yml_test_data)
         print_test_result(success, msg if not success else f"All tests in {test_file}, {graph_filename} passed")
 
-        if not success:
+        if not success:     # coverage: skip
             all_successful = False
 
     return all_successful, "[Backdoor module passed]" if all_successful else "[Backdoor module encountered errors]"

src/validation/backdoors/test_files/xi_xj.yml

@@ -26,6 +26,6 @@ tests:
     expect: false
 
   - type: independence
-    src: [ Xi ]
+    src: [ X5 ]
     dst: [ X2 ]
     expect: false

src/validation/inference/inference_tests.py

@@ -2,25 +2,15 @@
 from pathlib import Path
 from yaml import safe_load as load
 
-from src.config.settings import Settings
 from src.probability.structures.CausalGraph import CausalGraph, Outcome
-from src.util.ProbabilityExceptions import *
+from src.util.helpers import within_precision
 from src.util.ModelLoader import parse_model, parse_outcomes_and_interventions
+from src.util.ProbabilityExceptions import *
 from src.validation.test_util import print_test_result
 
 test_file_directory = Path(dirname(abspath(__file__))) / "test_files"
 
 
-def within_precision(a: float, b: float) -> bool:
-    """
-    Check whether two values differ by an amount less than some number of digits of precision
-    @param a: The first value
-    @param b: The second value
-    @return: True if the values are within the margin of error acceptable, False otherwise
-    """
-    return abs(a - b) < 1 / (10 ** Settings.regression_levels_of_precision)
-
-
 def model_inference_validation(cg: CausalGraph) -> (bool, str):
     """
     Validate all distributions in the given Causal Graph