:py:mod:`causal_testing.main` ============================= .. py:module:: causal_testing.main .. autoapi-nested-parse:: This module contains the classes that executes the Causal Testing Framework. Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: causal_testing.main.Command causal_testing.main.CausalTestingPaths causal_testing.main.CausalTestingFramework Functions ~~~~~~~~~ .. autoapisummary:: causal_testing.main.setup_logging causal_testing.main.parse_args Attributes ~~~~~~~~~~ .. autoapisummary:: causal_testing.main.logger .. py:data:: logger .. py:class:: Command(*args, **kwds) Bases: :py:obj:`enum.Enum` Enum for supported CTF commands. .. py:attribute:: TEST :value: 'test' .. py:attribute:: GENERATE :value: 'generate' .. py:class:: CausalTestingPaths(dag_path: Union[str, pathlib.Path], data_paths: List[Union[str, pathlib.Path]], test_config_path: Union[str, pathlib.Path], output_path: Union[str, pathlib.Path]) Class for managing paths for causal testing inputs and outputs. :param dag_path: Path to the DAG definition file :param data_paths: List of paths to input data files :param test_config_path: Path to the test configuration file :param output_path: Path where test results will be written .. py:attribute:: dag_path :type: pathlib.Path .. py:attribute:: data_paths :type: List[pathlib.Path] .. py:attribute:: test_config_path :type: pathlib.Path .. py:attribute:: output_path :type: pathlib.Path .. py:method:: validate_paths() -> None Validate existence of all input paths and writability of output path. :raises: FileNotFoundError if any required input file is missing. .. py:class:: CausalTestingFramework(paths: CausalTestingPaths, ignore_cycles: bool = False, query: Optional[str] = None) Main class for running causal tests. :param paths: CausalTestingPaths object containing required file paths :param ignore_cycles: Flag to ignore cycles in the DAG :param query: Optional query string to filter the input dataframe .. py:method:: setup() -> None Set up the framework by loading DAG, runtime csv data, creating the scenario and causal specification. :raises: FileNotFoundError if required files are missing .. py:method:: load_dag() -> causal_testing.specification.causal_dag.CausalDAG Load the causal DAG from the specified file path. .. py:method:: _read_dataframe(data_path) .. py:method:: load_data(query: Optional[str] = None) -> pandas.DataFrame Load and combine all data sources with optional filtering. :param query: Optional pandas query string to filter the loaded data :return: Combined pandas DataFrame containing all loaded and filtered data .. py:method:: create_variables() -> None Create variable objects from DAG nodes based on their connectivity. .. py:method:: load_tests() -> None Load and prepare test configurations from file. .. py:method:: create_base_test(test: dict) -> causal_testing.testing.base_test_case.BaseTestCase Create base test case from test configuration. :param test: Dictionary containing test configuration parameters :return: BaseTestCase object :raises: KeyError if required variables are not found in inputs or outputs .. py:method:: create_test_cases(test_configs: dict) -> List[causal_testing.testing.causal_test_case.CausalTestCase] Create test case objects from configuration dictionary. :param test_configs: Dictionary containing test configurations :return: List of CausalTestCase objects containing the initialised test cases :raises: KeyError if required variables are not found :raises: ValueError if invalid test configuration is provided .. py:method:: create_causal_test(test: dict, base_test: causal_testing.testing.base_test_case.BaseTestCase) -> causal_testing.testing.causal_test_case.CausalTestCase Create causal test case from test configuration and base test. :param test: Dictionary containing test configuration parameters :param base_test: BaseTestCase object :return: CausalTestCase object :raises: ValueError if invalid estimator or configuration is provided .. py:method:: run_tests_in_batches(batch_size: int = 100, silent: bool = False, adequacy: bool = False, bootstrap_size: int = 100) -> List[causal_testing.testing.causal_test_result.CausalTestResult] Run tests in batches to reduce memory usage. :param batch_size: Number of tests to run in each batch :param silent: Whether to suppress errors :param adequacy: Whether to calculate causal test adequacy (defaults to False) :param bootstrap_size: The number of bootstrap samples to use when calculating causal test adequacy (defaults to 100) :return: List of all test results :raises: ValueError if no tests are loaded .. py:method:: run_tests(silent: bool = False, adequacy: bool = False, bootstrap_size: int = 100) -> List[causal_testing.testing.causal_test_result.CausalTestResult] Run all test cases and return their results. :param silent: Whether to suppress errors :param adequacy: Whether to calculate causal test adequacy (defaults to False) :param bootstrap_size: The number of bootstrap samples to use when calculating causal test adequacy (defaults to 100) :return: List of CausalTestResult objects :raises: ValueError if no tests are loaded :raises: Exception if test execution fails .. py:method:: save_results(results: List[causal_testing.testing.causal_test_result.CausalTestResult], output_path: str = None) -> list Save test results to JSON file in the expected format. .. py:function:: setup_logging(verbose: bool = False) -> None Set up logging configuration. .. py:function:: parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace Parse command line arguments.