cellxgene_ontology_guide.supported_versions

  1import functools
  2import gzip
  3import json
  4import os
  5import warnings
  6from datetime import datetime
  7from typing import Any, Dict, List, Optional
  8
  9from semantic_version import Version
 10
 11from cellxgene_ontology_guide._constants import DATA_ROOT, ONTOLOGY_FILENAME_SUFFIX, ONTOLOGY_INFO_FILENAME
 12from cellxgene_ontology_guide.entities import Ontology
 13
 14
 15@functools.cache
 16def load_ontology_file(file_name: str) -> Any:
 17    """Load the ontology file from the data directory and return it as a dict."""
 18    with gzip.open(os.path.join(DATA_ROOT, file_name), "rt") as f:
 19        return json.load(f)
 20
 21
 22def clear_ontology_file_cache() -> None:
 23    """Clear the cache for the load_ontology_file function."""
 24    load_ontology_file.cache_clear()
 25
 26
 27def get_latest_schema_version(versions: List[str]) -> str:
 28    """Given a list of schema versions, return the latest version.
 29
 30    :param versions: List[str] list of schema versions. Versions can be in the format "v5.0.0" or "5.0.0"
 31    :return: str latest version without the leading "v"
 32    """
 33
 34    return str(sorted([coerce_version(version) for version in versions])[-1])
 35
 36
 37def coerce_version(version: str) -> Version:
 38    """Coerce a version string into a semantic_version.Version object.
 39
 40    :param version: str version string to coerce
 41    :return: Version coerced version object
 42    """
 43    v = version[1:] if version[0] == "v" else version
 44    return Version.coerce(v)
 45
 46
 47def load_supported_versions() -> Any:
 48    """Load the ontology_info.json file and return it as a dict."""
 49    with open(os.path.join(DATA_ROOT, ONTOLOGY_INFO_FILENAME)) as f:
 50        return json.load(f)
 51
 52
 53class CXGSchema:
 54    """A class to represent the ontology information used by a cellxgene schema version."""
 55
 56    version: str
 57    """The schema version used by the class instance."""
 58    supported_ontologies: Dict[str, Any]
 59    """A dictionary of supported ontologies for the schema version."""
 60    imported_ontologies: Dict[str, str]
 61    """In our supported ontologies, the CxG schema can support terms imported from different ontologies. 
 62    This dictionary maps these 'additional ontologies' to their supported ontology name. For example, 
 63    for ZFS ontology terms imported into the ZFA ontology, imported_ontologies would be {"ZFS":"ZFA", ...}"""
 64    ontology_file_names: Dict[str, str]
 65    """A dictionary of ontology names and their corresponding file names."""
 66
 67    def __init__(self, version: Optional[str] = None):
 68        """
 69
 70        :param version: The schema version to use. If not provided, the latest schema version will be used.
 71        """
 72        ontology_info = load_supported_versions()
 73        if version is None:
 74            _version = get_latest_schema_version(ontology_info.keys())
 75        else:
 76            _version = str(coerce_version(version))
 77            if str(_version) not in ontology_info:
 78                raise ValueError(f"Schema version {_version} is not supported in this package version.")
 79
 80        self.version = _version
 81        self.supported_ontologies = ontology_info[_version]["ontologies"]
 82        self.imported_ontologies = {
 83            imported_ontology: ontology
 84            for ontology, info in self.supported_ontologies.items()
 85            for imported_ontology in info.get("additional_ontologies", [])
 86        }
 87        self.cross_ontology_mappings = {
 88            ontology for ontology, info in self.supported_ontologies.items() if info.get("cross_ontology_mapping")
 89        }
 90        self.ontology_file_names: Dict[str, str] = {}
 91        self.deprecated_on = ontology_info[_version].get("deprecated_on")
 92        if self.deprecated_on:
 93            parsed_date = datetime.strptime(self.deprecated_on, "%Y-%m-%d")
 94            warnings.warn(
 95                f"Schema version {_version} is deprecated as of {parsed_date}. It will be removed in a future version.",
 96                DeprecationWarning,
 97                stacklevel=1,
 98            )
 99
100    def ontology(self, name: str) -> Any:
101        """Return the ontology terms for the given ontology name. Load from the file cache if available.
102
103        Does not support "additional ontologies" of another ontology.
104
105        :param name: str name of the ontology to get the terms for
106        :return: dict representation of the ontology terms
107        """
108        if name not in self.ontology_file_names:
109            if getattr(Ontology, name, None) is None:
110                raise ValueError(f"Ontology {name} is not supported in this package version.")
111
112            try:
113                onto_version = self.supported_ontologies[name]["version"]
114            except KeyError as e:
115                raise ValueError(f"Ontology {name} is not supported for schema version {self.version}") from e
116            file_name = f"{name}-ontology-{onto_version}{ONTOLOGY_FILENAME_SUFFIX}"
117            self.ontology_file_names[name] = file_name  # save to file name to access from cache
118        return load_ontology_file(self.ontology_file_names[name])
119
120    def get_ontology_download_url(self, ontology: Ontology) -> str:
121        """
122        Get the download URL for a given ontology file.
123
124        Examples:
125        get_ontology_download_url("CL") -> "http://example.com/2024-01-01/cl.owl"
126
127        :param ontology: Ontology enum of the ontology to fetch
128        :return: str download URL for the requested ontology file
129        """
130        source_url = self.supported_ontologies[ontology.name]["source"]
131        version = self.supported_ontologies[ontology.name]["version"]
132        filename = self.supported_ontologies[ontology.name]["filename"]
133        return f"{source_url}/{version}/{filename}"