[codex] fix: support info and tree for functional owl inputs (#876)

* fix: support info and tree for functional owl inputs

* fix: support mappings for functional owl inputs

Author: cmungall

src/oaklib/implementations/funowl/funowl_implementation.py

@@ -7,6 +7,7 @@
 
 import pyhornedowl
 import rdflib
+import sssom_schema as sssom
 from kgcl_schema.datamodel import kgcl
 from pyhornedowl.model import (
     IRI,
@@ -38,6 +39,8 @@
 )
 
 from oaklib.datamodels import obograph
+from oaklib.datamodels.search import SearchConfiguration
+from oaklib.datamodels.search_datamodel import SearchTermSyntax
 from oaklib.datamodels.vocabulary import (
     DEPRECATED_PREDICATE,
     EQUIVALENT_CLASS,
@@ -59,15 +62,19 @@
     RDF_TYPE,
     RDFS_DOMAIN,
     RDFS_RANGE,
+    SEMAPV,
+    SKOS_MATCH_PREDICATES,
     SUBPROPERTY_OF,
 )
 from oaklib.interfaces import SearchInterface
 from oaklib.interfaces.basic_ontology_interface import LANGUAGE_TAG, RELATIONSHIP
+from oaklib.interfaces.mapping_provider_interface import MappingProviderInterface
 from oaklib.interfaces.obograph_interface import OboGraphInterface
 from oaklib.interfaces.owl_interface import OwlInterface, ReasonerConfiguration
 from oaklib.interfaces.patcher_interface import PatcherInterface
 from oaklib.types import CURIE, PRED_CURIE
 from oaklib.utilities.axioms.logical_definition_utilities import logical_definition_matches
+from oaklib.utilities.mapping.sssom_utils import inject_mapping_sources
 
 logger = logging.getLogger(__name__)
 SERIALIZATION_ALIASES = {
@@ -139,6 +146,7 @@ class FunOwlImplementation(
     OboGraphInterface,
     PatcherInterface,
     SearchInterface,
+    MappingProviderInterface,
 ):
     """
     An experimental partial implementation of :ref:`OwlInterface`
@@ -329,7 +337,14 @@ def _coerce_annotation_value(self, value: Any):
         return SimpleLiteral(str(value))
 
     def _single_valued_assignment(self, curie: CURIE, property: CURIE) -> Optional[str]:
-        values = self._ontology.get_annotations(self.curie_to_uri(curie), self.curie_to_uri(property))
+        subject_iri = self.curie_to_uri(curie)
+        property_iri = self.curie_to_uri(property)
+        if subject_iri is None or property_iri is None:
+            return None
+        try:
+            values = self._ontology.get_annotations(subject_iri, property_iri)
+        except TypeError:
+            return None
         if values:
             if len(values) > 1:
                 logger.warning("Multiple values for %s %s = %s", curie, property, values)
@@ -429,15 +444,113 @@ def synonym_property_values(
                 for value in alias_map.get(predicate, []):
                     yield curie, obograph.SynonymPropertyValue(pred=pred_text, val=value)
 
+    def simple_mappings_by_curie(self, curie: CURIE) -> Iterable[tuple[PRED_CURIE, CURIE]]:
+        metadata = self.entity_metadata_map(curie)
+        for xref in metadata.get(HAS_DBXREF, []):
+            yield HAS_DBXREF, cast(CURIE, xref)
+        for predicate in SKOS_MATCH_PREDICATES:
+            for mapped_curie in metadata.get(predicate, []):
+                yield predicate, cast(CURIE, mapped_curie)
+
+    def get_sssom_mappings_by_curie(self, curie: CURIE) -> Iterable[sssom.Mapping]:
+        seen = set()
+
+        def _mapping(subject_id: CURIE, predicate_id: PRED_CURIE, object_id: CURIE) -> sssom.Mapping:
+            mapping = sssom.Mapping(
+                subject_id=subject_id,
+                predicate_id=predicate_id,
+                object_id=object_id,
+                mapping_justification=sssom.EntityReference(SEMAPV.UnspecifiedMatching.value),
+            )
+            inject_mapping_sources(mapping)
+            return mapping
+
+        direct_mappings = list(self.simple_mappings_by_curie(curie))
+        for predicate_id, object_id in direct_mappings:
+            key = (curie, predicate_id, object_id)
+            if key in seen:
+                continue
+            seen.add(key)
+            yield _mapping(curie, predicate_id, object_id)
+
+        if direct_mappings or self.label(curie) is not None or set(self.owl_type(curie)):
+            return
+
+        for entity in self.entities(filter_obsoletes=False):
+            for predicate_id, object_id in self.simple_mappings_by_curie(entity):
+                if object_id != curie:
+                    continue
+                key = (entity, predicate_id, object_id)
+                if key in seen:
+                    continue
+                seen.add(key)
+                yield _mapping(entity, predicate_id, object_id)
+
+    def basic_search(
+        self, search_term: str, config: Optional[SearchConfiguration] = None
+    ) -> Iterable[CURIE]:
+        if config is None:
+            config = SearchConfiguration()
+        property_names = {str(p) for p in config.properties}
+        if not property_names:
+            property_names = {"LABEL", "ALIAS"}
+
+        flags = re.IGNORECASE if config.force_case_insensitive else 0
+        normalized_search_term = (
+            search_term.lower() if config.force_case_insensitive else search_term
+        )
+
+        def _normalize(value: str) -> str:
+            return value.lower() if config.force_case_insensitive else value
+
+        if config.syntax == SearchTermSyntax.STARTS_WITH:
+            matches = lambda value: _normalize(value).startswith(normalized_search_term)
+        elif config.syntax == SearchTermSyntax.REGULAR_EXPRESSION:
+            prog = re.compile(search_term, flags=flags)
+            matches = lambda value: prog.search(value) is not None
+        elif config.is_partial:
+            matches = lambda value: normalized_search_term in _normalize(value)
+        else:
+            matches = lambda value: _normalize(value) == normalized_search_term
+
+        search_all = "ANYTHING" in property_names
+        seen = set()
+        for curie in self.entities(filter_obsoletes=not config.include_obsoletes_in_results):
+            if (search_all or "LABEL" in property_names) and (label := self.label(curie)) and matches(label):
+                if curie not in seen:
+                    seen.add(curie)
+                    yield curie
+                continue
+            if (search_all or "IDENTIFIER" in property_names) and matches(curie):
+                if curie not in seen:
+                    seen.add(curie)
+                    yield curie
+                continue
+            if search_all or "ALIAS" in property_names:
+                if any(matches(alias) for alias in self.entity_aliases(curie)):
+                    if curie not in seen:
+                        seen.add(curie)
+                        yield curie
+                    continue
+            if search_all or "MAPPED_IDENTIFIER" in property_names:
+                metadata = self.entity_metadata_map(curie)
+                if any(matches(xref) for xref in metadata.get(HAS_DBXREF, [])):
+                    if curie not in seen:
+                        seen.add(curie)
+                        yield curie
+
     def node(
         self, curie: CURIE, strict=False, include_metadata=False, expand_curies=False
     ) -> Optional[obograph.Node]:
         entity_types = set(self.owl_type(curie))
         label = self.label(curie)
+        node_id = cast(CURIE, self.curie_to_uri(curie)) if expand_curies else curie
+        if node_id is None:
+            node_id = curie
         if not entity_types and label is None:
             if strict:
                 raise ValueError(f"Unknown entity: {curie}")
-            return None
+            return obograph.Node(id=node_id)
         if any(
             owl_type in entity_types
             for owl_type in [OWL_OBJECT_PROPERTY, OWL_ANNOTATION_PROPERTY, OWL_DATATYPE_PROPERTY]
@@ -447,7 +560,6 @@ def node(
             node_type = "INDIVIDUAL"
         else:
             node_type = "CLASS"
-        node_id = cast(CURIE, self.curie_to_uri(curie)) if expand_curies else curie
         meta = None
         if include_metadata:
             meta = obograph.Meta()

tests/test_cli.py

@@ -66,6 +66,21 @@
 TEST_SYNONYMIZER_OBO = "simpleobo:" + str(INPUT_DIR / "synonym-test.obo")
 RULES_FILE = INPUT_DIR / "matcher_rules.yaml"
 SYNONYMIZER_RULES_FILE = INPUT_DIR / "cli-synonymizer-rules.yaml"
+MINIMAL_CL_FUNOWL = """\
+Prefix(obo:=<http://purl.obolibrary.org/obo/>)
+Prefix(oboInOwl:=<http://www.geneontology.org/formats/oboInOwl#>)
+Prefix(owl:=<http://www.w3.org/2002/07/owl#>)
+Prefix(rdfs:=<http://www.w3.org/2000/01/rdf-schema#>)
+Prefix(xsd:=<http://www.w3.org/2001/XMLSchema#>)
+
+Ontology(<http://example.org/cl-edit.owl>
+Declaration(Class(obo:CL_0000540))
+AnnotationAssertion(oboInOwl:hasDbXref obo:CL_0000540 "BTO:0000938")
+AnnotationAssertion(oboInOwl:hasDbXref obo:CL_0000540 "CALOHA:TS-0683")
+AnnotationAssertion(rdfs:label obo:CL_0000540 "neuron"^^xsd:string)
+SubClassOf(obo:CL_0000540 obo:CL_0000000)
+)
+"""
 
 
 def _outpath(test: str, fmt: str = "tmp") -> str:
@@ -94,6 +109,10 @@ def _out(self, path: Optional[str] = TEST_OUT) -> str:
         with open(path) as f:
             return "".join(f.readlines())
 
+    def _write_minimal_cl_funowl(self, path: Path) -> Path:
+        path.write_text(MINIMAL_CL_FUNOWL, encoding="utf-8")
+        return path
+
     def test_main_help(self):
         result = self.runner.invoke(main, ["--help"])
         out = result.stdout
@@ -115,6 +134,74 @@ def test_input_type_and_sniff_for_functional_owl_suffix(self):
                 self.assertEqual(0, result.exit_code, result.output)
                 self.assertIn("nucleus", result.stdout)
 
+    def test_functional_owl_info_and_tree_with_search_and_undeclared_ancestors(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            disguised_path = self._write_minimal_cl_funowl(Path(tmpdir) / "cl-edit.owl")
+            outpath = Path(tmpdir) / "info.tsv"
+
+            result = self.runner.invoke(
+                main, ["-I", "ofn", "-i", str(disguised_path), "info", "neuron", "-o", str(outpath)]
+            )
+            self.assertEqual(0, result.exit_code, result.output)
+            self.assertIn("CL:0000540", outpath.read_text(encoding="utf-8"))
+            self.assertIn("neuron", outpath.read_text(encoding="utf-8"))
+
+            result = self.runner.invoke(
+                main, ["-I", "ofn", "-i", str(disguised_path), "tree", "-p", "i", "CL:0000540"]
+            )
+            self.assertEqual(0, result.exit_code, result.output)
+            self.assertIn("CL:0000540", result.stdout)
+            self.assertIn("CL:0000000", result.stdout)
+
+    def test_functional_owl_mappings(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            disguised_path = self._write_minimal_cl_funowl(Path(tmpdir) / "cl-edit.owl")
+            outpath = Path(tmpdir) / "mappings.csv"
+            reverse_outpath = Path(tmpdir) / "reverse-mappings.csv"
+
+            result = self.runner.invoke(
+                main,
+                [
+                    "-I",
+                    "ofn",
+                    "-i",
+                    str(disguised_path),
+                    "mappings",
+                    "CL:0000540",
+                    "-O",
+                    "csv",
+                    "-o",
+                    str(outpath),
+                ],
+            )
+
+            self.assertEqual(0, result.exit_code, result.output)
+            output = outpath.read_text(encoding="utf-8")
+            self.assertIn("CL:0000540", output)
+            self.assertIn("BTO:0000938", output)
+            self.assertIn("CALOHA:TS-0683", output)
+
+            reverse_result = self.runner.invoke(
+                main,
+                [
+                    "-I",
+                    "ofn",
+                    "-i",
+                    str(disguised_path),
+                    "mappings",
+                    "BTO:0000938",
+                    "-O",
+                    "csv",
+                    "-o",
+                    str(reverse_outpath),
+                ],
+            )
+
+            self.assertEqual(0, reverse_result.exit_code, reverse_result.output)
+            reverse_output = reverse_outpath.read_text(encoding="utf-8")
+            self.assertIn("CL:0000540", reverse_output)
+            self.assertIn("BTO:0000938", reverse_output)
+
     def test_multilingual(self):
         for input_arg in [INPUT_DIR / "hp-international-test.db"]:
             results = self.runner.invoke(main, ["-i", str(input_arg), "languages"])