Merge pull request #67 from CITCOM-project/json_poisson_example

Json front end

Author: christopher-wild

causal_testing/data_collection/data_collector.py

@@ -43,7 +43,7 @@ def filter_valid_data(self, data: pd.DataFrame, check_pos: bool = True) -> pd.Da
         solver = z3.Solver()
         for c in self.scenario.constraints:
             solver.assert_and_track(c, f"background: {c}")
-        sat = []
+        sat = list()
         unsat_core = None
         for _, row in data.iterrows():
             solver.push()

causal_testing/json_front/README.md

@@ -0,0 +1,34 @@
+# JSON Causal Testing Framework Frontend
+
+The JSON frontend allows Causal Tests and parameters to be specified in JSON to allow for tests to be quickly written
+whilst retaining the flexibility of the Causal Testing Framework (CTF). 
+
+An example is provided in `examples/poisson` which will be walked through in this README to better understand
+the framework
+
+`examples/poisson/run_causal_tests.py` contains python code written by the user to implement scenario specific features
+such as:
+1. Custom Estimators
+2. Causal Variable specification
+3. Causal test case outcomes
+4. Meta constraint functions
+5. Mapping JSON distributions, effects, and estimators to python objects
+
+Use case specific information is also declared here such as the paths to the relevant files needed for the tests.
+
+`examples/poisson/causal_tests.json` is the JSON file that allows for the easy specification of multiple causal tests.
+Each test requires:
+1. Test name
+2. Mutations
+3. Estimator
+4. Estimate_type
+5. Effect modifiers
+6. Expected effects
+7. Skip: boolean that if set true the test won't be executed and will be skipped
+
+To run the JSON frontend example from the root directory of the project, use 
+`python examples/poisson/run_causal_tests.py"`
+
+A failure flag `-f` can be specified to stop the framework running if a test is failed
+`python examples/poisson/run_causal_tests.py -f"`
+

causal_testing/json_front/__init__.py

@@ -0,0 +1,210 @@
+from pathlib import Path
+
+from abc import ABC
+import json
+from fitter import Fitter, get_common_distributions
+import pandas as pd
+import scipy
+
+from causal_testing.specification.variable import Input, Output, Meta
+from causal_testing.specification.scenario import Scenario
+from causal_testing.specification.causal_dag import CausalDAG
+from causal_testing.specification.causal_specification import CausalSpecification
+from causal_testing.generation.abstract_causal_test_case import AbstractCausalTestCase
+from causal_testing.data_collection.data_collector import ObservationalDataCollector
+from causal_testing.testing.causal_test_engine import CausalTestEngine
+from causal_testing.testing.causal_test_case import CausalTestCase
+from causal_testing.testing.estimators import Estimator
+
+
+class JsonUtility(ABC):
+    """
+    The JsonUtility Class provides the functionality to use structured JSON to setup and run causal tests on the
+    CausalTestingFramework.
+
+    :attr {Path} json_path: Path to the JSON input file.
+    :attr {Path} dag_path: Path to the dag.dot file containing the Causal DAG.
+    :attr {Path} data_path: Path to the csv data file.
+    :attr {Input} inputs: Causal variables representing inputs.
+    :attr {Output} outputs: Causal variables representing outputs.
+    :attr {Meta} metas: Causal variables representing metavariables.
+    :attr {pd.DataFrame}: Pandas DataFrame containing runtime data.
+    :attr {dict} test_plan: Dictionary containing the key value pairs from the loaded json test plan.
+    :attr {Scenario} modelling_scenario:
+    :attr {CausalSpecification} causal_specification:
+    """
+
+    def __init__(self):
+        self.json_path = None
+        self.dag_path = None
+        self.data_path = None
+        self.inputs = None
+        self.outputs = None
+        self.metas = None
+        self.data = None
+        self.test_plan = None
+        self.modelling_scenario = None
+        self.causal_specification = None
+
+    def set_path(self, json_path: str, dag_path: str, data_path: str):
+        """
+            Takes a path of the directory containing all scenario specific files and creates individual paths for each file
+            :param json_path: string path representation to .json file containing test specifications
+            :param dag_path: string path representation to the .dot file containing the Causal DAG
+            :param data_path: string path representation to the data file
+            :returns:
+                - json_path -
+                - dag_path -
+                - data_path -
+            """
+        self.json_path = Path(json_path)
+        self.dag_path = Path(dag_path)
+        self.data_path = Path(data_path)
+
+    def set_variables(self, inputs: dict, outputs: dict, metas: dict, distributions: dict, populates: dict):
+        """ Populate the Causal Variables
+            :param inputs:
+            :param outputs:
+            :param metas:
+            :param distributions:
+            :param populates:
+        """
+        self.inputs = [Input(i['name'], i['type'], distributions[i['distribution']]) for i in
+                       inputs]
+        self.outputs = [Output(i['name'], i['type']) for i in outputs]
+        self.metas = [Meta(i['name'], i['type'], populates[i['populate']]) for i in
+                      metas] if metas else list()
+
+    def setup(self):
+        """ Function to populate all the necessary parts of the json_class needed to execute tests
+        """
+        self.modelling_scenario = Scenario(self.inputs + self.outputs + self.metas, None)
+        self.modelling_scenario.setup_treatment_variables()
+        self.causal_specification = CausalSpecification(scenario=self.modelling_scenario,
+                                                        causal_dag=CausalDAG(self.dag_path))
+        self._json_parse()
+        self._populate_metas()
+
+    def execute_tests(self, effects: dict, mutates: dict, estimators: dict, f_flag: bool):
+        """ Runs and evaluates each test case specified in the JSON input
+
+        :param effects: Dictionary mapping effect class instances to string representations.
+        :param mutates: Dictionary mapping mutation functions to string representations.
+        :param estimators: Dictionary mapping estimator classes to string representations.
+        :param f_flag: Failure flag that if True the script will stop executing when a test fails.
+        """
+        executed_tests = 0
+        failures = 0
+        for test in self.test_plan['tests']:
+            if "skip" in test and test['skip']:
+                continue
+
+            abstract_test = AbstractCausalTestCase(
+                scenario=self.modelling_scenario,
+                intervention_constraints=[mutates[v](k) for k, v in test['mutations'].items()],
+                treatment_variables={self.modelling_scenario.variables[v] for v in test['mutations']},
+                expected_causal_effect={self.modelling_scenario.variables[variable]: effects[effect] for
+                                        variable, effect
+                                        in
+                                        test["expectedEffect"].items()},
+                effect_modifiers={self.modelling_scenario.variables[v] for v in
+                                  test['effect_modifiers']} if "effect_modifiers" in test else {},
+                estimate_type=test['estimate_type']
+            )
+
+            concrete_tests, dummy = abstract_test.generate_concrete_tests(5, 0.05)
+            print(abstract_test)
+            print([(v.name, v.distribution) for v in abstract_test.treatment_variables])
+            print(len(concrete_tests))
+            for concrete_test in concrete_tests:
+                executed_tests += 1
+                failed = self._execute_test_case(concrete_test, estimators[test['estimator']], f_flag)
+                if failed:
+                    failures += 1
+
+        print(f"{failures}/{executed_tests} failed")
+
+    def _json_parse(self):
+        """Parse a JSON input file into inputs, outputs, metas and a test plan
+            :param distributions: dictionary of user defined scipy distributions
+            :param populates: dictionary of user defined populate functions
+        """
+        with open(self.json_path) as f:
+            self.test_plan = json.load(f)
+
+        self.data = pd.read_csv(self.data_path)
+
+    def _populate_metas(self):
+        """
+        Populate data with meta-variable values and add distributions to Causal Testing Framework Variables
+        """
+
+        for meta in self.metas:
+            meta.populate(self.data)
+
+        for var in self.metas + self.outputs:
+            f = Fitter(self.data[var.name], distributions=get_common_distributions())
+            f.fit()
+            (dist, params) = list(f.get_best(method="sumsquare_error").items())[0]
+            var.distribution = getattr(scipy.stats, dist)(**params)
+            print(var.name, f"{dist}({params})")
+
+    def _execute_test_case(self, causal_test_case: CausalTestCase, estimator: Estimator, f_flag: bool) -> bool:
+        """ Executes a singular test case, prints the results and returns the test case result
+        :param causal_test_case: The concrete test case to be executed
+        :param f_flag: Failure flag that if True the script will stop executing when a test fails.
+        :return: A boolean that if True indicates the causal test case passed and if false indicates the test case failed.
+        :rtype: bool
+        """
+        failed = False
+
+        causal_test_engine, estimation_model = self._setup_test(causal_test_case, estimator)
+        causal_test_result = causal_test_engine.execute_test(estimation_model,
+                                                             estimate_type=causal_test_case.estimate_type)
+
+        test_passes = causal_test_case.expected_causal_effect.apply(causal_test_result)
+
+        result_string = str()
+        if causal_test_result.ci_low() and causal_test_result.ci_high():
+            result_string = f"{causal_test_result.ci_low()} < {causal_test_result.ate} <  {causal_test_result.ci_high()}"
+        else:
+            result_string = causal_test_result.ate
+        if f_flag:
+            assert test_passes, f"{causal_test_case}\n    FAILED - expected {causal_test_case.expected_causal_effect}, " \
+                                f"got {result_string}"
+        if not test_passes:
+            failed = True
+            print(f"    FAILED - expected {causal_test_case.expected_causal_effect}, got {causal_test_result.ate}")
+        return failed
+
+    def _setup_test(self, causal_test_case: CausalTestCase, estimator: Estimator) -> tuple[CausalTestEngine, Estimator]:
+        """ Create the necessary inputs for a single test case
+        :param causal_test_case: The concrete test case to be executed
+        :returns:
+                - causal_test_engine - Test Engine instance for the test being run
+                - estimation_model - Estimator instance for the test being run
+        """
+        data_collector = ObservationalDataCollector(self.modelling_scenario, self.data_path)
+        causal_test_engine = CausalTestEngine(causal_test_case, self.causal_specification, data_collector)
+        minimal_adjustment_set = causal_test_engine.load_data(index_col=0)
+        treatment_vars = list(causal_test_case.treatment_input_configuration)
+        minimal_adjustment_set = minimal_adjustment_set - {v.name for v in treatment_vars}
+        estimation_model = estimator((list(treatment_vars)[0].name,),
+                                     [causal_test_case.treatment_input_configuration[v] for v in treatment_vars][0],
+                                     [causal_test_case.control_input_configuration[v] for v in treatment_vars][0],
+                                     minimal_adjustment_set,
+                                     (list(causal_test_case.outcome_variables)[0].name,),
+                                     causal_test_engine.scenario_execution_data_df,
+                                     effect_modifiers=causal_test_case.effect_modifier_configuration
+                                     )
+
+        self.add_modelling_assumptions(estimation_model)
+
+        return causal_test_engine, estimation_model
+
+    def add_modelling_assumptions(self, estimation_model: Estimator):
+        """ Optional abstract method where user functionality can be written to determine what assumptions are required
+        for specific test cases
+        :param estimation_model: estimator model instance for the current running test.
+        """
+        return

causal_testing/json_front/json_class.py

@@ -55,7 +55,6 @@ def to_dict(self):
             base_dict["ci_high"] = max(self.confidence_intervals)
         return base_dict
 
-
     def ci_low(self):
         """Return the lower bracket of the confidence intervals."""
         if not self.confidence_intervals:

causal_testing/testing/causal_test_outcome.py

@@ -21,6 +21,12 @@ Causal testing is a causal inference-driven framework for functional black-box t
 
    /autoapi/causal_testing/index
 
+.. toctree::
+   :maxdepth: 1
+   :caption: Examples
+
+   json_frontend
+
 Indices and tables
 ==================