Compare revisions

def _phen_diff(args: argparse.Namespace):

    """Handle the `phen diff` command."""

    phen.diff(args.phen_dir, args.version, args.old_phen_dir, args.old_version, args.not_check_config)

    phen.diff(

        args.phen_dir,

        args.version,

        args.old_phen_dir,

        args.old_version,

        args.not_check_config,

    )

def main():

    )

    phen_map_parser.add_argument(

        "--not-translate",

        action='store_true',

        action="store_true",

        default=False,

        help="(Optional) Prevent any phenotype translation using NHS TRUD vocabularies.",

    )

    phen_map_parser.add_argument(

        "--no-metadata",

        action='store_true',

        action="store_true",

        default=False,

        help="(Optional) Prevent copying of metadata columns to output.",

    )

    )

    phen_diff_parser.add_argument(

        "--not-check-config",

        action='store_true',

        action="store_true",

        default=False,

        help="(Optional) Prevent loading and comparing config file, in the case where one does not exist",

    )

                                        "schema": {

                                            "divide_col": {"type": "string"},

                                            "split_col": {"type": "string"},

                                            "codes_col": {"type": "string"}

                                            "codes_col": {"type": "string"},

                                        },

                                    },

                                },

                            },

                        },

# Translate Df with multiple codes into single code type Series

def translate_codes(

    source_df: pd.DataFrame, target_code_type: str, concept_name: str, not_translate:bool

    source_df: pd.DataFrame,

    target_code_type: str,

    concept_name: str,

    not_translate: bool,

) -> pd.DataFrame:

    """Translates each source code type the source coding list into a target type and returns all conversions as a concept set"""

                "SOURCE": err.codes_file,

                "CAUSE": err.message,

            }

            for err in code_errors if err.mask is not None

            for err in code_errors

            if err.mask is not None

        ]

    )

        )

    if target_code_type is not None:

        _map_target_code_type(phen_path, phenotype, target_code_type, not_translate, no_metadata)

        _map_target_code_type(

            phen_path, phenotype, target_code_type, not_translate, no_metadata

        )

    else:

        for t in phenotype["map"]:

            _map_target_code_type(phen_path, phenotype, t, not_translate, no_metadata)

    _logger.info(f"Phenotype processed successfully")

def _map_target_code_type(phen_path: Path, phenotype: dict, target_code_type: str, not_translate:bool, no_metadata:bool):

def _map_target_code_type(

    phen_path: Path,

    phenotype: dict,

    target_code_type: str,

    not_translate: bool,

    no_metadata: bool,

):

    _logger.debug(f"Target coding format: {target_code_type}")

    concepts_path = phen_path / CONCEPTS_DIR

    # Create output dataframe

# Add metadata dict to each row of Df codes

def add_metadata(

    codes: pd.DataFrame, metadata: dict, no_metadata:bool,

    codes: pd.DataFrame,

    metadata: dict,

    no_metadata: bool,

) -> pd.DataFrame:

    """Add concept set metadata, stored as a dictionary, to each concept row"""

        report += f"  - {new_version}\n"

        report += f"  - {str(new_phen_path.resolve())}\n"

        # Convert list of dicts into a dict: {name: file}

        report += diff_config(old_config, new_config)

    _logger.info(f"Phenotypes diff'd successfully")

def diff(phen_dir: str, version: str, old_phen_dir: str, old_version: str, not_check_config:bool):

def diff(

    phen_dir: str,

    version: str,

    old_phen_dir: str,

    old_version: str,

    not_check_config: bool,

):

    # make tmp directory .acmc

    timestamp = time.strftime("%Y%m%d_%H%M%S")

    temp_dir = Path(f".acmc/diff_{timestamp}")

            f"Old phenotype directory does not exist: {str(old_phen_path.resolve())}"

        )

    t_path = old_phen_path / "config.yml"

    with t_path.open("r") as file:

        c = yaml.safe_load(file)

    #    t_path = old_phen_path / "config.yml"

    #    with t_path.open("r") as file:

    #        c = yaml.safe_load(file)

    try:

        # Create the directory

        report_filename = f"{version}_{old_version}_diff.md"

        report_path = changed_phen_path / report_filename

        # diff old with new

        diff_phen(changed_path, version, old_path, old_version, report_path, not_check_config)

        diff_phen(

            changed_path, version, old_path, old_version, report_path, not_check_config

        )

    finally:

        # clean up tmp directory

        if temp_dir.exists():

            shutil.rmtree(temp_dir)

            print(f"Temporary directory removed: {temp_dir}")

def test_phen_workflow(tmp_dir, monkeypatch, caplog, config_file):

    print(f"Temporary directory: {tmp_dir}")  # Prints path for debugging

    # init phenotype in temp directory

    with caplog.at_level(logging.DEBUG):

        phen_path = tmp_dir / "phen"

        phen_path = phen_path.resolve()

        main.main()

    assert "Phenotype initialised successfully" in caplog.text

    # validate phenotype and copy examples across

    with caplog.at_level(logging.DEBUG):

        # validate phenotype

        # copy examples across

        shutil.rmtree(phen_path / phen.CONCEPTS_DIR)

        ex_path = Path("./examples").resolve()

        for item in ex_path.iterdir():

        main.main()

    assert "Phenotype validated successfully" in caplog.text

    # map phenotype

    # map phenotypes at read2 and read3

    for code_type in ["read2", "read3"]:

        with caplog.at_level(logging.DEBUG):

            monkeypatch.setattr(

    # diff phenotype

    with caplog.at_level(logging.DEBUG):

        old_path = tmp_dir / "0.0.1"

        monkeypatch.setattr(

            sys,

            "argv",

                "diff",

                "-d",

                str(phen_path.resolve()),

                "-od",

                str(phen_path.resolve()),

                "-ov",

                "0.0.1",

            ],