[docs]@dataclassclassFunOwlImplementation(OwlInterface,PatcherInterface,SearchInterface):""" An experimental partial implementation of :ref:`OwlInterface` Wraps FunOWL `<https://github.com/hsolbrig/funowl>`_ """ontology_document:OntologyDocument=Nonedef__post_init__(self):ifself.ontology_documentisNone:resource=self.resourceifresourceisNoneorresource.local_pathisNone:doc=OntologyDocument()else:logging.info(f"Loading {resource.local_path} into FunOwl")doc=to_python(str(resource.local_path))self.ontology_document=docifself.functional_writerisNone:self.functional_writer=FunctionalWriter()forprefixindoc.prefixDeclarations.as_prefixes():self.functional_writer.bind(prefix.prefixName,prefix.fullIRI)@propertydef_ontology(self):returnself.ontology_document.ontologydefentity_iri_to_curie(self,entity:IRI)->CURIE:uri=entity.to_rdf(self.functional_writer.g)returnself.uri_to_curie(str(uri),use_uri_fallback=True)defcurie_to_entity_iri(self,curie:CURIE)->IRI:returnIRI(self.curie_to_uri(curie))def_single_valued_assignment(self,curie:CURIE,property:CURIE)->Optional[str]:labels=[a.valueforainself.annotation_assertion_axioms(curie,property=property)]iflabels:iflen(labels)>1:logging.warning(f"Multiple labels for {curie} = {labels}")val=labels[0]rdf_v=val.to_rdf(self.functional_writer.g)ifisinstance(rdf_v,rdflib.Literal):returnrdf_v.valueelse:raiseValueError(f"Label must be literal, not {val}")defdefinition(self,curie:CURIE,lang:Optional[LANGUAGE_TAG]=None)->Optional[str]:returnself._single_valued_assignment(curie,HAS_DEFINITION_CURIE)deflabel(self,curie:CURIE,lang:Optional[LANGUAGE_TAG]=None)->str:labels=[a.valueforainself.annotation_assertion_axioms(curie,property=LABEL_PREDICATE)]iflabels:iflen(labels)>1:logging.warning(f"Multiple labels for {curie} = {labels}")label=labels[0]rdf_v=label.to_rdf(self.functional_writer.g)ifisinstance(rdf_v,rdflib.Literal):returnrdf_v.valueelse:raiseValueError(f"Label must be literal, not {label}")defentities(self,filter_obsoletes=True,owl_type=None)->Iterable[CURIE]:foraxinself._ontology.axioms:ifisinstance(ax,Declaration):uri=ax.v.full_uri(self.functional_writer.g)try:yv=self.uri_to_curie(str(uri))exceptValueError:logging.warning("could not compress URI %s with functional writer context %s",uri,list(self.functional_writer.g.namespaces()),)continueelse:yieldyvdefaxioms(self,reasoner:Optional[ReasonerConfiguration]=None)->Iterable[Axiom]:ont=self._ontologyforaxiominont.axioms:yieldaxiomdefset_axioms(self,axioms:List[Axiom])->None:self._ontology.axioms=axiomsdefdump(self,path:str=None,syntax:str=None,**kwargs):ifsyntaxisNoneorsyntax=="ofn":out=self.ontology_document.to_functional(self.functional_writer)elifsyntax=="ttl"orsyntax=="turtle":g=rdflib.Graph()self.ontology_document.to_rdf(g)out=g.serialize(format="ttl")else:out=str(self.ontology_document)ifpathisNone:print(out)elifisinstance(path,str):withopen(path,"w",encoding="UTF-8")asfile:file.write(str(out))else:path.write(str(out))# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^# Implements: PatcherInterface# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^def_set_annotation_predicate_value(self,subject:CURIE,property:CURIE,value:Any):foraxiominself.annotation_assertion_axioms(subject,property):self._ontology.axioms.remove(axiom)self._ontology.axioms.append(AnnotationAssertion(subject=self.curie_to_entity_iri(subject),property=self.curie_to_entity_iri(property),value=value,))defapply_patch(self,patch:kgcl.Change,activity:kgcl.Activity=None,metadata:Mapping[PRED_CURIE,Any]=None,configuration:kgcl.Configuration=None,)->Optional[kgcl.Change]:ifisinstance(patch,kgcl.NodeChange):about=patch.about_nodeifisinstance(patch,kgcl.NodeRename):self._set_annotation_predicate_value(about,LABEL_PREDICATE,patch.new_value)elifisinstance(patch,kgcl.NodeTextDefinitionChange):self._set_annotation_predicate_value(about,HAS_DEFINITION_CURIE,patch.new_value)elifisinstance(patch,kgcl.NewSynonym):self._ontology.axioms.append(AnnotationAssertion(subject=about,property=self.curie_to_entity_iri(HAS_EXACT_SYNONYM),value=patch.new_value,))elifisinstance(patch,kgcl.NodeObsoletion):self._set_annotation_predicate_value(about,DEPRECATED_PREDICATE,value=True)elifisinstance(patch,kgcl.NodeDeletion):raiseNotImplementedError("Deletions not supported yet")elifisinstance(patch,kgcl.NodeCreation):self._set_annotation_predicate_value(about,LABEL_PREDICATE,patch.name)elifisinstance(patch,kgcl.NameBecomesSynonym):label=self.label(about)self.apply_patch(kgcl.NodeRename(id=f"{patch.id}-1",about_node=about,new_value=patch.new_value))self.apply_patch(kgcl.NewSynonym(id=f"{patch.id}-2",about_node=about,new_value=label))else:raiseNotImplementedError(f"Cannot handle patches of type {type(patch)}")elifisinstance(patch,kgcl.EdgeChange):about=patch.about_edgesubject=self.curie_to_uri(patch.subject)object=self.curie_to_uri(patch.object)ifisinstance(patch,kgcl.EdgeCreation):ifpatch.predicate==IS_Aorpatch.predicate=="is_a":self._ontology.axioms.append(SubClassOf(subject,object))else:predicate=self.curie_to_entity_iri(patch.predicate)self._ontology.axioms.append(SubClassOf(subject,ObjectSomeValuesFrom(predicate,object)))else:raiseNotImplementedError(f"Cannot handle patches of type {type(patch)}")else:raiseNotImplementedError(f"Cannot handle patches of type {type(patch)}")returnpatch