import argparse
import logging
import sys

import pyspark.sql.functions as f

from pyspark.sql.dataframe import DataFrame

from common.evidence import initialize_sparksession, write_evidence_strings


def main(chembl_evidence: str, predictions: str, output_file: str) -> None:
    """
    This module adds the studyStopReasonCategories to the ChEMBL evidence as a result of the categorisation of the clinical trial reason to stop.
    Args:
        chembl_evidence: Input gzipped JSON with the evidence submitted by ChEMBL.
        predictions: Input JSON containing the categories of the clinical trial reason to stop. 
        Instructions for applying the ML model here: https://github.com/ireneisdoomed/stopReasons.
        output_file: Output gzipped json file containing the ChEMBL evidence with the additional studyStopReasonCategories field.
        log_file: Destination of the logs generated by this script. Defaults to None.
    """
    logging.info(f'ChEMBL evidence JSON file: {chembl_evidence}')
    logging.info(f'Classes of reason to stop table: {predictions}')

    chembl_df = spark.read.json(chembl_evidence).repartition(200).persist()
    predictions_df = (
        spark.read.json(predictions)
        .transform(prettify_subclasses)
        .distinct()
    )

    # Join datasets
    early_stopped_evd_df = (
        # Evidence with a given reason to stop is always supported by a single NCT ID
        chembl_df.filter(f.col('studyStopReason').isNotNull())
        .select(
            "*", f.explode(f.col("urls.url")).alias("nct_id")
        ).withColumn("nct_id", f.element_at(f.split(f.col("nct_id"), "/"), -1))
        .join(predictions_df, on='nct_id', how='left').drop('nct_id')
        .distinct()
    )

    # We expect that ~10% of evidence strings have a reason to stop assigned
    # It is asserted that this fraction is between 9 and 11% of the total count
    total_count = chembl_df.count()
    early_stopped_count = early_stopped_evd_df.count()

    if not 0.08 < early_stopped_count / total_count < 0.11:
        raise AssertionError(f'The fraction of evidence with a CT reason to stop class is not as expected ({early_stopped_count / total_count}).')

    logging.info('Evidence strings have been processed. Saving...')
    enriched_chembl_df = (
        chembl_df.filter(f.col('studyStopReason').isNull())
        .unionByName(early_stopped_evd_df, allowMissingColumns=True)
    )
    assert enriched_chembl_df.count() == chembl_df.count()
    write_evidence_strings(enriched_chembl_df, output_file)

    logging.info(f'{total_count} evidence strings have been saved to {output_file}. Exiting.')

def prettify_subclasses(predictions_df: DataFrame) -> DataFrame:
    """
    List of categories must be converted formatted with a nice name.
    """

    CATEGORIESMAPPINGS = {
        'Business_Administrative': 'Business or administrative',
        'Logistics_Resources': 'Logistics or resources',
        'Covid19': 'COVID-19',
        'Safety_Sideeffects': 'Safety or side effects',
        'Endpoint_Met': 'Met endpoint',
        'Insufficient_Enrollment': 'Insufficient enrollment',
        'Negative': 'Negative',
        'Study_Design': 'Study design',
        'Invalid_Reason': 'Invalid reason',
        'Study_Staff_Moved': 'Study staff moved',
        'Another_Study': 'Another study',
        'No_Context': 'No context',
        'Regulatory': 'Regulatory',
        'Interim_Analysis': 'Interim analysis',
        'Success': 'Success',
        'Ethical_Reason': 'Ethical reason',
        'Insufficient_Data': 'Insufficient data',
        'Uncategorised': 'Uncategorised'
    }

    sub_mapping_col = f.map_from_entries(f.array(*[f.struct(f.lit(k), f.lit(v)) for k, v in CATEGORIESMAPPINGS.items()]))

    return (
        predictions_df  
        .select('nct_id', 'subclasses', sub_mapping_col.alias('prettyStopReasonsMap'))
        # Create a MapType column to convert each element of the subclasses array
        .withColumn('studyStopReasonCategories', f.expr('transform(subclasses, x -> element_at(prettyStopReasonsMap, x))'))
        .drop('subclasses', 'prettyStopReasonsMap')
    )

def get_parser():
    """Get parser object for script ChEMBL.py."""
    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument(
        '--chembl_evidence',
        help='Input gzipped JSON with the evidence submitted by ChEMBL',
        type=str,
        required=True,
    )
    parser.add_argument(
        '--predictions',
        help='Input TSV containing the categories of the clinical trial reason to stop. Instructions for applying the ML model here: https://github.com/ireneisdoomed/stopReasons.',
        type=str,
        required=True,
    )
    parser.add_argument(
        '--output', help='Output gzipped json file following the target safety liabilities data model.',
        type=str,
        required=True
    )
    parser.add_argument(
        '--log_file',
        help='Destination of the logs generated by this script. Defaults to None',
        type=str,
        nargs='?',
        default=None
    )

    return parser


if __name__ == '__main__':
    args = get_parser().parse_args()

    # Logger initializer. If no log_file is specified, logs are written to stderr
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s %(levelname)s %(module)s - %(funcName)s: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S',
    )
    if args.log_file:
        logging.config.fileConfig(filename=args.log_file)
    else:
        logging.StreamHandler(sys.stderr)

    global spark
    spark = initialize_sparksession()

    main(
        chembl_evidence=args.chembl_evidence,
        predictions=args.predictions,
        output_file=args.output
    )

import argparse
import logging

from pyspark.conf import SparkConf
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import array, col, lit, monotonically_increasing_id, struct, trim
from pyspark.sql.types import StringType, StructType, TimestampType

from common.ontology import add_efo_mapping
from common.evidence import write_evidence_strings


def main(input_file: str, output_file: str, cache_dir: str, local: bool = False) -> None:

    # Initialize spark session
    if local:
        sparkConf = (
            SparkConf()
            .set('spark.driver.memory', '15g')
            .set('spark.executor.memory', '15g')
            .set('spark.driver.maxResultSize', '0')
            .set('spark.debug.maxToStringFields', '2000')
            .set('spark.sql.execution.arrow.maxRecordsPerBatch', '500000')
        )
        spark = (
            SparkSession.builder
            .config(conf=sparkConf)
            .master('local[*]')
            .getOrCreate()
        )
    else:
        sparkConf = (
            SparkConf()
            .set('spark.driver.maxResultSize', '0')
            .set('spark.debug.maxToStringFields', '2000')
            .set('spark.sql.execution.arrow.maxRecordsPerBatch', '500000')
        )
        spark = (
            SparkSession.builder
            .config(conf=sparkConf)
            .getOrCreate()
        )

    # Read and process Clingen's table into evidence strings

    clingen_df = read_input_file(input_file, spark_instance=spark)
    logging.info('Gene Validity Curations table has been imported. Processing evidence strings.')

    evidence_df = process_clingen(clingen_df)

    evidence_df = add_efo_mapping(evidence_strings=evidence_df, spark_instance=spark, ontoma_cache_dir=cache_dir)
    logging.info('Disease mappings have been added.')

    write_evidence_strings(evidence_df, output_file)
    logging.info(f'{evidence_df.count()} evidence strings have been saved to {output_file}')


def read_input_file(input_file: str, spark_instance) -> DataFrame:
    """
    Reads Gene Validity Curations CSV file into a Spark DataFrame forcing the schema
    The first 6 rows of this file include metadata that needs to be dropped
    """

    clingen_schema = (
        StructType()
        .add('GENE SYMBOL', StringType())
        .add('GENE ID (HGNC)', StringType())
        .add('DISEASE LABEL', StringType())
        .add('DISEASE ID (MONDO)', StringType())
        .add('MOI', StringType())
        .add('SOP', StringType())
        .add('CLASSIFICATION', StringType())
        .add('ONLINE REPORT', StringType())
        .add('CLASSIFICATION DATE', TimestampType())
        .add('GCEP', StringType())
    )

    clingen_df = (
        spark_instance.read.csv(input_file, schema=clingen_schema)
        # The first 6 rows of the GVC file include metadata that needs to be dropped
        .withColumn('idx', monotonically_increasing_id())
        .filter(col('idx') > 5)
        .drop('idx')
    )

    return clingen_df


def process_clingen(clingen_df: DataFrame) -> DataFrame:
    """
    The JSON Schema format is applied to the df
    """

    return (
        clingen_df
        .withColumn('datasourceId', lit('clingen'))
        .withColumn('datatypeId', lit('genetic_literature'))
        .withColumn('urls', struct(col('ONLINE REPORT').alias('url')))

        .select(
            'datasourceId', 'datatypeId',
            trim(col('GENE SYMBOL')).alias('targetFromSourceId'),
            col('DISEASE LABEL').alias('diseaseFromSource'),
            col('DISEASE ID (MONDO)').alias('diseaseFromSourceId'),
            array(col('MOI')).alias('allelicRequirements'),
            array(col('urls')).alias('urls'),
            col('CLASSIFICATION').alias('confidence'),
            col('GCEP').alias('studyId')
        )
    )


if __name__ == "__main__":

    # Parse CLI arguments
    parser = argparse.ArgumentParser(description='Parse ClinGen gene-disease associations from Gene Validity Curations')
    parser.add_argument('--input_file',
                        help='Name of csv file downloaded from https://search.clinicalgenome.org/kb/gene-validity',
                        type=str, required=True)
    parser.add_argument('--output_file',
                        help='Absolute path of the gzipped JSON evidence file.',
                        type=str, required=True)
    parser.add_argument('--log_file', type=str,
                        help='Optional filename to redirect the logs into.')
    parser.add_argument('--cache_dir', required=False, help='Directory to store the OnToma cache files in.')
    parser.add_argument(
        '--local', action='store_true', required=False, default=False,
        help='Flag to indicate if the script is executed locally or on the cluster'
    )
    args = parser.parse_args()

    # Initialize logging:
    logging.basicConfig(
        level=logging.INFO,
        format='%(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S',
    )
    if args.log_file:
        logging.config.fileConfig(filename=args.log_file)

    # Report input data:
    logging.info(f'Clingen input file path: {args.input_file}')
    logging.info(f'Evidence output file path: {args.output_file}')
    logging.info(f'Cache directory: {args.cache_dir}')

    main(
        input_file=args.input_file, output_file=args.output_file,
        cache_dir=args.cache_dir, local=args.local
    )

import argparse
import logging
import sys

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.types import FloatType
from pyspark.sql.functions import col, collect_set, split, element_at, struct, lit

from common.evidence import write_evidence_strings

# A few genes do not have Ensembl IDs in the data file provided
CRISPR_SYMBOL_MAPPING = {
    'CASC5': 'ENSG00000137812',
    'CIRH1A': 'ENSG00000141076',
    'EFTUD1': 'ENSG00000140598',
    'ENSG00000163660': 'ENSG00000163660',
    'KIAA0947': 'ENSG00000164151',
    'KIAA1432': 'ENSG00000107036',
    'NDNL2': 'ENSG00000185115',
    'SRPR': 'ENSG00000182934',
    'ZNF259': 'ENSG00000109917'
}


def main(desc_file, evid_file, cell_file, out_file):
    sparkConf = (
        SparkConf()
        .set('spark.driver.memory', '15g')
        .set('spark.executor.memory', '15g')
        .set('spark.driver.maxResultSize', '0')
        .set('spark.debug.maxToStringFields', '2000')
        .set('spark.sql.execution.arrow.maxRecordsPerBatch', '500000')
    )
    spark = (
        SparkSession.builder
        .config(conf=sparkConf)
        .master('local[*]')
        .getOrCreate()
    )

    # Log parameters:
    logging.info(f'Evidence file: {evid_file}')
    logging.info(f'Description file: {desc_file}')
    logging.info(f'Cell type annotation: {cell_file}')
    logging.info(f'Output file: {out_file}')

    # Read files:
    evidence_df = (
        spark.read.csv(evid_file, sep='\t', header=True)
        .drop('pmid', 'gene_set_name', 'disease_name')
    )
    cell_lines_df = spark.read.csv(cell_file, sep='\t', header=True)
    description_df = spark.read.csv(desc_file, sep='\t', header=True)

    # Logging dataframe stats:
    logging.info(f'Number of evidence: {evidence_df.count()}')
    logging.info(f'Number of descriptions: {description_df.count()}')
    logging.info(f'Number of cell/tissue annotation: {cell_lines_df.count()}')

    # Tissues and cancer types are annotated together in the same column (tissue_or_cancer_type)
    # To disambiguate one from another, the column is combined with the cell lines
    # First on the tissue level:
    tissue_desc = (
        description_df
        .withColumnRenamed('tissue_or_cancer_type', 'tissue')
        .join(cell_lines_df, on='tissue', how='inner')
    )

    # And then on the disease level:
    cell_desc = (
        description_df
        .withColumnRenamed('tissue_or_cancer_type', 'diseaseFromSource')
        .join(cell_lines_df, on='diseaseFromSource', how='inner')
    )

    merged_annotation = (
        # Concatenating the above generated dataframes:
        cell_desc
        .union(tissue_desc)

        # Aggregating by disease and method:
        .groupBy('diseaseFromSource', 'efo_id', 'method')

        # The cell annotation is aggregated in a list of struct:
        .agg(
            collect_set(struct(
                col('name'), col('id'), col('tissue'), col('tissueId')
            )).alias('diseaseCellLines')
        )
        .drop('method')
    )

    # Joining merged annotation with evidence:
    pooled_evidence_df = (
        evidence_df
        .select(
            col('target_id').alias('targetFromSourceId'),
            col('disease_id').alias('efo_id'),
            col('score').alias('resourceScore').cast(FloatType()),
        )

        # Some of the target identifier are not Ensembl Gene id - replace them:
        .replace(to_replace=CRISPR_SYMBOL_MAPPING, subset=['targetFromSourceId'])

        # Merging with descriptions:
        .join(merged_annotation, on='efo_id', how='outer')

        # From EFO uri, generate EFO id:
        .withColumn(
            'diseaseFromSourceMappedId',
            element_at(split(col('efo_id'), '/'), -1).alias('diseaseFromSourceMappedId')
        )
        .drop('efo_id')

        # Adding constants:
        .withColumn('datasourceId', lit('crispr'))
        .withColumn('datatypeId', lit('affected_pathway'))
        .persist()
    )

    logging.info(f'Saving {pooled_evidence_df.count()} CRISPR evidence in JSON format, to: {out_file}')

    write_evidence_strings(pooled_evidence_df, out_file)


if __name__ == "__main__":

    # Parse CLI arguments:
    parser = argparse.ArgumentParser(
        description='Parse essential cancer genes identified using CRISPR assays and prioritised in Project Score'
    )
    parser.add_argument('-d', '--descriptions_file',
                        help='Name of tsv file with the description of the method per cancer type',
                        type=str, required=True)
    parser.add_argument('-e', '--evidence_file',
                        help='Name of tsv file with the priority score',
                        type=str, required=True)
    parser.add_argument('-c', '--cell_types_file',
                        help='Name of tsv file with cell line names per cancer type',
                        type=str, required=True)
    parser.add_argument('-o', '--output_file',
                        help='Name of evidence file. (gzip compressed json)',
                        type=str, required=True)
    parser.add_argument('-l', '--log_file',
                        help='Name of log file. If not provided logs are written to standard error.',
                        type=str, required=False)

    args = parser.parse_args()
    desc_file = args.descriptions_file
    evid_file = args.evidence_file
    cell_file = args.cell_types_file
    out_file = args.output_file

    # If no logfile is specified, logs are written to stderr
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s %(levelname)s %(module)s - %(funcName)s: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S',
    )
    if args.log_file:
        logging.config.fileConfig(filename=args.log_file)
    else:
        logging.StreamHandler(sys.stderr)

    main(desc_file, evid_file, cell_file, out_file)

import argparse
from functools import reduce
import logging
import sys
from pyspark.sql import SparkSession, DataFrame

from common.evidence import initialize_sparksession, write_evidence_strings
from BrainCRISPR import CRISPRBrain


def crispr_brain_wrapper(spark: SparkSession, crispr_brain_mapping: str) -> DataFrame:
    """Simple wrapper around CRISPRBrain class to generate and extract evidence."""
    crispr_brain_parser = CRISPRBrain(spark, crispr_brain_mapping)
    crispr_brain_parser.create_crispr_brain_evidence()
    return crispr_brain_parser.get_dataframe()


def main(spark: SparkSession, crispr_brain_mapping: str) -> DataFrame:
    crispr_screen_evidence_sets = [
        # Generate evidence from CRISPR Brain:
        crispr_brain_wrapper(spark, crispr_brain_mapping).persist(),
        # Further datasets can be added here:
    ]

    # Combining all sources into one single dataframe:
    combined_evidence = reduce(
        lambda df1, df2: df1.unionByName(df2, allowMissingColumns=True),
        crispr_screen_evidence_sets,
    )

    logging.info(f"Total number of CRISPR screen evidence: {combined_evidence.count()}")
    return combined_evidence


def get_parser():
    """Parsing command line argument for crispr screen ingestion."""
    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument(
        "--crispr_brain_mapping",
        help="Curation for CRISPR Brain dataset containing disease mapping and contrast.",
        type=str,
        required=True,
    )
    parser.add_argument(
        "--output",
        help="Output gzipped json file.",
        type=str,
        required=True,
    )
    parser.add_argument(
        "--log_file",
        help="Destination of the logs generated by this script. Defaults to None",
        type=str,
        nargs="?",
        default=None,
    )
    return parser


if __name__ == "__main__":
    args = get_parser().parse_args()

    # Logger initializer. If no log_file is specified, logs are written to stderr
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(module)s - %(funcName)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    if args.log_file:
        logging.FileHandler(args.log_file)
    else:
        logging.StreamHandler(sys.stderr)

    spark = initialize_sparksession()

    evidence_df = main(
        spark,
        args.crispr_brain_mapping,
    )

    write_evidence_strings(evidence_df, args.output)
    logging.info(f"Evidence strings have been saved to {args.output}. Exiting.")

import argparse
import logging
import sys

from pyspark.conf import SparkConf
from pyspark.sql import DataFrame, SparkSession, functions as f, types as t

from common.ontology import add_efo_mapping
from common.evidence import initialize_sparksession, write_evidence_strings


G2P_mutationCsq2functionalCsq = {
    "uncertain": "SO_0002220",  # function_uncertain_variant
    "absent gene product": "SO_0002317",  # absent_gene_product
    "altered gene product structure": "SO_0002318",  # altered_gene_product_structure
    "5_prime or 3_prime UTR mutation": "SO_0001622",  # UTR_variant
    "increased gene product level": "SO_0002315",  # increased_gene_product_level
    "cis-regulatory or promotor mutation": "SO_0001566",  # regulatory_region_variant
}


def main(
    dd_file: str,
    eye_file: str,
    skin_file: str,
    cancer_file: str,
    cardiac_file: str,
    output_file: str,
    cache_dir: str,
    local: bool = False,
) -> None:
    # Initialize spark session
    spark = initialize_sparksession()

    # Read and process G2P's tables into evidence strings
    gene2phenotype_df = read_input_files(
        spark, dd_file, eye_file, skin_file, cancer_file, cardiac_file
    )
    logging.info(
        "Gene2Phenotype panels have been imported. Processing evidence strings."
    )

    evidence_df = process_gene2phenotype(gene2phenotype_df)

    evidence_df = add_efo_mapping(
        evidence_strings=evidence_df, ontoma_cache_dir=cache_dir, spark_instance=spark
    )
    logging.info("Disease mappings have been added.")

    # Saving data:
    write_evidence_strings(evidence_df, output_file)
    logging.info(
        f"{evidence_df.count()} evidence strings have been saved to {output_file}"
    )


def read_input_files(
    spark: SparkSession,
    dd_file: str,
    eye_file: str,
    skin_file: str,
    cancer_file: str,
    cardiac_file: str,
) -> DataFrame:
    """
    Reads G2P's panel CSV files into a Spark DataFrame forcing the schema
    """

    gene2phenotype_schema = (
        t.StructType()
        .add("gene symbol", t.StringType())
        .add("gene mim", t.IntegerType())
        .add("disease name", t.StringType())
        .add("disease mim", t.StringType())
        .add("confidence category", t.StringType())
        .add("allelic requirement", t.StringType())
        .add("mutation consequence", t.StringType())
        .add("phenotypes", t.StringType())
        .add("organ specificity list", t.StringType())
        .add("pmids", t.StringType())
        .add("panel", t.StringType())
        .add("prev symbols", t.StringType())
        .add("hgnc id", t.IntegerType())
        .add("gene disease pair entry date", t.TimestampType())
        .add("cross cutting modifier", t.StringType())
        .add("mutation consequence flag", t.StringType())
        .add("confidence value flag", t.StringType())
        .add("comments", t.StringType())
        .add("variant consequence", t.StringType())
        .add("disease ontology", t.StringType())
    )

    return (
        spark.read.option("multiLine", True)
        .option("encoding", "UTF-8")
        .csv(
            [dd_file, eye_file, skin_file, cancer_file, cardiac_file],
            schema=gene2phenotype_schema,
            enforceSchema=True,
            header=True,
            sep=",",
            quote='"',
        )
        # Dropping one incomplete evidence due to parsing problem:
        # All data: 4094
        # Filtered data: 4093
        .filter(f.col("gene symbol").isNotNull() & f.col("panel").isNotNull())
    )


def process_gene2phenotype(gene2phenotype_df: DataFrame) -> DataFrame:
    """
    The JSON Schema format is applied to the df
    """

    return gene2phenotype_df.select(
        # Split pubmed IDs to list:
        f.split(f.col("pmids"), ";").alias("literature"),
        # Phenotypes are excluded from the schema:
        # f.split(f.col("phenotypes"), ";").alias("phenotypes"),
        # Renaming a few columns:
        f.col("gene symbol").alias("targetFromSourceId"),
        f.col("panel").alias("studyId"),
        f.col("confidence category").alias("confidence"),
        # Parsing allelic requirements:
        f.when(
            f.col("allelic requirement").isNotNull(),
            f.array(f.col("allelic requirement")),
        ).alias("allelicRequirements"),
        # Parsing disease from source identifier:
        f.when(f.col("disease ontology").isNotNull(), f.col("disease ontology"))
        .when(
            (~f.col("disease mim").contains("No disease mim"))
            & (f.col("disease mim").isNotNull()),
            f.concat(f.lit("OMIM:"), f.col("disease mim")),
        )
        .otherwise(f.lit(None))
        .alias("diseaseFromSourceId"),
        # Cleaning disease names:
        f.regexp_replace(f.col("disease name"), r".+-related ", "").alias(
            "diseaseFromSource"
        ),
        # Map functional consequences:
        translate(G2P_mutationCsq2functionalCsq)("mutation consequence").alias(
            "variantFunctionalConsequenceId"
        ),
        # Adding constant columns:
        f.lit("gene2phenotype").alias("datasourceId"),
        f.lit("genetic_literature").alias("datatypeId"),
    )


def translate(mapping):
    """
    Mapping consequences - to SO codes
    """

    def translate_(col):
        return mapping.get(col)

    return f.udf(translate_, t.StringType())


if __name__ == "__main__":
    # Parse CLI arguments
    parser = argparse.ArgumentParser(
        description="Parse Gene2Phenotype gene-disease files downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads/"
    )
    parser.add_argument(
        "-d",
        "--dd_panel",
        help="DD panel file downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-e",
        "--eye_panel",
        help="Eye panel file downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-s",
        "--skin_panel",
        help="Skin panel file downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-c",
        "--cancer_panel",
        help="Cancer panel file downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-cr",
        "--cardiac_panel",
        help="Cardiac panel file downloaded from https://www.ebi.ac.uk/gene2phenotype/downloads",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-o",
        "--output_file",
        help="Absolute path of the gzipped, JSON evidence file.",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-l", "--log_file", help="Filename to store the parser logs.", type=str
    )
    parser.add_argument(
        "--cache_dir",
        required=False,
        help="Directory to store the OnToma cache files in.",
    )
    parser.add_argument(
        "--local",
        action="store_true",
        required=False,
        default=False,
        help="Flag to indicate if the script is executed locally or on the cluster",
    )

    args = parser.parse_args()

    # Get parameters
    dd_file = args.dd_panel
    eye_file = args.eye_panel
    skin_file = args.skin_panel
    cancer_file = args.cancer_panel
    cardiac_file = args.cardiac_panel
    output_file = args.output_file
    log_file = args.log_file
    cache_dir = args.cache_dir
    local = args.local

    # Configure logger:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(module)s - %(funcName)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    if log_file:
        logging.config.fileConfig(fname=log_file)
    else:
        logging.StreamHandler(sys.stderr)

    # Report input data:
    logging.info(f"DD panel file: {dd_file}")
    logging.info(f"Eye panel file: {eye_file}")
    logging.info(f"Skin panel file: {skin_file}")
    logging.info(f"Cancer panel file: {cancer_file}")
    logging.info(f"Cardiac panel file: {cardiac_file}")

    # Calling main:
    main(
        dd_file, eye_file, skin_file, cancer_file, cardiac_file, output_file, cache_dir
    )