Removed CausalSpecification

Author: jmafoster1

causal_testing/main.py

@@ -15,7 +15,6 @@
 from causal_testing.estimation.linear_regression_estimator import LinearRegressionEstimator
 from causal_testing.estimation.logistic_regression_estimator import LogisticRegressionEstimator
 from causal_testing.specification.causal_dag import CausalDAG
-from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Output
 from causal_testing.testing.base_test_case import BaseTestCase
@@ -106,7 +105,6 @@ def __init__(self, paths: CausalTestingPaths, ignore_cycles: bool = False, query
         self.data: Optional[pd.DataFrame] = None
         self.variables: Dict[str, Any] = {"inputs": {}, "outputs": {}, "metas": {}}
         self.scenario: Optional[Scenario] = None
-        self.causal_specification: Optional[CausalSpecification] = None
         self.test_cases: Optional[List[CausalTestCase]] = None
 
     def setup(self) -> None:
@@ -130,8 +128,11 @@ def setup(self) -> None:
         # Create variables from DAG
         self.create_variables()
 
-        # Create scenario and specification
-        self.create_scenario_and_specification()
+        # Create scenario
+        self.scenario = Scenario(
+            list(self.variables["inputs"].values()) + list(self.variables["outputs"].values()),
+            {self.query} if self.query else None,
+        )
 
         logger.info("Setup completed successfully")
 
@@ -187,15 +188,6 @@ def create_variables(self) -> None:
             if self.dag.in_degree(node_name) > 0:
                 self.variables["outputs"][node_name] = Output(name=node_name, datatype=dtype)
 
-    def create_scenario_and_specification(self) -> None:
-        """Create scenario and causal specification objects from loaded data."""
-        # Create scenario
-        all_variables = list(self.variables["inputs"].values()) + list(self.variables["outputs"].values())
-        self.scenario = Scenario(variables=all_variables)
-
-        # Create causal specification
-        self.causal_specification = CausalSpecification(scenario=self.scenario, causal_dag=self.dag)
-
     def load_tests(self) -> None:
         """
         Load and prepare test configurations from file.
@@ -315,7 +307,7 @@ def create_causal_test(self, test: dict, base_test: BaseTestCase) -> CausalTestC
             control_value=test.get("control_value"),
             adjustment_set=test.get(
                 "adjustment_set",
-                self.causal_specification.causal_dag.identification(base_test, self.scenario.hidden_variables()),
+                self.dag.identification(base_test, self.scenario.hidden_variables()),
             ),
             df=filtered_df,
             effect_modifiers=None,

causal_testing/specification/causal_dag.py

@@ -493,8 +493,7 @@ def identification(self, base_test_case: BaseTestCase, avoid_variables: set[Vari
 
         :param base_test_case: A base test case instance containing the outcome_variable and the
                                treatment_variable required for identification.
-        :param scenario: The modelling scenario relating to the tests
-
+        :param avoid_variables: Variables not to be adjusted for (e.g. hidden variables).
         :return: The smallest set of variables which can be adjusted for to obtain a causal
                  estimate as opposed to a purely associational estimate.
         """

causal_testing/specification/causal_specification.py

@@ -3,13 +3,17 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Callable
+import logging
 
 import pandas as pd
 
 from causal_testing.estimation.cubic_spline_estimator import CubicSplineRegressionEstimator
-from causal_testing.specification.causal_specification import CausalSpecification
+from causal_testing.specification.scenario import Scenario
+from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.testing.base_test_case import BaseTestCase
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class SimulationResult:
@@ -30,13 +34,11 @@ class SearchAlgorithm(ABC):  # pylint: disable=too-few-public-methods
     space to be searched"""
 
     @abstractmethod
-    def search(
-        self, surrogate_models: list[CubicSplineRegressionEstimator], specification: CausalSpecification
-    ) -> list:
+    def search(self, surrogate_models: list[CubicSplineRegressionEstimator], scenario: Scenario) -> list:
         """Function which implements a search routine which searches for the optimal fitness value for the specified
         scenario
         :param surrogate_models: The surrogate models to be searched
-        :param specification:  The Causal Specification (combination of Scenario and Causal Dag)"""
+        :param scenario:  The modelling scenario"""
 
 
 class Simulator(ABC):
@@ -64,11 +66,13 @@ class CausalSurrogateAssistedTestCase:
 
     def __init__(
         self,
-        specification: CausalSpecification,
+        scenario: Scenario,
+        causal_dag: CausalDAG,
         search_algorithm: SearchAlgorithm,
         simulator: Simulator,
     ):
-        self.specification = specification
+        self.scenario = scenario
+        self.causal_dag = causal_dag
         self.search_algorithm = search_algorithm
         self.simulator = simulator
 
@@ -87,8 +91,8 @@ def execute(
         :return: tuple containing SimulationResult or str, execution number and dataframe"""
 
         for i in range(max_executions):
-            surrogate_models = self.generate_surrogates(self.specification, df)
-            candidate_test_case, _, surrogate_model = self.search_algorithm.search(surrogate_models, self.specification)
+            surrogate_models = self.generate_surrogates(df)
+            candidate_test_case, _, surrogate_model = self.search_algorithm.search(surrogate_models, self.scenario)
 
             self.simulator.startup()
             test_result = self.simulator.run_with_config(candidate_test_case)
@@ -101,7 +105,7 @@ def execute(
             else:
                 df = pd.concat([df, test_result_df], ignore_index=True)
             if test_result.fault:
-                print(
+                logger.info(
                     f"Fault found between {surrogate_model.base_test_case.treatment_variable.name} causing "
                     f"{surrogate_model.base_test_case.outcome_variable.name}. Contradiction with "
                     f"expected {surrogate_model.expected_relationship}."
@@ -113,28 +117,25 @@ def execute(
                 )
                 return test_result, i + 1, df
 
-        print("No fault found")
+        logger.info("No fault found")
         return "No fault found", i + 1, df
 
-    def generate_surrogates(
-        self, specification: CausalSpecification, df: pd.DataFrame
-    ) -> list[CubicSplineRegressionEstimator]:
-        """Generate a surrogate model for each edge of the dag that specifies it is included in the DAG metadata.
-        :param specification: The Causal Specification (combination of Scenario and Causal Dag)
+    def generate_surrogates(self, df: pd.DataFrame) -> list[CubicSplineRegressionEstimator]:
+        """Generate a surrogate model for each edge of the DAG that specifies it is included in the DAG metadata.
         :param df: An dataframe which contains data relevant to the specified scenario
         :return: A list of surrogate models
         """
         surrogate_models = []
 
-        for u, v in specification.causal_dag.edges:
-            edge_metadata = specification.causal_dag.adj[u][v]
+        for u, v in self.causal_dag.edges:
+            edge_metadata = self.causal_dag.adj[u][v]
             if "included" in edge_metadata:
-                from_var = specification.scenario.variables.get(u)
-                to_var = specification.scenario.variables.get(v)
+                from_var = self.scenario.variables.get(u)
+                to_var = self.scenario.variables.get(v)
                 base_test_case = BaseTestCase(from_var, to_var)
 
-                minimal_adjustment_set = specification.causal_dag.identification(
-                    base_test_case, specification.scenario.hidden_variables()
+                minimal_adjustment_set = self.causal_dag.identification(
+                    base_test_case, self.scenario.hidden_variables()
                 )
 
                 surrogate = CubicSplineRegressionEstimator(

causal_testing/surrogate/causal_surrogate_assisted.py

@@ -7,8 +7,8 @@
 from pygad import GA
 
 from causal_testing.estimation.cubic_spline_estimator import CubicSplineRegressionEstimator
-from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.surrogate.causal_surrogate_assisted import SearchAlgorithm
+from causal_testing.specification.scenario import Scenario
 
 
 class GeneticSearchAlgorithm(SearchAlgorithm):
@@ -27,9 +27,7 @@ def __init__(self, delta=0.05, config: dict = None) -> None:
         }
 
     # pylint: disable=too-many-locals
-    def search(
-        self, surrogate_models: list[CubicSplineRegressionEstimator], specification: CausalSpecification
-    ) -> list:
+    def search(self, surrogate_models: list[CubicSplineRegressionEstimator], scenario: Scenario) -> list:
         solutions = []
 
         for surrogate_model in surrogate_models:
@@ -53,7 +51,7 @@ def fitness_function(ga, solution, idx):  # pylint: disable=unused-argument
                     )
                 return contradiction_function(ate[0])
 
-            gene_types, gene_space = self.create_gene_types(surrogate_model, specification)
+            gene_types, gene_space = self.create_gene_types(surrogate_model, scenario)
 
             num_genes = 1 + len(surrogate_model.adjustment_set)
 
@@ -71,8 +69,7 @@ def fitness_function(ga, solution, idx):  # pylint: disable=unused-argument
                 for k, v in self.config.items():
                     if k == "gene_space":
                         raise ValueError(
-                            "Gene space should not be set through config. This is generated from the causal "
-                            "specification"
+                            "Gene space should not be set through config. This is generated from the modelling scenario"
                         )
                     setattr(ga, k, v)
 
@@ -88,24 +85,22 @@ def fitness_function(ga, solution, idx):  # pylint: disable=unused-argument
         return max(solutions, key=itemgetter(1))  # This can be done better with fitness normalisation between edges
 
     @staticmethod
-    def create_gene_types(
-        surrogate_model: CubicSplineRegressionEstimator, specification: CausalSpecification
-    ) -> tuple[list, list]:
-        """Generate the gene_types and gene_space for a given fitness function and specification
+    def create_gene_types(surrogate_model: CubicSplineRegressionEstimator, scenario: Scenario) -> tuple[list, list]:
+        """Generate the gene_types and gene_space for a given fitness function and scenario
         :param surrogate_model: Instance of a CubicSplineRegressionEstimator
-        :param specification: The Causal Specification (combination of Scenario and Causal Dag)"""
+        :param scenario: The modelling scenario"""
 
         var_space = {}
         var_space[surrogate_model.base_test_case.treatment_variable.name] = {}
         for adj in surrogate_model.adjustment_set:
             var_space[adj] = {}
 
-        for relationship in list(specification.scenario.constraints):
+        for relationship in list(scenario.constraints):
             print(relationship)
             rel_split = str(relationship).split(" ")
 
             if rel_split[0] in var_space:
-                datatype = specification.scenario.variables.get(rel_split[0]).datatype
+                datatype = scenario.variables.get(rel_split[0]).datatype
                 if rel_split[1] == ">=":
                     var_space[rel_split[0]]["low"] = datatype(rel_split[2])
                 elif rel_split[1] == "<=":
@@ -119,9 +114,7 @@ def create_gene_types(
             gene_space.append(var_space[adj])
 
         gene_types = []
-        gene_types.append(
-            specification.scenario.variables.get(surrogate_model.base_test_case.treatment_variable.name).datatype
-        )
+        gene_types.append(scenario.variables.get(surrogate_model.base_test_case.treatment_variable.name).datatype)
         for adj in surrogate_model.adjustment_set:
-            gene_types.append(specification.scenario.variables.get(adj).datatype)
+            gene_types.append(scenario.variables.get(adj).datatype)
         return gene_types, gene_space

causal_testing/surrogate/surrogate_search_algorithms.py

@@ -12,7 +12,7 @@
 
 import networkx as nx
 
-from causal_testing.specification.causal_specification import CausalDAG, Node
+from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.testing.base_test_case import BaseTestCase
 
 logger = logging.getLogger(__name__)
@@ -23,7 +23,7 @@ class MetamorphicRelation:
     """Class representing a metamorphic relation."""
 
     base_test_case: BaseTestCase
-    adjustment_vars: Iterable[Node]
+    adjustment_vars: Iterable[str]
 
     def __eq__(self, other):
         same_type = self.__class__ == other.__class__

causal_testing/testing/metamorphic_relation.py

@@ -2,9 +2,7 @@
 import logging
 import pandas as pd
 from causal_testing.specification.causal_dag import CausalDAG
-from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Output
-from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.testing.causal_test_case import CausalTestCase
 from causal_testing.testing.causal_effect import Positive, Negative, NoEffect
 from causal_testing.estimation.linear_regression_estimator import LinearRegressionEstimator
@@ -28,32 +26,12 @@ def setup_test_case(verbose: bool = False):
     causal_dag = CausalDAG(f"{ROOT}/dag.dot")
 
     # 2. Create variables
-    pop_size = Input("pop_size", int)
-    pop_infected = Input("pop_infected", int)
-    n_days = Input("n_days", int)
     vaccine = Input("vaccine", int)
     cum_infections = Output("cum_infections", int)
     cum_vaccinations = Output("cum_vaccinations", int)
     cum_vaccinated = Output("cum_vaccinated", int)
     max_doses = Output("max_doses", int)
 
-    # 3. Create scenario by applying constraints over a subset of the input variables
-    scenario = Scenario(
-        variables={
-            pop_size,
-            pop_infected,
-            n_days,
-            cum_infections,
-            vaccine,
-            cum_vaccinated,
-            cum_vaccinations,
-            max_doses,
-        },
-    )
-
-    # 4. Construct a causal specification from the scenario and causal DAG
-    causal_specification = CausalSpecification(scenario, causal_dag)
-
     # 5. Read the previously simulated data
     obs_df = pd.read_csv("simulated_data.csv")
 

examples/covasim_/vaccinating_elderly/example_vaccine.py

@@ -4,7 +4,6 @@
 from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Output
-from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.testing.causal_test_case import CausalTestCase
 from causal_testing.testing.causal_effect import Positive, Negative, NoEffect
 from causal_testing.estimation.linear_regression_estimator import LinearRegressionEstimator

examples/lr91/example_max_conductances.py

@@ -6,7 +6,6 @@
 from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Output
-from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.testing.causal_test_case import CausalTestCase
 from causal_testing.testing.causal_effect import ExactValue, Positive
 from causal_testing.estimation.linear_regression_estimator import LinearRegressionEstimator
@@ -79,9 +78,6 @@ def estimate_risk_ratio(self) -> float:
     }
 )
 
-# 4. Construct a causal specification from the scenario and causal DAG
-causal_specification = CausalSpecification(scenario, causal_dag)
-
 observational_data_path = f"{ROOT}/data/random/data_random_1000.csv"
 
 
@@ -99,7 +95,7 @@ def test_poisson_intensity_num_shapes(save=False):
                     base_test_case=base_test_case,
                     treatment_value=treatment_value,
                     control_value=control_value,
-                    adjustment_set=causal_specification.causal_dag.identification(base_test_case),
+                    adjustment_set=causal_dag.identification(base_test_case),
                     df=pd.read_csv(f"{ROOT}/data/smt_100/data_smt_wh{wh}_100.csv", index_col=0).astype(float),
                     effect_modifiers=None,
                     alpha=0.05,
@@ -114,7 +110,7 @@ def test_poisson_intensity_num_shapes(save=False):
                     base_test_case=base_test_case,
                     treatment_value=treatment_value,
                     control_value=control_value,
-                    adjustment_set=causal_specification.causal_dag.identification(base_test_case),
+                    adjustment_set=causal_dag.identification(base_test_case),
                     df=observational_df,
                     effect_modifiers=None,
                     formula="num_shapes_unit ~ I(intensity ** 2) + intensity - 1",
@@ -158,7 +154,7 @@ def test_poisson_width_num_shapes(save=False):
                 base_test_case=base_test_case,
                 treatment_value=w + 1.0,
                 control_value=float(w),
-                adjustment_set=causal_specification.causal_dag.identification(base_test_case),
+                adjustment_set=causal_dag.identification(base_test_case),
                 df=df,
                 effect_modifiers={"intensity": i},
                 formula="num_shapes_unit ~ width + I(intensity ** 2)+I(width ** -1)+intensity-1",