diff --git a/README.md b/README.md index e92a633..c5c8bd5 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ Build the same RAG-based alignment yourself for full control over every stage: ```python from ontoaligner.ontology import GenericOMDataset -from ontoaligner.utils import metrics, xmlify +from ontoaligner.utils import metrics, xmlify, sssom_pydantic from ontoaligner.aligner import MistralLLMBERTRetrieverRAG from ontoaligner.encoder import ConceptParentRAGEncoder from ontoaligner.postprocess import rag_hybrid_postprocessor @@ -129,6 +129,9 @@ print("Hybrid Matching Evaluation Report:", evaluation) # Step 7: Convert matchings to XML format and save the XML representation xml_str = xmlify.xml_alignment_generator(matchings=hybrid_matchings) open("matchings.xml", "w", encoding="utf-8").write(xml_str) + +# Or you can explicitly convert to SSSOM and write to disk +sssom_pydantic.sssom_alignment_generator(hybrid_matchings, "matchings.sssom.tsv") ``` ### Advanced `AlignerPipeline` diff --git a/docs/source/aligner/kge.rst b/docs/source/aligner/kge.rst index 558d54c..bd0ec0a 100644 --- a/docs/source/aligner/kge.rst +++ b/docs/source/aligner/kge.rst @@ -187,55 +187,55 @@ The following table lists the available KGE aligners: * - ``ConvEAligner`` - Based on ConvE, which uses 2D convolutions over reshaped entity and relation embeddings to model complex interactions. - - `Source `_ + - `Source `_ * - ``TransDAligner`` - Based on TransD, which constructs relation-specific projection matrices dynamically from both entity and relation vectors. - - `Source `_ + - `Source `_ * - ``TransEAligner`` - Based on TransE, a translation-based model that learns embeddings where :math:`h + r \approx t`. - - `Source `_ + - `Source `_ * - ``TransFAligner`` - Based on TransF, which enables flexible translations for complex relations without increasing model complexity. - - `Source `_ + - `Source `_ * - ``TransHAligner`` - Based on TransH, which projects entities onto relation-specific hyperplanes before translation. - - `Source `_ + - `Source `_ * - ``TransRAligner`` - Based on TransR, which embeds entities and relations in separate spaces using relation-specific projections. - - `Source `_ + - `Source `_ * - ``DistMultAligner`` - Based on DistMult, a bilinear model that uses diagonal matrices for efficient relational modeling. - - `Source `_ + - `Source `_ * - ``ComplExAligner`` - Based on ComplEx, which uses complex-valued embeddings to model symmetric and antisymmetric relations; includes a custom similarity function using real parts of complex dot products. - - `Source `_ + - `Source `_ * - ``HolEAligner`` - Based on HolE, which combines compositional and holographic representations using circular correlation. - - `Source `_ + - `Source `_ * - ``RotatEAligner`` - Based on RotatE, which models relations as rotations in complex space and supports rich relational patterns; includes a similarity override. - - `Source `_ + - `Source `_ * - ``SimplEAligner`` - Based on SimplE, which learns dependent embeddings for each entity and supports fully expressive factorization. - - `Source `_ + - `Source `_ * - ``CrossEAligner`` - Based on CrossE, which learns both general and triple-specific embeddings to capture bidirectional interactions. - - `Source `_ + - `Source `_ * - ``BoxEAligner`` - Based on BoxE, which models relations as boxes in vector space to support hierarchies and logical rules. - - `Source `_ + - `Source `_ * - ``CompGCNAligner`` - Based on CompGCN, a graph convolutional network designed for multi-relational graphs using composition operations. - - `Source `_ + - `Source `_ * - ``MuREAligner`` - Based on MuRE, which embeds entities in hyperbolic space to better model hierarchies and relation-specific transformations. - - `Source `_ + - `Source `_ * - ``QuatEAligner`` - Based on QuatE, which uses quaternion embeddings and custom similarity logic to model expressive 4D rotations and relational structure. - - `Source `_ + - `Source `_ * - ``SEAligner`` - Based on SE, a neural model that embeds symbolic knowledge into vector space using learned neural transformations. - - `Source `_ + - `Source `_ To use KGE aligner based technique: @@ -279,8 +279,8 @@ KGE Retriever .. sidebar:: Key Parameters: - - ``retruever``: boolean - - ``top_K``: integer + - ``retriever``: boolean + - ``top_k``: integer In addition to one-to-one alignments, OntoAligner also supports retriever-based alignment. When retriever mode is enabled (``retriever=True``), the aligner returns the top-k candidate target entities for each source entity, along with their similarity scores (similar to retriever aligner). This model is useful if you want to build downstream candidate filtering pipelines, apply human-in-the-loop validation, or integrate with reranking modules (e.g., LLMs or supervised classifiers). diff --git a/docs/source/aligner/olala.rst b/docs/source/aligner/olala.rst index 85dc87b..09d46d3 100644 --- a/docs/source/aligner/olala.rst +++ b/docs/source/aligner/olala.rst @@ -94,7 +94,7 @@ Usage .. code-block:: python - task = OLaLaOMDataset(language="en") + task = OLaLaOMDataset() print("Task:", task) dataset = task.collect( diff --git a/docs/source/aligner/rag.rst b/docs/source/aligner/rag.rst index 6a464d8..ddd40d0 100644 --- a/docs/source/aligner/rag.rst +++ b/docs/source/aligner/rag.rst @@ -282,7 +282,7 @@ ICV-RAG Aligner [1] Liu, S., Ye, H., Xing, L., & Zou, J. (2023). `In-context vectors: Making in context learning more effective and controllable through latent space steering `_. arXiv preprint arXiv:2311.06668. -This RAG variant performs ontology matching using ``ConceptRAGEncoder`` only. The In-Contect Vectors introduced by [1](https://github.com/shengliu66/ICV) tackle in-context learning as in-context vectors (ICV). We used LLMs in this perspective in the RAG module. The workflow is the same as RAG or FewShot RAG with the following differences: +This RAG variant performs ontology matching using ``ConceptRAGEncoder`` only. The In-Context Vectors introduced by [1](https://github.com/shengliu66/ICV) tackle in-context learning as in-context vectors (ICV). We used LLMs in this perspective in the RAG module. The workflow is the same as RAG or FewShot RAG with the following differences: 1. Incorporate the ``ConceptRAGEncoder`` and also provide reference (or examples to build up the ICV vectors). @@ -314,7 +314,7 @@ Embedded ICV-RAG aligners within OntoAligner: - Link * - ``FalconLLMAdaRetrieverICVRAG`` - - Falcon LLM with Ada retriever for iterative consistency verification (ICV) alignment. + - Falcon LLM with Ada retriever for ICV-driven alignment. - `Source `__ * - ``FalconLLMBERTRetrieverICVRAG`` @@ -342,7 +342,7 @@ Embedded ICV-RAG aligners within OntoAligner: - `Source `__ * - ``VicunaLLMBERTRetrieverICVRAG`` - - Vicuna model paired with BERT-based retrieval for iterative consistency verification. + - Vicuna model paired with BERT-based retrieval for ICV-driven alignment. - `Source `__ diff --git a/examples/retriever_sssom.py b/examples/retriever_sssom.py new file mode 100644 index 0000000..b660a67 --- /dev/null +++ b/examples/retriever_sssom.py @@ -0,0 +1,130 @@ +import json + +# Import necessary modules from the 'ontoaligner' library +# The library provides tools for ontology alignment tasks, including dataset management, encoding, retrieval models, and postprocessing. +from ontoaligner import ontology, encoder +from ontoaligner.utils import metrics +from ontoaligner.utils.sssom import sssom_alignment_generator +from ontoaligner.aligner import SBERTRetrieval # Other available modules: AdaRetrieval, SVMBERTRetrieval, BM25Retrieval +from ontoaligner.postprocess import retriever_postprocessor + + +# Step 1: Initialize the ontology matching task +# The task is created using the Mouse-Human ontology matching dataset, +# which includes source and target anatomy ontologies and reference matchings for evaluation. +task = ontology.MouseHumanOMDataset() + +# Confirm the task initialization by printing its details +print("Test Task:", task) + + +# Step 2: Collect the ontology dataset +# The dataset includes paths to the source ontology, target ontology, and reference matching files. +dataset = task.collect( + source_ontology_path="../assets/mouse-human/source.xml", + target_ontology_path="../assets/mouse-human/target.xml", + reference_matching_path="../assets/mouse-human/reference.xml", +) + + +# Step 3: Initialize the encoder model +# The encoder prepares concept representations from the source and target ontologies. +# Here, the 'ConceptParentLightweightEncoder' is used for lightweight encoding, which can be used here as well. +encoder_model = encoder.ConceptParentLightweightEncoder() + +# Encode the source and target ontologies +# The encoder processes the concepts in both ontologies and returns embeddings for further alignment. +encoder_output = encoder_model( + source=dataset["source"], + target=dataset["target"], +) + + +# Step 4: Set up the retrieval model +# The retrieval model aligns the source and target ontologies using semantic similarity techniques. +# 'SBERTRetrieval' is selected with a pre-trained model ('all-MiniLM-L6-v2') for embedding retrieval. +model = SBERTRetrieval( + device="cpu", + top_k=1, +) + +model.load( + path="all-MiniLM-L6-v2", +) + +# Generate ontology matchings +# The retrieval model compares encoded embeddings from the source and target datasets to predict matchings. +matchings = model.generate( + input_data=encoder_output, +) + + +# Step 5: Post-process the matchings +# Apply the 'retriever_postprocessor' function to refine the predicted matchings. +# Postprocessing helps filter or adjust the matchings for improved alignment quality. +threshold = 0.2 + +matchings = retriever_postprocessor( + predicts=matchings, + threshold=threshold, +) + + +# Step 6: Evaluate the matchings +# The evaluation report compares the predicted matchings against the reference matchings +# provided in the dataset using metrics such as precision, recall, and F1-score. +evaluation = metrics.evaluation_report( + predicts=matchings, + references=dataset["reference"], +) + +# Print the evaluation report in a human-readable JSON format +print( + "Evaluation Report:", + json.dumps(evaluation, indent=4), +) + + +# Step 7: Export matchings in SSSOM format +# Convert the generated matchings into an SSSOM TSV alignment file using +# the 'sssom_alignment_generator' utility. +sssom_str = sssom_alignment_generator( + matchings=matchings, + source=dataset["source"], + target=dataset["target"], + predicate_id="owl:equivalentClass", + mapping_set_metadata={ + "mapping_set_id": "https://example.org/mappings/mouse-human-sbert", + "license": "https://creativecommons.org/licenses/by/4.0/", + "subject_type": "owl:Class", + "object_type": "owl:Class", + }, + # Explicit CURIE maps must include all used prefixes; if omitted, Bioregistry is used as fallback. + curie_map={ + "mouse": "http://mouse.owl#", + "human": "http://human.owl#", + "owl": "http://www.w3.org/2002/07/owl#", + "semapv": "https://w3id.org/semapv/vocab/", + }, + aligner=model, + postprocessor=retriever_postprocessor, + postprocessor_params={ + "threshold": threshold, + }, + # If you don't want per-matching aligner/postprocessor metadata (e.g. similarity + # scores or mapping_justification inferred from the aligner), set + # include_aligner_metadata=False to produce a minimal SSSOM output. + include_aligner_metadata=False, +) + + +# Save the SSSOM alignment to a file for further use or analysis +output_file_path = "mouse-human-sbert.sssom.tsv" + +with open(output_file_path,"w",encoding="utf-8",newline="",) as sssom_file: + sssom_file.write(sssom_str) + +print( + f"Matchings in SSSOM format have been successfully written to " + f"'{output_file_path}'." +) diff --git a/ontoaligner/aligner/ensemble/voting/base.py b/ontoaligner/aligner/ensemble/voting/base.py index 6099655..af0d4f3 100644 --- a/ontoaligner/aligner/ensemble/voting/base.py +++ b/ontoaligner/aligner/ensemble/voting/base.py @@ -121,6 +121,8 @@ def _select(self, fused: List[Dict]) -> List[Dict]: Returns: List[Dict]: The filtered/selected predictions. """ + if self.selection == "none": + return fused if self.selection == "top1_source": return self._select_top1_per_source(fused=fused) if self.selection == "bijective": diff --git a/ontoaligner/aligner/ensemble/voting/reciprocal_rank_fusion.py b/ontoaligner/aligner/ensemble/voting/reciprocal_rank_fusion.py index 7ec8bb9..3624ce5 100644 --- a/ontoaligner/aligner/ensemble/voting/reciprocal_rank_fusion.py +++ b/ontoaligner/aligner/ensemble/voting/reciprocal_rank_fusion.py @@ -42,9 +42,9 @@ def __str__(self): Returns a string representation of the voting method. Returns: - dict: A dictionary with the class name as key and voting configuration as value. + str: A simple string representation of the class ("ReciprocalRankFusionVoting"). """ - return {"ReciprocalRankFusion": {"k": self.k}} + return "ReciprocalRankFusionVoting" def fuse(self, aligner_outputs: List[Tuple[List[Dict], float]]) -> List[Dict]: """ diff --git a/ontoaligner/aligner/ensemble/voting/weighted.py b/ontoaligner/aligner/ensemble/voting/weighted.py index 494f672..65ce3a0 100644 --- a/ontoaligner/aligner/ensemble/voting/weighted.py +++ b/ontoaligner/aligner/ensemble/voting/weighted.py @@ -42,14 +42,9 @@ def __str__(self): Returns a string representation of the voting method. Returns: - dict: A dictionary with the class name as key and voting configuration as value. + str: A simple string representation of the class ("WeightedVoting"). """ - return { - "WeightedVoting": { - "min_votes": self.min_votes, - "score_threshold": self.score_threshold, - } - } + return "WeightedVoting" def fuse(self, aligner_outputs: List[Tuple[List[Dict], float]]) -> List[Dict]: """ diff --git a/ontoaligner/aligner/fewshot/fewshot.py b/ontoaligner/aligner/fewshot/fewshot.py index b7f88f2..594fd25 100644 --- a/ontoaligner/aligner/fewshot/fewshot.py +++ b/ontoaligner/aligner/fewshot/fewshot.py @@ -77,7 +77,11 @@ def generate(self, input_data: List[Dict]) -> List: """ # IR generation ir_output = self.ir_generate(input_data=input_data) - ir_output_cleaned = process.retriever_postprocessor(predicts=ir_output) + if 'threshold' in self.kwargs['retriever_config']: + threshold = self.kwargs['retriever_config']['threshold'] + else: + threshold = 0.0 + ir_output_cleaned = process.retriever_postprocessor(predicts=ir_output, threshold=threshold) examples = self.build_fewshots(input_data=input_data) input_data['examples'] = examples # LLm generation diff --git a/ontoaligner/aligner/flora/flora.py b/ontoaligner/aligner/flora/flora.py index 99cb274..d5a1359 100644 --- a/ontoaligner/aligner/flora/flora.py +++ b/ontoaligner/aligner/flora/flora.py @@ -189,7 +189,7 @@ def __init__( string_identity: bool = False, relinit: float = 0.1, ngrams: Optional[List[int]] = None, - model_id: Optional[str] = None, + model_id: Optional[str] = "Lihuchen/pearl_small", emb_path: Optional[str] = None, training_data: Optional[str] = None, device: Optional[str] = None, diff --git a/ontoaligner/aligner/flora/literals.py b/ontoaligner/aligner/flora/literals.py index 5bccd1f..8e1d713 100644 --- a/ontoaligner/aligner/flora/literals.py +++ b/ontoaligner/aligner/flora/literals.py @@ -406,8 +406,7 @@ def __init__( if not identity: if emb_path is not None: try: - self.kg1_pretrained = self.load_embeddings(emb_path) - self.kg2_pretrained = self.load_embeddings(emb_path) + self.kg1_pretrained, self.kg2_pretrained = self.load_embeddings(emb_path) except Exception as e: raise RuntimeError(f"Error loading embeddings from {emb_path}: {e} \n " f"Please ensure the embedding files exist and are in the correct format. ") diff --git a/ontoaligner/aligner/icv/icv.py b/ontoaligner/aligner/icv/icv.py index 1972bff..0bd1faf 100644 --- a/ontoaligner/aligner/icv/icv.py +++ b/ontoaligner/aligner/icv/icv.py @@ -35,6 +35,7 @@ from .tasks.demo import DemoProbInferenceForStyle from ..rag import RAG, RAGBasedDecoderLLMArch from ...postprocess import process +from .utils.llm_layers import get_layers def tokenize_each_demonstration(tok, demonstration_list, dataset_name=None): """ @@ -141,6 +142,18 @@ def __init__(self, model): for params in self.model.parameters(): params.requires_grad = False + def _get_mlp_attr_name(self, layer): + """ + Returns the feed-forward/MLP attribute name used by the transformer layer. + """ + for attr_name in ("mlp", "ffn", "feed_forward", "feedforward"): + if hasattr(layer, attr_name): + return attr_name + + raise AttributeError( + f"Could not find MLP/FFN module in layer {layer.__class__.__name__}" + ) + def get_model(self, icvs, alpha): """ Adds ICV-based adapter layers to the model for ICV-based embedding adjustment. @@ -155,23 +168,28 @@ def get_model(self, icvs, alpha): torch.nn.Module The model with ICV-based adapter layers integrated. """ - for i in range(0, len(self.model.transformer.h)): - icvs_ = icvs[i] - self.model.transformer.h[i].mlp = torch.nn.Sequential( - self.model.transformer.h[i].mlp, AdapterLayer(icvs_, alpha) - ) + layers = get_layers(self.model) + + for i, layer in enumerate(layers): + mlp_attr_name = self._get_mlp_attr_name(layer) + mlp_layer = getattr(layer, mlp_attr_name) + + setattr(layer,mlp_attr_name,torch.nn.Sequential(mlp_layer, AdapterLayer(icvs[i], alpha))) return self.model def remove_adapter(self): """ Removes adapter layers from the model and restores the original architecture. - """ weight_all = [] + layers = get_layers(self.model) - for i in range(0, len(self.model.transformer.h)): - weight_all.append(self.model.transformer.h[i].mlp[1].weight_all) - self.model.transformer.h[i].mlp = self.model.transformer.h[i].mlp[0] + for layer in layers: + mlp_attr_name = self._get_mlp_attr_name(layer) + mlp_layer = getattr(layer, mlp_attr_name) + + weight_all.append(mlp_layer[1].weight_all) + setattr(layer, mlp_attr_name, mlp_layer[0]) return weight_all @@ -222,7 +240,7 @@ def build_icv(self, examples): icvs_to_shift = [icv_examples] updated_wrapper = ICVAdapter(self.model) _ = updated_wrapper.get_model( - torch.stack(icvs_to_shift, dim=1).cuda(), alpha=[self.icv_alpha] + torch.stack(icvs_to_shift, dim=1).to(self.kwargs.get("device", "cpu")), alpha=[self.icv_alpha] ) @@ -278,7 +296,11 @@ def generate(self, input_data: List) -> List: """ # IR generation ir_output = self.ir_generate(input_data=input_data) - ir_output_cleaned = process.retriever_postprocessor(predicts=ir_output) + if 'threshold' in self.kwargs['retriever_config']: + threshold = self.kwargs['retriever_config']['threshold'] + else: + threshold = 0.0 + ir_output_cleaned = process.retriever_postprocessor(predicts=ir_output, threshold=threshold) examples = self.build_icv_examples(input_data=input_data) self.LLM.build_icv(examples=examples) # LLm generation diff --git a/ontoaligner/aligner/llm/dataset.py b/ontoaligner/aligner/llm/dataset.py index 2075139..cfe6db1 100644 --- a/ontoaligner/aligner/llm/dataset.py +++ b/ontoaligner/aligner/llm/dataset.py @@ -160,54 +160,14 @@ def fill_one_sample(self, input_data: Any) -> str: source = self.preprocess(input_data["source"].get("label", "")) target = self.preprocess(input_data["target"].get("label", "")) - source_domain = ( - " ".join(input_data["source"].get("domain_text", [])) - if len(input_data["source"].get("domain_text", [])) > 0 - else "" - ) - - target_domain = ( - " ".join(input_data["target"].get("domain_text", [])) - if len(input_data["target"].get("domain_text", [])) > 0 - else "" - ) + source_domain = self.preprocess(input_data["source"].get("domain_text", "")) + target_domain = self.preprocess(input_data["target"].get("domain_text", "")) - source_range = ( - " ".join(input_data["source"].get("range_text", [])) - if len(input_data["source"].get("range_text", [])) > 0 - else "" - ) - - target_range = ( - " ".join(input_data["target"].get("range_text", [])) - if len(input_data["target"].get("range_text", [])) > 0 - else "" - ) + source_range = self.preprocess(input_data["source"].get("range_text", "")) + target_range = self.preprocess(input_data["target"].get("range_text", "")) - source_inverse = "" - if input_data["source"].get("inverse_of"): - source_inverse = ( - " ".join(input_data["source"].get("inverse_label", [])) - if len(input_data["source"].get("inverse_label", [])) > 0 - else "" - ) - - target_inverse = "" - if input_data["target"].get("inverse_of"): - target_inverse = ( - " ".join(input_data["target"].get("inverse_label", [])) - if len(input_data["target"].get("inverse_label", [])) > 0 - else "" - ) - - source_domain = self.preprocess(source_domain) - target_domain = self.preprocess(target_domain) - - source_range = self.preprocess(source_range) - target_range = self.preprocess(target_range) - - source_inverse = self.preprocess(source_inverse) - target_inverse = self.preprocess(target_inverse) + source_inverse = self.preprocess(input_data["source"].get("inverse_text", "")) + target_inverse = self.preprocess(input_data["target"].get("inverse_text", "")) template = ( template.replace("{source}", source) diff --git a/ontoaligner/aligner/llm/llm.py b/ontoaligner/aligner/llm/llm.py index 76d4670..076f59a 100644 --- a/ontoaligner/aligner/llm/llm.py +++ b/ontoaligner/aligner/llm/llm.py @@ -232,7 +232,7 @@ def load_model(self, path: str) -> None: path (str): The path to the pretrained model. """ if self.kwargs["device"] != "cpu": - self.model = self.model.from_pretrained(self.path, load_in_8bit=True, device_map=self.kwargs['device_map']) + self.model = self.model.from_pretrained(path, load_in_8bit=True, device_map=self.kwargs['device_map']) else: super().load_model(path=path) @@ -276,6 +276,7 @@ def generate_for_multiple_input(self, tokenized_input_data: Any) -> List: sequence_ids = self.model.generate( input_ids=tokenized_input_data["input_ids"], attention_mask=tokenized_input_data["attention_mask"], + num_beams=self.kwargs["num_beams"], max_new_tokens=self.kwargs["max_new_tokens"], temperature=self.kwargs["temperature"], top_p=self.kwargs["top_p"], @@ -403,6 +404,68 @@ def __str__(self): """ return "EncoderDecoderLLMArch" + def generate_for_one_input(self, tokenized_input_data: Any) -> List: + """ + Generates output for a single input using beam search. + + Args: + tokenized_input_data (Any): The tokenized input data. + + Returns: + List: A list of generated texts for the single input. + """ + with torch.no_grad(): + sequence_ids = self.model.generate( + **tokenized_input_data, + num_beams=self.kwargs["num_beams"], + max_new_tokens=self.kwargs["max_new_tokens"], + temperature=self.kwargs["temperature"], + top_p=self.kwargs["top_p"], + pad_token_id=self.tokenizer.eos_token_id, + return_dict_in_generate=True, + output_scores=False, + ) + + generated_ids = sequence_ids["sequences"] + + sequences = self.tokenizer.batch_decode( + generated_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=False, + ) + return sequences + + def generate_for_multiple_input(self, tokenized_input_data: Any) -> List: + """ + Generates output for multiple inputs. + + Args: + tokenized_input_data (Any): The tokenized input data. + + Returns: + List: A list of generated texts for the multiple inputs. + """ + with torch.no_grad(): + sequence_ids = self.model.generate( + input_ids=tokenized_input_data["input_ids"], + attention_mask=tokenized_input_data["attention_mask"], + num_beams=self.kwargs["num_beams"], + max_new_tokens=self.kwargs["max_new_tokens"], + temperature=self.kwargs["temperature"], + top_p=self.kwargs["top_p"], + pad_token_id=self.tokenizer.eos_token_id, + return_dict_in_generate=True, + output_scores=False, + ) + generated_ids = sequence_ids["sequences"] + + sequences = self.tokenizer.batch_decode( + generated_ids, + skip_special_tokens=True, + clean_up_tokenization_spaces=False, + ) + return sequences + class DecoderLLMArch(BaseLLMArch): """ diff --git a/ontoaligner/aligner/retrieval/retrieval.py b/ontoaligner/aligner/retrieval/retrieval.py index 790b5c3..9f6352c 100644 --- a/ontoaligner/aligner/retrieval/retrieval.py +++ b/ontoaligner/aligner/retrieval/retrieval.py @@ -268,7 +268,7 @@ def load(self, path: str): Returns: None """ - self.model = SentenceTransformer(self.path, device=self.kwargs["device"]) + self.model = SentenceTransformer(path, device=self.kwargs["device"]) def fit(self, inputs: Any) -> Any: """ @@ -330,7 +330,7 @@ def generate(self, input_data: List) -> List: ) clf.fit(x, y) similarities = clf.decision_function(x)[1:] - values, indexes = torch.topk(torch.Tensor(similarities), k=self.kwargs["top_k"], axis=-1) + values, indexes = torch.topk(torch.Tensor(similarities), k=min(self.kwargs["top_k"], len(similarities)), axis=-1) scores = [float(value) for value in values] ids = [int(index) for index in indexes] candidates_iris, candidates_scores = [], [] diff --git a/ontoaligner/encoder/llm.py b/ontoaligner/encoder/llm.py index 388e02d..8e20044 100644 --- a/ontoaligner/encoder/llm.py +++ b/ontoaligner/encoder/llm.py @@ -36,12 +36,12 @@ def parse(self, **kwargs) -> Any: source_onto, target_onto = kwargs["source"], kwargs["target"] source_ontos = [] for source in source_onto: - encoded_source = self.get_owl_items(owl=source) + encoded_source = self.get_owl_items(source) # encoded_source["concept"] = self.preprocess(encoded_source["text"]) source_ontos.append(encoded_source) target_ontos = [] for target in target_onto: - encoded_target = self.get_owl_items(owl=target) + encoded_target = self.get_owl_items(target) # encoded_target["concept"] = self.preprocess(encoded_target["text"]) target_ontos.append(encoded_target) return [source_ontos, target_ontos] diff --git a/ontoaligner/encoder/property.py b/ontoaligner/encoder/property.py index 32935a7..566a4f3 100644 --- a/ontoaligner/encoder/property.py +++ b/ontoaligner/encoder/property.py @@ -133,7 +133,7 @@ def get_owl_items(self, property) -> Dict: range_text = " ".join(property['range_text']) if len(property['range_text']) > 0 else "" inverse_text = "" if property['inverse_of']: - inverse_text = " ".join(property['inverse_label']) if len(property['inverse_label']) > 0 else "" + inverse_text = property.get("inverse_label") or "" combined_text = label if domain_text: @@ -261,11 +261,7 @@ def get_owl_items(self, prop: Dict) -> Any: inverse_text = "" if prop.get("inverse_of"): - inverse_text = ( - " ".join(prop.get("inverse_label", [])) - if len(prop.get("inverse_label", [])) > 0 - else "" - ) + inverse_text = prop.get("inverse_label") or "" combined_text = label @@ -279,10 +275,9 @@ def get_owl_items(self, prop: Dict) -> Any: combined_text += " inverse: " + inverse_text return { - "iri": prop["iri"], - "label": label, - "domain": domain_text, - "range": range_text, - "inverse": inverse_text, + **prop, + "domain_text": domain_text, + "range_text": range_text, + "inverse_text": inverse_text, "text": combined_text, - } \ No newline at end of file + } \ No newline at end of file diff --git a/ontoaligner/ontology/kg.py b/ontoaligner/ontology/kg.py index 981bb84..94e6c48 100644 --- a/ontoaligner/ontology/kg.py +++ b/ontoaligner/ontology/kg.py @@ -296,9 +296,10 @@ def is_attribute(self, pred): bool: ``True`` if the predicate has any literal objects. """ if pred in self.relindex: - for literal in self.relindex[pred]: - if is_literal(literal): - return True + for subject in self.relindex[pred]: + for literal in self.relindex[pred][subject]: + if is_literal(literal): + return True return False def local_functionality(self, subjects, preds): @@ -1025,7 +1026,7 @@ def load_ontology(self, input_file_path: str, attribute_path: str) -> Any: for line in file.readlines(): head, rel, tail = line.strip().split('\t') kg.add((head, rel, tail)) - if not attribute_path: + if attribute_path: with open(attribute_path, 'r', encoding='UTF-8') as attribute_file: for line in attribute_file.readlines(): head, attribute, literal = line.strip().split('\t') @@ -1185,7 +1186,7 @@ def load_ontology(self, id2rel[int(ids)] = rel with open(ent_ids_path, encoding='UTF-8') as ent_ids_file: - if (not translated_name_path) and ent_ids_path.endswith('_1'): + if translated_name_path and ent_ids_path.endswith('_1'): ent_name_trans = {} with open(translated_name_path, encoding='UTF-8') as f2: for line1, line2 in zip(ent_ids_file.readlines(), f2.readlines()): @@ -1219,7 +1220,7 @@ def load_ontology(self, head, rel, tail = int(head), int(rel), int(tail) kg.add((id2ent[head], id2rel[rel], id2ent[tail])) - if not att_triples_path: + if att_triples_path: for triple in parse_turtle_triples(att_triples_path): head, attribute, literal = triple literal = literal.replace('\n','\\n')\ @@ -1641,8 +1642,8 @@ def collect(self, target_kg = self.target_ontology.parse(target_kg_path, target_kg_attribute_path) return { "dataset-info": {"track": self.track, "ontology-name": self.ontology_name}, - "source": source_kg, - "target": target_kg, + "source": [source_kg], + "target": [target_kg], "reference": self.alignments.parse(input_file_path=reference_matching_path), } diff --git a/ontoaligner/utils/__init__.py b/ontoaligner/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ontoaligner/utils/metrics.py b/ontoaligner/utils/metrics.py index 54eedf9..bade5ea 100644 --- a/ontoaligner/utils/metrics.py +++ b/ontoaligner/utils/metrics.py @@ -78,7 +78,8 @@ def precision_score(predicts: List, references: List) -> float: - **`precision`** (*float*): The calculated precision score. """ intersection = calculate_intersection(predicts=predicts, references=references) - return intersection / len(predicts) if len(predicts) != 0 else 0 + predict_set = {(predict["source"], predict["target"]) for predict in predicts} + return intersection / len(predict_set) if len(predict_set) != 0 else 0 def recall_score(predicts: List, references: List) -> float: r""" @@ -97,7 +98,8 @@ def recall_score(predicts: List, references: List) -> float: - **`recall`** (*float*): The calculated recall score. """ intersection = calculate_intersection(predicts=predicts, references=references) - return intersection / len(references) if len(references) != 0 else 0 + reference_set = {(reference["source"], reference["target"]) for reference in references} + return intersection / len(reference_set) if len(reference_set) != 0 else 0 def f1_measurement(predicts: List, references: List, beta: int = 1) -> float: r""" @@ -137,14 +139,16 @@ def evaluation_report(predicts: List, references: List, beta: int = 1) -> Dict: precision = precision_score(predicts=predicts, references=references) recall = recall_score(predicts=predicts, references=references) f_score = f1_measurement(predicts=predicts, references=references, beta=beta) + predict_set = {(predict["source"], predict["target"]) for predict in predicts} + reference_set = {(reference["source"], reference["target"]) for reference in references} evaluations_dict = { "intersection": intersection, "precision": precision * 100, "recall": recall * 100, "f-score": f_score * 100, - "predictions-len": len(predicts), - "reference-len": len(references), + "predictions-len": len(predict_set), + "reference-len": len(reference_set), } return evaluations_dict diff --git a/ontoaligner/utils/sssom.py b/ontoaligner/utils/sssom.py new file mode 100644 index 0000000..3acd21e --- /dev/null +++ b/ontoaligner/utils/sssom.py @@ -0,0 +1,391 @@ +# Copyright 2025 Scientific Knowledge Organization (SciKnowOrg) Research Group. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from io import StringIO +from typing import Any, Dict, List, Optional +import pandas as pd +import bioregistry +from curies import Converter +import sssom as sss +from ontoaligner import __version__ + + +def _get_label_lookup(entities: Optional[List[Dict]]) -> Dict[str, str]: + """ + Create an IRI-to-label lookup from OntoAligner parser output. + + Parameters: + entities (Optional[List[Dict]]): Parsed ontology entities containing + 'iri' and optionally 'label'. + + Returns: + Dict[str, str]: Mapping from entity IRI to label. + """ + if entities is None: + return {} + + return { + entity["iri"]: str(entity["label"]) + for entity in entities + if "iri" in entity and "label" in entity and entity["label"] is not None + } + +def _get_converter( + curie_map: Optional[Dict[str, str]] = None, +) -> Converter: + """ + Get a CURIE converter for SSSOM serialization. + + If no CURIE map is provided, Bioregistry is used as a fallback for + registered namespaces. Custom or unregistered namespaces must be + supplied explicitly through ``curie_map``. + + Parameters: + curie_map (Optional[Dict[str, str]]): Prefix-to-IRI mappings. + + Returns: + Converter: CURIE converter used for SSSOM serialization. + """ + if curie_map is not None: + return Converter.from_prefix_map(curie_map) + + return bioregistry.get_default_converter() + +def _get_aligner_metadata( + matching: Dict, + aligner: Optional[Any] = None, + postprocessor: Optional[Any] = None, + postprocessor_params: Optional[Dict] = None, +) -> Dict: + """ + Get SSSOM metadata based on the aligner and postprocessor. + + Parameters: + matching (Dict): OntoAligner matching result. + + aligner (Optional[Any]): Aligner used to generate the matching. + + postprocessor (Optional[Any]): Postprocessor applied to the aligner + output. + + postprocessor_params (Optional[Dict]): Parameters supplied to the + postprocessor. + + Returns: + Dict: SSSOM mapping metadata. + """ + metadata = { + "mapping_justification": "semapv:UnspecifiedMatching", + } + + if aligner is None: + return metadata + + aligner_name = aligner.__class__.__name__ + postprocessor_name = ( + postprocessor.__name__ if postprocessor is not None else None + ) + postprocessor_params = postprocessor_params or {} + + score = matching.get("score") + has_retrieval_threshold = ( + postprocessor_name == "retriever_postprocessor" + and "threshold" in postprocessor_params + ) + + fuzzy_aligners = { + "SimpleFuzzySMLightweight": "RapidFuzz fuzz.ratio", + "WeightedFuzzySMLightweight": "RapidFuzz fuzz.WRatio", + "TokenSetFuzzySMLightweight": "RapidFuzz fuzz.token_set_ratio", + } + + graph_aligners = { + "ConvEAligner", + "TransDAligner", + "TransEAligner", + "TransFAligner", + "TransHAligner", + "TransRAligner", + "DistMultAligner", + "ComplExAligner", + "HolEAligner", + "RotatEAligner", + "SimplEAligner", + "CrossEAligner", + "BoxEAligner", + "CompGCNAligner", + "MuREAligner", + "QuatEAligner", + "SEAligner", + } + + if postprocessor_name in { + "rag_heuristic_postprocessor", + "rag_hybrid_postprocessor", + }: + metadata["mapping_justification"] = "semapv:CompositeMatching" + + elif aligner_name in fuzzy_aligners: + metadata["mapping_justification"] = ( + "semapv:LexicalSimilarityThresholdMatching" + ) + + elif aligner_name in {"TFIDFRetrieval", "BM25Retrieval"} and has_retrieval_threshold: + metadata["mapping_justification"] = ( + "semapv:LexicalSimilarityThresholdMatching" + ) + + elif aligner_name in {"SBERTRetrieval", "AdaRetrieval"} and has_retrieval_threshold: + metadata["mapping_justification"] = ( + "semapv:SemanticSimilarityThresholdMatching" + ) + + elif aligner_name in graph_aligners: + metadata["mapping_justification"] = "semapv:StructuralMatching" + + elif aligner_name == "PropMatchAligner": + disable_domain_range = getattr( + aligner, + "disable_domain_range", + getattr(aligner, "kwargs", {}).get("disable_domain_range", False), + ) + metadata["mapping_justification"] = ( + "semapv:LexicalSimilarityThresholdMatching" + if disable_domain_range + else "semapv:CompositeMatching" + ) + + elif aligner_name == "OLaLaHighPrecisionMatcher": + metadata["mapping_justification"] = "semapv:LexicalMatching" + + elif aligner_name in {"FLORAAligner", "EnsembleLearningAligner"}: + metadata["mapping_justification"] = "semapv:CompositeMatching" + + if score is not None: + score = float(score) + + if aligner_name in fuzzy_aligners: + metadata["similarity_score"] = score + metadata["similarity_measure"] = fuzzy_aligners[aligner_name] + + elif aligner_name == "TFIDFRetrieval": + metadata["similarity_score"] = score + metadata["similarity_measure"] = "TF-IDF cosine similarity" + + elif aligner_name == "SBERTRetrieval" and 0 <= score <= 1: + metadata["similarity_score"] = score + metadata["similarity_measure"] = ( + "cosine similarity over SentenceTransformer embeddings" + ) + + elif aligner_name == "AdaRetrieval" and 0 <= score <= 1: + metadata["similarity_score"] = score + metadata["similarity_measure"] = ( + "cosine similarity over OpenAI embeddings" + ) + + elif aligner_name in graph_aligners and 0 <= score <= 1: + metadata["similarity_score"] = score + metadata["similarity_measure"] = ( + "cosine similarity over graph embeddings" + ) + + elif ( + aligner_name + in { + "OLaLaHighPrecisionMatcher", + "OLaLaLLMAligner", + "OLaLaAligner", + } + and 0 <= score <= 1 + ): + metadata["confidence"] = score + + if ( + postprocessor_name == "rag_heuristic_postprocessor" + and matching.get("confidence") is not None + ): + confidence = float(matching["confidence"]) + if 0 <= confidence <= 1: + metadata["confidence"] = confidence + + if ( + postprocessor_name + in {"rag_heuristic_postprocessor", "rag_hybrid_postprocessor"} + and score is not None + and 0 <= score <= 1 + ): + if "BERTRetriever" in aligner_name: + metadata["similarity_score"] = score + metadata["similarity_measure"] = ( + "cosine similarity over SentenceTransformer embeddings" + ) + elif "AdaRetriever" in aligner_name: + metadata["similarity_score"] = score + metadata["similarity_measure"] = ( + "cosine similarity over OpenAI embeddings" + ) + + return metadata + + +def sssom_alignment_generator( + matchings: List[Dict], + source: Optional[List[Dict]] = None, + target: Optional[List[Dict]] = None, + *, + predicate_id: str, + mapping_set_metadata: Dict, + curie_map: Optional[Dict[str, str]] = None, + pipeline: Optional[Any] = None, + aligner: Optional[Any] = None, + postprocessor: Optional[Any] = None, + postprocessor_params: Optional[Dict] = None, + mapping_justification: Optional[str] = None, + include_aligner_metadata: bool = True, + ) -> str: + """ + Generate ontology alignments in SSSOM TSV format. + + Parameters: + matchings (List[Dict]): OntoAligner matching results containing + 'source', 'target', and optionally score information. + + source (Optional[List[Dict]]): Parsed source ontology entities. + Entities containing 'iri' and 'label' are used to populate + 'subject_label'. + + target (Optional[List[Dict]]): Parsed target ontology entities. + Entities containing 'iri' and 'label' are used to populate + 'object_label'. + + predicate_id (str): SSSOM mapping predicate. + + mapping_set_metadata (Dict): SSSOM MappingSet metadata. It must + contain 'mapping_set_id' and 'license'. + + curie_map(Optional[Dict[str, str]]): Prefix-to-IRI mappings used to convert + OntoAligner entity IRIs into SSSOM CURIEs. + + pipeline (Optional[Any]): AlignerPipeline used to generate the + mappings. + + aligner (Optional[Any]): Aligner used when mappings were generated + without AlignerPipeline. + + postprocessor (Optional[Any]): Postprocessor used when mappings were + generated without AlignerPipeline. + + postprocessor_params (Optional[Dict]): Parameters supplied to the + postprocessor. + + mapping_justification (Optional[str]): Explicit SSSOM mapping + justification. If not supplied, it is inferred from the aligner + and postprocessor. + + Returns: + str: Ontology alignments serialized as SSSOM TSV. + """ + if "mapping_set_id" not in mapping_set_metadata: + raise ValueError("'mapping_set_id' is required in mapping_set_metadata.") + + if "license" not in mapping_set_metadata: + raise ValueError("'license' is required in mapping_set_metadata.") + + if pipeline is not None: + aligner = pipeline.reranker or pipeline.aligner + postprocessor = pipeline.postprocessor + postprocessor_params = pipeline.postprocessor_params + + if source is None and pipeline.om_dataset is not None: + source = pipeline.om_dataset.get("source") + + if target is None and pipeline.om_dataset is not None: + target = pipeline.om_dataset.get("target") + + converter = _get_converter(curie_map=curie_map) + + source_labels = _get_label_lookup(source) + target_labels = _get_label_lookup(target) + + rows = [] + + for matching in matchings: + source_iri = matching["source"] + target_iri = matching["target"] + + if include_aligner_metadata: + aligner_metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + postprocessor=postprocessor, + postprocessor_params=postprocessor_params, + ) + justification = ( + mapping_justification + or aligner_metadata.get("mapping_justification", "semapv:UnspecifiedMatching") + ) + else: + aligner_metadata = {} + justification = mapping_justification or "semapv:UnspecifiedMatching" + + row = { + "subject_id": sss.util.safe_compress(source_iri, converter), + "predicate_id": sss.util.safe_compress(predicate_id, converter), + "object_id": sss.util.safe_compress(target_iri, converter), + "mapping_justification": sss.util.safe_compress( + justification, + converter, + ), + } + + if source_iri in source_labels: + row["subject_label"] = source_labels[source_iri] + + if target_iri in target_labels: + row["object_label"] = target_labels[target_iri] + + if include_aligner_metadata and "similarity_score" in aligner_metadata: + row["similarity_score"] = aligner_metadata["similarity_score"] + row["similarity_measure"] = aligner_metadata[ + "similarity_measure" + ] + + if include_aligner_metadata and "confidence" in aligner_metadata: + row["confidence"] = aligner_metadata["confidence"] + + rows.append(row) + + metadata = dict(mapping_set_metadata) + metadata.setdefault("mapping_tool", "OntoAligner") + metadata.setdefault("mapping_tool_version", __version__) + + if pipeline is not None and pipeline.om_dataset is not None: + dataset_info = pipeline.om_dataset.get("dataset-info", {}) + ontology_name = dataset_info.get("ontology-name") + + if ontology_name is not None: + metadata.setdefault("mapping_set_title", ontology_name) + + sssom_df = pd.DataFrame(rows) + + sssom_mapping_set = sss.util.MappingSetDataFrame( + df=sssom_df, + converter=converter, + metadata=metadata, + ) + + sssom_output = StringIO() + sss.writers.write_tsv(sssom_mapping_set, sssom_output) + + return sssom_output.getvalue() diff --git a/ontoaligner/utils/sssom_pydantic.py b/ontoaligner/utils/sssom_pydantic.py new file mode 100644 index 0000000..0b0abcc --- /dev/null +++ b/ontoaligner/utils/sssom_pydantic.py @@ -0,0 +1,125 @@ +# Copyright 2025 Scientific Knowledge Organization (SciKnowOrg) Research Group. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import datetime +import logging +from pathlib import Path +from typing import Any, Iterable, List, Mapping, Optional, TextIO +import bioregistry +import curies +import sssom_pydantic as spd + +logger = logging.getLogger(__name__) + +# Mapping of simple relation tokens to SSSOM predicate CURIEs +PREDICATE_LOOKUPS = {"=": curies.vocabulary.exact_match} + +def sssom_alignment_generator( + matchings: Iterable[Mapping[str, Any]], + path: str | Path | TextIO, + *, + converter: Optional[curies.Converter] = None, + metadata: Optional[spd.MappingSet] = None, + **kwargs: Any, +) -> None: + """Convert matchings to SSSOM objects and write a TSV to disk. + + Parameters + - matchings: iterable of dict-like matchings (must contain at least + 'source' and 'target'). + - path: file path or open file-like object to write to. + - converter: optional curies.Converter to parse URIs into CURIEs. + - metadata: optional SSSOM sssom_pydantic.MappingSet metadata object. + - kwargs: forwarded to sssom_pydantic.write. + """ + if converter is None: + converter = bioregistry.get_default_converter() + + mappings = to_semantic_mappings(matchings, converter) + try: + spd.write(mappings, path, converter=converter, metadata=metadata, **kwargs) + except Exception as exc: + logger.exception("Failed to write SSSOM output: %s", exc) + raise + + +def to_semantic_mappings( + matchings: Iterable[Mapping[str, Any]], + converter: Optional[curies.Converter] = None +) -> List[spd.SemanticMapping]: + """Convert an iterable of matchings to a list of SSSOM sssom_pydantic.SemanticMapping. + + Invalid or unparsable matchings are skipped but logged at debug level. + """ + if converter is None: + converter = bioregistry.get_default_converter() + + mappings: List[spd.SemanticMapping] = [] + for idx, matching in enumerate(matchings): + try: + sm = convert_matching(matching, converter) + except Exception: # be explicit about skipping only problematic entries + logger.debug("Skipping mapping at index %d due to conversion error", idx, exc_info=True) + continue + if sm is not None: + mappings.append(sm) + return mappings + + +def convert_matching(matching: Mapping[str, Any], converter: curies.Converter) -> Optional[spd.SemanticMapping]: + """Convert a single OntoAligner matching dict to a sssom_pydantic.SemanticMapping. + + Expected keys in ``matching``: + - 'source': source URI/IRI + - 'target': target URI/IRI + - optional 'score': confidence value between 0 and 1 + - optional 'relation': one of the keys in PREDICATE_LOOKUPS (defaults to "=") + + Returns None if the subject or object cannot be parsed by the converter. + """ + # Required fields + source = matching.get("source") + target = matching.get("target") + if source is None or target is None: + logger.debug("Matching missing source/target: %r", matching) + return None + + subject = converter.parse_uri(source) + object_ = converter.parse_uri(target) + if subject is None or object_ is None: + logger.debug("Could not parse source/target into CURIEs: %r -> %r", source, target) + return None + + # Confidence handling + confidence: Optional[float] = None + if "score" in matching and matching.get("score") is not None: + try: + confidence = float(matching.get("score")) + except (TypeError, ValueError): + logger.debug("Invalid confidence value: %r", matching.get("score")) + confidence = None + + relation = str(matching.get("relation", "=")) + predicate = PREDICATE_LOOKUPS.get(relation) + if predicate is None: + logger.debug("Unknown relation token %r, falling back to unspecified_matching_process", relation) + predicate = curies.vocabulary.unspecified_matching_process + + return spd.SemanticMapping( + subject=subject, + predicate=predicate, + object=object_, + justification= curies.vocabulary.unspecified_matching_process, + confidence=confidence, + mapping_date=datetime.date.today(), # noqa: DTZ011 + ) diff --git a/pyproject.toml b/pyproject.toml index 81f841f..5473282 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,10 @@ pykeen = "^1.11.1" multiprocess = "*" scipy = "^1.15.3" cohere = "*" +sssom = "^0.4.21" +sssom-pydantic = "*" +bioregistry = "*" +curies = "*" [tool.poetry.dev-dependencies] ruff = "*" diff --git a/requirements.txt b/requirements.txt index 96b34c7..9625d9b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -26,3 +26,6 @@ wheel twine pytest cohere +sssom-pydantic +bioregistry +curies diff --git a/setup.py b/setup.py index 598d063..09d6412 100644 --- a/setup.py +++ b/setup.py @@ -37,6 +37,9 @@ "scipy==1.15.3", "multiprocess", "cohere", + "sssom-pydantic", + "bioregistry", + "curies", ], classifiers=[ "Development Status :: 5 - Production/Stable", diff --git a/tests/data/ensemble_alignments.json b/tests/data/ensemble_alignments.json new file mode 100644 index 0000000..2260149 --- /dev/null +++ b/tests/data/ensemble_alignments.json @@ -0,0 +1,57 @@ +[ + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_4", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_1", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_44", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_15", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_53", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_16", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_6", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_3", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_7", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_4", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_82", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_86", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_9", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_24", + "score": 5.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_49", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_20", + "score": 4.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_51", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_19", + "score": 3.0 + }, + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_52", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_18", + "score": 3.0 + } +] \ No newline at end of file diff --git a/tests/utils/test_sssom.py b/tests/utils/test_sssom.py new file mode 100644 index 0000000..b6b7938 --- /dev/null +++ b/tests/utils/test_sssom.py @@ -0,0 +1,722 @@ +"""Tests for SSSOM alignment generator.""" + +import unittest +from unittest.mock import Mock +from ontoaligner.utils.sssom import ( + sssom_alignment_generator, + _get_label_lookup, + _get_converter, + _get_aligner_metadata, +) + + +class TestGetLabelLookup(unittest.TestCase): + """Test cases for _get_label_lookup function.""" + + def test_empty_entities_list(self): + """Test with empty entities list.""" + result = _get_label_lookup([]) + self.assertEqual(result, {}) + + def test_none_entities(self): + """Test with None entities.""" + result = _get_label_lookup(None) + self.assertEqual(result, {}) + + def test_single_entity_with_label(self): + """Test with a single entity containing IRI and label.""" + entities = [{"iri": "http://example.org/entity1", "label": "Entity 1"}] + result = _get_label_lookup(entities) + self.assertEqual( + result, + {"http://example.org/entity1": "Entity 1"} + ) + + def test_multiple_entities_with_labels(self): + """Test with multiple entities containing IRIs and labels.""" + entities = [ + {"iri": "http://example.org/entity1", "label": "Entity 1"}, + {"iri": "http://example.org/entity2", "label": "Entity 2"}, + {"iri": "http://example.org/entity3", "label": "Entity 3"}, + ] + result = _get_label_lookup(entities) + self.assertEqual(len(result), 3) + self.assertEqual(result["http://example.org/entity1"], "Entity 1") + self.assertEqual(result["http://example.org/entity2"], "Entity 2") + self.assertEqual(result["http://example.org/entity3"], "Entity 3") + + def test_entity_without_iri(self): + """Test with entity missing IRI field.""" + entities = [ + {"label": "Entity 1"}, # No IRI + {"iri": "http://example.org/entity2", "label": "Entity 2"}, + ] + result = _get_label_lookup(entities) + self.assertEqual(len(result), 1) + self.assertEqual(result["http://example.org/entity2"], "Entity 2") + + def test_entity_without_label(self): + """Test with entity missing label field.""" + entities = [ + {"iri": "http://example.org/entity1"}, # No label + {"iri": "http://example.org/entity2", "label": "Entity 2"}, + ] + result = _get_label_lookup(entities) + self.assertEqual(len(result), 1) + self.assertEqual(result["http://example.org/entity2"], "Entity 2") + + def test_entity_with_none_label(self): + """Test with entity having None as label.""" + entities = [ + {"iri": "http://example.org/entity1", "label": None}, + {"iri": "http://example.org/entity2", "label": "Entity 2"}, + ] + result = _get_label_lookup(entities) + self.assertEqual(len(result), 1) + self.assertEqual(result["http://example.org/entity2"], "Entity 2") + + def test_label_conversion_to_string(self): + """Test that labels are converted to strings.""" + entities = [{"iri": "http://example.org/entity1", "label": 123}] + result = _get_label_lookup(entities) + self.assertEqual(result["http://example.org/entity1"], "123") + self.assertIsInstance(result["http://example.org/entity1"], str) + + +class TestGetConverter(unittest.TestCase): + """Test cases for _get_converter function.""" + + def test_converter_with_custom_curie_map(self): + """Test converter creation with custom CURIE map.""" + curie_map = { + "ex": "http://example.org/", + "test": "http://test.org/", + } + converter = _get_converter(curie_map=curie_map) + self.assertIsNotNone(converter) + + def test_converter_without_curie_map(self): + """Test converter creation without custom CURIE map (uses bioregistry).""" + converter = _get_converter(curie_map=None) + self.assertIsNotNone(converter) + + def test_converter_empty_curie_map(self): + """Test converter creation with empty CURIE map.""" + converter = _get_converter(curie_map={}) + self.assertIsNotNone(converter) + + +class TestGetAlignerMetadata(unittest.TestCase): + """Test cases for _get_aligner_metadata function.""" + + def test_no_aligner(self): + """Test with no aligner provided.""" + matching = {"score": 0.8} + metadata = _get_aligner_metadata(matching=matching) + self.assertEqual(metadata["mapping_justification"], "semapv:UnspecifiedMatching") + + def test_fuzzy_aligner_simple(self): + """Test with SimpleFuzzySMLightweight aligner.""" + aligner = Mock() + aligner.__class__.__name__ = "SimpleFuzzySMLightweight" + matching = {"score": 0.9} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual( + metadata["mapping_justification"], + "semapv:LexicalSimilarityThresholdMatching" + ) + self.assertEqual(metadata["similarity_score"], 0.9) + self.assertEqual(metadata["similarity_measure"], "RapidFuzz fuzz.ratio") + + def test_fuzzy_aligner_weighted(self): + """Test with WeightedFuzzySMLightweight aligner.""" + aligner = Mock() + aligner.__class__.__name__ = "WeightedFuzzySMLightweight" + matching = {"score": 0.85} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual( + metadata["mapping_justification"], + "semapv:LexicalSimilarityThresholdMatching" + ) + self.assertEqual(metadata["similarity_measure"], "RapidFuzz fuzz.WRatio") + + def test_graph_aligner(self): + """Test with graph aligner.""" + aligner = Mock() + aligner.__class__.__name__ = "TransEAligner" + matching = {"score": 0.75} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["mapping_justification"], "semapv:StructuralMatching") + self.assertEqual(metadata["similarity_score"], 0.75) + self.assertEqual(metadata["similarity_measure"], "cosine similarity over graph embeddings") + + def test_sbert_retrieval_with_score(self): + """Test with SBERTRetrieval aligner and valid score.""" + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + matching = {"score": 0.8} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["similarity_score"], 0.8) + self.assertIn("SentenceTransformer", metadata["similarity_measure"]) + + def test_sbert_retrieval_with_threshold(self): + """Test with SBERTRetrieval and retriever_postprocessor threshold.""" + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + postprocessor = Mock() + postprocessor.__name__ = "retriever_postprocessor" + matching = {"score": 0.85} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + postprocessor=postprocessor, + postprocessor_params={"threshold": 0.5}, + ) + self.assertEqual( + metadata["mapping_justification"], + "semapv:SemanticSimilarityThresholdMatching" + ) + + def test_tfidf_retrieval(self): + """Test with TFIDFRetrieval aligner.""" + aligner = Mock() + aligner.__class__.__name__ = "TFIDFRetrieval" + matching = {"score": 0.7} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["similarity_score"], 0.7) + self.assertEqual(metadata["similarity_measure"], "TF-IDF cosine similarity") + + def test_rag_heuristic_postprocessor(self): + """Test with RAG heuristic postprocessor.""" + aligner = Mock() + aligner.__class__.__name__ = "SomeAligner" + postprocessor = Mock() + postprocessor.__name__ = "rag_heuristic_postprocessor" + matching = {"score": 0.8, "confidence": 0.9} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + postprocessor=postprocessor, + postprocessor_params={}, + ) + self.assertEqual(metadata["mapping_justification"], "semapv:CompositeMatching") + self.assertEqual(metadata["confidence"], 0.9) + + def test_prop_match_aligner_without_domain_range(self): + """Test PropMatchAligner without domain_range disabled.""" + aligner = Mock() + aligner.__class__.__name__ = "PropMatchAligner" + aligner.disable_domain_range = False + matching = {"score": 0.8} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["mapping_justification"], "semapv:CompositeMatching") + + def test_prop_match_aligner_with_domain_range_disabled(self): + """Test PropMatchAligner with domain_range disabled.""" + aligner = Mock() + aligner.__class__.__name__ = "PropMatchAligner" + aligner.disable_domain_range = True + matching = {"score": 0.8} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual( + metadata["mapping_justification"], + "semapv:LexicalSimilarityThresholdMatching" + ) + + def test_olala_aligner(self): + """Test with OLaLaHighPrecisionMatcher aligner.""" + aligner = Mock() + aligner.__class__.__name__ = "OLaLaHighPrecisionMatcher" + matching = {"score": 0.95} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["mapping_justification"], "semapv:LexicalMatching") + self.assertEqual(metadata["confidence"], 0.95) + + def test_ensemble_aligner(self): + """Test with EnsembleLearningAligner.""" + aligner = Mock() + aligner.__class__.__name__ = "EnsembleLearningAligner" + matching = {"score": 0.85} + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + self.assertEqual(metadata["mapping_justification"], "semapv:CompositeMatching") + + def test_score_out_of_range(self): + """Test with score outside valid range.""" + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + matching = {"score": 2.5} # Out of range + + metadata = _get_aligner_metadata( + matching=matching, + aligner=aligner, + ) + # Should not add similarity_score if out of range + self.assertNotIn("similarity_score", metadata) + + +class TestSSSOMAlignmentGenerator(unittest.TestCase): + """Test cases for sssom_alignment_generator function.""" + + def setUp(self): + """Set up test fixtures.""" + self.base_metadata = { + "mapping_set_id": "http://example.org/mapping1", + "license": "CC-BY-4.0", + } + self.simple_matchings = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + } + ] + self.predicate_id = "http://www.w3.org/2004/02/skos/core#exactMatch" + self.curie_map = { + "ex": "http://example.org/", + "skos": "http://www.w3.org/2004/02/skos/core#", + "semapv": "https://w3id.org/semapv/", + } + + def test_basic_alignment_generation(self): + """Test basic SSSOM alignment generation.""" + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIsInstance(result, str) + self.assertIn("subject_id", result) + self.assertIn("predicate_id", result) + self.assertIn("object_id", result) + + def test_missing_mapping_set_id(self): + """Test error handling for missing mapping_set_id.""" + incomplete_metadata = {"license": "CC-BY-4.0"} + with self.assertRaises(ValueError) as context: + sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=incomplete_metadata, + curie_map=self.curie_map, + ) + self.assertIn("mapping_set_id", str(context.exception)) + + def test_missing_license(self): + """Test error handling for missing license.""" + incomplete_metadata = {"mapping_set_id": "http://example.org/mapping1"} + with self.assertRaises(ValueError) as context: + sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=incomplete_metadata, + curie_map=self.curie_map, + ) + self.assertIn("license", str(context.exception)) + + def test_alignment_with_source_labels(self): + """Test alignment generation with source labels.""" + source_entities = [ + { + "iri": "http://example.org/source1", + "label": "Source Entity 1" + } + ] + result = sssom_alignment_generator( + matchings=self.simple_matchings, + source=source_entities, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIn("subject_label", result) + self.assertIn("Source Entity 1", result) + + def test_alignment_with_target_labels(self): + """Test alignment generation with target labels.""" + target_entities = [ + { + "iri": "http://example.org/target1", + "label": "Target Entity 1" + } + ] + result = sssom_alignment_generator( + matchings=self.simple_matchings, + target=target_entities, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIn("object_label", result) + self.assertIn("Target Entity 1", result) + + def test_alignment_with_both_labels(self): + """Test alignment generation with both source and target labels.""" + source_entities = [ + {"iri": "http://example.org/source1", "label": "Source Entity 1"} + ] + target_entities = [ + {"iri": "http://example.org/target1", "label": "Target Entity 1"} + ] + result = sssom_alignment_generator( + matchings=self.simple_matchings, + source=source_entities, + target=target_entities, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIn("subject_label", result) + self.assertIn("object_label", result) + + def test_alignment_with_score(self): + """Test alignment generation with similarity score.""" + matchings_with_score = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.95, + } + ] + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + + result = sssom_alignment_generator( + matchings=matchings_with_score, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + curie_map=self.curie_map, + ) + self.assertIn("similarity_score", result) + + def test_alignment_with_multiple_matchings(self): + """Test alignment generation with multiple matchings.""" + multiple_matchings = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + }, + { + "source": "http://example.org/source2", + "target": "http://example.org/target2", + }, + { + "source": "http://example.org/source3", + "target": "http://example.org/target3", + }, + ] + result = sssom_alignment_generator( + matchings=multiple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + # Check that all sources and targets are in the output + for matching in multiple_matchings: + self.assertIn("source1", result) + self.assertIn("source2", result) + self.assertIn("source3", result) + + def test_custom_curie_map(self): + """Test alignment generation with custom CURIE map.""" + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIsInstance(result, str) + + def test_explicit_mapping_justification(self): + """Test alignment generation with explicit mapping justification.""" + explicit_justification = "semapv:LexicalMatching" + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + mapping_justification=explicit_justification, + curie_map=self.curie_map, + ) + self.assertIn("semapv:LexicalMatching", result) + + def test_include_aligner_metadata_false(self): + """Test alignment generation with include_aligner_metadata=False.""" + matchings_with_score = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.95, + } + ] + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + + result = sssom_alignment_generator( + matchings=matchings_with_score, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + include_aligner_metadata=False, + curie_map=self.curie_map, + ) + # Should not include similarity_score when metadata inclusion is disabled + lines = result.split('\n') + # Check the data lines (not header) + data_lines = [line for line in lines if not line.startswith('#')] + for line in data_lines: + if line.strip(): + # Even with aligner and score, metadata shouldn't be included + parts = line.split('\t') + # The structure should still be valid + self.assertGreater(len(parts), 0) + + def test_metadata_defaults(self): + """Test that metadata defaults are applied.""" + extended_metadata = dict(self.base_metadata) + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=extended_metadata, + curie_map=self.curie_map, + ) + # Should have OntoAligner as mapping tool + self.assertIn("mapping_tool", result) + self.assertIn("OntoAligner", result) + + def test_with_pipeline_parameter(self): + """Test alignment generation with pipeline parameter.""" + pipeline = Mock() + pipeline.reranker = None + pipeline.aligner = Mock() + pipeline.aligner.__class__.__name__ = "SBERTRetrieval" + pipeline.postprocessor = None + pipeline.postprocessor_params = {} + pipeline.om_dataset = None + + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + pipeline=pipeline, + curie_map=self.curie_map, + ) + self.assertIsInstance(result, str) + + def test_with_pipeline_and_dataset_info(self): + """Test alignment generation with pipeline containing dataset info.""" + pipeline = Mock() + pipeline.reranker = None + pipeline.aligner = Mock() + pipeline.aligner.__class__.__name__ = "SBERTRetrieval" + pipeline.postprocessor = None + pipeline.postprocessor_params = {} + pipeline.om_dataset = { + "source": [{"iri": "http://example.org/source1", "label": "Source"}], + "target": [{"iri": "http://example.org/target1", "label": "Target"}], + "dataset-info": {"ontology-name": "Test Ontology"}, + } + + result = sssom_alignment_generator( + matchings=self.simple_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + pipeline=pipeline, + curie_map=self.curie_map, + ) + self.assertIn("Test Ontology", result) + + def test_empty_matchings(self): + """Test alignment generation with empty matchings list.""" + result = sssom_alignment_generator( + matchings=[], + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + self.assertIsInstance(result, str) + # Should still have header + self.assertIn("#", result) + + def test_matching_without_labels_for_entities(self): + """Test alignment when entity labels are not found.""" + source_entities = [ + {"iri": "http://example.org/other", "label": "Other Entity"} + ] + target_entities = [ + {"iri": "http://example.org/other_target", "label": "Other Target"} + ] + result = sssom_alignment_generator( + matchings=self.simple_matchings, + source=source_entities, + target=target_entities, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + curie_map=self.curie_map, + ) + # Should still work, just without the labels for these specific IRIs + self.assertIsInstance(result, str) + + def test_large_matching_set(self): + """Test alignment generation with large number of matchings.""" + large_matchings = [ + { + "source": f"http://example.org/source{i}", + "target": f"http://example.org/target{i}", + "score": 0.5 + (i * 0.01), + } + for i in range(100) + ] + aligner = Mock() + aligner.__class__.__name__ = "SBERTRetrieval" + + result = sssom_alignment_generator( + matchings=large_matchings, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + curie_map=self.curie_map, + ) + # Check that output contains data for all matchings + for i in range(10): # Check a subset + self.assertIn(f"source{i}", result) + + def test_fuzzy_aligner_integration(self): + """Test with fuzzy aligner.""" + matchings_with_score = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.85, + } + ] + aligner = Mock() + aligner.__class__.__name__ = "SimpleFuzzySMLightweight" + + result = sssom_alignment_generator( + matchings=matchings_with_score, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + curie_map=self.curie_map, + ) + self.assertIn("similarity_measure", result) + self.assertIn("fuzz.ratio", result) + + def test_graph_aligner_integration(self): + """Test with graph aligner.""" + matchings_with_score = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.75, + } + ] + aligner = Mock() + aligner.__class__.__name__ = "TransEAligner" + + result = sssom_alignment_generator( + matchings=matchings_with_score, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + curie_map=self.curie_map, + ) + self.assertIn("graph embeddings", result) + + def test_confidence_score_included(self): + """Test that confidence score is included when available.""" + matchings_with_confidence = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.9, # OLaLa aligners use score -> confidence + } + ] + aligner = Mock() + aligner.__class__.__name__ = "OLaLaHighPrecisionMatcher" + + result = sssom_alignment_generator( + matchings=matchings_with_confidence, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=aligner, + curie_map=self.curie_map, + ) + self.assertIn("confidence", result) + + def test_pipeline_overrides_aligner_params(self): + """Test that pipeline parameters override standalone aligner/postprocessor.""" + standalone_aligner = Mock() + standalone_aligner.__class__.__name__ = "SimpleFuzzySMLightweight" + + pipeline = Mock() + pipeline_aligner = Mock() + pipeline_aligner.__class__.__name__ = "SBERTRetrieval" + pipeline.reranker = None + pipeline.aligner = pipeline_aligner + postprocessor = Mock() + postprocessor.__name__ = "rag_heuristic_postprocessor" + pipeline.postprocessor = postprocessor + pipeline.postprocessor_params = {"threshold": 0.5} + pipeline.om_dataset = None + + matchings_with_score = [ + { + "source": "http://example.org/source1", + "target": "http://example.org/target1", + "score": 0.8, + } + ] + + result = sssom_alignment_generator( + matchings=matchings_with_score, + predicate_id=self.predicate_id, + mapping_set_metadata=self.base_metadata, + aligner=standalone_aligner, # Should be overridden + pipeline=pipeline, # Should take precedence + curie_map=self.curie_map, + ) + # Pipeline aligner should be used, resulting in SentenceTransformer reference + self.assertIn("SentenceTransformer", result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/utils/test_sssom_paydantic.py b/tests/utils/test_sssom_paydantic.py new file mode 100644 index 0000000..4d0c154 --- /dev/null +++ b/tests/utils/test_sssom_paydantic.py @@ -0,0 +1,375 @@ +"""Tests for SSSOM export using sssom_pydantic.""" + +import unittest +from pathlib import Path +import json +from io import StringIO +from unittest.mock import Mock, patch + +from ontoaligner.utils.sssom_pydantic import ( + to_semantic_mappings, + convert_matching, + sssom_alignment_generator, + PREDICATE_LOOKUPS, +) +from curies import Converter +import curies + +HERE = Path(__file__).parent.resolve() +TEST_DATA = HERE.parent.joinpath("data", "ensemble_alignments.json") + + +class TestPredicateLookups(unittest.TestCase): + """Test cases for PREDICATE_LOOKUPS dictionary.""" + + def test_predicate_lookups_contains_exact_match(self): + """Test that PREDICATE_LOOKUPS contains '='.""" + self.assertIn("=", PREDICATE_LOOKUPS) + + def test_exact_match_value(self): + """Test that '=' maps to exact_match.""" + self.assertEqual( + PREDICATE_LOOKUPS["="], + curies.vocabulary.exact_match + ) + + def test_predicate_lookups_not_empty(self): + """Test that PREDICATE_LOOKUPS is not empty.""" + self.assertGreater(len(PREDICATE_LOOKUPS), 0) + + +class TestConvertMatchingEdgeCases(unittest.TestCase): + """Test cases for convert_matching function edge cases.""" + + def setUp(self): + """Set up test fixtures.""" + self.converter = Converter.from_prefix_map( + { + "fishtraits": "https://kos.lifewatch.eu/thesauri/fishtraits/", + "zooplanktontraits": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/", + } + ) + + def test_matching_missing_source_returns_none(self): + """Test that matching without source returns None.""" + matching = { + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + def test_matching_missing_target_returns_none(self): + """Test that matching without target returns None.""" + matching = { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + def test_matching_with_none_source_returns_none(self): + """Test that matching with None source returns None.""" + matching = { + "source": None, + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + def test_matching_with_none_target_returns_none(self): + """Test that matching with None target returns None.""" + matching = { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": None, + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + def test_matching_with_unparseable_source_returns_none(self): + """Test that unparseable source returns None.""" + matching = { + "source": "http://unparseable.example.org/unknown/entity", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + def test_matching_with_unparseable_target_returns_none(self): + """Test that unparseable target returns None.""" + matching = { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": "http://unparseable.example.org/unknown/entity", + } + result = convert_matching(matching, self.converter) + self.assertIsNone(result) + + +class TestToSemanticMappings(unittest.TestCase): + """Test cases for to_semantic_mappings function.""" + + def setUp(self): + """Set up test fixtures.""" + self.converter = Converter.from_prefix_map( + { + "fishtraits": "https://kos.lifewatch.eu/thesauri/fishtraits/", + "zooplanktontraits": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/", + } + ) + + def test_empty_matchings_returns_empty_list(self): + """Test with empty matchings list.""" + result = to_semantic_mappings([], converter=self.converter) + self.assertEqual(result, []) + self.assertIsInstance(result, list) + + def test_returns_list(self): + """Test that to_semantic_mappings returns a list.""" + result = to_semantic_mappings([], converter=self.converter) + self.assertIsInstance(result, list) + + def test_skips_matching_with_none_source(self): + """Test that matchings with None source are skipped.""" + matchings = [ + { + "source": None, + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + ] + result = to_semantic_mappings(matchings, converter=self.converter) + self.assertEqual(len(result), 0) + + def test_skips_matching_with_none_target(self): + """Test that matchings with None target are skipped.""" + matchings = [ + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": None, + } + ] + result = to_semantic_mappings(matchings, converter=self.converter) + self.assertEqual(len(result), 0) + + def test_skips_unparseable_matchings(self): + """Test that unparseable matchings are skipped.""" + matchings = [ + { + "source": "http://unknown.org/entity1", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + ] + result = to_semantic_mappings(matchings, converter=self.converter) + self.assertEqual(len(result), 0) + + def test_default_converter_when_none(self): + """Test that default converter is used when None is passed.""" + # Use GOinically known URIs + matchings = [ + { + "source": "http://purl.obolibrary.org/obo/GO_0008150", + "target": "http://purl.obolibrary.org/obo/GO_0008151", + } + ] + result = to_semantic_mappings(matchings, converter=None) + self.assertIsInstance(result, list) + + def test_generator_input_is_processed(self): + """Test that generator input is properly iterated.""" + def matching_generator(): + yield { + "source": None, + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + yield { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": None, + } + + result = to_semantic_mappings(matching_generator(), converter=self.converter) + + # Both should be skipped + self.assertEqual(len(result), 0) + + def test_processes_multiple_matchings(self): + """Test processing multiple matchings.""" + matchings = [ + {"source": None, "target": "target1"}, # Skip + {"source": "source2", "target": None}, # Skip + {"source": None, "target": None}, # Skip + ] + result = to_semantic_mappings(matchings, converter=self.converter) + self.assertEqual(len(result), 0) + + +class TestSSSOMAlignmentGeneratorBasic(unittest.TestCase): + """Test cases for sssom_alignment_generator function.""" + + def setUp(self): + """Set up test fixtures.""" + self.converter = Converter.from_prefix_map( + { + "fishtraits": "https://kos.lifewatch.eu/thesauri/fishtraits/", + "zooplanktontraits": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/", + } + ) + + @patch('ontoaligner.utils.sssom_pydantic.spd.write') + def test_write_is_called(self, mock_write): + """Test that spd.write is called.""" + output = StringIO() + matchings = [ + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + ] + + sssom_alignment_generator( + matchings, + output, + converter=self.converter, + ) + + mock_write.assert_called_once() + + @patch('ontoaligner.utils.sssom_pydantic.to_semantic_mappings') + @patch('ontoaligner.utils.sssom_pydantic.spd.write') + def test_to_semantic_mappings_called(self, mock_write, mock_to_mappings): + """Test that to_semantic_mappings is called.""" + mock_to_mappings.return_value = [] + + output = StringIO() + matchings = [ + { + "source": "https://kos.lifewatch.eu/thesauri/fishtraits/c_10", + "target": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13", + } + ] + + sssom_alignment_generator( + matchings, + output, + converter=self.converter, + ) + + mock_to_mappings.assert_called_once() + + @patch('ontoaligner.utils.sssom_pydantic.bioregistry.get_default_converter') + def test_default_converter_used_when_none(self, mock_default_converter): + """Test that default converter is used when None is passed.""" + mock_converter = Mock() + mock_default_converter.return_value = mock_converter + + with patch('ontoaligner.utils.sssom_pydantic.to_semantic_mappings') as mock_to_mappings: + with patch('ontoaligner.utils.sssom_pydantic.spd.write'): + mock_to_mappings.return_value = [] + + output = StringIO() + matchings = [] + + sssom_alignment_generator( + matchings, + output, + converter=None, + ) + + mock_default_converter.assert_called_once() + + +class TestTestDataIntegrity(unittest.TestCase): + """Test the integrity and structure of test data.""" + + def test_test_data_file_exists(self): + """Test that test data file exists.""" + self.assertTrue(TEST_DATA.exists()) + + def test_test_data_is_valid_json(self): + """Test that test data is valid JSON.""" + with TEST_DATA.open() as file: + matchings = json.load(file) + + self.assertIsInstance(matchings, list) + + def test_test_data_has_content(self): + """Test that test data has content.""" + with TEST_DATA.open() as file: + matchings = json.load(file) + + self.assertGreater(len(matchings), 0) + + def test_test_data_structure(self): + """Test structure of test data entries.""" + with TEST_DATA.open() as file: + matchings = json.load(file) + + for matching in matchings: + self.assertIn("source", matching) + self.assertIn("target", matching) + self.assertIsInstance(matching["source"], str) + self.assertIsInstance(matching["target"], str) + + def test_test_data_has_scores(self): + """Test that test data contains score values.""" + with TEST_DATA.open() as file: + matchings = json.load(file) + + matchings_with_score = [m for m in matchings if "score" in m] + self.assertGreater(len(matchings_with_score), 0) + + +class TestConverterParsing(unittest.TestCase): + """Test CURIE converter parsing of test data.""" + + def setUp(self): + """Set up test fixtures.""" + self.converter = Converter.from_prefix_map( + { + "fishtraits": "https://kos.lifewatch.eu/thesauri/fishtraits/", + "zooplanktontraits": "https://kos.lifewatch.eu/thesauri/zooplanktontraits/", + } + ) + + def test_converter_parses_fishtraits_uri(self): + """Test that converter can parse fishtraits URIs.""" + uri = "https://kos.lifewatch.eu/thesauri/fishtraits/c_10" + result = self.converter.parse_uri(uri) + + self.assertIsNotNone(result) + self.assertEqual(result.prefix, "fishtraits") + self.assertEqual(result.identifier, "c_10") + + def test_converter_parses_zooplanktontraits_uri(self): + """Test that converter can parse zooplanktontraits URIs.""" + uri = "https://kos.lifewatch.eu/thesauri/zooplanktontraits/c_13" + result = self.converter.parse_uri(uri) + + self.assertIsNotNone(result) + self.assertEqual(result.prefix, "zooplanktontraits") + self.assertEqual(result.identifier, "c_13") + + def test_converter_cannot_parse_unknown_uri(self): + """Test that converter cannot parse unknown URIs.""" + uri = "http://unknown.example.org/entity" + result = self.converter.parse_uri(uri) + + self.assertIsNone(result) + + def test_converter_from_test_data(self): + """Test converter with actual test data.""" + with TEST_DATA.open() as file: + matchings = json.load(file) + + # Test first few matchings + for matching in matchings[:3]: + source_uri = matching["source"] + target_uri = matching["target"] + + source_result = self.converter.parse_uri(source_uri) + target_result = self.converter.parse_uri(target_uri) + + self.assertIsNotNone(source_result) + self.assertIsNotNone(target_result) + self.assertEqual(source_result.prefix, "fishtraits") + self.assertEqual(target_result.prefix, "zooplanktontraits") + + +if __name__ == "__main__": + unittest.main()