Merge pull request #418 from PDCMFinder/develop

Develop

Author: mauroz77

etl/entities_registry.py

@@ -366,6 +366,7 @@ def get_all_entities_names_to_store_db():
         "spark_job": etl.jobs.transformation.platform_transformer_job.main,
         "expected_database_columns": [
             "id",
+            "platform_id",
             "library_strategy",
             "provider_group_id",
             "instrument_model",
@@ -695,6 +696,7 @@ def get_all_entities_names_to_store_db():
             "cancer_annotation_resources",
             "model_availability",
             "date_submitted",
+            "email_list",
             "model_generator",
             "view_data_at",
             "scores",

etl/jobs/transformation/model_metadata_transformer_job.py

@@ -171,6 +171,7 @@ def get_formatted_model(model_df: DataFrame) -> DataFrame:
         "drug_concentration",
         "model_availability",
         "date_submitted",
+        "email_list",
         Constants.DATA_SOURCE_COLUMN,
     )
     return model_df

etl/jobs/transformation/model_transformer_job.py

@@ -10,7 +10,7 @@
 
 def main(argv):
     """
-    
+
     :param list argv: the list elements should be:
                     [1]: Parquet file path with raw external_model_ids configuration
                     [2]: Parquet file path with initial df with model_information data
@@ -33,8 +33,12 @@ def main(argv):
     output_path = argv[9]
 
     spark = SparkSession.builder.getOrCreate()
-    raw_external_model_ids_df = spark.read.parquet(raw_external_model_ids_resources_parquet_path)
-    initial_model_information_df = spark.read.parquet(initial_model_information_parquet_path)
+    raw_external_model_ids_df = spark.read.parquet(
+        raw_external_model_ids_resources_parquet_path
+    )
+    initial_model_information_df = spark.read.parquet(
+        initial_model_information_parquet_path
+    )
     publication_group_df = spark.read.parquet(publication_group_parquet_path)
     accessibility_group_df = spark.read.parquet(accessibility_group_parquet_path)
     contact_people_df = spark.read.parquet(contact_people_parquet_path)
@@ -50,21 +54,22 @@ def main(argv):
         contact_people_df,
         contact_form_df,
         source_database_df,
-        license_df)
+        license_df,
+    )
 
     model_df.write.mode("overwrite").parquet(output_path)
 
 
 def transform_model(
-        raw_external_model_ids_df: DataFrame,
-        initial_model_information_df: DataFrame,
-        publication_group_df: DataFrame,
-        accessibility_group_df: DataFrame,
-        contact_people_df: DataFrame,
-        contact_form_df: DataFrame,
-        source_database_df: DataFrame,
-        license_df: DataFrame) -> DataFrame:
-
+    raw_external_model_ids_df: DataFrame,
+    initial_model_information_df: DataFrame,
+    publication_group_df: DataFrame,
+    accessibility_group_df: DataFrame,
+    contact_people_df: DataFrame,
+    contact_form_df: DataFrame,
+    source_database_df: DataFrame,
+    license_df: DataFrame,
+) -> DataFrame:
     model_df = initial_model_information_df
     model_df = set_fk_publication_group(model_df, publication_group_df)
     model_df = set_fk_accessibility_group(model_df, accessibility_group_df)
@@ -73,51 +78,85 @@ def transform_model(
     model_df = set_fk_source_database(model_df, source_database_df)
     model_df = set_fk_license(model_df, license_df)
     model_df = add_model_links(model_df, raw_external_model_ids_df)
-    
+
     model_df = get_columns_expected_order(model_df)
 
     return model_df
 
 
-def set_fk_publication_group(model_df: DataFrame, publication_group_df: DataFrame) -> DataFrame:
+def set_fk_publication_group(
+    model_df: DataFrame, publication_group_df: DataFrame
+) -> DataFrame:
     model_df = transform_to_fk(
-        model_df, publication_group_df, "publications", "pubmed_ids", "id", "publication_group_id")
+        model_df,
+        publication_group_df,
+        "publications",
+        "pubmed_ids",
+        "id",
+        "publication_group_id",
+    )
     return model_df
 
 
-def set_fk_accessibility_group(model_df: DataFrame, accessibility_group_df: DataFrame) -> DataFrame:
-    model_df = model_df.withColumnRenamed("europdx_access_modality", "europdx_access_modalities")
-    accessibility_group_df = accessibility_group_df.withColumnRenamed("id", "accessibility_group_id")
+def set_fk_accessibility_group(
+    model_df: DataFrame, accessibility_group_df: DataFrame
+) -> DataFrame:
+    model_df = model_df.withColumnRenamed(
+        "europdx_access_modality", "europdx_access_modalities"
+    )
+    accessibility_group_df = accessibility_group_df.withColumnRenamed(
+        "id", "accessibility_group_id"
+    )
     model_df = model_df.join(
         accessibility_group_df,
-        on=['accessibility', 'europdx_access_modalities'], how='left')
+        on=["accessibility", "europdx_access_modalities"],
+        how="left",
+    )
     return model_df
 
 
-def set_fk_contact_people(model_df: DataFrame, contact_people_df: DataFrame) -> DataFrame:
-    contact_people_df = contact_people_df.select("id", "email_list", "name_list", Constants.DATA_SOURCE_COLUMN)
+def set_fk_contact_people(
+    model_df: DataFrame, contact_people_df: DataFrame
+) -> DataFrame:
+    contact_people_df = contact_people_df.select(
+        "id", "email_list", "name_list", Constants.DATA_SOURCE_COLUMN
+    )
     model_df = model_df.withColumnRenamed("email", "email_list")
     model_df = model_df.withColumnRenamed("name", "name_list")
     contact_people_df = contact_people_df.withColumnRenamed("id", "contact_people_id")
 
-    cond = [model_df.name_list.eqNullSafe(contact_people_df.name_list),
-            model_df.email_list.eqNullSafe(contact_people_df.email_list),
-            model_df[Constants.DATA_SOURCE_COLUMN] == contact_people_df[Constants.DATA_SOURCE_COLUMN]]
+    cond = [
+        model_df.name_list.eqNullSafe(contact_people_df.name_list),
+        model_df.email_list.eqNullSafe(contact_people_df.email_list),
+        model_df[Constants.DATA_SOURCE_COLUMN]
+        == contact_people_df[Constants.DATA_SOURCE_COLUMN],
+    ]
 
-    model_df = model_df.join(contact_people_df, cond, how='left')
+    model_df = model_df.join(contact_people_df, cond, how="left")
+    model_df = model_df.drop(contact_people_df.email_list)
+    model_df = model_df.drop(contact_people_df.name_list)
     model_df = model_df.drop(contact_people_df[Constants.DATA_SOURCE_COLUMN])
     return model_df
 
 
 def set_fk_contact_form(model_df: DataFrame, contact_form_df: DataFrame) -> DataFrame:
     model_df = transform_to_fk(
-        model_df, contact_form_df, "form_url", "form_url", "id", "contact_form_id")
+        model_df, contact_form_df, "form_url", "form_url", "id", "contact_form_id"
+    )
     return model_df
 
 
-def set_fk_source_database(model_df: DataFrame, source_database_df: DataFrame) -> DataFrame:
+def set_fk_source_database(
+    model_df: DataFrame, source_database_df: DataFrame
+) -> DataFrame:
     model_df = transform_to_fk(
-        model_df, source_database_df, "database_url", "database_url", "id", "source_database_id")
+        model_df,
+        source_database_df,
+        "database_url",
+        "database_url",
+        "id",
+        "source_database_id",
+    )
     return model_df
 
 
@@ -126,12 +165,16 @@ def set_fk_license(model_df: DataFrame, license_df: DataFrame) -> DataFrame:
     license_df = license_df.withColumnRenamed("name", "license_name")
     license_df = license_df.withColumnRenamed("url", "license_url")
 
-    model_df = model_df.join(license_df, model_df.license == license_df.license_name, how='left')
+    model_df = model_df.join(
+        license_df, model_df.license == license_df.license_name, how="left"
+    )
     return model_df
 
 
 def get_provider_type_from_sharing(raw_sharing_df: DataFrame) -> DataFrame:
-    provider_type_df = raw_sharing_df.select(format_name_column("provider_type").alias("name"))
+    provider_type_df = raw_sharing_df.select(
+        format_name_column("provider_type").alias("name")
+    )
     provider_type_df = provider_type_df.select("name").where("name is not null")
     provider_type_df = provider_type_df.drop_duplicates()
     return provider_type_df
@@ -179,8 +222,9 @@ def get_columns_expected_order(model_df: DataFrame) -> DataFrame:
         "drug_concentration",
         "other_model_links",
         "date_submitted",
-        "model_availability"
-        )
+        "model_availability",
+        "email_list",
+    )
 
 
 if __name__ == "__main__":

etl/jobs/transformation/scoring/calculation_methods/generic_metadata_calculator.py

@@ -10,7 +10,7 @@
 raw_data_score_weight = 0.07
 cancer_annotation_score_weight = 0.03
 
-columns_with_multiple_values = ['quality_assurance', 'xenograft_model_specimens']
+columns_with_multiple_values = ["quality_assurance", "xenograft_model_specimens"]
 
 
 def get_list_resources_available_molecular_data(resources_df: DataFrame):
@@ -33,29 +33,35 @@ def get_metadata_max_score(column_weights):
         if value is None:
             value = 0
         total_score += value
-    
+
     return total_score
 
 
 def is_valid_value(attribute_value: str) -> bool:
-    lc_attribute_value = attribute_value.lower() if attribute_value is not None else ''
-    return (lc_attribute_value != ''
-            and lc_attribute_value != 'not provided'
-            and lc_attribute_value != 'not collected'
-            and lc_attribute_value != 'unknown')
-
-
-def calculate_score_single_value_column(column_name: str, column_value: str, column_weights) -> float:
+    lc_attribute_value = attribute_value.lower() if attribute_value is not None else ""
+    return (
+        lc_attribute_value != ""
+        and lc_attribute_value != "not provided"
+        and lc_attribute_value != "not collected"
+        and lc_attribute_value != "unknown"
+    )
+
+
+def calculate_score_single_value_column(
+    column_name: str, column_value: str, column_weights
+) -> float:
     column_weight = column_weights.get(column_name)
     if is_valid_value(column_value):
         return column_weight
     else:
         return 0
 
 
-def calculate_score_multiple_value_column(column_name: str, column_value: str, column_weights) -> float:
+def calculate_score_multiple_value_column(
+    column_name: str, column_value: str, column_weights
+) -> float:
     score = 0
-    if column_value == '[]' or column_value is None:
+    if column_value == "[]" or column_value is None:
         return score
 
     # `column_value` is expected to be a string representing a JSON array with
@@ -67,7 +73,6 @@ def calculate_score_multiple_value_column(column_name: str, column_value: str, c
     rows_count = len(json_array)
     for obj in json_array:
         for attribute, value in obj.items():
-
             if attribute not in valid_elements_per_column:
                 valid_elements_per_column[attribute] = 0
 
@@ -86,21 +91,29 @@ def calculate_score_multiple_value_column(column_name: str, column_value: str, c
     return score
 
 
-def calculate_score_by_column(column_name: str, column_value: str, column_weights) -> float:
+def calculate_score_by_column(
+    column_name: str, column_value: str, column_weights
+) -> float:
     score = 0
     if column_name in column_weights.keys():
         if is_valid_value(column_value):
-            score += calculate_score_single_value_column(column_name, column_value, column_weights)
+            score += calculate_score_single_value_column(
+                column_name, column_value, column_weights
+            )
     elif column_name in columns_with_multiple_values:
-        score += calculate_score_multiple_value_column(column_name, column_value, column_weights)
+        score += calculate_score_multiple_value_column(
+            column_name, column_value, column_weights
+        )
     return score
 
 
 def calculate_metadata_score(row, column_weights):
     score = 0
     row_as_dict = row.asDict()
     for column_name in row_as_dict:
-        score += calculate_score_by_column(column_name, row_as_dict[column_name], column_weights)
+        score += calculate_score_by_column(
+            column_name, row_as_dict[column_name], column_weights
+        )
     return score / get_metadata_max_score(column_weights) * 100
 
 
@@ -130,10 +143,14 @@ def calculate_cancer_annotation_score(row, total_cancer_annotation_resources):
 def calculate_score_for_row(row, total_cancer_annotation_resources, column_weights):
     columns = {"pdcm_model_id": row["pdcm_model_id"]}
 
-    metadata_score = calculate_metadata_score(row, column_weights) * metadata_score_weight
+    metadata_score = (
+        calculate_metadata_score(row, column_weights) * metadata_score_weight
+    )
     raw_data_score = calculate_raw_data_score(row) * raw_data_score_weight
-    cancer_annotation_score = calculate_cancer_annotation_score(
-        row, total_cancer_annotation_resources) * cancer_annotation_score_weight
+    cancer_annotation_score = (
+        calculate_cancer_annotation_score(row, total_cancer_annotation_resources)
+        * cancer_annotation_score_weight
+    )
 
     score = int(metadata_score + raw_data_score + cancer_annotation_score)
 
@@ -142,21 +159,31 @@ def calculate_score_for_row(row, total_cancer_annotation_resources, column_weigh
     return output
 
 
-def calculate_model_metadata_score(input_df: DataFrame, raw_external_resources_df: DataFrame, column_weights: dict) -> DataFrame:
+def calculate_model_metadata_score(
+    input_df: DataFrame, raw_external_resources_df: DataFrame, column_weights: dict
+) -> DataFrame:
     """
-    Calculates metadata score. It receives a dataframe `input_df` (a subset of `search_index_df` filtered by a model type) 
+    Calculates metadata score. It receives a dataframe `input_df` (a subset of `search_index_df` filtered by a model type)
     and returns a dataset with (pdcm_model_id, score)
     """
     input_df = input_df.drop_duplicates()
-    
-    total_cancer_annotation_resources = count_cancer_annotation_resources(raw_external_resources_df)
-
-    rdd_with_score = input_df.rdd.map(lambda x: calculate_score_for_row(x, total_cancer_annotation_resources, column_weights))
 
-    score_schema = StructType([
-        StructField('pdcm_model_id', LongType(), True),
-        StructField('score', IntegerType(), True)
-    ])
+    total_cancer_annotation_resources = count_cancer_annotation_resources(
+        raw_external_resources_df
+    )
+
+    rdd_with_score = input_df.rdd.map(
+        lambda x: calculate_score_for_row(
+            x, total_cancer_annotation_resources, column_weights
+        )
+    )
+
+    score_schema = StructType(
+        [
+            StructField("pdcm_model_id", LongType(), True),
+            StructField("score", IntegerType(), True),
+        ]
+    )
 
     score_df = rdd_with_score.toDF(score_schema)
 

etl/jobs/transformation/scoring/weights_per_fields.py

@@ -30,8 +30,7 @@
     "quality_assurance.STR_analysis": 0,
     "quality_assurance.comments": 0,
     "supplier": 0,
-    "supplier_type": 0
-
+    "supplier_type": 0,
 }
 
 # Weights for fields that only apply to PDX models
@@ -54,8 +53,8 @@
     "growth_properties": 1,
     "growth_media": 1,
     "media_id": 1,
-    "plate_coating": 1, 
-    "other_plate_coating" :1,
+    "plate_coating": 1,
+    "other_plate_coating": 1,
     "passage_number": 1,
     "contaminated": 1,
     "contamination_details": 0.5,
@@ -66,4 +65,3 @@
     "quality_assurance.tumour_status": 1,
     "quality_assurance.model_purity": 1,
 }
-