Merge pull request #23 from bradendubois/develop

feat: Type hinting / API improvements / wiki

.github/workflows/update_wiki.yml

@@ -5,7 +5,7 @@ on:
   push:
     # Trigger only when wiki directory changes
     paths:
-      - 'doc/**'
+      - 'wiki/**'
 
     # Trigger only on main/beta
     branches: [ main ]
@@ -19,7 +19,7 @@ jobs:
     - name: Push Wiki Changes
       uses: Andrew-Chen-Wang/github-wiki-action@v2
       env:
-        WIKI_DIR: doc/
+        WIKI_DIR: docs/
         GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         GH_MAIL: ${{ secrets.EMAIL }}
         GH_NAME: ${{ github.repository_owner }}

README.md

@@ -1,11 +1,11 @@
 <h1 align="center" style="border-bottom: none;">do-calculus</h1>
 <h3 align="center">A Python implementation of the <i>do-calculus</i> of Judea Pearl et. al.</h3>
 <p align="center">
-    <a href="https://github.com/bradendubois/probability-code/actions?query=workflow%3ATest+branch%3Amain">
-        <img alt="Test Workflows" src="https://github.com/bradendubois/probability-code/workflows/Test and Release/badge.svg">
+    <a href="https://github.com/bradendubois/do-calculus/actions?query=workflow%3ATest+branch%3Amain">
+        <img alt="Test Workflows" src="https://github.com/bradendubois/do-calculus/workflows/Test and Release/badge.svg">
     </a>
-    <a href='https://coveralls.io/github/bradendubois/probability-code?branch=main'>
-        <img src='https://coveralls.io/repos/github/bradendubois/probability-code/badge.svg?branch=main' alt='Coverage Status' />
+    <a href='https://coveralls.io/github/bradendubois/do-calculus?branch=main'>
+        <img src='https://coveralls.io/repos/github/bradendubois/do-calculus/badge.svg?branch=main' alt='Coverage Status' />
     </a>
     <a href="https://pypi.org/project/do-calculus/">
         <img alt="" src="https://pypip.in/v/do-calculus/badge.svg">
@@ -20,11 +20,11 @@
 
 ## Resources
 
-* **Documentation / Wiki**: [github.com/bradendubois/probability-code/wiki](https://github.com/bradendubois/probability-code/wiki)
-* **Source Code**: [github.com/bradendubois/probability-code](https://github.com/bradendubois/probability-code)
+* **Documentation / Wiki**: [github.com/bradendubois/do-calculus/wiki](https://github.com/bradendubois/do-calculus/wiki)
+* **Source Code**: [github.com/bradendubois/do-calculus](https://github.com/bradendubois/do-calculus)
 * **PyPI**: [pypi.org/project/do-calculus/](https://pypi.org/project/do-calculus/)
-* **Releases**: [github.com/bradendubois/probability-code/releases](https://github.com/bradendubois/probability-code/releases)
-* **Bug reports**: [github.com/bradendubois/probability-code/issues](https://github.com/bradendubois/probability-code/issues)
+* **Releases**: [github.com/bradendubois/do-calculus/releases](https://github.com/bradendubois/do-calculus/releases)
+* **Bug reports**: [github.com/bradendubois/do-calculus/issues](https://github.com/bradendubois/do-calculus/issues)
 * **Contact**: [[email protected]](mailto:[email protected])
 
-See the [wiki](https://github.com/bradendubois/probability-code/wiki) to get started.
+See the [wiki](https://github.com/bradendubois/do-calculus/wiki) to get started.

do/graphs/dataset_generator/model_generator.py → debug/dataset_generator/model_generator.py

@@ -26,7 +26,7 @@
 except ValueError:
     print("Could not convert", argv[1], "to int; defaulting to", N)
 
-destination_directory = Path(".", argv[2])
+destination_directory = Path("", argv[2])
 
 if not destination_directory.is_dir():
     print("Cannot resolve", destination_directory)

do/API.py

@@ -2,18 +2,19 @@
 #                   probability-code API                  #
 ###########################################################
 
-from typing import Union
+from typing import Collection, List, Optional, Set, Union
 from pathlib import Path
 
 from .api.backdoor_paths import api_backdoor_paths
 from .api.deconfounding_sets import api_deconfounding_sets
 from .api.joint_distribution_table import api_joint_distribution_table
 from .api.probability_query import api_probability_query
 
-from .probability.structures.BackdoorController import BackdoorController
-from .probability.structures.CausalGraph import CausalGraph
-from .probability.structures.ConditionalProbabilityTable import ConditionalProbabilityTable
-from .probability.structures.VariableStructures import Variable
+from .structures.BackdoorController import BackdoorController
+from .structures.CausalGraph import CausalGraph
+from .structures.ConditionalProbabilityTable import ConditionalProbabilityTable
+from .structures.Types import Vertex, Vertices
+from .structures.VariableStructures import Outcome, Intervention
 
 from .util.ModelLoader import parse_model
 from .util.OutputLogger import OutputLogger
@@ -32,6 +33,9 @@ def __init__(self, model: dict or None, print_detail=False, print_result=False,
             an argument to log_fd, or can be done later with a call to set_log_fd.
         @param log_fd: An open file descriptor to write to, if log_details is enabled.
         """
+        self._print_result = print_result
+        self._output = OutputLogger(print_result, print_detail, log, log_fd)
+
         if model:
             self.load_model(model)
 
@@ -40,9 +44,6 @@ def __init__(self, model: dict or None, print_detail=False, print_result=False,
             self._g = None
             self._bc = None
 
-        self._print_result = print_result
-        self._output = OutputLogger(print_result, print_detail, log, log_fd)
-
     ################################################################
     #                       API Modifications                      #
     ################################################################
@@ -93,27 +94,31 @@ def set_log_fd(self, log_fd):
     #                         Distributions                        #
     ################################################################
 
-    def p(self, y: set, x: set) -> float:
+    def p(self, y: Collection[Outcome], x: Collection[Union[Outcome, Intervention]]) -> Optional[float]:
         """
         Compute a probability query of Y, given X.
         @param y: Head of query; a set of Outcome objects
         @param x: Body of query; a set of Outcome and/or Variable objects
         @return: The probability of P(Y | X), in the range [0.0, 1.0]
         @raise ProbabilityException when the given probability cannot be computed, such as an invalid Outcome
         """
-        # All deconfounding is handled by the CG
-        result = api_probability_query(self._cg, y, x)
-        self._output.result(result)
+        try:
+            # All deconfounding is handled by the CG
+            result = api_probability_query(self._cg, y, x)
+            self._output.result(result)
+            return result
 
-        return result
+        except AssertionError as e:
+            self._output.detail(e)
+            return None
 
     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
             P is the corresponding probability.
         """
-        result = api_joint_distribution_table(self._cg)
+        result: ConditionalProbabilityTable = api_joint_distribution_table(self._cg)
 
         if self._print_result:
             keys = sorted(self._cg.variables.keys())
@@ -126,7 +131,7 @@ def joint_distribution_table(self) -> ConditionalProbabilityTable:
     #               Pathfinding (Backdoor Controller)              #
     ################################################################
 
-    def backdoor_paths(self, src: set, dst: set, dcf: set) -> list:
+    def backdoor_paths(self, src: Vertices, dst: Vertices, dcf: Optional[Vertices]) -> List[Path]:
         """
         Find all the "backdoor paths" between two sets of variables.
         @param src: A set of (string) vertices defined in the loaded model, which will be the source to begin searching
@@ -145,9 +150,10 @@ def backdoor_paths(self, src: set, dst: set, dcf: set) -> list:
                 for left, right in zip(path[:-1], path[1:]):
                     print(left, "<-" if right in self._g.parents(left) else "->", end=" ")
                 print(path[-1])
+
         return result
 
-    def deconfounding_sets(self, src: set, dst: set) -> list:
+    def deconfounding_sets(self, src: set, dst: set) -> List[Set[str]]:
         """
         Find the sets of vertices in the loaded model that are sufficient at blocking all backdoor paths from all
         vertices in src to any vertices in dst

do/__main__.py

@@ -12,7 +12,7 @@
 #   there being path issues depending on the working directory
 
 
-def main(graph_location=Path(".", "src", "graphs", "full")):
+def main(graph_location=Path(".", "graphs")):
     """
     Run an interactive IO prompt allowing full use of the causality software.
     @param graph_location: A string of the path from the working directory to a directory of graphs

do/api/backdoor_paths.py

@@ -1,9 +1,10 @@
-from itertools import product
+from typing import Collection, Dict, List, Optional
 
-from ..probability.structures.BackdoorController import BackdoorController
+from ..structures.BackdoorController import BackdoorController
+from ..structures.Types import Path, Vertices
 
 
-def api_backdoor_paths_parse(query: str) -> (set, set):
+def api_backdoor_paths_parse(query: str) -> Dict[str, Collection[str]]:
     """
     Convert a given query string into a pair of sets to compute all backdoor paths between
     @param query: A string of the form "X, Y, Z -> A, B, C" or "X, Y, Z -> A, B, C | I, J, K"
@@ -28,7 +29,7 @@ def clean(x):
     }
 
 
-def api_backdoor_paths(bc: BackdoorController, src: set, dst: set, dcf: set) -> list:
+def api_backdoor_paths(bc: BackdoorController, src: Vertices, dst: Vertices, dcf: Optional[Vertices]) -> List[Path]:
     """
     Compute and return all the backdoor paths from any vertex in src to any vertex in dst
     @param bc: A Backdoor Controller with a graph conforming to the given source and destination sets.
@@ -42,8 +43,4 @@ def api_backdoor_paths(bc: BackdoorController, src: set, dst: set, dcf: set) ->
         list containing each vertex (as a string) from the source vertex to the destination vertex, with dcf acting as
         a deconfounding set.
     """
-    # TODO Add a method in Backdoor Controller that can return all paths immediately
-    paths = []
-    for s, t in product(src, dst):
-        paths += bc.backdoor_paths_pair(s, t, dcf)
-    return paths
+    return bc.backdoor_paths(src, dst, dcf)

do/api/deconfounding_sets.py

@@ -1,7 +1,10 @@
-from ..probability.structures.BackdoorController import BackdoorController
+from typing import Collection, Dict, List, Set
 
+from ..structures.BackdoorController import BackdoorController
+from ..structures.Types import Vertices
 
-def api_deconfounding_sets_parse(query: str) -> (set, set):
+
+def api_deconfounding_sets_parse(query: str) -> Dict[str, Collection[str]]:
     """
     Convert a given query string into a pair of sets to find all sufficient deconfounding sets between.
     @param query: A string of the form "X, Y, Z -> A, B, C"
@@ -19,7 +22,7 @@ def clean(x):
     }
 
 
-def api_deconfounding_sets(bc: BackdoorController, src: set, dst: set) -> list:
+def api_deconfounding_sets(bc: BackdoorController, src: Vertices, dst: Vertices) -> List[Set[str]]:
     """
     Compute and return all the backdoor paths from any vertex in src to any vertex is dst
     @param bc: A Backdoor Controller with a graph conforming to the given source and destination sets.

do/api/joint_distribution_table.py

@@ -1,8 +1,8 @@
 from itertools import product
 
-from ..probability.structures.CausalGraph import CausalGraph
-from ..probability.structures.ConditionalProbabilityTable import ConditionalProbabilityTable
-from ..probability.structures.VariableStructures import Outcome, Variable
+from ..structures.CausalGraph import CausalGraph
+from ..structures.ConditionalProbabilityTable import ConditionalProbabilityTable
+from ..structures.VariableStructures import Outcome, Variable
 
 
 def api_joint_distribution_table(cg: CausalGraph) -> ConditionalProbabilityTable:

do/api/probability_query.py

@@ -1,8 +1,10 @@
-from ..probability.structures.CausalGraph import CausalGraph
-from ..probability.structures.VariableStructures import parse_outcomes_and_interventions
+from typing import Collection, Dict, Union
 
+from ..structures.CausalGraph import CausalGraph
+from ..structures.VariableStructures import Outcome, Intervention, parse_outcomes_and_interventions
 
-def api_probability_query_parse(query: str) -> (tuple, tuple):
+
+def api_probability_query_parse(query: str) -> Dict[str, Collection[str]]:
     """
     Parse a query string into Outcome and Intervention structures.
     @param query: A string of the form "Y=y, X=x | W=w", or just "Y=y, X=x"
@@ -21,7 +23,7 @@ def api_probability_query_parse(query: str) -> (tuple, tuple):
     }
 
 
-def api_probability_query(cg: CausalGraph, y: set, x: set) -> float:
+def api_probability_query(cg: CausalGraph, y: Collection[Outcome], x: Collection[Union[Outcome, Intervention]]) -> float:
     """
     Compute a probability query for the currently loaded causal graph.
     @param cg: A Causal Graph containing variables, distributions, etc.

do/config/config_manager.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from yaml import safe_load as load, dump
 
-from ..config.primary_configuration import *
+from ..config.primary_configuration import primary_config_file
 
 path = Path(".", "config.yml")
 

do/config/generate_config_docs.py

@@ -4,7 +4,7 @@
 
 from pathlib import Path
 
-from .primary_configuration import *
+from .primary_configuration import primary_config_file
 
 documentation_file = Path(".", "doc", "Configuration.md")
 

do/config/primary_configuration.py

@@ -61,4 +61,3 @@
         }]
     }
 ]
-