diff --git a/who_l3_smart_tools/cli/indicators.py b/who_l3_smart_tools/cli/indicators.py new file mode 100755 index 0000000..4ef3eae --- /dev/null +++ b/who_l3_smart_tools/cli/indicators.py @@ -0,0 +1,39 @@ +#! /usr/bin/env python +import argparse +from os import makedirs + +from who_l3_smart_tools.core.l2.indicators import IndicatorLibrary + + +def main(): + parser = argparse.ArgumentParser( + description="Generate Questionnaire FSH from L3 Data Dictionary Excel file." + ) + parser.add_argument( + "-dd", + "--data_dictionary", + required=True, + help="Path to the L2 Data Dictionary", + ) + parser.add_argument( + "-i", + "--input", + required=True, + help="Path to the L2 Data Dictionary", + ) + parser.add_argument( + "-o", + "--output", + required=True, + help="Path to the output directory.", + ) + args = parser.parse_args() + + makedirs(args.output, exist_ok=True) + + indicators = IndicatorLibrary(args.input, args.output, args.data_dictionary) + indicators.generate_cql_scaffolds() + + +if __name__ == "__main__": + main() diff --git a/who_l3_smart_tools/core/cql_tools/cql_template_generator.py b/who_l3_smart_tools/core/cql_tools/cql_template_generator.py index 80c2cc5..1ef512d 100644 --- a/who_l3_smart_tools/core/cql_tools/cql_template_generator.py +++ b/who_l3_smart_tools/core/cql_tools/cql_template_generator.py @@ -349,15 +349,15 @@ def print_to_files(self, output_dir: str, update_existing: bool = True): # File is empty after the last generated line output_file_contents += scaffold["default_content"] - with open(f"{output_dir}/{file_name}Logic.cql", "w") as file: + with open(f"{output_dir}/{file_name}Logic.new1.cql", "w") as file: file.write(output_file_contents) - if create_template_file: - # Create or Overwrite the .template file - with open( - f"{output_dir}/suggested_templates/{file_name}Logic-template.cql", - "w", - ) as file: - file.write(additional_template_file_contents) + # if create_template_file: + # # Create or Overwrite the .template file + # with open( + # f"{output_dir}/suggested_templates/{file_name}Logic-template.cql", + # "w", + # ) as file: + # file.write(additional_template_file_contents) def generate_cql_scaffolds(self): """ diff --git a/who_l3_smart_tools/core/l2/data_dictionary.py b/who_l3_smart_tools/core/l2/data_dictionary.py index 3c0f4cd..89edc52 100644 --- a/who_l3_smart_tools/core/l2/data_dictionary.py +++ b/who_l3_smart_tools/core/l2/data_dictionary.py @@ -66,6 +66,8 @@ def __init__( self.input_options = raw_row["Input Options"] self.validation_condition = raw_row["Validation Condition"] self.required = raw_row["Required"] + self.linkages_dst = raw_row["Linkages to Decision Support Tables"] + self.linkages_ind = raw_row["Linkages to Aggregate Indicators"] self.coding_data_element = coding_data_element @property @@ -102,6 +104,18 @@ def question_instance(self) -> str: else remove_special_characters(parts[0]) ) + @property + def linkages(self) -> list[str]: + _linkages = [] + if self.linkages_dst: + _linkages.extend(item.strip() for item in self.linkages_dst.split(",")) + if self.linkages_ind: + _linkages.extend(item.strip() for item in self.linkages_ind.split(",")) + return _linkages + + def concept_by_linkage(self) -> dict[str, str]: + return {linkage: self.to_concept_item() for linkage in self.linkages} + def to_invariant(self) -> Optional[dict[str, str]]: if self.validation_condition and self.validation_condition.lower() != "none": return { @@ -187,6 +201,7 @@ def __init__( self.models = {} self.questionnaires = {} self.valuesets = {} + self.linkage_concepts = defaultdict(list) def set_active_coding(self, row: L2Row) -> None: if self.active_coding_data_element and row.data_type != "Codes": @@ -259,6 +274,10 @@ def format_concepts_for_cql(self) -> list[dict[str, str]]: reformatted_concepts.extend(concepts) return reformatted_concepts + def merge_linkage_concepts(self, row: L2Row) -> None: + for linkage, concept in row.concept_by_linkage().items(): + self.linkage_concepts[linkage].append(concept) + def process(self): for sheet_name in self.workbook.sheetnames: if not sheet_name.startswith(self.sheet_name_prefix): @@ -276,6 +295,7 @@ def process(self): self.add_to_model(sheet_name, l2_row) self.add_to_questionnaire(l2_row) self.add_to_valueset(l2_row) + self.merge_linkage_concepts(l2_row) def write_concepts(self): for _type in ["cql", "fsh"]: @@ -288,13 +308,13 @@ def write_concepts(self): self.output_path, concepts_dir, f"HIVConcepts.{_type}" ) os.makedirs(os.path.join(self.output_path, concepts_dir), exist_ok=True) - template = jinja_env.get_template(f"concepts.{_type}.j2") + template = jinja_env.get_template(f"data_dictionary/concepts.{_type}.j2") render_to_file(template, {"concepts": concepts}, output_path) def write_models(self): models_dir = "models" os.makedirs(os.path.join(self.output_path, models_dir), exist_ok=True) - template = jinja_env.get_template("model.fsh.j2") + template = jinja_env.get_template("data_dictionary/model.fsh.j2") for model in self.models.values(): output_path = os.path.join( self.output_path, models_dir, f"{model['id']}.fsh" @@ -304,7 +324,7 @@ def write_models(self): def write_questionnaires(self): questionnaires_dir = "questionnaires" os.makedirs(os.path.join(self.output_path, questionnaires_dir), exist_ok=True) - template = jinja_env.get_template("questionnaire.fsh.j2") + template = jinja_env.get_template("data_dictionary/questionnaire.fsh.j2") for questionnaire in self.questionnaires.values(): output_path = os.path.join( self.output_path, @@ -316,7 +336,7 @@ def write_questionnaires(self): def write_valuesets(self): valuesets_dir = "valuesets" os.makedirs(os.path.join(self.output_path, valuesets_dir), exist_ok=True) - template = jinja_env.get_template("valueset.fsh.j2") + template = jinja_env.get_template("data_dictionary/valueset.fsh.j2") for valueset in self.valuesets.values(): output_path = os.path.join( self.output_path, valuesets_dir, f"{valueset['name']}.fsh" diff --git a/who_l3_smart_tools/core/l2/indicators.py b/who_l3_smart_tools/core/l2/indicators.py new file mode 100644 index 0000000..8323a2b --- /dev/null +++ b/who_l3_smart_tools/core/l2/indicators.py @@ -0,0 +1,156 @@ +import os +from openpyxl import load_workbook +import stringcase + +from who_l3_smart_tools.core.l2.data_dictionary import L2Dictionary +from who_l3_smart_tools.utils.jinja2 import initalize_jinja_env, render_to_file + +jinja_env = initalize_jinja_env(__name__) + + +# pylint: disable=too-many-instance-attributes +# pylint: disable=no-member +class IndicatorRow: + """ + Class representing an indicator row. + + Attributes: + raw_row (dict): The raw row data. + dak_id (str): The DAK ID. + library_name (str): The library name. + ref_no (str): The reference number. + denominator (str): The denominator calculation. + all_data_elements (list): The list of all data elements. + included_in_dak (bool): Indicates if the indicator is included in DAK. + + Methods: + set_other_columns_as_attributes(self) -> None: + Sets the other columns as attributes. + + determine_scoring_suggestion(self) -> str: + Determines the scoring suggestion based on the denominator value. + + to_cql(self) -> str: + Converts the indicator to CQL format. + """ + + scoring_measure_instance = { + "proportion": "http://hl7.org/fhir/us/cqfmeasures/StructureDefinition/proportion-measure-cqfm", + "continuous-variable": "http://hl7.org/fhir/us/cqfmeasures/StructureDefinition/cv-measure-cqfm", + } + + measure_required_elements = { + "proportion": ["initialPopulation", "Numerator", "Denominator"], + "continuous-variable": [ + "initialPopulation", + "measurePopulation", + "measureObservation", + ], + } + + def __init__( + self, raw_row: dict, out_dir: str, linkage_concepts: dict[str, list] + ) -> None: + self.raw_row = raw_row + self.out_dir = out_dir + self.dak_id = raw_row.pop("DAK ID") + self.library_name = f'{self.dak_id.replace(".", "")}Logic' + self.ref_no = raw_row.pop("Ref no.") + self.all_data_elements = ( + raw_row.pop( + "List of all data elements included in numerator and denominator", "" + ) + or "" + ).split("\n") + self.included_in_dak = raw_row.pop("Included in DAK") + self.data_concepts = linkage_concepts.get(self.ref_no, []) + self.set_other_columns_as_attributes() + + @property + def disaggregations(self) -> str: + return self.disaggregation_description.split("|") + + @property + def scoring_method(self) -> str: + return self.determine_scoring_suggestion() + + @property + def proportion(self) -> bool: + return self.scoring_method == "proportion" + + @property + def continuous_variable(self) -> bool: + return self.scoring_method == "continuous-variable" + + @property + def scoring_instance(self) -> str: + return self.scoring_measure_instance[self.scoring_method] + + def set_other_columns_as_attributes(self) -> None: + for key, value in self.raw_row.items(): + key = stringcase.snakecase(key.lower()) + value = value.replace("\n", "|") if isinstance(value, str) else value + setattr(self, key, value) + + def determine_scoring_suggestion(self) -> str: + if ( + not self.denominator_calculation.strip() + or self.denominator_calculation.strip() == "1" + ): + return "continuous-variable" + return "proportion" + + def to_cql(self) -> str: + output_path = os.path.join(self.out_dir, f"{self.library_name}.cql") + template = jinja_env.get_template("indicators/cql_library.cql.j2") + render_to_file(template, {"indicator": self}, output_path) + + +# pylint: disable=too-few-public-methods +class IndicatorLibrary: + """ + A class representing an indicator library. + + Parameters: + - indicator_file (str): The file path of the indicator file. + - output_dir (str): The directory where the output will be saved. + - data_dictionary_file (str): The file path of the data dictionary file. + - sheet_name (str, optional): The name of the sheet in the indicator file. + Defaults to "Indicator definitions". + + Methods: + - generate_cql_scaffolds(): Generates CQL scaffolds based on the indicator definitions. + """ + + def __init__( + self, + indicator_file: str, + output_dir: str, + data_dictionary_file: str, + sheet_name: str = "Indicator definitions", + ) -> None: + self.sheet = load_workbook(indicator_file)[sheet_name] + self.data_dictionary = L2Dictionary(data_dictionary_file, output_dir) + self.output_dir = output_dir + + def generate_cql_scaffolds(self) -> None: + self.data_dictionary.process() + header = None + x = 0 + for row in self.sheet.iter_rows(values_only=True): + if not header: + header = row + continue + row = dict(zip(header, row)) + indicator_row = IndicatorRow( + row, self.output_dir, self.data_dictionary.linkage_concepts + ) + indicator_row.to_cql() + print( + indicator_row.numerator_exclusions, + type(indicator_row.numerator_exclusions), + ) + + x += 1 + if x == 2: + break diff --git a/who_l3_smart_tools/core/l2/templates/concepts.cql.j2 b/who_l3_smart_tools/core/l2/templates/data_dictionary/concepts.cql.j2 similarity index 100% rename from who_l3_smart_tools/core/l2/templates/concepts.cql.j2 rename to who_l3_smart_tools/core/l2/templates/data_dictionary/concepts.cql.j2 diff --git a/who_l3_smart_tools/core/l2/templates/concepts.fsh.j2 b/who_l3_smart_tools/core/l2/templates/data_dictionary/concepts.fsh.j2 similarity index 100% rename from who_l3_smart_tools/core/l2/templates/concepts.fsh.j2 rename to who_l3_smart_tools/core/l2/templates/data_dictionary/concepts.fsh.j2 diff --git a/who_l3_smart_tools/core/l2/templates/model.fsh.j2 b/who_l3_smart_tools/core/l2/templates/data_dictionary/model.fsh.j2 similarity index 100% rename from who_l3_smart_tools/core/l2/templates/model.fsh.j2 rename to who_l3_smart_tools/core/l2/templates/data_dictionary/model.fsh.j2 diff --git a/who_l3_smart_tools/core/l2/templates/questionnaire.fsh.j2 b/who_l3_smart_tools/core/l2/templates/data_dictionary/questionnaire.fsh.j2 similarity index 100% rename from who_l3_smart_tools/core/l2/templates/questionnaire.fsh.j2 rename to who_l3_smart_tools/core/l2/templates/data_dictionary/questionnaire.fsh.j2 diff --git a/who_l3_smart_tools/core/l2/templates/valueset.fsh.j2 b/who_l3_smart_tools/core/l2/templates/data_dictionary/valueset.fsh.j2 similarity index 100% rename from who_l3_smart_tools/core/l2/templates/valueset.fsh.j2 rename to who_l3_smart_tools/core/l2/templates/data_dictionary/valueset.fsh.j2 diff --git a/who_l3_smart_tools/core/l2/templates/indicators/cql_library.cql.j2 b/who_l3_smart_tools/core/l2/templates/indicators/cql_library.cql.j2 new file mode 100644 index 0000000..491f9e2 --- /dev/null +++ b/who_l3_smart_tools/core/l2/templates/indicators/cql_library.cql.j2 @@ -0,0 +1,158 @@ +/** + * Library: {{ indicator.dak_id }} Logic + * Ref No: {{ indicator.ref_no }} + * Short Name: {{ indicator.short_name }} + * + * Definition: {{ indicator.indicator_definition }} + * + * Numerator: {{ indicator.numerator_definition }} + * Numerator Calculation: {{ indicator.numerator_calculation }} + * Numerator Exclusions: {% if indicator.numerator_exclusions %} {{ indicator.numerator_exclusions }} {% endif %} + * + * Denominator: {{ indicator.denominator_definition }} + * Denominator Calculation: {{ indicator.denominator_calculation }} + * Denominator Exclusions: {% if indicator.denominator_exclusions %}{{ indicator.denominator_exclusions }} {% endif %} + * + * Disaggregations: +{% for disaggregation in indicator.disaggregations %} + * {{ disaggregation }} +{% endfor %} + * + * Disaggregation Elements: {{ indicator.disaggregation_data_elements }} + * + * Numerator and Denominator Elements: +{% for element in indicator.all_data_elements %} + * {{ element }} +{% endfor %} + * + * Reference: {{ indicator.reference }} + * + * Data Concepts: +{% for concept in indicator.data_concepts %} + * {{ concept.id }}: {{ concept.label }} | {{ concept.description }} +{% endfor %} + * + * Additional Context + * - what it measures: {{ indicator.what_it_measures }} + * - rationale: {{ indicator.rationale }} + * - method: {{ indicator.method_of_measurement }} + * + * Suggested Scoring Method: {{ indicator.scoring_method }} | {{ indicator.scoring_instance }} + */ + +library {{ indicator.library_name }} + +// Included Libraries +using FHIR version '4.0.1' + +include HIVCommon version '0.0.1' called HIC +include FHIRHelpers version '4.0.1' +include FHIRCommon called FC +include WHOCommon called WCom + +// Indicator Definition +parameter "Measurement Period" Interval default Interval[@2023-01-01, @2023-01-30] + +context Patient + +/* Populations */ +{% if indicator.proportion %} +/* + *Initial Population + */ + +define "Initial Population": + true + +/** + * Numerator + * + * Definition: {{ indicator.numerator_definition }} + * Calculation: {{ indicator.numerator_calculation }} + */ + +define "Numerator": + true + +{% if indicator.numerator_exclusions %} +/** + * Numerator Exclusions + * + * Calculation: {{ indicator.numerator_exclusions }} + */ + +define "Numerator Exclusions": + false +{% endif %} + +/** + * Denominator + * + * Definition: {{ indicator.denominator_definition }} + * Calculation: {{ indicator.denominator_calculation }} + */ + +define "Denominator": + true + +{% if indicator.denominator_exclusions %} +/** +* Denominator Exclusions +* +* Calculation: {{ indicator.denominator_exclusions }} +*/ + +define "Denominator Exclusions": + false +{% endif %} +{% elif indicator.continuous_variable %} +/* + *Initial Population + */ + +define "Initial Population": + true + +/** + * Measure Population + * + * Definition: {{ indicator.numerator_definition }} + * Calculation: {{ indicator.numerator_definition }} + */ + +define "Measure Population": + true + +{% if indicator.measure_population_exclusions %} +/** + * Measure Population Exclusions + * + * Calculation: {{ indicator.measure_population_exclusions }} + */ + define "Measure Population Exclusions": + false +{% endif %} + +/** + * Measure Observation + * Definition: {{ indicator.measure_observation_definition }} + * Calculation: {{ indicator.measure_observation_calculation }} + */ + +define function "Measure Observation"(Patient "Patient"): + 1 +{% endif %} +/* end Populations */ + +{% if indicator.disaggregation_data_elements %} +/** + * Disaggregators + * + */ +{% for disaggregation in indicator.disaggregation_data_elements.split("|") %} +define "{{ disaggregation }} Stratifier": + HIC."{{ disaggregation }} Stratifier" + +{% endfor %} +/* end Disaggregators */ +{% endif %} \ No newline at end of file