Source code for osp.core.ontology.entity

"""Abstract superclass of any entity in the ontology."""

from abc import ABC, abstractmethod
from functools import lru_cache
import rdflib
import logging

logger = logging.getLogger(__name__)

# The properties of the instances of the metaclass OntologyEntity defined below
# may be cached by applying the decorator @lru_cache after the @property
# decorator. The following parameter fixes the maximum number of different
# instances of OntologyEntity for which a property may be cached.
entity_cache_size = 1024


[docs]class OntologyEntity(ABC): """Abstract superclass of any entity in the ontology.""" @abstractmethod def __init__(self, namespace_registry, namespace_iri, name, iri_suffix): """Initialise the ontology entity. Args: namespace_registry (OntologyNamespaceRegistry): The namespace registry. namespace_iri (rdflib.URIRef): The namespace of the entity name (str): The name of the entity iri_suffix (str): namespace_iri + iri_suffix is namespace of the entity. """ self._name = name self._iri_suffix = iri_suffix self._namespace_iri = namespace_iri self._namespace_registry = namespace_registry @property def _namespace_name(self): return self._namespace_registry._get_namespace_name_and_iri( self.iri )[0] def __str__(self): """Transform the entity into a human readable string.""" return "%s.%s" % (self._namespace_name, self._name) def __repr__(self): """Transform the entity into a string.""" return "<%s %s.%s>" % (self.__class__.__name__, self._namespace_name, self._name) def __eq__(self, other): """Check whether two entities are the same. Args: other (OntologyEntity): The other entity. Returns: bool: Whether the two entities are the same. """ return isinstance(other, OntologyEntity) and self.iri == other.iri @property def name(self): """Get the name of the entity.""" return self._name @property def iri(self): """Get the IRI of the Entity.""" return rdflib.URIRef(self._namespace_iri + self._iri_suffix) @property def tblname(self): """Get the name used in storage backends to store instances.""" return "%s___%s" % (self._namespace_name, self._iri_suffix) @property def namespace(self): """Get the namespace object of the entity.""" return self._namespace_registry.namespace_from_iri(self._namespace_iri) @property def direct_superclasses(self): """Get the direct superclass of the entity. Returns: List[OntologyEntity]: The direct superclasses of the entity """ return set(self._direct_superclasses()) @property def direct_subclasses(self): """Get the direct subclasses of the entity. Returns: Set[OntologyEntity]: The direct subclasses of the entity """ return set(self._direct_subclasses()) @property @lru_cache(maxsize=entity_cache_size) def subclasses(self): """Get the subclasses of the entity. Returns: Set[OntologyEntity]: The direct subclasses of the entity """ return set(self._subclasses()) @property @lru_cache(maxsize=entity_cache_size) def superclasses(self): """Get the superclass of the entity. Returns: Set[OntologyEntity]: The direct superclasses of the entity """ return set(self._superclasses()) @property def description(self): """Get the description of the entity. Returns: str: The description of the entity """ desc = self.namespace._graph.value( self.iri, rdflib.RDFS.isDefinedBy, None) if desc is None: return "To Be Determined" return str(desc)
[docs] def get_triples(self): """Get the triples of the entity.""" return self.namespace._graph.triples((self.iri, None, None))
[docs] def is_superclass_of(self, other): """Perform a superclass check. Args: other (Entity): The other entity. Returns: bool: Whether self is a superclass of other. """ return self in other.superclasses
[docs] def is_subclass_of(self, other): """Perform a subclass check. Args: other (Entity): The other entity. Returns: bool: Whether self is a subclass of other. """ return self in other.subclasses
@abstractmethod def _direct_superclasses(self): pass @abstractmethod def _direct_subclasses(self): pass @abstractmethod def _superclasses(self): pass @abstractmethod def _subclasses(self): pass def _transitive_hull(self, predicate_iri, inverse=False, blacklist=()): """Get all the entities connected with the given predicate. Args: predicate_iri (URIRef): The IRI of the predicate inverse (bool, optional): Use the inverse instead. Defaults to False. blacklist (collection): A collection of IRIs not to return. Yields: OntologyEntity: The connected entities """ visited = {self.iri} frontier = {self.iri} while frontier: current = frontier.pop() yield from self._directly_connected(predicate_iri=predicate_iri, inverse=inverse, blacklist=blacklist, _frontier=frontier, _visited=visited, _iri=current) def _special_cases(self, triple): """Some supclass statements are often omitted in the ontology. Replace these with safer triple patterns. Args: triple (Tuple[rdflib.term]): A triple pattern to possibly replace. Returns: triple (Tuple[rdflib.term]): Possibly replaced triple. """ if triple == (None, rdflib.RDFS.subClassOf, rdflib.OWL.Thing): return (None, rdflib.RDF.type, rdflib.OWL.Class) if triple == (rdflib.OWL.Nothing, rdflib.RDFS.subClassOf, None): return (None, rdflib.RDF.type, rdflib.OWL.Class) if triple == (None, rdflib.RDFS.subPropertyOf, rdflib.OWL.topObjectProperty): return (None, rdflib.RDF.type, rdflib.OWL.ObjectProperty) if triple == (rdflib.OWL.bottomObjectProperty, rdflib.RDFS.subPropertyOf, None): return (None, rdflib.RDF.type, rdflib.OWL.ObjectProperty) if triple == (None, rdflib.RDFS.subPropertyOf, rdflib.OWL.topDataProperty): return (None, rdflib.RDF.type, rdflib.OWL.DataProperty) if triple == (rdflib.OWL.bottomDataProperty, rdflib.RDFS.subPropertyOf, None): return (None, rdflib.RDF.type, rdflib.OWL.DataProperty) return triple def _directly_connected(self, predicate_iri, inverse=False, blacklist=(), _frontier=None, _visited=None, _iri=None): """Get all the entities directly connected with the given predicate. Args: predicate_iri (URIRef): The IRI of the predicate inverse (bool, optional): Use the inverse instead. Defaults to False. blacklist (collection): A collection of IRIs not to return. Others: Helper for _transitive_hull method. Yields: OntologyEntity: The connected entities """ triple = (_iri or self.iri, predicate_iri, None) if inverse: triple = (None, predicate_iri, _iri or self.iri) if predicate_iri in [rdflib.RDFS.subClassOf, rdflib.RDFS.subPropertyOf]: triple = self._special_cases(triple) for x in self.namespace._graph.triples(triple): o = x[0 if triple[0] is None else 2] if _visited and o in _visited: continue if not isinstance(o, rdflib.BNode): if _visited is not None: _visited.add(o) if _frontier is not None: _frontier.add(o) if o not in blacklist: x = self.namespace._namespace_registry.from_iri( o, raise_error=False) if x: yield x def __hash__(self): """Make the entity hashable.""" return hash(self.iri)