Merge pull request #57 from monarch-initiative/56-add-random-terms-to…

…-disease-profile

56-add-random-terms-to-disease-profile

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "phenotype2phenopacket"
-version = "0.6.12"
+version = "0.6.13"
 description = ""
 authors = ["Yasemin Bridges <[email protected]>"]
 readme = "README.md"

src/phenotype2phenopacket/create/create.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from typing import List
 
 import polars as pl
 from oaklib.implementations import ProntoImplementation
@@ -16,12 +17,29 @@
 )
 
 
+def _get_terms_for_randomisation(
+    human_phenotype_ontology: ProntoImplementation,
+) -> List[str]:
+    """
+    Get the terms for randomisation.
+    Returns:
+        Set[str]: A set of terms for randomisation.
+    """
+    descendents = set(human_phenotype_ontology.descendants("HP:0000118"))
+    descendents.discard("HP:0000118")
+    direct_children = {
+        term[1] for term in set(human_phenotype_ontology.incoming_relationships("HP:0000118"))
+    }
+    return list(descendents - direct_children)
+
+
 def create_synthetic_patient_phenopacket(
     human_phenotype_ontology: ProntoImplementation,
     omim_disease: pl.DataFrame,
     output_dir: Path,
     pt_id: str,
     hpoa_version: str,
+    random_terms: List[str],
 ):
     """
     Create a synthetic patient phenopacket from a set of phenotype entries for a specific OMIM disease.
@@ -34,7 +52,9 @@ def create_synthetic_patient_phenopacket(
         hpoa_version (str): The version of the Human Phenotype Ontology Annotation.
 
     """
-    synthetic_patient_generator = SyntheticPatientGenerator(omim_disease, human_phenotype_ontology)
+    synthetic_patient_generator = SyntheticPatientGenerator(
+        omim_disease, human_phenotype_ontology, random_terms
+    )
     patient_terms = synthetic_patient_generator.patient_term_annotation_set()
     phenopacket_file = PhenotypeAnnotationToPhenopacketConverter(
         human_phenotype_ontology
@@ -45,7 +65,8 @@ def create_synthetic_patient_phenopacket(
         onset=synthetic_patient_generator.get_onset_range(),
     )
     write_phenopacket(
-        phenopacket_file.phenopacket, output_dir.joinpath(phenopacket_file.phenopacket_path)
+        phenopacket_file.phenopacket,
+        output_dir.joinpath(phenopacket_file.phenopacket_path),
     )
 
 
@@ -72,6 +93,7 @@ def create_synthetic_patients(
     """
     phenotype_annotation_data = return_phenotype_annotation_data(phenotype_annotation)
     human_phenotype_ontology = load_ontology(local_cached_ontology)
+    random_terms = _get_terms_for_randomisation(human_phenotype_ontology)
     grouped_omim_diseases = filter_diseases(
         num_disease, omim_id, omim_id_list, phenotype_annotation_data
     )
@@ -89,4 +111,5 @@ def create_synthetic_patients(
             output_dir,
             None,
             phenotype_annotation_data.version,
+            random_terms,
         )

src/phenotype2phenopacket/utils/phenopacket_utils.py

@@ -2,7 +2,7 @@
 import secrets
 import threading
 import warnings
-from copy import copy
+from copy import copy, deepcopy
 from dataclasses import dataclass
 from fractions import Fraction
 from pathlib import Path
@@ -149,7 +149,12 @@ def write_phenopacket(phenopacket: Phenopacket, output_file: Path) -> None:
 class SyntheticPatientGenerator:
     """Class for generating synthetic patients."""
 
-    def __init__(self, disease_df: pl.DataFrame, ontology: ProntoImplementation):
+    def __init__(
+        self,
+        disease_df: pl.DataFrame,
+        ontology: ProntoImplementation,
+        random_terms: List[str] = None,
+    ):
         """
         Initialise the SyntheticPatientGenerator class
 
@@ -164,6 +169,7 @@ def __init__(self, disease_df: pl.DataFrame, ontology: ProntoImplementation):
         self.upper_age = 0
         self.filtered_df = []
         self.secret_rand = secrets.SystemRandom()
+        self.random_terms = random_terms
 
     def get_number_of_terms(self) -> int:
         """
@@ -486,8 +492,8 @@ def get_parents_of_terms(self, phenotype_entry: dict, steps: int) -> dict:
             term = "".join(rels[(list(rels.keys())[0])]) if rels else ""
             if (
                 term.startswith("Abnormality of")
-                or term_id == "HP:0000118"
-                or term_id == "HP:0032443"
+                or parent == "HP:0000118"
+                or parent == "HP:0032443"
             ):
                 break
             else:
@@ -545,6 +551,40 @@ def alter_term_specificity(
             )
         return new_phenotype_terms
 
+    def calculate_number_of_randomised_terms(self, num_terms: int) -> int:
+        """
+        Calculate the number of terms to randomise.
+        Args:
+            num_terms: The total number of terms.
+
+        Returns:
+            int: Number of terms to randomise.
+
+        """
+        return round(num_terms * 0.33 * self.secret_rand.uniform(0, 1))
+
+    def randomised_terms(self, num_terms: int, template_dict: dict) -> List[dict]:
+        """
+        Create randomised terms from a template.
+        Args:
+            num_terms: The total number of terms.
+            template_dict: The template dictionary.
+        Returns:
+            List[dict]: List of randomised terms.
+        """
+        num_terms_to_randomise = self.calculate_number_of_randomised_terms(num_terms)
+        random_terms_subset = self.secret_rand.choices(self.random_terms, k=num_terms_to_randomise)
+        return [
+            {
+                **deepcopy(template_dict),
+                "hpo_id": random_term,
+                "modifier": None,
+                "sex": None,
+                "onset": None,
+            }
+            for random_term in random_terms_subset
+        ]
+
     def patient_term_annotation_set(self) -> pl.DataFrame:
         """
         Get the final patient term annotation set.
@@ -569,7 +609,10 @@ def patient_term_annotation_set(self) -> pl.DataFrame:
         patient_terms_filtered = self.remove_terms_to_be_randomised(
             patient_terms, patient_terms_sub_sample
         )
-        final_patient_terms = patient_terms_filtered.to_dicts() + new_phenotype_terms
+        altered_patient_terms = patient_terms_filtered.to_dicts() + new_phenotype_terms
+        final_patient_terms = altered_patient_terms + self.randomised_terms(
+            len(altered_patient_terms), altered_patient_terms[0]
+        )
         return pl.from_dicts(final_patient_terms, infer_schema_length=len(final_patient_terms))