Added event_group_id to process_measurement

src/cehrbert_data/apps/generate_training_data.py

@@ -119,10 +119,9 @@ def main(
     if MEASUREMENT in domain_table_list:
         measurement = preprocess_domain_table(spark, input_folder, MEASUREMENT)
         required_measurement = preprocess_domain_table(spark, input_folder, REQUIRED_MEASUREMENT)
-        if os.path.exists(os.path.join(input_folder, CONCEPT)):
-            concept = preprocess_domain_table(spark, input_folder, CONCEPT)
-        else:
-            concept = None
+        if not os.path.exists(os.path.join(input_folder, CONCEPT)):
+            raise ValueError("concept needs to be provided when measurement is included!")
+        concept = preprocess_domain_table(spark, input_folder, CONCEPT)
         # The select is necessary to make sure the order of the columns is the same as the
         # original dataframe, otherwise the union might use the wrong columns
         filtered_measurement = process_measurement(spark, measurement, required_measurement, concept)

src/cehrbert_data/utils/spark_utils.py

@@ -1387,7 +1387,8 @@ def process_measurement(
             m.visit_occurrence_id,
             'measurement' AS domain,
             m.unit, 
-            m.value_as_number AS concept_value
+            m.value_as_number AS concept_value,
+            CONCAT('measurement-', CAST(m.measurement_id AS STRING)) AS event_group_id
         FROM measurement AS m
         WHERE m.visit_occurrence_id IS NOT NULL
             AND m.value_as_number IS NOT NULL
@@ -1410,7 +1411,8 @@ def process_measurement(
             m.visit_occurrence_id,
             'categorical_measurement' AS domain,
             CAST(NULL AS STRING) AS unit,
-            -1.0 AS concept_value
+            -1.0 AS concept_value,
+            CONCAT('measurement-', CAST(m.measurement_id AS STRING)) AS event_group_id
         FROM measurement AS m
         WHERE EXISTS (
             SELECT