Merge pull request #319 from Knowledge-Graph-Hub/use-metpo-oxygen-tolerance-terms

Use BacDive/METPO mapping file in `kg_microbe/transform_utils/bacdive/bacdive.py`

Author: realmarcin

kg_microbe/transform_utils/bacdive/bacdive.py

@@ -180,6 +180,7 @@
 )
 from kg_microbe.transform_utils.transform import Transform
 from kg_microbe.utils.dummy_tqdm import DummyTqdm
+from kg_microbe.utils.mapping_file_utils import load_oxygen_phenotype_mappings
 from kg_microbe.utils.oak_utils import get_label
 from kg_microbe.utils.pandas_utils import drop_duplicates
 from kg_microbe.utils.string_coding import remove_nextlines
@@ -205,6 +206,7 @@ def __init__(
         source_name = BACDIVE
         super().__init__(source_name, input_dir, output_dir)
         self.ncbi_impl = get_adapter(f"sqlite:{NCBITAXON_SOURCE}")
+        self.oxygen_phenotype_mappings = load_oxygen_phenotype_mappings()
 
     def _flatten_to_dicts(self, obj):
         if isinstance(obj, dict):
@@ -1358,15 +1360,25 @@ def run(self, data_file: Union[Optional[Path], Optional[str]] = None, show_statu
                             # e.g. ot_rec might look like {"@ref": 4562, "oxygen tolerance": "microaerophile"}
                             ot_label = ot_rec.get("oxygen tolerance", "").strip()
                             if ot_label:
-                                # Create a node for this oxygen tolerance
-                                # Category is typically "biolink:PhenotypicQuality"
-                                # ID can be something like "oxygen:microaerophile"
+                                # Check if we have a METPO mapping for this oxygen tolerance term
+                                mapping = None
+                                for map_key, map_value in self.oxygen_phenotype_mappings.items():
+                                    if map_key == ot_label:
+                                        mapping = map_value
+                                        break
+                                if mapping:
+                                    # Use METPO term
+                                    ot_id = mapping['curie']
+                                    ot_display_label = mapping['label']
+                                    # print(f"DEBUG: Mapped '{ot_label}' -> {ot_id} ({ot_display_label})")
+                                else:
+                                    # Raise exception if no mapping found
+                                    raise ValueError(f"No METPO mapping found for oxygen tolerance term: '{ot_label}'")
 
-                                ot_id = f"oxygen:{ot_label.replace(' ', '_').lower()}"
                                 node_writer.writerow([
                                     ot_id,
                                     PHENOTYPIC_CATEGORY,
-                                    ot_label
+                                    ot_display_label
                                 ] + [None]*(len(self.node_header) - 3))
 
                                 # Now create an edge from each organism in species_with_strains

kg_microbe/utils/mapping_file_utils.py

@@ -0,0 +1,45 @@
+"""Utilities for handling mapping files from remote sources."""
+import csv
+from typing import Dict
+
+import requests
+
+# remote URL location in metpo GitHub repository for oxygen phenotype mappings file
+BACDIVE_OXYGEN_PHENOTYPE_MAPPINGS_URL = "https://raw.githubusercontent.com/berkeleybop/metpo/refs/heads/main/generated/bacdive_oxygen_phenotype_mappings.tsv"
+
+
+def load_oxygen_phenotype_mappings() -> Dict[str, Dict[str, str]]:
+    """
+    Load METPO oxygen phenotype mappings file from remote location in metpo repository.
+
+    :return: Dictionary mapping BacDive labels to METPO curie and label information.
+             Format: {bacdive_label: {'curie': metpo_curie, 'label': metpo_label}}
+    :rtype: Dict[str, Dict[str, str]]
+    :raises requests.exceptions.HTTPError: If unable to fetch from remote URL
+    :raises ValueError: If the response content is empty or invalid
+    """
+    mappings = {}
+
+    try:
+        response = requests.get(BACDIVE_OXYGEN_PHENOTYPE_MAPPINGS_URL, timeout=30)
+        response.raise_for_status()
+
+        if not response.text.strip():
+            raise ValueError("The contents of the file at the remote URL are empty or invalid.")
+
+        reader = csv.DictReader(response.text.splitlines(), delimiter='\t')
+        for row in reader:
+            bacdive_label = row.get('?bacdive_label', '').strip().strip('"')
+            metpo_curie = row.get('?metpo_curie', '').strip().strip('"')
+            metpo_label = row.get('?metpo_label', '').strip().strip('"')
+
+            if bacdive_label and metpo_curie:
+                mappings[bacdive_label] = {
+                    'curie': metpo_curie,
+                    'label': metpo_label
+                }
+
+        return mappings
+
+    except requests.exceptions.HTTPError as e:
+        raise requests.exceptions.HTTPError(f"Please ensure the remote URL is accessible: {e}") from e