phen.py

import argparse
import pandas as pd
import numpy as np
import json
import os
import sqlite3
import sys
import shutil
import git
import re
import logging
import requests
import yaml
from cerberus import Validator
from deepdiff import DeepDiff
from pathlib import Path
from urllib.parse import urlparse, urlunparse

import acmc
from acmc import trud, omop, parse, util

# setup logging
import acmc.logging_config as lc

logger = lc.setup_logger()

pd.set_option("mode.chained_assignment", None)

PHEN_DIR = "phen"
DEFAULT_PHEN_PATH = Path("./workspace") / PHEN_DIR

CODES_DIR = "codes"
MAP_DIR = "map"
CONCEPT_SET_DIR = "concept-set"
OMOP_DIR = "omop"
DEFAULT_PHEN_DIR_LIST = [CODES_DIR, MAP_DIR, CONCEPT_SET_DIR, OMOP_DIR]
CONFIG_FILE = "config.yaml"
VOCAB_VERSION_FILE = "vocab_version.yaml"

DEFAULT_GIT_BRANCH = "main"

SPLIT_COL_ACTION = "split_col"
CODES_COL_ACTION = "codes_col"
DIVIDE_COL_ACTION = "divide_col"
COL_ACTIONS = [SPLIT_COL_ACTION, CODES_COL_ACTION, DIVIDE_COL_ACTION]

CODE_FILE_TYPES = [".xlsx", ".xls", ".csv"]

# config.yaml schema
CONFIG_SCHEMA = {
    "phenotype": {
        "type": "dict",
        "required": True,
        "schema": {
            "version": {
                "type": "string",
                "required": True,
                "regex": r"^v\d+\.\d+\.\d+$",  # Enforces 'vN.N.N' format
            },
            "omop": {
                "type": "dict",
                "required": True,
                "schema": {
                    "vocabulary_id": {"type": "string", "required": True},
                    "vocabulary_name": {"type": "string", "required": True},
                    "vocabulary_reference": {
                        "type": "string",
                        "required": True,
                        "regex": r"^https?://.*",  # Ensures it's a URL
                    },
                },
            },
            "map": {
                "type": "list",
                "schema": {
                    "type": "string",
                    "allowed": list(
                        parse.SUPPORTED_CODE_TYPES
                    ),  # Ensure only predefined values are allowed
                },
            },
            "concept_sets": {
                "type": "list",
                "required": True,
                "schema": {
                    "type": "dict",
                    "schema": {
                        "name": {"type": "string", "required": True},
                        "file": {
                            "type": "dict",
                            "required": False,
                            "schema": {
                                "path": {"type": "string", "required": True},
                                "columns": {"type": "dict", "required": True},
                                "category": {
                                    "type": "string"
                                },  # Optional but must be string if present
                                "actions": {
                                    "type": "dict",
                                    "schema": {"divide_col": {"type": "string"}},
                                },
                            },
                        },
                        "metadata": {"type": "dict", "required": True},
                    },
                },
            },
        },
    }
}


class PhenValidationException(Exception):
    """Custom exception class raised when validation errors in phenotype configuration file"""

    def __init__(self, message, validation_errors=None):
        super().__init__(message)
        self.validation_errors = validation_errors


def construct_git_url(remote_url):
    """Constructs a git url for github or gitlab including a PAT token environment variable"""
    # check the url
    parsed_url = urlparse(remote_url)

    # if github in the URL otherwise assume it's gitlab, if we want to use others such as codeberg we'd
    # need to update this function if the URL scheme is different.
    if "github.com" in parsed_url.netloc:
        # get GitHub PAT from environment variable
        auth = os.getenv("ACMC_GITHUB_PAT")
        if not auth:
            raise ValueError(
                "GitHub PAT not found. Set the ACMC_GITHUB_PAT environment variable."
            )
    else:
        # get GitLab PAT from environment variable
        auth = os.getenv("ACMC_GITLAB_PAT")
        if not auth:
            raise ValueError(
                "GitLab PAT not found. Set the ACMC_GITLAB_PAT environment variable."
            )
        auth = f"oauth2:{auth}"

    # Construct the new URL with credentials
    new_netloc = f"{auth}@{parsed_url.netloc}"
    return urlunparse(
        (
            parsed_url.scheme,
            new_netloc,
            parsed_url.path,
            parsed_url.params,
            parsed_url.query,
            parsed_url.fragment,
        )
    )


def create_empty_git_dir(path):
    """Creates a directory with a .gitkeep file so that it's tracked in git"""
    path.mkdir(exist_ok=True)
    keep_path = path / ".gitkeep"
    keep_path.touch(exist_ok=True)


def check_delete_dir(path, msg):
    deleted = False

    user_input = input(f"{msg}").strip().lower()
    if user_input in ["yes", "y"]:
        shutil.rmtree(path)
        deleted = True
    else:
        logger.info("Directory was not deleted.")

    return deleted


def init(phen_dir, remote_url):
    """Initial phenotype directory as git repo with standard structure"""
    logger.info(f"Initialising Phenotype in directory: {phen_dir}")
    phen_path = Path(phen_dir)

    # check if directory already exists and ask user if they want to recreate it
    if (
        phen_path.exists() and phen_path.is_dir()
    ):  # Check if it exists and is a directory
        configure = check_delete_dir(
            phen_path,
            f"The phen directory already exists. Do you want to reinitialise? (yes/no): ",
        )
    else:
        configure = True

    if not configure:
        logger.info(f"Exiting, phenotype not initiatised")
        return

    # Initialise repo from local or remote
    repo = None
    # if remote then clone the repo otherwise init a local repo
    if remote_url != None:
        # add PAT token to the URL
        git_url = construct_git_url(remote_url)

        # clone the repo
        repo = git.cmd.Git()
        repo.clone(git_url, phen_path)
        # open repo
        repo = git.Repo(phen_path)
        # check if there are any commits (new repo has no commits)
        if (
            len(repo.branches) == 0 or repo.head.is_detached
        ):  # Handle detached HEAD (e.g., after init)
            logger.debug("The phen repository has no commits yet.")
            commit_count = 0
        else:
            # Get the total number of commits in the default branch
            commit_count = sum(1 for _ in repo.iter_commits())
            logger.debug(f"Repo has previous commits: {commit_count}")
    else:
        # local repo, create the directories and init
        phen_path.mkdir(parents=True, exist_ok=True)
        logger.debug(f"Phen directory '{phen_path}' has been created.")
        repo = git.Repo.init(phen_path)
        commit_count = 0

    phen_path = phen_path.resolve()
    # initialise empty repos
    if commit_count == 0:
        # create initial commit
        initial_file_path = phen_path / "README.md"
        with open(initial_file_path, "w") as file:
            file.write(
                "# Initial commit\nThis is the first commit in the phen repository.\n"
            )
        repo.index.add([initial_file_path])
        repo.index.commit("Initial commit")
        commit_count = 1

    # Checkout the phens default branch, creating it if it does not exist
    if DEFAULT_GIT_BRANCH in repo.branches:
        main_branch = repo.heads[DEFAULT_GIT_BRANCH]
        main_branch.checkout()
    else:
        main_branch = repo.create_head(DEFAULT_GIT_BRANCH)
        main_branch.checkout()

    # if the phen path does not contain the config file then initialise the phen type
    config_path = phen_path / CONFIG_FILE
    if config_path.exists():
        logger.debug(f"Phenotype configuration files already exist")
        return

    logger.info("Creating phen directory structure and config files")
    for d in DEFAULT_PHEN_DIR_LIST:
        create_empty_git_dir(phen_path / d)

    # set initial version based on the number of commits in the repo, depending on how the repo was created
    # e.g., with a README.md, then there will be some initial commits before the phen config is added
    next_commit_count = commit_count + 1
    initial_version = f"v1.0.{next_commit_count}"

    # create empty phen config file
    config = {
        "phenotype": {
            "version": initial_version,
            "omop": {
                "vocabulary_id": "",
                "vocabulary_name": "",
                "vocabulary_reference": "",
            },
            "translate": [],
            "concept_sets": [],
        }
    }

    with open(phen_path / CONFIG_FILE, "w") as file:
        yaml.dump(
            config,
            file,
            Dumper=util.QuotedDumper,
            default_flow_style=False,
            sort_keys=False,
            default_style='"',
        )

    # add git ignore
    ignore_content = """# Ignore SQLite database files
*.db
*.sqlite3
 
# Ignore SQLite journal and metadata files
*.db-journal
*.sqlite3-journal

# python
.ipynb_checkpoints
 """
    ignore_path = phen_path / ".gitignore"
    with open(ignore_path, "w") as file:
        file.write(ignore_content)

    # add to git repo and commit
    for d in DEFAULT_PHEN_DIR_LIST:
        repo.git.add(phen_path / d)
    repo.git.add(all=True)
    repo.index.commit("initialised the phen git repo.")

    logger.info(f"Phenotype initialised successfully")


def validate(phen_dir):
    """Validates the phenotype directory is a git repo with standard structure"""
    logger.info(f"Validating phenotype: {phen_dir}")
    phen_path = Path(phen_dir)
    if not phen_path.is_dir():
        raise NotADirectoryError(
            f"Error: '{str(phen_path.resolve())}' is not a directory"
        )

    config_path = phen_path / CONFIG_FILE
    if not config_path.is_file():
        raise FileNotFoundError(
            f"Error: phen configuration file '{config_path}' does not exist."
        )

    codes_path = phen_path / CODES_DIR
    if not codes_path.is_dir():
        raise FileNotFoundError(
            f"Error: source codes directory {source_codes_dir} does not exist."
        )

    # Calidate the directory is a git repo
    try:
        git.Repo(phen_path)
    except (git.exc.InvalidGitRepositoryError, git.exc.NoSuchPathError):
        raise Exception(f"Phen directory {phen_path} is not a git repo")

    # Load configuration File
    if config_path.suffix == ".yaml":
        try:
            with config_path.open("r") as file:
                phenotype = yaml.safe_load(file)

            validator = Validator(CONFIG_SCHEMA)
            if validator.validate(phenotype):
                logger.debug("YAML structure is valid.")
            else:
                logger.error(f"YAML structure validation failed: {validator.errors}")
                raise Exception(f"YAML structure validation failed: {validator.errors}")
        except yaml.YAMLError as e:
            logger.error(f"YAML syntax error: {e}")
            raise e
    else:
        raise Exception(
            f"Unsupported configuration filetype: {str(config_path.resolve())}"
        )

    # initiatise
    validation_errors = []
    phenotype = phenotype["phenotype"]
    code_types = parse.CodeTypeParser().code_types

    # check the version number is of the format vn.n.n
    match = re.match(r"v(\d+\.\d+\.\d+)", phenotype["version"])
    if not match:
        validation_errors.append(
            f"Invalid version format in configuration file: {phenotype['version']}"
        )

    # create a list of all the concept set names defined in the concept set configuration
    concept_set_names = []
    for item in phenotype["concept_sets"]:
        if item["name"] in concept_set_names:
            validation_errors.append(
                f"Duplicate concept set defined in concept sets {item['name'] }"
            )
        else:
            concept_set_names.append(item["name"])

    # check codes definition
    for item in phenotype["concept_sets"]:
        # check concepte code file exists
        concept_code_file_path = codes_path / item["file"]["path"]
        if not concept_code_file_path.exists():
            validation_errors.append(
                f"Coding file {str(concept_code_file_path.resolve())} does not exist"
            )

        # check concepte code file is not empty
        if concept_code_file_path.stat().st_size == 0:
            validation_errors.append(
                f"Coding file {str(concept_code_file_path.resolve())} is an empty file"
            )

        # check code file type is supported
        if concept_code_file_path.suffix not in CODE_FILE_TYPES:
            raise ValueError(
                f"Unsupported filetype {concept_code_file_path.suffix}, only support csv, xlsx, xls code file types"
            )

        # check columns specified are a supported medical coding type
        for column in item["file"]["columns"]:
            if column not in code_types:
                validation_errors.append(
                    f"Column type {column} for file {concept_code_file_path} is not supported"
                )

        # check the actions are supported
        if "actions" in item["file"]:
            for action in item["file"]["actions"]:
                if action not in COL_ACTIONS:
                    validation_errors.append(f"Action {action} is not supported")

    if len(validation_errors) > 0:
        logger.error(validation_errors)
        raise PhenValidationException(
            f"Configuration file {str(config_path.resolve())} failed validation",
            validation_errors,
        )

    logger.info(f"Phenotype validated successfully")


def read_table_file(path, excel_sheet=None):
    """
    Load Code List File
    """

    path = path.resolve()
    if path.suffix == ".csv":
        df = pd.read_csv(path, dtype=str)
    elif path.suffix == ".xlsx" or path.suffix == ".xls":
        if excel_sheet:
            df = pd.read_excel(path, sheet_name=excel_sheet, dtype=str)
        else:
            df = pd.read_excel(path, dtype=str)
    elif path.suffix == ".dta":
        df = pd.read_stata(path, dtype=str)
    else:
        raise ValueError(
            f"Unsupported filetype {codes_file_path.suffix}, only support{CODE_FILE_TYPES} code file types"
        )

    return df


def process_actions(df, concept_set):
    # Perform Structural Changes to file before preprocessing
    logger.debug("Processing file structural actions")
    if (
        "actions" in concept_set["file"]
        and "split_col" in concept_set["file"]["actions"]
        and "codes_col" in concept_set["file"]["actions"]
    ):
        split_col = concept_set["file"]["actions"]["split_col"]
        codes_col = concept_set["file"]["actions"]["codes_col"]
        logger.debug(
            "Action: Splitting",
            split_col,
            "column into:",
            df[split_col].unique(),
        )
        codes = df[codes_col]
        oh = pd.get_dummies(df[split_col], dtype=bool)  # one hot encode
        oh = oh.where((oh != True), codes, axis=0)  # fill in 1s with codes
        oh[oh == False] = np.nan  # replace 0s with None
        df = pd.concat([df, oh], axis=1)  # merge in new columns

    return df


# Perform QA Checks on columns individually and append to df
def preprocess_codes(df, concept_set, code_file_path, target_code_type=None):
    """Parses each column individually - Order and length will not be preserved!"""
    out = pd.DataFrame([])  # create output df to append to
    code_errors = []  # list of errors from processing

    # TODO: Is there a better way of processing this action as it's distributed across
    # different parts of the programme.
    if (
        "actions" in concept_set["file"]
        and "divide_col" in concept_set["file"]["actions"]
    ):
        divide_col_df = df[concept_set["file"]["actions"]["divide_col"]]
    else:
        divide_col_df = pd.DataFrame()

    # Preprocess codes
    code_types = parse.CodeTypeParser().code_types
    for code_type in concept_set["file"]["columns"]:
        parser = code_types[code_type]
        logger.info(f"Processing {code_type} codes...")

        # get code types
        codes = df[concept_set["file"]["columns"][code_type]].dropna()
        codes = codes.astype(str)  # convert to string
        codes = codes.str.strip()  # remove excess spaces

        # process codes, validating them using parser and returning the errors
        codes, errors = parser.process(codes, code_file_path)
        if len(errors) > 0:
            code_errors.extend(errors)
            logger.warning(f"Codes validation failed with {len(errors)} errors")

        # append to output dataframe
        out = pd.concat(
            [out, pd.DataFrame({code_type: codes}).join(divide_col_df)],
            ignore_index=True,
        )

    return out, code_errors


# Translate Df with multiple codes into single code type Series
def translate_codes(df, target_code_type):
    codes = pd.Series([], dtype=str)

    # Convert codes to target type
    logger.info(f"Converting to target code type {target_code_type}")
    for col_name in df.columns:
        # if target code type is the same as thet source code type, no translation, just appending source as target
        if col_name == target_code_type:
            logger.debug(
                f"Target code type {target_code_type} has source code types {len(df)}, copying rather than translating"
            )
            codes = pd.concat([codes, df[target_code_type]])
        else:
            filename = f"{col_name}_to_{target_code_type}.parquet"
            map_path = trud.PROCESSED_PATH / filename
            if map_path.exists():
                col = df[col_name]
                df_map = pd.read_parquet(map_path)
                # merge on corresponding codes and take target column
                translated = pd.merge(col, df_map, how="left")[target_code_type]
                # TODO: BUG mask does not match column
                codes = pd.concat([codes, translated])  # merge to output
            else:
                logger.warning(
                    f"No mapping from {col_name} to {target_code_type}, file {str(map_path.resolve())} does not exist"
                )

    return codes


# Append file's codes to output Df with concept
def map_file(df, target_code_type, out, concept_name):

    # translate codes
    codes = translate_codes(df, target_code_type)
    codes = codes.dropna()  # delete NaNs

    # Append to output if translated
    if len(codes) > 0:
        codes = pd.DataFrame({"CONCEPT": codes})
        codes["CONCEPT_SET"] = np.repeat(concept_name.strip(), len(codes))
        out = pd.concat([out, codes])
    else:
        logger.debug(f"No codes converted with target code type {target_code_type}")

    return out


def sql_row_exist(conn, table, column, value):
    # Execute and check if a result exists
    cur = conn.cursor()
    query = f"SELECT 1 FROM {table} WHERE {column} = ? LIMIT 1;"
    cur.execute(query, (value,))
    exists = cur.fetchone() is not None

    return exists


def write_code_errors(code_errors, code_errors_path):
    err_df = pd.DataFrame(
        [
            {
                "CONCEPT": ", ".join(err.codes[~err.mask].tolist()),
                "VOCABULARY": err.code_type,
                "SOURCE": err.codes_file,
                "CAUSE": err.message,
            }
            for err in code_errors
        ]
    )

    err_df = err_df.drop_duplicates()  # Remove Duplicates from Error file
    err_df = err_df.sort_values(by=["SOURCE", "VOCABULARY", "CONCEPT"])
    err_df.to_csv(code_errors_path, index=False, mode="w")


def write_vocab_version(phen_path):
    # write the vocab version files

    if not trud.VERSION_PATH.exists():
        raise FileNotFoundError(
            f"TRUD version path {trud.VERSION_PATH} does not exist, please check TRUD is installed"
        )

    if not omop.VERSION_PATH.exists():
        raise FileNotFoundError(
            f"OMOP version path {omop.VERSION_PATH} does not exist, please check OMOP is installed"
        )

    with trud.VERSION_PATH.open("r") as file:
        trud_version = yaml.safe_load(file)

    with omop.VERSION_PATH.open("r") as file:
        omop_version = yaml.safe_load(file)

    # Create the combined YAML structure
    version_data = {
        "versions": {
            "acmc": acmc.__version__,
            "trud": trud_version,
            "omop": omop_version,
        }
    }

    with open(phen_path / VOCAB_VERSION_FILE, "w") as file:
        yaml.dump(
            version_data,
            file,
            Dumper=util.QuotedDumper,
            default_flow_style=False,
            sort_keys=False,
            default_style='"',
        )


def map(phen_dir, target_code_type):
    logger.info(f"Processing phenotype: {phen_dir}")

    # Validate configuration
    validate(phen_dir)

    # initialise paths
    phen_path = Path(phen_dir)
    config_path = phen_path / CONFIG_FILE

    # load configuration
    with config_path.open("r") as file:
        config = yaml.safe_load(file)
    phenotype = config["phenotype"]

    if len(phenotype["map"]) == 0:
        raise ValueError(f"No map codes defined in the phenotype configuration")

    if target_code_type is not None and target_code_type not in phenotype["map"]:
        raise ValueError(
            f"Target code type {target_code_type} not in phenotype configuration map {phenotype['map']}"
        )

    if target_code_type is not None:
        map_target_code_type(phen_path, phenotype, target_code_type)
    else:
        for t in phenotype["map"]:
            map_target_code_type(phen_path, phenotype, t)

    logger.info(f"Phenotype processed successfully")


def map_target_code_type(phen_path, phenotype, target_code_type):

    logger.debug(f"Target coding format: {target_code_type}")
    codes_path = phen_path / CODES_DIR
    # Create output dataframe
    out = pd.DataFrame([])
    code_errors = []

    # Process each folder in codes section
    for concept_set in phenotype["concept_sets"]:
        logger.debug(f"--- {concept_set['file']} ---")

        # Load code file
        codes_file_path = Path(codes_path / concept_set["file"]["path"])
        df = read_table_file(codes_file_path)

        # process structural actions
        df = process_actions(df, concept_set)

        # Preprocessing & Validation Checks
        logger.debug("Processing and validating code formats")
        df, errors = preprocess_codes(
            df,
            concept_set,
            codes_file_path,
            target_code_type=target_code_type,
        )

        logger.debug(f"Length of errors from preprocess {len(errors)}")
        if len(errors) > 0:
            code_errors.extend(errors)
        logger.debug(f" Length of code_errors {len(code_errors)}")

        # Map
        # if processing a source coding list with categorical data
        if (
            "actions" in concept_set["file"]
            and "divide_col" in concept_set["file"]["actions"]
            and len(df) > 0
        ):
            divide_col = concept_set["file"]["actions"]["divide_col"]
            logger.debug(f"Action: Dividing Table by {divide_col}")
            logger.debug(f"column into: {df[divide_col].unique()}")
            df_grp = df.groupby(divide_col)
            for cat, grp in df_grp:
                if cat == concept_set["file"]["category"]:
                    grp = grp.drop(columns=[divide_col])  # delete categorical column
                    out = map_file(
                        grp, target_code_type, out, concept_name=concept_set["name"]
                    )
        else:
            out = map_file(df, target_code_type, out, concept_name=concept_set["name"])

    if len(code_errors) > 0:
        logger.error(f"The map processing has {len(code_errors)} errors")
        error_path = phen_path / MAP_DIR / "errors"
        error_path.mkdir(parents=True, exist_ok=True)
        error_filename = f"{target_code_type}-code-errors.csv"
        write_code_errors(code_errors, error_path / error_filename)

    # Check there is output from processing
    if len(out.index) == 0:
        logger.error(f"No output after map processing")
        raise Exception(
            f"No output after map processing, check config {str(config_path.resolve())}"
        )

    # Final processing
    out = out.reset_index(drop=True)
    out = out.drop_duplicates(subset=["CONCEPT_SET", "CONCEPT"])
    out = out.sort_values(by=["CONCEPT_SET", "CONCEPT"])

    # Save output to map directory
    output_filename = target_code_type + ".csv"
    map_path = phen_path / MAP_DIR / output_filename
    out.to_csv(map_path, index=False)
    logger.info(f"Saved mapped concepts to {str(map_path.resolve())}")

    # save concept sets as separate files
    concept_set_path = phen_path / CONCEPT_SET_DIR / target_code_type

    # empty the concept-set directory if it exists but keep the .git file
    git_items = [".git", ".gitkeep"]
    if concept_set_path.exists():
        for item in concept_set_path.iterdir():
            if item not in git_items:
                item.unlink()
    else:
        concept_set_path.mkdir(parents=True, exist_ok=True)

    # write each concept as a separate file
    for name, concept in out.groupby("CONCEPT_SET"):
        concept = concept.sort_values(by="CONCEPT")  # sort rows
        concept = concept.dropna(how="all", axis=1)  # remove empty cols
        concept = concept.reindex(
            sorted(concept.columns), axis=1
        )  # sort cols alphabetically
        filename = f"{name}.csv"
        concept_path = concept_set_path / filename
        concept.to_csv(concept_path, index=False)

    write_vocab_version(phen_path)

    logger.info(f"Phenotype processed target code type {target_code_type}")


def publish(phen_dir, remote_url):
    """Publishes updates to the phenotype by commiting all changes to the repo directory"""

    # Validate config
    validate(phen_dir)
    phen_path = Path(phen_dir)

    # load git repo and set the branch
    repo = git.Repo(phen_path)
    if DEFAULT_GIT_BRANCH in repo.branches:
        main_branch = repo.heads[DEFAULT_GIT_BRANCH]
        main_branch.checkout()
    else:
        raise AttributeError(
            f"Phen repo does not contain the default branch {DEFAULT_GIT_BRANCH}"
        )

    # check if any changes to publish
    if not repo.is_dirty() and not repo.untracked_files:
        logger.info("Nothing to publish, no changes to the repo")
        return

    # get major version from configuration file
    config_path = phen_path / CONFIG_FILE
    with config_path.open("r") as file:
        config = yaml.safe_load(file)
    match = re.match(r"v(\d+\.\d+)", config["phenotype"]["version"])
    major_version = match.group(1)

    # get latest minor version from git commit count
    commit_count = len(list(repo.iter_commits("HEAD")))

    # set version and write to config file so consistent with repo version
    next_minor_version = commit_count + 1
    version = f"v{major_version}.{next_minor_version}"
    logger.debug(f"New version: {version}")
    config["phenotype"]["version"] = version
    with open(config_path, "w") as file:
        yaml.dump(
            config,
            file,
            Dumper=util.QuotedDumper,
            default_flow_style=False,
            sort_keys=False,
            default_style='"',
        )

    # Add and commit changes to repo
    commit_message = f"Committing updates to phenotype {phen_path}"
    repo.git.add("--all")
    repo.index.commit(commit_message)

    # Create and push the tag
    if version in repo.tags:
        raise Exception(f"Tag {version} already exists in repo {phen_path}")
    repo.create_tag(version, message=f"Release {version}")
    logger.info(f"New version: {version}")

    # push to origin if a remote repo
    if remote_url is not None and "origin" not in repo.remotes:
        repo.create_remote("origin", remote_url)

    try:
        if "origin" in repo.remotes:
            logger.debug(f"Remote 'origin' is already set {repo.remotes.origin.url}")
            origin = repo.remotes.origin
            logger.info(f"Pushing main branch to {repo.remotes.origin.url}")
            origin.push("main")
            logger.info(f"Pushing tags to {repo.remotes.origin.url}")
            origin.push(tags=True)
            logger.debug("Changes pushed to 'origin'")
        else:
            logger.debug("Remote 'origin' is not set")
    except Exception as e:
        repo.delete_tag(version)
        repo.git.reset("--soft", "HEAD~1")
        raise e

    logger.info(f"Phenotype published successfully")


def export(phen_dir, version):
    """Exports a phen repo at a specific tagged version into a target directory"""
    logger.info(f"Exporting phenotype {phen_dir} at version {version}")

    # validate configuration
    validate(phen_dir)
    phen_path = Path(phen_dir)

    # load configuration
    config_path = phen_path / CONFIG_FILE
    with config_path.open("r") as file:
        config = yaml.safe_load(file)

    map_path = phen_path / MAP_DIR
    if not map_path.exists():
        logger.warning(f"Map path does not exist '{map_path}'")

    export_path = phen_path / OMOP_DIR
    # check export directory exists and if not create it
    if not export_path.exists():
        export_path.mkdir(parents=True)
        logger.debug(f"OMOP export directory '{export_path}' created.")

    # omop export db
    export_db_path = omop.export(
        map_path,
        export_path,
        config["phenotype"]["version"],
        config["phenotype"]["omop"],
    )

    # write to tables
    # export as csv
    logger.info(f"Phenotype exported successfully")


def copy(phen_dir, target_dir, version):
    """Copys a phen repo at a specific tagged version into a target directory"""

    # Validate
    validate(phen_dir)
    phen_path = Path(phen_dir)

    # Check target directory exists
    target_path = Path(target_dir)
    if not target_path.exists():
        raise FileNotFoundError(f"The target directory {target_path} does not exist")

    # Set copy directory
    copy_path = target_path / version
    logger.info(f"Copying repo {phen_path} to {copy_path}")

    if (
        copy_path.exists() and copy_path.is_dir()
    ):  # Check if it exists and is a directory
        copy = check_delete_dir(
            copy_path,
            f"The directory {str(copy_path.resolve())} already exists. Do you want to overwrite? (yes/no): ",
        )
    else:
        copy = True

    if not copy:
        logger.info(f"Not copying the version {version}")
        return

    logger.debug(f"Cloning repo from {phen_path} into {copy_path}...")
    repo = git.Repo.clone_from(phen_path, copy_path)

    # Check out the latest commit or specified version
    if version:
        # Checkout a specific version (e.g., branch, tag, or commit hash)
        logger.info(f"Checking out version {version}...")
        repo.git.checkout(version)
    else:
        # Checkout the latest commit (HEAD)
        logger.info(f"Checking out the latest commit...")
        repo.git.checkout("HEAD")

    logger.debug(f"Copied {phen_path} {repo.head.commit.hexsha[:7]} in {copy_path}")

    logger.info(f"Phenotype copied successfully")


# Convert concept_sets list into dictionaries
def extract_concepts(config_data):
    """Extracts concepts as {name: file_path} dictionary and a name set."""
    concepts_dict = {
        item["name"]: item["file"]["path"]
        for item in config_data["phenotype"]["concept_sets"]
    }
    name_set = set(concepts_dict.keys())
    return concepts_dict, name_set


def extract_clean_deepdiff_keys(diff, key_type):
    """
    Extracts clean keys from a DeepDiff dictionary.

    :param diff: DeepDiff result dictionary
    :param key_type: The type of change to extract (e.g., "dictionary_item_added", "dictionary_item_removed")
    :return: A set of clean key names
    """
    return {key.split("root['")[1].split("']")[0] for key in diff.get(key_type, [])}


def diff_config(old_config, new_config):
    report = f"\n# Changes to phenotype configuration\n"
    report += f"This compares changes in the phenotype configuration including added, removed and renamed concept sets and changes to concept set source concept code file paths\n\n"

    old_concepts, old_names = extract_concepts(old_config)
    new_concepts, new_names = extract_concepts(new_config)

    # Check added and removed names
    added_names = new_names - old_names  # Names that appear in new but not in old
    removed_names = old_names - new_names  # Names that were in old but not in new

    # find file path changes for unchanged names
    unchanged_names = old_names & new_names  # Names that exist in both
    file_diff = DeepDiff(
        {name: old_concepts[name] for name in unchanged_names},
        {name: new_concepts[name] for name in unchanged_names},
    )

    # Find renamed concepts (same file, different name)
    renamed_concepts = []
    for removed in removed_names:
        old_path = old_concepts[removed]
        for added in added_names:
            new_path = new_concepts[added]
            if old_path == new_path:
                renamed_concepts.append((removed, added))

    # Remove renamed concepts from added and removed sets
    for old_name, new_name in renamed_concepts:
        added_names.discard(new_name)
        removed_names.discard(old_name)

    # generate config report
    if added_names:
        report += "## Added Concepts\n"
        for name in added_names:
            report += f"- `{name}` (File: `{new_concepts[name]}`)\n"
        report += "\n"

    if removed_names:
        report += "## Removed Concepts\n"
        for name in removed_names:
            report += f"- `{name}` (File: `{old_concepts[name]}`)\n"
        report += "\n"

    if renamed_concepts:
        report += "## Renamed Concepts\n"
        for old_name, new_name in renamed_concepts:
            report += (
                f"- `{old_name}` ➝ `{new_name}` (File: `{old_concepts[old_name]}`)\n"
            )
        report += "\n"

    if "values_changed" in file_diff:
        report += "## Updated File Paths\n"
        for name, change in file_diff["values_changed"].items():
            old_file = change["old_value"]
            new_file = change["new_value"]
            clean_name = name.split("root['")[1].split("']")[0]
            report += (
                f"- `{clean_name}` changed file from `{old_file}` ➝ `{new_file}`\n"
            )
        report += "\n"

    if not (
        added_names
        or removed_names
        or renamed_concepts
        or file_diff.get("values_changed")
    ):
        report += "No changes in concept sets.\n"

    return report


def diff_map_files(old_map_path, new_map_path):
    old_output_files = [
        file.name
        for file in old_map_path.iterdir()
        if file.is_file() and not file.name.startswith(".")
    ]
    new_output_files = [
        file.name
        for file in new_map_path.iterdir()
        if file.is_file() and not file.name.startswith(".")
    ]

    # Convert the lists to sets for easy comparison
    old_output_set = set(old_output_files)
    new_output_set = set(new_output_files)

    # Outputs that are in old_output_set but not in new_output_set (removed files)
    removed_outputs = old_output_set - new_output_set
    # Outputs that are in new_output_set but not in old_output_set (added files)
    added_outputs = new_output_set - old_output_set
    # Outputs that are the intersection of old_output_set and new_output_set
    common_outputs = old_output_set & new_output_set

    report = f"\n# Changes to available translations\n"
    report += f"This compares the coding translations files available.\n\n"
    report += f"- Removed outputs: {sorted(list(removed_outputs))}\n"
    report += f"- Added outputs: {sorted(list(added_outputs))}\n"
    report += f"- Common outputs: {sorted(list(common_outputs))}\n\n"

    # Step N: Compare common outputs between versions
    report += f"# Changes to concepts in translation files\n\n"
    report += f"This compares the added and removed concepts in each of the coding translation files. Note that this might be different to the config.yaml if the translations have not been run for the current config.\n\n"
    for file in common_outputs:
        old_output = old_map_path / file
        new_output = new_map_path / file

        logger.debug(f"Old ouptput: {str(old_output.resolve())}")
        logger.debug(f"New ouptput: {str(new_output.resolve())}")

        df1 = pd.read_csv(old_output)
        df1 = df1[["CONCEPT", "CONCEPT_SET"]].groupby("CONCEPT_SET").count()
        df2 = pd.read_csv(new_output)
        df2 = df2[["CONCEPT", "CONCEPT_SET"]].groupby("CONCEPT_SET").count()

        # Check for added and removed concepts
        report += f"- File {file}\n"
        sorted_list = sorted(list(set(df1.index) - set(df2.index)))
        report += f"- Removed concepts {sorted_list}\n"
        sorted_list = sorted(list(set(df2.index) - set(df1.index)))
        report += f"- Added concepts {sorted_list}\n"

        # Check for changed concepts
        diff = df2 - df1  # diff in counts
        diff = diff[
            (~(diff["CONCEPT"] == 0.0)) & diff["CONCEPT"].notna()
        ]  # get non-zero counts
        s = "\n"
        if len(diff.index) > 0:
            for concept, row in diff.iterrows():
                s += "\t - {} {}\n".format(concept, row["CONCEPT"])
            report += f"- Changed concepts {s}\n\n"
        else:
            report += f"- Changed concepts []\n\n"

    return report


def diff(phen_dir, phen_old_dir):
    """Compare the differences between two versions of a phenotype"""

    # validate phenotypes
    validate(phen_old_dir)
    validate(phen_dir)

    # get old and new config
    old_phen_path = Path(phen_old_dir)
    old_config = old_phen_path / CONFIG_FILE
    with old_config.open("r") as file:
        old_config = yaml.safe_load(file)
    new_phen_path = Path(phen_dir)
    new_config = new_phen_path / CONFIG_FILE
    with new_config.open("r") as file:
        new_config = yaml.safe_load(file)

    # write report heading
    report = f"# Phenotype Comparison Report\n"
    report += f"## Original phenotype\n"
    report += f"  - {old_config['phenotype']['omop']['vocabulary_id']}\n"
    report += f"  - {old_config['phenotype']['version']}\n"
    report += f"  - {str(old_phen_path.resolve())}\n"
    report += f"## Changed phenotype:\n"
    report += f"  - {new_config['phenotype']['omop']['vocabulary_id']}\n"
    report += f"  - {new_config['phenotype']['version']}\n"
    report += f"  - {str(new_phen_path.resolve())}\n"

    # Step 1: check differences configuration files
    # Convert list of dicts into a dict: {name: file}
    report += diff_config(old_config, new_config)

    # Step 2: check differences between map files
    # List files from output directories
    old_map_path = old_phen_path / MAP_DIR
    new_map_path = new_phen_path / MAP_DIR
    report += diff_map_files(old_map_path, new_map_path)

    # initialise report file
    report_file_name = old_phen_path.name + "_diff.md"
    report_path = new_phen_path / report_file_name
    logger.debug(f"Writing to report file {str(report_path.resolve())}")
    report_file = open(report_path, "w")
    report_file.write(report)
    report_file.close()

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