Merge pull request #179 from CITCOM-project/JSON_treatment_var

JSON concrete tests

Author: christopher-wild

causal_testing/json_front/json_class.py

@@ -20,6 +20,7 @@
 from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Meta, Output
+from causal_testing.testing.base_test_case import BaseTestCase
 from causal_testing.testing.causal_test_case import CausalTestCase
 from causal_testing.testing.causal_test_engine import CausalTestEngine
 from causal_testing.testing.estimators import Estimator
@@ -46,7 +47,7 @@ class JsonUtility:
 
     def __init__(self, output_path: str, output_overwrite: bool = False):
         self.input_paths = None
-        self.variables = None
+        self.variables = {"inputs": {}, "outputs": {}, "metas": {}}
         self.data = []
         self.test_plan = None
         self.scenario = None
@@ -66,13 +67,71 @@ def set_paths(self, json_path: str, dag_path: str, data_paths: str):
     def setup(self, scenario: Scenario):
         """Function to populate all the necessary parts of the json_class needed to execute tests"""
         self.scenario = scenario
+        self._get_scenario_variables()
         self.scenario.setup_treatment_variables()
         self.causal_specification = CausalSpecification(
             scenario=self.scenario, causal_dag=CausalDAG(self.input_paths.dag_path)
         )
         self._json_parse()
         self._populate_metas()
 
+    def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False, mutates: dict = None):
+        """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.
+        """
+        failures = 0
+        for test in self.test_plan["tests"]:
+            if "skip" in test and test["skip"]:
+                continue
+            test["estimator"] = estimators[test["estimator"]]
+            if "mutations" in test:
+                abstract_test = self._create_abstract_test_case(test, mutates, effects)
+
+                concrete_tests, dummy = abstract_test.generate_concrete_tests(5, 0.05)
+                failures = self._execute_tests(concrete_tests, test, f_flag)
+                msg = (
+                    f"Executing test: {test['name']}\n"
+                    + "abstract_test\n"
+                    + f"{abstract_test}\n"
+                    + f"{abstract_test.treatment_variable.name},{abstract_test.treatment_variable.distribution}\n"
+                    + f"Number of concrete tests for test case: {str(len(concrete_tests))}\n"
+                    + f"{failures}/{len(concrete_tests)} failed for {test['name']}"
+                )
+                self._append_to_file(msg, logging.INFO)
+            else:
+                outcome_variable = next(
+                    iter(test["expected_effect"])
+                )  # Take first key from dictionary of expected effect
+                base_test_case = BaseTestCase(
+                    treatment_variable=self.variables["inputs"][test["treatment_variable"]],
+                    outcome_variable=self.variables["outputs"][outcome_variable],
+                )
+
+                causal_test_case = CausalTestCase(
+                    base_test_case=base_test_case,
+                    expected_causal_effect=effects[test["expected_effect"][outcome_variable]],
+                    control_value=test["control_value"],
+                    treatment_value=test["treatment_value"],
+                    estimate_type=test["estimate_type"],
+                )
+                if self._execute_test_case(causal_test_case=causal_test_case, test=test, f_flag=f_flag):
+                    result = "failed"
+                else:
+                    result = "passed"
+
+                msg = (
+                    f"Executing concrete test: {test['name']} \n"
+                    + f"treatment variable: {test['treatment_variable']} \n"
+                    + f"outcome_variable = {outcome_variable} \n"
+                    + f"control value = {test['control_value']}, treatment value = {test['treatment_value']} \n"
+                    + f"result - {result}"
+                )
+                self._append_to_file(msg, logging.INFO)
+
     def _create_abstract_test_case(self, test, mutates, effects):
         assert len(test["mutations"]) == 1
         abstract_test = AbstractCausalTestCase(
@@ -81,7 +140,7 @@ def _create_abstract_test_case(self, test, mutates, effects):
             treatment_variable=next(self.scenario.variables[v] for v in test["mutations"]),
             expected_causal_effect={
                 self.scenario.variables[variable]: effects[effect]
-                for variable, effect in test["expectedEffect"].items()
+                for variable, effect in test["expected_effect"].items()
             },
             effect_modifiers={self.scenario.variables[v] for v in test["effect_modifiers"]}
             if "effect_modifiers" in test
@@ -91,35 +150,8 @@ def _create_abstract_test_case(self, test, mutates, effects):
         )
         return abstract_test
 
-    def generate_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.
-        """
+    def _execute_tests(self, concrete_tests, test, f_flag):
         failures = 0
-        for test in self.test_plan["tests"]:
-            if "skip" in test and test["skip"]:
-                continue
-            abstract_test = self._create_abstract_test_case(test, mutates, effects)
-
-            concrete_tests, dummy = abstract_test.generate_concrete_tests(5, 0.05)
-            failures = self._execute_tests(concrete_tests, estimators, test, f_flag)
-            msg = (
-                f"Executing test: {test['name']} \n"
-                + "abstract_test \n"
-                + f"{abstract_test} \n"
-                + f"{abstract_test.treatment_variable.name},{abstract_test.treatment_variable.distribution} \n"
-                + f"Number of concrete tests for test case: {str(len(concrete_tests))} \n"
-                + f"{failures}/{len(concrete_tests)} failed for {test['name']}"
-            )
-            self._append_to_file(msg, logging.INFO)
-
-    def _execute_tests(self, concrete_tests, estimators, test, f_flag):
-        failures = 0
-        test["estimator"] = estimators[test["estimator"]]
         if "formula" in test:
             self._append_to_file(f"Estimator formula used for test: {test['formula']}")
         for concrete_test in concrete_tests:
@@ -161,15 +193,13 @@ def _execute_test_case(self, causal_test_case: CausalTestCase, test: Iterable[Ma
         :rtype: bool
         """
         failed = False
-
         causal_test_engine, estimation_model = self._setup_test(causal_test_case, test)
         causal_test_result = causal_test_engine.execute_test(
             estimation_model, causal_test_case, 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.test_value.value} <  "
@@ -214,7 +244,6 @@ def _setup_test(self, causal_test_case: CausalTestCase, test: Mapping) -> tuple[
         }
         if "formula" in test:
             estimator_kwargs["formula"] = test["formula"]
-
         estimation_model = test["estimator"](**estimator_kwargs)
         return causal_test_engine, estimation_model
 
@@ -226,10 +255,18 @@ def _append_to_file(self, line: str, log_level: int = None):
         is possible to use the inbuilt logging level variables such as logging.INFO and logging.WARNING
         """
         with open(self.output_path, "a", encoding="utf-8") as f:
-            f.write(line)
+            f.write(line + "\n")
         if log_level:
             logger.log(level=log_level, msg=line)
 
+    def _get_scenario_variables(self):
+        for input_var in self.scenario.inputs():
+            self.variables["inputs"][input_var.name] = input_var
+        for output_var in self.scenario.outputs():
+            self.variables["outputs"][output_var.name] = output_var
+        for meta_var in self.scenario.metas():
+            self.variables["metas"][meta_var.name] = meta_var
+
     @staticmethod
     def check_file_exists(output_path: Path, overwrite: bool):
         """Method that checks if the given path to an output file already exists. If overwrite is true the check is

docs/source/frontends/json_front_end.rst

@@ -21,17 +21,28 @@ Use case specific information is also declared here such as the paths to the rel
 
 causal_tests.json
 -----------------
-`examples/poisson/causal_tests.json <https://github.com/CITCOM-project/CausalTestingFramework/blob/main/examples/poisson/causal_tests.json>`_ contains python code written by the user to implement scenario specific features
-is the JSON file that allows for the easy specification of multiple causal tests.
+`examples/poisson/causal_tests.json <https://github.c#om/CITCOM-project/CausalTestingFramework/blob/main/examples/poisson/causal_tests.json>`_ contains python code written by the user to implement scenario specific features
+is the JSON file that allows for the easy specification of multiple causal tests. Tests can be specified two ways; firstly by specifying a mutation lke in the example tests with the following structure:
 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
 
+#. name
+#. mutations
+#. estimator
+#. estimate_type
+#. effect_modifiers
+#. expected_effects
+#. skip: boolean that if set true the test won't be executed and will be skipped
+
+The second method of specifying a test is to specify the test in a concrete form with the following structure:
+
+#. name
+#. treatment_variable
+#. control_value
+#. treatment_value
+#. estimator
+#. estimate_type
+#. expected_effect
+#. skip
 
 Run Commands
 ------------

examples/poisson/README.md

@@ -6,6 +6,6 @@ To run this case study:
 1. Ensure all project dependencies are installed by running `pip install .` in the top level directory
    (instructions are provided in the project README).
 2. Change directory to `causal_testing/examples/poisson`.
-3. Run the command `python test_run_causal_tests.py --data_path data.csv --dag_path dag.dot --json_path causal_tests.json`
+3. Run the command `python example_run_causal_tests.py --data_path data.csv --dag_path dag.dot --json_path causal_tests.json`
 
 This should print a series of causal test results and produce two CSV files. `intensity_num_shapes_results_random_1000.csv` corresponds to table 1, and `width_num_shapes_results_random_1000.csv` relates to our findings regarding the relationship of width and `P_u`.