Wsg 105 merge all indicator and resources into one module

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()

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):
         """

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"

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