core.py

import asyncio
import json
import logging
import os
import pickle
from datetime import datetime
from multiprocessing import freeze_support
from typing import Tuple

import click
from pathlib import Path
from cdisc_rules_engine.config import config
from cdisc_rules_engine.enums.default_file_paths import DefaultFilePaths
from cdisc_rules_engine.enums.progress_parameter_options import ProgressParameterOptions
from cdisc_rules_engine.enums.report_types import ReportTypes
from cdisc_rules_engine.enums.dataformat_types import DataFormatTypes
from cdisc_rules_engine.models.validation_args import Validation_args
from cdisc_rules_engine.models.test_args import TestArgs
from scripts.run_validation import run_validation
from scripts.test_rule import test as test_rule
from cdisc_rules_engine.services.cache.cache_populator_service import CachePopulator
from cdisc_rules_engine.services.cache.cache_service_factory import CacheServiceFactory
from cdisc_rules_engine.services.cdisc_library_service import CDISCLibraryService
from cdisc_rules_engine.utilities.utils import (
    generate_report_filename,
    get_rules_cache_key,
    get_local_cache_key,
)
from scripts.list_dataset_metadata_handler import list_dataset_metadata_handler
from version import __version__


def valid_data_file(data_path: list) -> Tuple[list, set]:
    allowed_formats = [format.value for format in DataFormatTypes]
    found_formats = set()
    file_list = []
    for file in data_path:
        file_extension = os.path.splitext(file)[1][1:].upper()
        if file_extension in allowed_formats:
            found_formats.add(file_extension)
            file_list.append(file)
    if len(found_formats) > 1:
        return [], found_formats
    elif len(found_formats) == 1:
        return file_list, found_formats


@click.group()
def cli():
    pass


@click.command()
@click.option(
    "-ca",
    "--cache",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.option(
    "-ps",
    "--pool-size",
    default=10,
    type=int,
    help="Number of parallel processes for validation",
)
@click.option(
    "-d",
    "--data",
    required=False,
    help="Path to directory containing data files",
)
@click.option(
    "-dp",
    "--dataset-path",
    required=False,
    multiple=True,
    help="Absolute path to dataset file",
)
@click.option(
    "-l",
    "--log-level",
    default="disabled",
    type=click.Choice(["info", "debug", "error", "critical", "disabled", "warn"]),
    help="Sets log level for engine logs, logs are disabled by default",
)
@click.option(
    "-rt",
    "--report-template",
    default=DefaultFilePaths.EXCEL_TEMPLATE_FILE.value,
    help="File path of report template to use for excel output",
)
@click.option(
    "-s",
    "--standard",
    required=True,
    default=None,
    help="CDISC standard to validate against",
)
@click.option(
    "-v",
    "--version",
    required=True,
    default=None,
    help="Standard version to validate against",
)
@click.option(
    "-ct",
    "--controlled-terminology-package",
    multiple=True,
    help=(
        "Controlled terminology package to validate against, "
        "can provide more than one"
    ),
)
@click.option(
    "-o",
    "--output",
    default=generate_report_filename(datetime.now().isoformat()),
    help="Report output file destination",
)
@click.option(
    "-of",
    "--output-format",
    multiple=True,
    default=[ReportTypes.XLSX.value],
    type=click.Choice(ReportTypes.values(), case_sensitive=False),
    help="Output file format",
)
@click.option(
    "-rr",
    "--raw-report",
    default=False,
    show_default=True,
    is_flag=True,
    help=(
        "Report in a raw format as it is generated by the engine. "
        "This flag must be used only with --output-format JSON."
    ),
)
@click.option(
    "-dv",
    "--define-version",
    type=click.Choice(["2-1", "2-0", "2.0", "2.1"]),
    help="Define-XML version used for validation",
)
@click.option("--whodrug", help="Path to directory with WHODrug dictionary files")
@click.option("--meddra", help="Path to directory with MedDRA dictionary files")
@click.option("--loinc", help="Path to directory with LOINC dictionary files")
@click.option("--medrt", help="Path to directory with MEDRT dictionary files")
@click.option(
    "--rules",
    "-r",
    multiple=True,
    help="specify rule core ID ex. CORE-000001. Can be specified multiple times",
)
@click.option(
    "--local_rules",
    "-lr",
    required=False,
    type=click.Path(exists=True, readable=True, resolve_path=True),
    help="path to directory containing local rules.",
)
@click.option(
    "--local_rules_cache",
    "-lrc",
    required=False,
    is_flag=True,
    default=False,
    help=(
        "flag to run a validation using the local rules in the cache"
        "must be provided with a local rules id -lri to specify the local rules to use"
    ),
)
@click.option(
    "--local_rules_id",
    "-lri",
    required=False,
    help=(
        "local rule ID of rules to use from the local rules cache"
        "for the validation run. Must be provided with the -lrc flag"
    ),
)
@click.option(
    "-p",
    "--progress",
    default=ProgressParameterOptions.BAR.value,
    type=click.Choice(ProgressParameterOptions.values()),
    help=(
        "Defines how to display the validation progress. "
        'By default a progress bar like "[████████████████████████████--------]   78%"'
        "is printed."
    ),
)
@click.option("-dxp", "--define-xml-path", required=False, help="Path to Define-XML")
@click.pass_context
def validate(
    ctx,
    cache: str,
    pool_size: int,
    data: str,
    dataset_path: Tuple[str],
    log_level: str,
    report_template: str,
    standard: str,
    version: str,
    controlled_terminology_package: Tuple[str],
    output: str,
    output_format: Tuple[str],
    raw_report: bool,
    define_version: str,
    whodrug: str,
    meddra: str,
    loinc: str,
    medrt: str,
    rules: Tuple[str],
    local_rules: str,
    local_rules_cache: bool,
    local_rules_id: str,
    progress: str,
    define_xml_path: str,
):
    """
    Validate data using CDISC Rules Engine

    Example:

    python core.py -s SDTM -v 3.4 -d /path/to/datasets
    """

    # Validate conditional options
    logger = logging.getLogger("validator")

    if raw_report is True:
        if not (len(output_format) == 1 and output_format[0] == ReportTypes.JSON.value):
            logger.error(
                "Flag --raw-report can be used only when --output-format is JSON"
            )
            ctx.exit()

    cache_path: str = os.path.join(os.path.dirname(__file__), cache)

    print(os.path.dirname(__file__))

    if data:
        if dataset_path:
            logger.error(
                "Argument --dataset-path cannot be used together with argument --data"
            )
            ctx.exit()
        dataset_paths, found_formats = valid_data_file(
            [str(Path(data).joinpath(fn)) for fn in os.listdir(data)]
        )
        if len(found_formats) > 1:
            logger.error(
                f"Argument --data contains more than one allowed file format ({', '.join(found_formats)})."  # noqa: E501
            )
            ctx.exit()
    elif dataset_path:
        dataset_paths, found_formats = valid_data_file([dp for dp in dataset_path])
        if len(found_formats) > 1:
            logger.error(
                f"Argument --dataset_path contains more than one allowed file format ({', '.join(found_formats)})."  # noqa: E501
            )
            ctx.exit()
    else:
        logger.error(
            "You must pass one of the following arguments: --dataset-path, --data"
        )
        # no need to define dataset_paths here, the program execution will stop
        ctx.exit()

    run_validation(
        Validation_args(
            cache_path,
            pool_size,
            dataset_paths,
            log_level,
            report_template,
            standard,
            version,
            set(controlled_terminology_package),  # avoiding duplicates
            output,
            set(output_format),  # avoiding duplicates
            raw_report,
            define_version,
            whodrug,
            meddra,
            loinc,
            medrt,
            rules,
            local_rules,
            local_rules_cache,
            local_rules_id,
            progress,
            define_xml_path,
        )
    )


@click.command()
@click.option(
    "-c",
    "--cache_path",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.option(
    "--apikey",
    envvar="CDISC_LIBRARY_API_KEY",
    help=(
        "CDISC Library api key. "
        "Can be provided in the environment "
        "variable CDISC_LIBRARY_API_KEY"
    ),
    required=True,
)
@click.option(
    "-lr",
    "--local_rules",
    help=(
        "Relative path to folder containing local rules in yaml or JSON formats"
        "to be added to the cache.  Must be provided in conjunction with -lri"
    ),
)
@click.option(
    "-lri",
    "--local_rules_id",
    help=(
        "Custom ID attached to local rules added to the cache"
        "used for granular control of local rules when removing"
        "and validating from the cache.  Must be given when adding"
        "local rules to the cache."
    ),
)
@click.option(
    "-rlr",
    "--remove_rules",
    help="removes all local rules from the cache",
)
@click.pass_context
def update_cache(
    ctx: click.Context,
    cache_path: str,
    apikey: str,
    local_rules: str,
    local_rules_id: str,
    remove_rules: str,
):
    cache = CacheServiceFactory(config).get_cache_service()
    library_service = CDISCLibraryService(apikey, cache)
    cache_populator = CachePopulator(
        cache, library_service, local_rules, local_rules_id, remove_rules, cache_path
    )
    cache = asyncio.run(cache_populator.load_cache_data())
    if remove_rules:
        cache_populator.save_removed_rules_locally(
            os.path.join(cache_path, DefaultFilePaths.LOCAL_RULES_CACHE_FILE.value),
            remove_rules,
        )
        print("Local rules removed from cache")
    elif local_rules and local_rules_id:
        cache_populator.save_local_rules_locally(
            os.path.join(cache_path, DefaultFilePaths.LOCAL_RULES_CACHE_FILE.value),
            local_rules_id,
        )
        print("Local rules saved to cache")
    else:
        cache_populator.save_rules_locally(
            os.path.join(cache_path, DefaultFilePaths.RULES_CACHE_FILE.value)
        )
        cache_populator.save_ct_packages_locally(f"{cache_path}")
        cache_populator.save_standards_metadata_locally(
            os.path.join(cache_path, DefaultFilePaths.STANDARD_DETAILS_CACHE_FILE.value)
        )
        cache_populator.save_standards_models_locally(
            os.path.join(cache_path, DefaultFilePaths.STANDARD_MODELS_CACHE_FILE.value)
        )
        cache_populator.save_variable_codelist_maps_locally(
            os.path.join(
                cache_path, DefaultFilePaths.VARIABLE_CODELIST_CACHE_FILE.value
            )
        )
        cache_populator.save_variables_metadata_locally(
            os.path.join(
                cache_path, DefaultFilePaths.VARIABLE_METADATA_CACHE_FILE.value
            )
        )
    print("Cache updated successfully")


@click.command()
@click.option(
    "-c",
    "--cache_path",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.option(
    "-s", "--standard", required=False, help="CDISC standard to get rules for"
)
@click.option(
    "-v", "--version", required=False, help="Standard version to get rules for"
)
@click.option(
    "-lr",
    "--local_rules",
    is_flag=True,
    default=False,
    required=False,
    help="flag to list local rules in the cache",
)
@click.option(
    "-lri",
    "--local_rules_id",
    required=False,
    help="local rule id to list from the local rules cache",
)
@click.pass_context
def list_rules(
    ctx: click.Context,
    cache_path: str,
    standard: str,
    version: str,
    local_rules: bool,
    local_rules_id: str,
):
    # Load all rules
    if local_rules:
        rules_file = DefaultFilePaths.LOCAL_RULES_CACHE_FILE.value
    else:
        rules_file = DefaultFilePaths.RULES_CACHE_FILE.value
    with open(os.path.join(cache_path, rules_file), "rb") as f:
        rules_data = pickle.load(f)
    if not local_rules and (standard and version):
        key_prefix = get_rules_cache_key(standard, version.replace(".", "-"))
        rules = [rule for key, rule in rules_data.items() if key.startswith(key_prefix)]
    elif local_rules and local_rules_id:
        key_prefix = get_local_cache_key(local_rules_id)
        rules = [rule for key, rule in rules_data.items() if key.startswith(key_prefix)]
    else:
        # Print all rules
        rules = list(rules_data.values())
    print(json.dumps(rules, indent=4))


@click.command()
@click.option(
    "-c",
    "--cache_path",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.option(
    "-dp",
    "--dataset-path",
    required=True,
    help="Absolute path to dataset file",
)
@click.option(
    "-r",
    "--rule",
    required=True,
    help="Absolute path to rule file",
)
@click.option(
    "-s", "--standard", required=False, help="CDISC standard to get rules for"
)
@click.option(
    "-v", "--version", required=False, help="Standard version to get rules for"
)
@click.option(
    "-ct",
    "--controlled-terminology-package",
    multiple=True,
    help=(
        "Controlled terminology package to validate against, "
        "can provide more than one"
    ),
)
@click.option(
    "-dv",
    "--define-version",
    type=click.Choice(["2-1", "2-0", "2.0", "2.1"]),
    help="Define-XML version used for validation",
)
@click.option("--whodrug", help="Path to directory with WHODrug dictionary files")
@click.option("--meddra", help="Path to directory with MedDRA dictionary files")
@click.option("--loinc", help="Path to directory with LOINC dictionary files")
@click.option("--medrt", help="Path to directory with MEDRT dictionary files")
@click.option(
    "-vx",
    "--validate-xml",
    default="y",
    help="Enable XML validation (default: 'y' to enable, otherwise disable)",
)
@click.option("-dxp", "--define-xml-path", required=False, help="Path to Define-XML")
@click.pass_context
def test(
    ctx,
    cache_path: str,
    dataset_path: Tuple[str],
    rule: str,
    standard: str,
    version: str,
    controlled_terminology_package: Tuple[str],
    define_version: str,
    whodrug: str,
    meddra: str,
    loinc: str,
    medrt: str,
    validate_xml,
    define_xml_path: str,
):
    validate_xml = True if validate_xml.lower() in ("y", "yes") else False
    args = TestArgs(
        cache_path,
        dataset_path,
        rule,
        standard,
        version,
        whodrug,
        meddra,
        loinc,
        medrt,
        controlled_terminology_package,
        define_version,
        define_xml_path,
        validate_xml,
    )
    test_rule(args)


@click.command()
@click.option(
    "-c",
    "--cache_path",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.pass_context
def list_rule_sets(ctx: click.Context, cache_path: str):
    # Load all rules
    rules_file = DefaultFilePaths.RULES_CACHE_FILE.value
    with open(os.path.join(cache_path, rules_file), "rb") as f:
        rules_data = pickle.load(f)
    rule_sets = set()
    for rule in rules_data.keys():
        standard, version = rule.split("/")[1:3]
        rule_set = f"{standard.upper()}, {version}"
        if rule_set not in rule_sets:
            print(rule_set)
            rule_sets.add(rule_set)


@click.command()
@click.option(
    "-dp",
    "--dataset-path",
    required=True,
    multiple=True,
)
@click.pass_context
def list_dataset_metadata(ctx: click.Context, dataset_path: Tuple[str]):
    """
    Command that lists metadata of given datasets.

    Input:
        core.py list-ds-metadata -dp=path_1 -dp=path_2 -dp=path_3 ...
    Output:
        [
           {
              "domain":"AE",
              "filename":"ae.xpt",
              "full_path":"/Users/Aleksei_Furmenkov/PycharmProjects/cdisc-rules-engine/resources/data/ae.xpt",
              "size":"38000",
              "label":"Adverse Events",
              "modification_date":"2020-08-21T09:14:26"
           },
           {
              "domain":"EX",
              "filename":"ex.xpt",
              "full_path":"/Users/Aleksei_Furmenkov/PycharmProjects/cdisc-rules-engine/resources/data/ex.xpt",
              "size":"78050",
              "label":"Exposure",
              "modification_date":"2021-09-17T09:23:22"
           },
           ...
        ]
    """
    print(json.dumps(list_dataset_metadata_handler(dataset_path), indent=4))


@click.command()
def version():
    print(__version__)


@click.command()
@click.option(
    "-c",
    "--cache_path",
    default=DefaultFilePaths.CACHE.value,
    help="Relative path to cache files containing pre loaded metadata and rules",
)
@click.option(
    "-s",
    "--subsets",
    help="CT package subset type. Ex: sdtmct. Multiple values allowed",
    required=False,
    multiple=True,
)
def list_ct(cache_path: str, subsets: Tuple[str]):
    """
    Command to list the ct packages available in the cache.
    """
    if subsets:
        subsets = set([subset.lower() for subset in subsets])

    for file in os.listdir(cache_path):
        file_prefix = file[0 : file.find("ct-") + 2]
        if file_prefix.endswith("ct") and (not subsets or file_prefix in subsets):
            print(os.path.splitext(file)[0])


cli.add_command(validate)
cli.add_command(update_cache)
cli.add_command(list_rules)
cli.add_command(list_rule_sets)
cli.add_command(list_dataset_metadata)
cli.add_command(test)
cli.add_command(version)
cli.add_command(list_ct)

if __name__ == "__main__":
    freeze_support()
    cli()