Format files.

Author: np-complete0

apps/bfd-model-idr/claims_generator.py

@@ -39,12 +39,9 @@ def get_referenced_maps(compiled_map_path):
 
 def get_structure_definitions():
     try:
-        source_dir = (
-            Path(__file__).parent.absolute() / "StructureDefinitions" / "Source"
-        )
+        source_dir = Path(__file__).parent.absolute() / "StructureDefinitions" / "Source"
         structure_defs = [
-            f"StructureDefinitions/Source/{file.name}"
-            for file in source_dir.glob("*.json")
+            f"StructureDefinitions/Source/{file.name}" for file in source_dir.glob("*.json")
         ]
         return " ".join([f"-ig {def_file}" for def_file in structure_defs])
     except Exception as e:
@@ -54,12 +51,9 @@ def get_structure_definitions():
 
 def get_sushi_resources():
     try:
-        sushi_dir = (
-            Path(__file__).parent.absolute() / "sushi" / "fsh-generated" / "resources"
-        )
+        sushi_dir = Path(__file__).parent.absolute() / "sushi" / "fsh-generated" / "resources"
         sushi_resources = [
-            f"sushi/fsh-generated/resources/{file.name}"
-            for file in sushi_dir.glob("*.json")
+            f"sushi/fsh-generated/resources/{file.name}" for file in sushi_dir.glob("*.json")
         ]
         return " ".join([f"-ig {resource}" for resource in sushi_resources])
     except Exception as e:
@@ -75,10 +69,8 @@ def run_conformance_test(input_file, output_file):
         sushi_resources = get_sushi_resources()
 
         print("Running conformance testing")
-        test_cmd = (
-            f"java -jar validator_cli.jar {input_file} -output {output_file} -version 4.0.1 \
+        test_cmd = f"java -jar validator_cli.jar {input_file} -output {output_file} -version 4.0.1 \
             {structure_defs} -ig hl7.fhir.us.carin-bb#2.1.0 {sushi_resources}"
-        )
         stdout, stderr = run_command(test_cmd, cwd=script_dir)
 
         print("Conformance test output:")
@@ -100,9 +92,7 @@ def run_conformance_test(input_file, output_file):
 
 def main():
     # Parse args instead of just putting everything in separate READMEs.
-    parser = argparse.ArgumentParser(
-        description="Compile and execute FHIR structure maps."
-    )
+    parser = argparse.ArgumentParser(description="Compile and execute FHIR structure maps.")
     parser.add_argument(
         "--map", "-m", type=str, help="Path to the structure map file", required=True
     )
@@ -148,8 +138,10 @@ def main():
     # Compile FML files.
     print("Compiling FML ")
     compiled_map_path = f"StructureMaps/BFD-{Path(args.map).stem}-StructureMap.json"
-    compile_cmd = f"java -jar validator_cli.jar -version 4.0.1 -ig {args.map} -compile {args.resource} \
+    compile_cmd = (
+        f"java -jar validator_cli.jar -version 4.0.1 -ig {args.map} -compile {args.resource} \
         -output {compiled_map_path}"
+    )
     print("Input compilation command was:" + compile_cmd)
     stdout, stderr = run_command(compile_cmd, cwd=script_dir)
     print("Compilation output:")
@@ -163,11 +155,9 @@ def main():
     map_imports = " ".join([f"-ig {map_file}" for map_file in referenced_maps])
 
     print("Executing Transform")
-    execute_cmd = (
-        f"java -jar validator_cli.jar {args.input} -output {args.output} -transform \
+    execute_cmd = f"java -jar validator_cli.jar {args.input} -output {args.output} -transform \
         {args.resource} -version 4.0.1 -ig {compiled_map_path} {structure_defs} \
             -ig hl7.fhir.us.carin-bb#2.1.0 {map_imports} {sushi_resources}"
-    )
     stdout, stderr = run_command(execute_cmd, cwd=script_dir)
 
     print("Execution output:")

apps/bfd-model-idr/compile_resources.py

@@ -37,9 +37,7 @@
         usecols=["Target Table", "Target Column"],
     )
     for _, row in df.iterrows():
-        element_concatenated = (
-            f"{row['Target Table'].strip()}.{row['Target Column'].strip()}"
-        )
+        element_concatenated = f"{row['Target Table'].strip()}.{row['Target Column'].strip()}"
         # there are newlines in some, we should ask IDR to consider changing the DD structure?
         if (
             row["Target Table"] != "-"
@@ -48,12 +46,9 @@
         ):
             if (
                 element_concatenated in applies_to
-                and translations[cur_profile]
-                not in applies_to[element_concatenated]["profiles"]
+                and translations[cur_profile] not in applies_to[element_concatenated]["profiles"]
             ):
-                applies_to[element_concatenated]["profiles"].append(
-                    translations[cur_profile]
-                )
+                applies_to[element_concatenated]["profiles"].append(translations[cur_profile])
             else:
                 applies_to[element_concatenated] = {
                     "profiles": [translations[cur_profile]],

apps/bfd-model-idr/dd_helper_script.py

@@ -59,23 +59,19 @@
             test_resource = json.load(file)
             for element in test_resource["differential"]["element"]:
                 structure_def_names_descriptions[element["id"]] = {}
-                structure_def_names_descriptions[element["id"]]["name"] = element[
-                    "label"
-                ]
+                structure_def_names_descriptions[element["id"]]["name"] = element["label"]
                 if "definition" in element:
-                    structure_def_names_descriptions[element["id"]]["definition"] = (
-                        element["definition"]
-                    )
+                    structure_def_names_descriptions[element["id"]]["definition"] = element[
+                        "definition"
+                    ]
 
 for walk_info in os.walk(idr_ref_folder):
     files = list(filter(lambda file: ".csv" in file, walk_info[2]))
     for file_name in files:
         idr_table_descriptors[file_name[0 : len(file_name) - 4]] = {}
         df = pd.read_csv(idr_ref_folder + "/" + str(file_name))
         for _, row in df.iterrows():
-            idr_table_descriptors[file_name[0 : len(file_name) - 4]][row["name"]] = row[
-                "comment"
-            ]
+            idr_table_descriptors[file_name[0 : len(file_name) - 4]][row["name"]] = row["comment"]
 
 coverage_parts = ["PartA", "PartB", "PartC", "PartD", "DUAL"]
 claim_profiles = [
@@ -115,9 +111,7 @@
                         [
                             "node",
                             "eval_fhirpath.js",
-                            json.dumps(
-                                sample_resources_by_profile[entry["appliesTo"][0]]
-                            ),
+                            json.dumps(sample_resources_by_profile[entry["appliesTo"][0]]),
                             entry["fhirPath"],
                         ],
                         check=True,
@@ -130,23 +124,17 @@
                         entry["example"] = entry["example"][0]
                     else:
                         entry["example"] = ""
-                    if (
-                        "sourceView" in entry
-                        and entry["sourceView"] in idr_table_descriptors
-                    ):
-                        entry["Description"] = idr_table_descriptors[
-                            entry["sourceView"]
-                        ][entry["sourceColumn"]]
+                    if "sourceView" in entry and entry["sourceView"] in idr_table_descriptors:
+                        entry["Description"] = idr_table_descriptors[entry["sourceView"]][
+                            entry["sourceColumn"]
+                        ]
 
                     # Populate the element names + missing descriptions
                     if entry["inputPath"] in structure_def_names_descriptions:
-                        entry["Field Name"] = structure_def_names_descriptions[
-                            entry["inputPath"]
-                        ]["name"]
-                        if (
-                            "definition"
-                            in structure_def_names_descriptions[entry["inputPath"]]
-                        ):
+                        entry["Field Name"] = structure_def_names_descriptions[entry["inputPath"]][
+                            "name"
+                        ]
+                        if "definition" in structure_def_names_descriptions[entry["inputPath"]]:
                             entry["Description"] = structure_def_names_descriptions[
                                 entry["inputPath"]
                             ]["definition"]
@@ -162,9 +150,7 @@
 def replace_str(input_str):
     # Yes, the below is intentional.
     if input_str == input_str and len(str(input_str)) > 0:
-        return "https://bluebutton.cms.gov/fhir/CodeSystem/" + str(input_str).replace(
-            "_", "-"
-        )
+        return "https://bluebutton.cms.gov/fhir/CodeSystem/" + str(input_str).replace("_", "-")
     return ""
 
 
@@ -213,9 +199,7 @@ def replace_str(input_str):
 
     workbook = writer.book
     worksheet = writer.sheets["Data Dictionary"]
-    header_format = workbook.add_format(
-        {"bold": True, "bg_color": "#DCE6F2", "border": 1}
-    )
+    header_format = workbook.add_format({"bold": True, "bg_color": "#DCE6F2", "border": 1})
     text_format = workbook.add_format({"border": 1})
 
     worksheet.write(0, 0, "Row", header_format)

apps/bfd-model-idr/gen_dd.py

@@ -119,9 +119,7 @@ def gen_bene_sk(self):
         return bene_sk
 
     def generate_bene_xref(self, new_bene_sk, old_bene_sk):
-        bene_hicn_num = str(random.randint(1000, 100000000)) + random.choice(
-            string.ascii_letters
-        )
+        bene_hicn_num = str(random.randint(1000, 100000000)) + random.choice(string.ascii_letters)
 
         # 10% chance for invalid xref.
         kill_cred_cd = 1 if random.randint(1, 10) == 1 else 2
@@ -248,9 +246,7 @@ def handle_mbis(self, patient, num_mbis, custom_first_mbi=None):
             patient["BENE_MBI_ID"] = current_mbi
 
     def generate_coverages(self, patient):
-        parts = random.choices(
-            [["A"], ["B"], ["A", "B"], []], weights=[0.2, 0.2, 0.5, 0.1]
-        )[0]
+        parts = random.choices([["A"], ["B"], ["A", "B"], []], weights=[0.2, 0.2, 0.5, 0.1])[0]
         include_tp = random.random() > 0.2
         expired = random.random() < 0.2
         future = random.random() < 0.2
@@ -314,12 +310,9 @@ def _generate_coverages(self, patient, coverage_parts, include_tp, expired, futu
                 "BENE_SK": patient["BENE_SK"],
                 "IDR_LTST_TRANS_FLG": "Y",
                 "BENE_MDCR_ENTLMT_TYPE_CD": coverage_type,
-                "BENE_MDCR_ENRLMT_RSN_CD": random.choice(
-                    self.code_systems["BENE_ENRLMT_RSN_CD"]
-                ),
+                "BENE_MDCR_ENRLMT_RSN_CD": random.choice(self.code_systems["BENE_ENRLMT_RSN_CD"]),
                 "BENE_MDCR_ENTLMT_STUS_CD": "Y",
-                "IDR_TRANS_EFCTV_TS": str(medicare_start_date)
-                + "T00:00:00.000000+0000",
+                "IDR_TRANS_EFCTV_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                 "IDR_INSRT_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                 "IDR_UPDT_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                 "IDR_TRANS_OBSLT_TS": "9999-12-31T00:00:00.000000+0000",
@@ -333,8 +326,7 @@ def _generate_coverages(self, patient, coverage_parts, include_tp, expired, futu
                     "BENE_SK": patient["BENE_SK"],
                     "IDR_LTST_TRANS_FLG": "Y",
                     "BENE_TP_TYPE_CD": coverage_type,
-                    "IDR_TRANS_EFCTV_TS": str(medicare_start_date)
-                    + "T00:00:00.000000+0000",
+                    "IDR_TRANS_EFCTV_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                     "IDR_INSRT_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                     "IDR_UPDT_TS": str(medicare_start_date) + "T00:00:00.000000+0000",
                     "IDR_TRANS_OBSLT_TS": "9999-12-31T00:00:00.000000+0000",

apps/bfd-model-idr/generator_util.py

@@ -13,9 +13,7 @@
 
 # Command line argument parsing
 parser = argparse.ArgumentParser(description="Generate synthetic patient data")
-parser.add_argument(
-    "--benes", type=str, help="Path to CSV file containing beneficiary data"
-)
+parser.add_argument("--benes", type=str, help="Path to CSV file containing beneficiary data")
 parser.add_argument(
     "--claims",
     action="store_true",
@@ -48,9 +46,7 @@
 csv_data = None
 if args.benes:
     try:
-        csv_data = pd.read_csv(
-            args.benes, dtype={"BENE_SEX_CD": "Int64", "BENE_RACE_CD": "Int64"}
-        )
+        csv_data = pd.read_csv(args.benes, dtype={"BENE_SEX_CD": "Int64", "BENE_RACE_CD": "Int64"})
         print(f"Loaded {len(csv_data)} rows from CSV file: {args.benes}")
 
         patients_to_generate = len(csv_data)
@@ -100,9 +96,7 @@
             if pd.notna(row.get("BENE_VRFY_DEATH_DAY_SW")):
                 patient["BENE_VRFY_DEATH_DAY_SW"] = str(row["BENE_VRFY_DEATH_DAY_SW"])
             else:
-                patient["BENE_VRFY_DEATH_DAY_SW"] = (
-                    "Y" if random.randint(0, 1) == 1 else "N"
-                )
+                patient["BENE_VRFY_DEATH_DAY_SW"] = "Y" if random.randint(0, 1) == 1 else "N"
         else:
             if random.randint(0, 10) < 2:
                 # death!
@@ -187,9 +181,7 @@
             patient["BENE_VRFY_DEATH_DAY_SW"] = "~"
 
         patient["BENE_SEX_CD"] = str(random.randint(1, 2))
-        patient["BENE_RACE_CD"] = random.choice(
-            ["~", "0", "1", "2", "3", "4", "5", "6", "7", "8"]
-        )
+        patient["BENE_RACE_CD"] = random.choice(["~", "0", "1", "2", "3", "4", "5", "6", "7", "8"])
 
         pt_bene_sk = generator.gen_bene_sk()
         patient["BENE_SK"] = str(pt_bene_sk)
@@ -227,14 +219,10 @@
 
         generator.generate_bene_xref(pt_bene_sk, old_bene_sk)
 
-        generator.set_timestamps(
-            prior_patient, datetime.date(year=2017, month=5, day=20)
-        )
+        generator.set_timestamps(prior_patient, datetime.date(year=2017, month=5, day=20))
 
         # Override the obsolete timestamp to be in the past year instead of future
-        past_year_date = datetime.date.today() - datetime.timedelta(
-            days=random.randint(30, 365)
-        )
+        past_year_date = datetime.date.today() - datetime.timedelta(days=random.randint(30, 365))
         prior_patient["IDR_TRANS_OBSLT_TS"] = f"{past_year_date}T00:00:00.000000+0000"
 
         generator.bene_hstry_table.append(prior_patient)

apps/bfd-model-idr/patient_generator.py

@@ -125,9 +125,7 @@
 # Helpers
 
 
-def make_bene_id_pattern(
-    bene_ids: list[str], pattern: str, cast_id: bool = False
-) -> str:
+def make_bene_id_pattern(bene_ids: list[str], pattern: str, cast_id: bool = False) -> str:
     """Make the WHERE clause to handle bene_ids that are in the list or fit the pattern."""
     output = "  bene_id IN ('" + "', '".join(bene_ids) + "')"
     if pattern is not None:
@@ -150,9 +148,7 @@ def make_claims_sql(
     if is_count:
         output += f"SELECT COUNT(lines.*) AS {claim_lines_table}\n"
         output += f"FROM {claim_lines_table} AS lines\n"
-        output += (
-            f"LEFT JOIN {claims_table} AS claims ON (claims.clm_id = lines.clm_id)\n"
-        )
+        output += f"LEFT JOIN {claims_table} AS claims ON (claims.clm_id = lines.clm_id)\n"
         output += f"WHERE\n{bene_id_clause};\n\n\n"
     else:
         output += f"DELETE FROM {claim_lines_table} AS lines\n"
@@ -212,21 +208,13 @@ def main(args: list):
         sys.exit()
 
     bene_id_pattern = make_bene_id_pattern(bene_ids["ids"], bene_ids["pattern"], False)
-    new_bene_id_pattern = make_bene_id_pattern(
-        bene_ids["ids"], bene_ids["pattern"], True
-    )
+    new_bene_id_pattern = make_bene_id_pattern(bene_ids["ids"], bene_ids["pattern"], True)
 
     for claims_table, claim_lines_table in CLAIMS_TABLES:
-        print(
-            make_claims_sql(bene_id_pattern, claims_table, claim_lines_table, is_count)
-        )
+        print(make_claims_sql(bene_id_pattern, claims_table, claim_lines_table, is_count))
 
     for claims_table, claim_lines_table in NEW_CLAIMS_TABLES:
-        print(
-            make_claims_sql(
-                new_bene_id_pattern, claims_table, claim_lines_table, is_count
-            )
-        )
+        print(make_claims_sql(new_bene_id_pattern, claims_table, claim_lines_table, is_count))
 
     for base_table in BASE_TABLES:
         print(make_base_sql(bene_id_pattern, base_table, is_count))