Support for hidden variables

Author: jmafoster1

causal_testing/data_collection/data_collector.py

@@ -38,11 +38,11 @@ def filter_valid_data(self, data: pd.DataFrame, check_pos: bool = True) -> pd.Da
         """
 
         # Check positivity
-        scenario_variables = set(self.scenario.variables)
+        scenario_variables = set(self.scenario.variables) - {x.name for x in self.scenario.hidden_variables()}
 
-        if check_pos and not scenario_variables.issubset(data.columns):
+        if check_pos and not (scenario_variables - {x.name for x in self.scenario.hidden_variables()}).issubset(set(data.columns)):
             missing_variables = scenario_variables - set(data.columns)
-            raise IndexError(f"Positivity violation: missing data for variables {missing_variables}.")
+            raise IndexError(f"Missing columns: missing data for variables {missing_variables}. Should they be marked as hidden?")
 
         # For each row, does it satisfy the constraints?
         solver = z3.Solver()
@@ -57,6 +57,7 @@ def filter_valid_data(self, data: pd.DataFrame, check_pos: bool = True) -> pd.Da
                 self.scenario.variables[var].z3
                 == self.scenario.variables[var].z3_val(self.scenario.variables[var].z3, row[var])
                 for var in self.scenario.variables
+                if var in row
             ]
             for c in model:
                 solver.assert_and_track(c, f"model: {c}")

causal_testing/json_front/json_class.py

@@ -296,6 +296,6 @@ class CausalVariables:
     metas: list[Meta]
 
     def __init__(self, inputs: list[dict], outputs: list[dict], metas: list[dict]):
-        self.inputs = [Input(i["name"], i["type"], i["distribution"]) for i in inputs]
-        self.outputs = [Output(i["name"], i["type"], i.get("distribution", None)) for i in outputs]
-        self.metas = [Meta(i["name"], i["type"], i["populate"]) for i in metas] if metas else []
+        self.inputs = [Input(**i) for i in inputs]
+        self.outputs = [Output(**o) for o in outputs]
+        self.metas = [Meta(**m) for m in metas] if metas else []

causal_testing/specification/scenario.py

@@ -142,6 +142,9 @@ def metas(self) -> set[Meta]:
         """
         return self.variables_of_type(Meta)
 
+    def hidden_variables(self) -> set[Variable]:
+        return {v for v in self.variables.values() if v.hidden}
+
     def add_variable(self, v: Variable) -> None:
         """Add variable to variables attribute
         :param v: Variable to be added

causal_testing/specification/variable.py

@@ -60,18 +60,19 @@ class Variable(ABC):
     :attr name:
     :attr datatype:
     :attr distribution:
-
+    :attr hidden:
     """
 
     name: str
     datatype: T
     distribution: rv_generic
 
-    def __init__(self, name: str, datatype: T, distribution: rv_generic = None):
+    def __init__(self, name: str, datatype: T, distribution: rv_generic = None, hidden: bool =  False):
         self.name = name
         self.datatype = datatype
         self.z3 = z3_types(datatype)(name)
         self.distribution = distribution
+        self.hidden = hidden
 
     def __repr__(self):
         return f"{self.typestring()}: {self.name}::{self.datatype.__name__}"

tests/json_front_tests/test_json_class.py

@@ -31,9 +31,9 @@ def setUp(self) -> None:
         self.data_path = test_data_dir_path / data_file_name
         self.json_class = JsonUtility("logs.log")
         self.example_distribution = scipy.stats.uniform(1, 10)
-        self.input_dict_list = [{"name": "test_input", "type": float, "distribution": self.example_distribution}]
-        self.output_dict_list = [{"name": "test_output", "type": float}]
-        self.meta_dict_list = [{"name": "test_meta", "type": float, "populate": populate_example}]
+        self.input_dict_list = [{"name": "test_input", "datatype": float, "distribution": self.example_distribution}]
+        self.output_dict_list = [{"name": "test_output", "datatype": float}]
+        self.meta_dict_list = [{"name": "test_meta", "datatype": float, "populate": populate_example}]
         self.json_class.set_variables(self.input_dict_list, self.output_dict_list, None)
         self.json_class.set_paths(self.json_path, self.dag_path, self.data_path)