"""OWL Ontology"""
import json
import logging
import os
import types
from datetime import date, datetime
from functools import singledispatch, singledispatchmethod
from itertools import chain, combinations, islice
from pathlib import Path
from types import MappingProxyType
from typing import Any, Dict, Final, Iterable, List, Optional, Tuple, Union, cast
import jpype
import rdflib
from pandas import Timedelta
from owlapy import namespaces
from owlapy._lazy_owlready2 import import_owlready2
from owlapy.abstracts.abstract_owl_ontology import _OI, AbstractOWLOntology
from owlapy.class_expression import (
OWLClass,
OWLClassExpression,
OWLDataAllValuesFrom,
OWLDataExactCardinality,
OWLDataHasValue,
OWLDataMaxCardinality,
OWLDataMinCardinality,
OWLDataOneOf,
OWLDataRestriction,
OWLDataSomeValuesFrom,
OWLDatatypeRestriction,
OWLFacetRestriction,
OWLNaryBooleanClassExpression,
OWLObjectAllValuesFrom,
OWLObjectComplementOf,
OWLObjectExactCardinality,
OWLObjectHasValue,
OWLObjectIntersectionOf,
OWLObjectMaxCardinality,
OWLObjectMinCardinality,
OWLObjectOneOf,
OWLObjectRestriction,
OWLObjectSomeValuesFrom,
OWLObjectUnionOf,
OWLQuantifiedDataRestriction,
OWLQuantifiedObjectRestriction,
OWLRestriction,
OWLThing,
)
from owlapy.iri import IRI
from owlapy.owl_axiom import (
OWLAnnotation,
OWLAnnotationAssertionAxiom,
OWLAnnotationProperty,
OWLAsymmetricObjectPropertyAxiom,
OWLAxiom,
OWLClassAssertionAxiom,
OWLClassAxiom,
OWLDataPropertyAssertionAxiom,
OWLDataPropertyCharacteristicAxiom,
OWLDataPropertyDomainAxiom,
OWLDataPropertyRangeAxiom,
OWLDeclarationAxiom,
OWLDifferentIndividualsAxiom,
OWLDisjointClassesAxiom,
OWLDisjointDataPropertiesAxiom,
OWLDisjointObjectPropertiesAxiom,
OWLDisjointUnionAxiom,
OWLEquivalentClassesAxiom,
OWLEquivalentDataPropertiesAxiom,
OWLEquivalentObjectPropertiesAxiom,
OWLFunctionalDataPropertyAxiom,
OWLFunctionalObjectPropertyAxiom,
OWLInverseFunctionalObjectPropertyAxiom,
OWLInverseObjectPropertiesAxiom,
OWLIrreflexiveObjectPropertyAxiom,
OWLNaryIndividualAxiom,
OWLNaryPropertyAxiom,
OWLObjectPropertyAssertionAxiom,
OWLObjectPropertyCharacteristicAxiom,
OWLObjectPropertyDomainAxiom,
OWLObjectPropertyRangeAxiom,
OWLPropertyDomainAxiom,
OWLPropertyRangeAxiom,
OWLReflexiveObjectPropertyAxiom,
OWLSameIndividualAxiom,
OWLSubClassOfAxiom,
OWLSubPropertyAxiom,
OWLSubPropertyChainAxiom,
OWLSymmetricObjectPropertyAxiom,
OWLTransitiveObjectPropertyAxiom,
)
from owlapy.owl_data_ranges import OWLDataComplementOf, OWLDataIntersectionOf, OWLDataRange, OWLDataUnionOf
from owlapy.owl_datatype import OWLDatatype
from owlapy.owl_individual import OWLIndividual, OWLNamedIndividual
from owlapy.owl_literal import BooleanOWLDatatype, DateOWLDatatype, DateTimeOWLDatatype, DoubleOWLDatatype, DurationOWLDatatype, IntegerOWLDatatype, OWLLiteral, StringOWLDatatype, TopOWLDatatype
from owlapy.owl_object import OWLEntity, OWLObject
from owlapy.owl_property import OWLDataProperty, OWLDataPropertyExpression, OWLObjectInverseOf, OWLObjectProperty, OWLObjectPropertyExpression, OWLProperty, OWLPropertyExpression
from owlapy.static_funcs import startJVM
from owlapy.vocab import OWLFacet
logger = logging.getLogger(__name__)
# owlready2 is an optional dependency (see owlapy._lazy_owlready2 / issue #205): `owlready2`
# below is either the real module or a placeholder that raises a clear ImportError the moment
# anything below actually tries to call into it (module-level attribute access, e.g. for type
# annotations on the owlready2-backed Ontology/ToOwlready2/FromOwlready2 classes below, always
# succeeds either way).
owlready2 = import_owlready2()
AllDifferent = owlready2.AllDifferent
AllDisjoint = owlready2.AllDisjoint
GeneralClassAxiom = owlready2.GeneralClassAxiom
destroy_entity = owlready2.destroy_entity
_Datatype_map: Final = MappingProxyType({
int: IntegerOWLDatatype,
float: DoubleOWLDatatype,
bool: BooleanOWLDatatype,
str: StringOWLDatatype,
date: DateOWLDatatype,
datetime: DateTimeOWLDatatype,
Timedelta: DurationOWLDatatype,
})
_VERSION_IRI: Final = IRI.create(namespaces.OWL, "versionIRI")
# Mapping of format string identifiers to OWL API document format class names.
# All formats are from org.semanticweb.owlapi.formats.
_DOCUMENT_FORMATS: Final = MappingProxyType({
"rdf/xml": "RDFXMLDocumentFormat",
"rdfxml": "RDFXMLDocumentFormat",
"owl/xml": "OWLXMLDocumentFormat",
"owlxml": "OWLXMLDocumentFormat",
"turtle": "TurtleDocumentFormat",
"ttl": "TurtleDocumentFormat",
"functional": "FunctionalSyntaxDocumentFormat",
"fss": "FunctionalSyntaxDocumentFormat",
"manchester": "ManchesterSyntaxDocumentFormat",
"ms": "ManchesterSyntaxDocumentFormat",
"latex": "LatexDocumentFormat",
"dlsyntax": "DLSyntaxDocumentFormat",
"dl": "DLSyntaxDocumentFormat",
"krss2": "KRSS2DocumentFormat",
"obo": "OBODocumentFormat",
})
# Formats handled by rdflib (not natively supported by the bundled OWL API).
# Keys are the user-facing format strings; values are the rdflib format names
# passed to ``rdflib.Graph.serialize(format=...)``.
# Formats that need a ConjunctiveGraph (context-aware store) are listed in
# _RDFLIB_CONJUNCTIVE_FORMATS.
_RDFLIB_FORMATS: Final = MappingProxyType({
"turtle2": "turtle", # rdflib's Turtle serializer (alias for clarity)
"json-ld": "json-ld", # JSON-LD (requires rdflib[jsonld] or jsonld pkg)
"jsonld": "json-ld",
"ntriples": "ntriples",
"nt": "ntriples",
"nt11": "nt11", # N-Triples 1.1
"n3": "n3", # Notation3
"trig": "trig", # TriG
"trix": "trix", # TriX (requires ConjunctiveGraph)
"nquads": "nquads", # N-Quads (requires ConjunctiveGraph)
"nq": "nquads",
})
# rdflib formats that require a ConjunctiveGraph (context-aware store)
# instead of a plain Graph.
_RDFLIB_CONJUNCTIVE_FORMATS: Final = frozenset({"trix", "nquads"})
# Format table for RDFLibOntology.save(), which serializes natively via rdflib (no OWL API
# fallback dance needed, unlike Ontology/SyncOntology above). Keys are user-facing format
# strings -- both rdflib's own names and the OWL-API-style aliases used by Ontology/SyncOntology's
# save(), so the three ontology classes accept the same format vocabulary. Values are the format
# names passed to ``rdflib.Graph.serialize(format=...)``.
_RDFLIB_NATIVE_FORMATS: Final = MappingProxyType({
"rdfxml": "xml", "rdf/xml": "xml", "xml": "xml",
"turtle": "turtle", "ttl": "turtle", "turtle2": "turtle",
"n3": "n3",
"ntriples": "nt", "nt": "nt", "nt11": "nt11",
"trig": "trig",
"trix": "trix",
"nquads": "nquads", "nq": "nquads",
"json-ld": "json-ld", "jsonld": "json-ld",
})
[docs]
class OWLOntologyID:
"""An object that identifies an ontology. Since OWL 2, ontologies do not have to have an ontology IRI, or if they
have an ontology IRI then they can optionally also have a version IRI. Instances of this OWLOntologyID class bundle
identifying information of an ontology together. If an ontology doesn't have an ontology IRI then we say that it is
"anonymous".
"""
__slots__ = '_ontology_iri', '_version_iri'
_ontology_iri: Optional[IRI]
_version_iri: Optional[IRI]
def __init__(self, ontology_iri: Optional[IRI] = None, version_iri: Optional[IRI] = None):
"""Constructs an ontology identifier specifying the ontology IRI and version IRI.
Args:
ontology_iri: The ontology IRI (optional).
version_iri: The version IRI (must be None if no ontology_iri is provided).
"""
if ontology_iri is not None and not isinstance(ontology_iri, IRI):
raise TypeError(f"Expected 'ontology_iri' to be an instance of IRI or None, got {type(ontology_iri).__name__} instead ({ontology_iri!r}).")
if version_iri is not None and not isinstance(version_iri, IRI):
raise TypeError(f"Expected 'version_iri' to be an instance of IRI or None, got {type(version_iri).__name__} instead ({version_iri!r}).")
self._ontology_iri = ontology_iri
self._version_iri = version_iri
[docs]
def get_ontology_iri(self) -> Optional[IRI]:
"""Gets the ontology IRI.
Returns:
Ontology IRI. If the ontology is anonymous, it will return None.
"""
return self._ontology_iri
[docs]
def get_version_iri(self) -> Optional[IRI]:
"""Gets the version IRI.
Returns:
Version IRI or None.
"""
return self._version_iri
[docs]
def get_default_document_iri(self) -> Optional[IRI]:
"""Gets the IRI which is used as a default for the document that contain a representation of an ontology with
this ID. This will be the version IRI if there is an ontology IRI and version IRI, else it will be the ontology
IRI if there is an ontology IRI but no version IRI, else it will be None if there is no ontology IRI. See
Ontology Documents in the OWL 2 Structural Specification.
Returns:
the IRI that can be used as a default for an ontology document, or None.
"""
if self._ontology_iri is not None:
if self._version_iri is not None:
return self._version_iri
return self._ontology_iri
[docs]
def is_anonymous(self) -> bool:
return self._ontology_iri is None
[docs]
def __repr__(self):
return f"OWLOntologyID({repr(self._ontology_iri)}, {repr(self._version_iri)})"
[docs]
def __eq__(self, other):
if type(other) is type(self):
return self._ontology_iri == other._ontology_iri and self._version_iri == other._version_iri
return NotImplemented
def _check_expression(expr: OWLObject, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
"""
Creates all entities (individuals, classes, properties) that appear in the given (complex) class expression
and do not exist in the given ontology yet
"""
if isinstance(expr, (OWLClass, OWLProperty, OWLNamedIndividual,)):
_add_axiom(OWLDeclarationAxiom(expr), ontology, world)
elif isinstance(expr, (OWLNaryBooleanClassExpression, OWLObjectComplementOf, OWLObjectOneOf,)):
for op in expr.operands():
_check_expression(op, ontology, world)
elif isinstance(expr, (OWLQuantifiedObjectRestriction, OWLObjectHasValue,)):
_check_expression(expr.get_property(), ontology, world)
_check_expression(expr.get_filler(), ontology, world)
elif isinstance(expr, OWLObjectInverseOf):
_check_expression(expr.get_named_property(), ontology, world)
_check_expression(expr.get_inverse_property(), ontology, world)
elif isinstance(expr, (OWLQuantifiedDataRestriction, OWLDataHasValue,)):
_check_expression(expr.get_property(), ontology, world)
elif not isinstance(expr, OWLObject):
raise ValueError(f'({expr}) is not an OWLObject.')
@singledispatch
def _add_axiom(axiom: OWLAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
raise NotImplementedError(f'Axiom type {axiom} is not implemented yet.')
@_add_axiom.register
def _(axiom: OWLDeclarationAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
entity = axiom.get_entity()
with ont_x:
entity_x = world[entity.to_string_id()]
# Entity already exists
if entity_x is not None:
return
thing_x: owlready2.entity.ThingClass = conv.map_concept(OWLThing)
if isinstance(entity, OWLClass):
if entity.is_owl_thing() or entity.is_owl_nothing():
return
entity_x = types.new_class(name=entity.iri.get_remainder(), bases=(thing_x,))
elif isinstance(entity, OWLIndividual):
entity_x = thing_x(entity.iri.get_remainder())
elif isinstance(entity, OWLObjectProperty):
entity_x = types.new_class(name=entity.iri.get_remainder(), bases=(owlready2.ObjectProperty,))
elif isinstance(entity, OWLDataProperty):
entity_x = types.new_class(name=entity.iri.get_remainder(), bases=(owlready2.DatatypeProperty,))
elif isinstance(entity, OWLAnnotationProperty):
entity_x = types.new_class(name=entity.iri.get_remainder(), bases=(owlready2.AnnotationProperty,))
else:
raise ValueError(f'Cannot add ({entity}). Not an atomic class, property, or individual.')
entity_x.namespace = ont_x.get_namespace(entity.iri.get_namespace())
entity_x.namespace.world._refactor(entity_x.storid, entity_x.iri)
@_add_axiom.register
def _(axiom: OWLClassAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
individual = axiom.get_individual()
cls_ = axiom.get_class_expression()
_check_expression(cls_, ontology, world)
_add_axiom(OWLDeclarationAxiom(individual), ontology, world)
with ont_x:
cls_x = conv.map_concept(cls_)
ind_x = conv._to_owlready2_individual(individual)
thing_x = conv.map_concept(OWLThing)
if thing_x in ind_x.is_a:
ind_x.is_a.remove(thing_x)
ind_x.is_a.append(cls_x)
@_add_axiom.register
def _(axiom: OWLObjectPropertyAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
subject = axiom.get_subject()
property_ = axiom.get_property()
object_ = axiom.get_object()
_add_axiom(OWLDeclarationAxiom(subject), ontology, world)
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
_add_axiom(OWLDeclarationAxiom(object_), ontology, world)
with ont_x:
subject_x = conv._to_owlready2_individual(subject)
property_x = conv._to_owlready2_property(property_)
object_x = conv._to_owlready2_individual(object_)
property_x[subject_x].append(object_x)
@_add_axiom.register
def _(axiom: OWLDataPropertyAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
subject = axiom.get_subject()
property_ = axiom.get_property()
_add_axiom(OWLDeclarationAxiom(subject), ontology, world)
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
with ont_x:
subject_x = conv._to_owlready2_individual(subject)
property_x = conv._to_owlready2_property(property_)
property_x[subject_x].append(axiom.get_object().to_python())
@_add_axiom.register
def _(axiom: OWLSubClassOfAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
sub_class = axiom.get_sub_class()
super_class = axiom.get_super_class()
_check_expression(sub_class, ontology, world)
_check_expression(super_class, ontology, world)
with ont_x:
thing_x = conv.map_concept(OWLThing)
sub_class_x = conv.map_concept(sub_class)
super_class_x = conv.map_concept(super_class)
if isinstance(sub_class, OWLClass):
if thing_x in sub_class_x.is_a:
sub_class_x.is_a.remove(thing_x)
else:
# Currently owlready2 seems to expect that we make a new GeneralClassAxiom object each time.
# Another option would be to check whether a GeneralClassAxiom with the sub_class_x already exists and just
# add the super_class_x to its is_a attribute
sub_class_x = GeneralClassAxiom(sub_class_x)
sub_class_x.is_a.append(super_class_x)
# NOTE: owlready2 limitation - This implementation is constrained by owlready2's current lack of support
# for EquivalentClasses general class axioms. Only named classes can be marked as equivalent.
# See: https://owlready2.readthedocs.io/en/latest/class.html#equivalent-classes
# This may be updated in future owlready2 releases.
@_add_axiom.register
def _(axiom: OWLEquivalentClassesAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x = conv.map_object(ontology)
assert axiom.contains_named_equivalent_class(), 'Owlready2 does not support general' \
'class axioms for equivalent classes.'
for ce in axiom.class_expressions():
_check_expression(ce, ontology, world)
with ont_x:
for ce_1, ce_2 in combinations(axiom.class_expressions(), 2):
assert ce_1 is not None, f"ce_1 cannot be None: {ce_1}, {type(ce_1)}"
assert ce_2 is not None, f"ce_2_x cannot be None: {ce_2}, {type(ce_2)}"
ce_1_x = conv.map_concept(ce_1)
ce_2_x = conv.map_concept(ce_2)
try:
assert ce_1_x is not None, f"ce_1_x cannot be None: {ce_1_x}, {type(ce_1_x)}"
assert ce_2_x is not None, f"ce_2_x cannot be None: {ce_2_x}, {type(ce_2_x)}"
except AssertionError:
logger.warning(
f"ToOwlready2.map_concept() returned None; reinitializing ce_1_x=ce_2_x. "
f"ce_1={ce_1} ce_1_x={ce_1_x} ce_2={ce_2} ce_2_x={ce_2_x} axiom={axiom}"
)
ce_1_x=ce_2_x
if isinstance(ce_1_x, owlready2.ThingClass):
ce_1_x.equivalent_to.append(ce_2_x)
if isinstance(ce_2_x, owlready2.ThingClass):
ce_2_x.equivalent_to.append(ce_1_x)
@_add_axiom.register
def _(axiom: OWLDisjointClassesAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
for cls_ in axiom.class_expressions():
_check_expression(cls_, ontology, world)
with ont_x:
# TODO: If the first element in the list is a complex class expression owlready2 is bugged
# and creates an AllDifferent axiom
AllDisjoint(list(map(conv.map_concept, axiom.class_expressions())))
@_add_axiom.register
def _(axiom: OWLDisjointUnionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
assert isinstance(axiom.get_owl_class(), OWLClass), f'({axiom.get_owl_class()}) is not a named class.'
_add_axiom(OWLDeclarationAxiom(axiom.get_owl_class()), ontology, world)
for cls_ in axiom.get_class_expressions():
_check_expression(cls_, ontology, world)
with ont_x:
cls_x = conv.map_concept(axiom.get_owl_class())
cls_x.disjoint_unions.append(list(map(conv.map_concept, axiom.get_class_expressions())))
@_add_axiom.register
def _(axiom: OWLAnnotationAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
prop_x = conv.map_object(axiom.get_property())
if prop_x is None:
with ont_x:
prop_x: owlready2.annotation.AnnotationPropertyClass = cast(
owlready2.AnnotationProperty,
types.new_class(
name=axiom.get_property().iri.get_remainder(),
bases=(owlready2.AnnotationProperty,)))
prop_x.namespace = ont_x.get_namespace(axiom.get_property().iri.get_namespace())
sub_x = world[axiom.get_subject().as_iri().as_str()]
assert sub_x is not None, f'{axiom.get_subject} not found in {ontology}'
with ont_x:
if axiom.get_value().is_literal():
literal = axiom.get_value().as_literal()
setattr(sub_x, prop_x.python_name, literal.to_python())
else:
o_x = world[axiom.get_value().as_iri().as_str()]
assert o_x is not None, f'{axiom.get_value()} not found in {ontology}'
setattr(sub_x, prop_x.python_name, o_x)
@_add_axiom.register
def _(axiom: OWLNaryIndividualAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
for ind in axiom.individuals():
_add_axiom(OWLDeclarationAxiom(ind), ontology, world)
with ont_x:
if isinstance(axiom, OWLSameIndividualAxiom):
for idx, ind in enumerate(axiom.individuals()):
ind_x = conv._to_owlready2_individual(ind)
for ind_2 in islice(axiom.individuals(), idx + 1, None):
ind_2_x = conv._to_owlready2_individual(ind_2)
ind_x.equivalent_to.append(ind_2_x)
elif isinstance(axiom, OWLDifferentIndividualsAxiom):
AllDifferent(list(map(conv._to_owlready2_individual, axiom.individuals())))
else:
raise ValueError(f'OWLNaryIndividualAxiom ({axiom}) is not defined.')
@_add_axiom.register
def _(axiom: OWLSubPropertyAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
sub_property = axiom.get_sub_property()
super_property = axiom.get_super_property()
_add_axiom(OWLDeclarationAxiom(sub_property), ontology, world)
_add_axiom(OWLDeclarationAxiom(super_property), ontology, world)
with ont_x:
sub_property_x = conv._to_owlready2_property(sub_property)
super_property_x = conv._to_owlready2_property(super_property)
sub_property_x.is_a.append(super_property_x)
@_add_axiom.register
def _(axiom: OWLPropertyDomainAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
property_ = axiom.get_property()
domain = axiom.get_domain()
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
_check_expression(domain, ontology, world)
with ont_x:
property_x = conv._to_owlready2_property(property_)
domain_x = conv.map_concept(domain)
property_x.domain.append(domain_x)
@_add_axiom.register
def _(axiom: OWLPropertyRangeAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
property_ = axiom.get_property()
range_ = axiom.get_range()
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
if isinstance(axiom, OWLObjectPropertyRangeAxiom):
_check_expression(range_, ontology, world)
with ont_x:
property_x = conv._to_owlready2_property(property_)
range_x = conv.map_concept(range_) if isinstance(axiom, OWLObjectPropertyRangeAxiom) \
else conv.map_datarange(range_)
property_x.range.append(range_x)
@_add_axiom.register
def _(axiom: OWLNaryPropertyAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
for property_ in axiom.properties():
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
with ont_x:
if isinstance(axiom, (OWLEquivalentObjectPropertiesAxiom, OWLEquivalentDataPropertiesAxiom,)):
for idx, property_ in enumerate(axiom.properties()):
property_x = conv._to_owlready2_property(property_)
for property_2 in islice(axiom.properties(), idx + 1, None):
property_2_x = conv._to_owlready2_property(property_2)
property_x.equivalent_to.append(property_2_x)
elif isinstance(axiom, (OWLDisjointObjectPropertiesAxiom, OWLDisjointDataPropertiesAxiom,)):
AllDisjoint(list(map(conv._to_owlready2_property, axiom.properties())))
elif isinstance(axiom, OWLInverseObjectPropertiesAxiom):
property_first_x = conv._to_owlready2_property(axiom.get_first_property())
property_second_x = conv._to_owlready2_property(axiom.get_second_property())
if property_second_x.inverse_property is not None:
property_second_x.inverse_property = None
property_first_x.inverse_property = property_second_x
else:
raise ValueError(f'OWLNaryPropertyAxiom ({axiom}) is not defined.')
@_add_axiom.register
def _(axiom: OWLObjectPropertyCharacteristicAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
property_ = axiom.get_property()
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
with ont_x:
property_x = conv._to_owlready2_property(property_)
if isinstance(axiom, OWLFunctionalObjectPropertyAxiom):
property_x.is_a.append(owlready2.FunctionalProperty)
elif isinstance(axiom, OWLAsymmetricObjectPropertyAxiom):
property_x.is_a.append(owlready2.AsymmetricProperty)
elif isinstance(axiom, OWLInverseFunctionalObjectPropertyAxiom):
property_x.is_a.append(owlready2.InverseFunctionalProperty)
elif isinstance(axiom, OWLIrreflexiveObjectPropertyAxiom):
property_x.is_a.append(owlready2.IrreflexiveProperty)
elif isinstance(axiom, OWLReflexiveObjectPropertyAxiom):
property_x.is_a.append(owlready2.ReflexiveProperty)
elif isinstance(axiom, OWLSymmetricObjectPropertyAxiom):
property_x.is_a.append(owlready2.SymmetricProperty)
elif isinstance(axiom, OWLTransitiveObjectPropertyAxiom):
property_x.is_a.append(owlready2.TransitiveProperty)
else:
raise ValueError(f'ObjectPropertyCharacteristicAxiom ({axiom}) is not defined.')
@_add_axiom.register
def _(axiom: OWLSubPropertyChainAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
super_property = axiom.get_super_property()
property_chain = list(axiom.get_property_chain())
# Ensure all properties in the chain and the super property are declared
_add_axiom(OWLDeclarationAxiom(super_property), ontology, world)
for prop in property_chain:
if isinstance(prop, OWLObjectInverseOf):
_add_axiom(OWLDeclarationAxiom(prop.get_named_property()), ontology, world)
else:
_add_axiom(OWLDeclarationAxiom(prop), ontology, world)
with ont_x:
super_property_x = conv._to_owlready2_property(super_property)
property_chain_x = [conv._to_owlready2_property(prop) for prop in property_chain]
super_property_x.property_chain.append(owlready2.PropertyChain(property_chain_x))
@_add_axiom.register
def _(axiom: OWLDataPropertyCharacteristicAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
property_ = axiom.get_property()
_add_axiom(OWLDeclarationAxiom(property_), ontology, world)
with ont_x:
property_x = conv._to_owlready2_property(property_)
if isinstance(axiom, OWLFunctionalDataPropertyAxiom):
property_x.is_a.append(owlready2.FunctionalProperty)
else:
raise ValueError(f'DataPropertyCharacteristicAxiom ({axiom}) is not defined.')
@singledispatch
def _remove_axiom(axiom: OWLAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
raise NotImplementedError(f'Axiom type {axiom} is not implemented yet.')
@_remove_axiom.register
def _(axiom: OWLDeclarationAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
with ont_x:
entity_x = world[axiom.get_entity().to_string_id()]
if entity_x is not None:
# TODO: owlready2 seems to be bugged for properties here
destroy_entity(entity_x)
@_remove_axiom.register
def _(axiom: OWLClassAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
with ont_x:
cls_x = conv.map_concept(axiom.get_class_expression())
ind_x = conv._to_owlready2_individual(axiom.get_individual())
if cls_x is None or ind_x is None:
return
if cls_x in ind_x.is_a:
ind_x.is_a.remove(cls_x)
elif isinstance(axiom.get_class_expression(), OWLClass):
ont_x._del_obj_triple_spo(ind_x.storid, owlready2.rdf_type, cls_x.storid)
@_remove_axiom.register
def _(axiom: OWLObjectPropertyAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
with ont_x:
subject_x = conv._to_owlready2_individual(axiom.get_subject())
property_x = conv._to_owlready2_property(axiom.get_property())
object_x = conv._to_owlready2_individual(axiom.get_object())
if all([subject_x, property_x, object_x]) and object_x in property_x[subject_x]:
property_x[subject_x].remove(object_x)
@_remove_axiom.register
def _(axiom: OWLDataPropertyAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
with ont_x:
subject_x = conv._to_owlready2_individual(axiom.get_subject())
property_x = conv._to_owlready2_property(axiom.get_property())
object_ = axiom.get_object().to_python()
if subject_x is not None and property_x is not None and object_ in property_x[subject_x]:
property_x[subject_x].remove(object_)
@_remove_axiom.register
def _(axiom: OWLSubClassOfAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
sub_class = axiom.get_sub_class()
super_class = axiom.get_super_class()
with ont_x:
sub_class_x = conv.map_concept(sub_class)
super_class_x = conv.map_concept(super_class)
if sub_class_x is None or super_class_x is None:
return
if isinstance(sub_class, OWLClass):
if super_class_x in sub_class_x.is_a:
sub_class_x.is_a.remove(super_class_x)
elif isinstance(axiom.get_sub_class(), OWLClass) and isinstance(axiom.get_super_class(), OWLClass):
ont_x._del_obj_triple_spo(sub_class_x.storid, owlready2.rdfs_subclassof, super_class_x.storid)
else:
for ca in ont_x.general_class_axioms():
if ca.left_side == sub_class_x and super_class_x in ca.is_a:
ca.is_a.remove(super_class_x)
# TODO: Update as soons as owlready2 adds support for EquivalentClasses general class axioms
@_remove_axiom.register
def _(axiom: OWLEquivalentClassesAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x = conv.map_object(ontology)
if not axiom.contains_named_equivalent_class():
return
with ont_x:
ces_x = list(map(conv.map_concept, axiom.class_expressions()))
if len(ces_x) < 2 or not all(ces_x):
return
for ce_1_x, ce_2_x in combinations(ces_x, 2):
if isinstance(ce_2_x, owlready2.ThingClass) and ce_1_x in ce_2_x.equivalent_to:
ce_2_x.equivalent_to.remove(ce_1_x)
if isinstance(ce_1_x, owlready2.ThingClass) and ce_2_x in ce_1_x.equivalent_to:
ce_1_x.equivalent_to.remove(ce_2_x)
@_remove_axiom.register
def _(axiom: OWLDisjointClassesAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
class_expressions_x = set(map(conv.map_concept, axiom.class_expressions()))
if len(class_expressions_x) < 2 or not all(class_expressions_x):
return
for disjoints_x in ont_x.disjoint_classes():
if set(disjoints_x.entities) == class_expressions_x:
del disjoints_x.entities[:-1]
break
@_remove_axiom.register
def _(axiom: OWLDisjointUnionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
assert isinstance(axiom.get_owl_class(), OWLClass), f'({axiom.get_owl_class()}) is not a named class.'
with ont_x:
cls_x = conv.map_concept(axiom.get_owl_class())
union_expressions_x = set(map(conv.map_concept, axiom.get_class_expressions()))
if cls_x is not None and all(union_expressions_x):
for union_x in cls_x.disjoint_unions:
if union_expressions_x == set(union_x):
cls_x.disjoint_unions.remove(union_x)
break
@_remove_axiom.register
def _(axiom: OWLAnnotationAssertionAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
sub_x = world[axiom.get_subject().as_iri().as_str()]
if sub_x is None:
return
name = axiom.get_property().iri.get_remainder()
with ont_x:
if axiom.get_value().is_literal():
o_x = axiom.get_value().as_literal().to_python()
else:
o_x = world[axiom.get_value().as_iri().as_str()]
value = getattr(sub_x, name, None)
if value is not None and o_x in value:
value.remove(o_x)
@_remove_axiom.register
def _(axiom: OWLNaryIndividualAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
individuals_x = list(map(conv._to_owlready2_individual, axiom.individuals()))
if len(individuals_x) < 2 or not all(individuals_x):
return
if isinstance(axiom, OWLSameIndividualAxiom):
if set(individuals_x[1:-1]) <= set(individuals_x[0].INDIRECT_equivalent_to):
for individual_1_x, individual_2_x in combinations(individuals_x, 2):
if individual_1_x in individual_2_x.equivalent_to:
individual_2_x.equivalent_to.remove(individual_1_x)
if individual_2_x in individual_1_x.equivalent_to:
individual_1_x.equivalent_to.remove(individual_2_x)
elif isinstance(axiom, OWLDifferentIndividualsAxiom):
individuals_x = set(individuals_x)
for different_x in ont_x.different_individuals():
if set(different_x.entities) == individuals_x:
del different_x.entities[:-1]
break
else:
raise ValueError(f'OWLNaryIndividualAxiom ({axiom}) is not defined.')
@_remove_axiom.register
def _(axiom: OWLSubPropertyAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.namespace.Ontology = conv.map_object(ontology)
with ont_x:
sub_property_x = conv._to_owlready2_property(axiom.get_sub_property())
super_property_x = conv._to_owlready2_property(axiom.get_super_property())
if sub_property_x is None or super_property_x is None:
return
if super_property_x in sub_property_x.is_a:
sub_property_x.is_a.remove(super_property_x)
else:
ont_x._del_obj_triple_spo(sub_property_x.storid, owlready2.rdfs_subpropertyof, super_property_x.storid)
@_remove_axiom.register
def _(axiom: OWLPropertyDomainAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
property_x = conv._to_owlready2_property(axiom.get_property())
domain_x = conv.map_concept(axiom.get_domain())
if domain_x is not None and property_x is not None and domain_x in property_x.domain:
property_x.domain.remove(domain_x)
@_remove_axiom.register
def _(axiom: OWLPropertyRangeAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
property_x = conv._to_owlready2_property(axiom.get_property())
range_x = conv.map_concept(axiom.get_range()) \
if isinstance(axiom, OWLObjectPropertyRangeAxiom) else conv.map_datarange(axiom.get_range())
if range_x is not None and property_x is not None and range_x in property_x.range:
property_x.range.remove(range_x)
@_remove_axiom.register
def _(axiom: OWLNaryPropertyAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
properties_x = list(map(conv._to_owlready2_property, axiom.properties()))
if len(properties_x) < 2 or not all(properties_x):
return
if isinstance(axiom, (OWLEquivalentObjectPropertiesAxiom, OWLEquivalentDataPropertiesAxiom,)):
# Check if all equivalent properties are defined in the ontology
if set(properties_x[1:-1]) <= set(properties_x[0].INDIRECT_equivalent_to):
for property_1_x, property_2_x in combinations(properties_x, 2):
if property_1_x in property_2_x.equivalent_to:
property_2_x.equivalent_to.remove(property_1_x)
if property_2_x in property_1_x.equivalent_to:
property_1_x.equivalent_to.remove(property_2_x)
elif isinstance(axiom, (OWLDisjointObjectPropertiesAxiom, OWLDisjointDataPropertiesAxiom,)):
properties_x = set(properties_x)
for disjoints_x in ont_x.disjoint_properties():
if set(disjoints_x.entities) == properties_x:
del disjoints_x.entities[:-1]
break
elif isinstance(axiom, OWLInverseObjectPropertiesAxiom):
if len(properties_x) != 2:
return
first = properties_x[0]
second = properties_x[1]
if first.inverse_property == second and second.inverse_property == first:
first.inverse_property = None
else:
raise ValueError(f'OWLNaryPropertyAxiom ({axiom}) is not defined.')
@_remove_axiom.register
def _(axiom: OWLObjectPropertyCharacteristicAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
property_x = conv._to_owlready2_property(axiom.get_property())
if property_x is None:
return
if isinstance(axiom, OWLFunctionalObjectPropertyAxiom) and owlready2.FunctionalProperty in property_x.is_a:
property_x.is_a.remove(owlready2.FunctionalProperty)
elif isinstance(axiom, OWLAsymmetricObjectPropertyAxiom) and owlready2.AsymmetricProperty in property_x.is_a:
property_x.is_a.remove(owlready2.AsymmetricProperty)
elif isinstance(axiom, OWLInverseFunctionalObjectPropertyAxiom) \
and owlready2.InverseFunctionalProperty in property_x.is_a:
property_x.is_a.remove(owlready2.InverseFunctionalProperty)
elif isinstance(axiom, OWLIrreflexiveObjectPropertyAxiom) and owlready2.IrreflexiveProperty in property_x.is_a:
property_x.is_a.remove(owlready2.IrreflexiveProperty)
elif isinstance(axiom, OWLReflexiveObjectPropertyAxiom) and owlready2.ReflexiveProperty in property_x.is_a:
property_x.is_a.remove(owlready2.ReflexiveProperty)
elif isinstance(axiom, OWLSymmetricObjectPropertyAxiom) and owlready2.SymmetricProperty in property_x.is_a:
property_x.is_a.remove(owlready2.SymmetricProperty)
elif isinstance(axiom, OWLTransitiveObjectPropertyAxiom) and owlready2.TransitiveProperty in property_x.is_a:
property_x.is_a.remove(owlready2.TransitiveProperty)
else:
raise ValueError(f'OWLObjectPropertyCharacteristicAxiom ({axiom}) is not defined.')
@_remove_axiom.register
def _(axiom: OWLDataPropertyCharacteristicAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
with ont_x:
property_x = conv._to_owlready2_property(axiom.get_property())
if property_x is not None and isinstance(axiom, OWLFunctionalDataPropertyAxiom) \
and owlready2.FunctionalProperty in property_x.is_a:
property_x.is_a.remove(owlready2.FunctionalProperty)
@_remove_axiom.register
def _(axiom: OWLSubPropertyChainAxiom, ontology: AbstractOWLOntology, world: owlready2.namespace.World):
conv = ToOwlready2(world)
ont_x: owlready2.Ontology = conv.map_object(ontology)
super_property = axiom.get_super_property()
property_chain = list(axiom.get_property_chain())
with ont_x:
super_property_x = conv._to_owlready2_property(super_property)
if super_property_x is None:
return
property_chain_x = [conv._to_owlready2_property(prop) for prop in property_chain]
if not all(property_chain_x):
return
# Find and remove the matching property chain
for pc in super_property_x.property_chain:
if list(pc.properties) == property_chain_x:
super_property_x.property_chain.remove(pc)
break
[docs]
class Ontology(AbstractOWLOntology):
__slots__ = '_iri', '_world', '_onto', 'is_modified'
_onto: owlready2.Ontology
is_modified: bool
def __init__(self, ontology_iri: IRI | str, load: bool = True, world_store=None):
"""Represents an Ontology in Ontolearn.
Args:
ontology_iri: IRI of the ontology.
load: Whether to load the ontology or not.
"""
self._iri = ontology_iri
if world_store is None:
self._world = owlready2.World()
else:
self._world = owlready2.World(filename=world_store)
self.is_modified = False
if isinstance(ontology_iri, str):
onto = self._world.get_ontology(ontology_iri)
self._iri = IRI.create(ontology_iri)
else:
onto = self._world.get_ontology(ontology_iri.as_str())
if load:
onto = onto.load()
self._onto = onto
[docs]
def __len__(self) -> int:
return len([t for t in self._onto.get_triples()])
[docs]
def classes_in_signature(self) -> Iterable[OWLClass]:
for c in self._onto.classes():
yield OWLClass(IRI.create(c.iri))
[docs]
def data_properties_in_signature(self) -> Iterable[OWLDataProperty]:
for dp in self._onto.data_properties():
yield OWLDataProperty(IRI.create(dp.iri))
[docs]
def object_properties_in_signature(self) -> Iterable[OWLObjectProperty]:
for op in self._onto.object_properties():
yield OWLObjectProperty(IRI.create(op.iri))
[docs]
def properties_in_signature(self) -> Iterable[OWLProperty]:
yield from self.object_properties_in_signature()
yield from self.data_properties_in_signature()
[docs]
def individuals_in_signature(self) -> Iterable[OWLNamedIndividual]:
for i in self._onto.individuals():
yield OWLNamedIndividual(IRI.create(i.iri))
[docs]
def get_abox_axioms(self) -> Iterable:
"""Get all ABox (assertional) axioms: class assertions, object- and
data-property assertions."""
results = list(self.get_abox_axioms_between_individuals_and_classes())
results.extend(self.get_abox_axioms_between_individuals())
for i_x in self._onto.individuals():
subject = OWLNamedIndividual(IRI.create(i_x.iri))
for prop_x in i_x.get_properties():
if isinstance(prop_x, owlready2.DataPropertyClass):
property_ = OWLDataProperty(IRI.create(prop_x.iri))
for value in prop_x[i_x]:
results.append(OWLDataPropertyAssertionAxiom(subject, property_, OWLLiteral(value)))
return results
[docs]
def get_tbox_axioms(self) -> Iterable:
"""Get all TBox (schema-level) axioms: class declarations, SubClassOf,
EquivalentClasses and DisjointClasses axioms."""
results = []
for c in self.classes_in_signature():
results.append(OWLDeclarationAxiom(c))
c_x: owlready2.ThingClass = self._world[c.str]
for parent_x in c_x.is_a:
if parent_x is owlready2.Thing:
continue
if isinstance(parent_x, (owlready2.ThingClass, owlready2.ClassConstruct)):
results.append(OWLSubClassOfAxiom(c, _parse_concept_to_owlapy(parent_x)))
results.extend(self.equivalent_classes_axioms(c))
for ca in self._onto.general_class_axioms():
results.extend(
OWLSubClassOfAxiom(_parse_concept_to_owlapy(ca.left_side), _parse_concept_to_owlapy(super_x))
for super_x in ca.is_a
)
for disjoints_x in self._onto.disjoint_classes():
results.append(OWLDisjointClassesAxiom([_parse_concept_to_owlapy(e) for e in disjoints_x.entities]))
return results
[docs]
def get_abox_axioms_between_individuals(self) -> Iterable:
"""Get all object-property assertion axioms, i.e. triples between two individuals."""
results = []
for i_x in self._onto.individuals():
subject = OWLNamedIndividual(IRI.create(i_x.iri))
for prop_x in i_x.get_properties():
if isinstance(prop_x, owlready2.ObjectPropertyClass):
property_ = OWLObjectProperty(IRI.create(prop_x.iri))
for obj_x in prop_x[i_x]:
obj = OWLNamedIndividual(IRI.create(obj_x.iri))
results.append(OWLObjectPropertyAssertionAxiom(subject, property_, obj))
return results
[docs]
def get_abox_axioms_between_individuals_and_classes(self) -> Iterable:
"""Get all class assertion axioms, i.e. triples of the form `individual rdf:type Class`."""
results = []
for i_x in self._onto.individuals():
ind = OWLNamedIndividual(IRI.create(i_x.iri))
for cls_x in i_x.is_a:
if isinstance(cls_x, (owlready2.ThingClass, owlready2.ClassConstruct)):
results.append(OWLClassAssertionAxiom(ind, _parse_concept_to_owlapy(cls_x)))
return results
# @TODO:CD:Unsure it is working
[docs]
def equivalent_classes_axioms(self, c: OWLClass) -> Iterable[OWLEquivalentClassesAxiom]:
c_x: owlready2.ThingClass = self._world[c.str]
# TODO: Should this also return EquivalentClasses general class axioms? Compare to java owlapi
for ec_x in c_x.equivalent_to:
yield OWLEquivalentClassesAxiom([c, _parse_concept_to_owlapy(ec_x)])
# @TODO:CD:Unsure it is working
[docs]
def general_class_axioms(self) -> Iterable[OWLClassAxiom]:
# TODO: At the moment owlready2 only supports SubClassOf general class axioms. (18.02.2023)
for ca in self._onto.general_class_axioms():
yield from (OWLSubClassOfAxiom(_parse_concept_to_owlapy(ca.left_side), _parse_concept_to_owlapy(c))
for c in ca.is_a)
[docs]
def get_ontology_id(self) -> OWLOntologyID:
onto_iri = self._world._unabbreviate(self._onto.storid)
look_version = self._world._get_obj_triple_sp_o(
self._onto.storid,
self._world._abbreviate(_VERSION_IRI.as_str()))
if look_version is not None:
version_iri = self._world._unabbreviate(look_version)
else:
version_iri = None
return OWLOntologyID(IRI.create(onto_iri) if onto_iri is not None else None,
IRI.create(version_iri) if version_iri is not None else None)
[docs]
def data_property_domain_axioms(self, pe: OWLDataProperty) -> Iterable[OWLDataPropertyDomainAxiom]:
p_x: owlready2.DataPropertyClass = self._world[pe.str]
domains = set(p_x.domains_indirect())
if len(domains) == 0:
yield OWLDataPropertyDomainAxiom(pe, OWLThing)
else:
for dom in domains:
if isinstance(dom, (owlready2.ThingClass, owlready2.ClassConstruct)):
yield OWLDataPropertyDomainAxiom(pe, _parse_concept_to_owlapy(dom))
else:
logger.warning("Construct %s not implemented at %s", dom, pe)
pass # XXX TODO
[docs]
def data_property_range_axioms(self, pe: OWLDataProperty) -> Iterable[OWLDataPropertyRangeAxiom]:
p_x: owlready2.DataPropertyClass = self._world[pe.str]
ranges = set(chain.from_iterable(super_prop.range for super_prop in p_x.ancestors()))
if len(ranges) == 0:
pass
# TODO
else:
for rng in ranges:
if rng in _Datatype_map:
yield OWLDataPropertyRangeAxiom(pe, _Datatype_map[rng])
elif isinstance(rng, owlready2.ClassConstruct):
yield OWLDataPropertyRangeAxiom(pe, _parse_datarange_to_owlapy(rng))
else:
logger.warning("Datatype %s not implemented at %s", rng, pe)
pass # XXX TODO
[docs]
def object_property_domain_axioms(self, pe: OWLObjectProperty) -> Iterable[OWLObjectPropertyDomainAxiom]:
p_x: owlready2.ObjectPropertyClass = self._world[pe.str]
domains = set(p_x.domains_indirect())
if len(domains) == 0:
yield OWLObjectPropertyDomainAxiom(pe, OWLThing)
else:
for dom in domains:
if isinstance(dom, (owlready2.ThingClass, owlready2.ClassConstruct)):
yield OWLObjectPropertyDomainAxiom(pe, _parse_concept_to_owlapy(dom))
else:
logger.warning("Construct %s not implemented at %s", dom, pe)
pass # XXX TODO
[docs]
def object_property_range_axioms(self, pe: OWLObjectProperty) -> Iterable[OWLObjectPropertyRangeAxiom]:
p_x: owlready2.ObjectPropertyClass = self._world[pe.str]
ranges = set(chain.from_iterable(super_prop.range for super_prop in p_x.ancestors()))
if len(ranges) == 0:
yield OWLObjectPropertyRangeAxiom(pe, OWLThing)
else:
for rng in ranges:
if isinstance(rng, (owlready2.ThingClass, owlready2.ClassConstruct)):
yield OWLObjectPropertyRangeAxiom(pe, _parse_concept_to_owlapy(rng))
else:
logger.warning("Construct %s not implemented at %s", rng, pe)
pass # XXX TODO
[docs]
def add_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
self.is_modified = True
if isinstance(axiom, OWLAxiom):
_add_axiom(axiom, self, self._world)
else:
for ax in axiom:
_add_axiom(ax, self, self._world)
[docs]
def remove_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
self.is_modified = True
if isinstance(axiom, OWLAxiom):
_remove_axiom(axiom, self, self._world)
else:
for ax in axiom:
_remove_axiom(ax, self, self._world)
[docs]
def save(self, path: Union[str, IRI] = None, inplace: bool = False, document_format: Optional[str] = None):
"""Save the ontology to a file, optionally in a different serialisation format.
owlready2 natively supports ``"rdfxml"`` and ``"ntriples"``. Any other
format is handled by loading the RDF/XML output into rdflib and
re-serialising there.
**owlready2-native formats:**
======================== ==========================================
Format string(s) Serialisation
======================== ==========================================
``"rdfxml"`` RDF/XML *(default)*
``"ntriples"`` / ``"nt"`` N-Triples
======================== ==========================================
**rdflib-backed formats** – saved as RDF/XML first, then converted:
============================= ==========================================
Format string(s) Serialisation
============================= ==========================================
``"turtle"`` / ``"ttl"`` Turtle
``"json-ld"`` / ``"jsonld"`` JSON-LD
``"nt11"`` N-Triples 1.1
``"n3"`` Notation3
``"trig"`` TriG
``"trix"`` TriX
``"nquads"`` / ``"nq"`` N-Quads
============================= ==========================================
Args:
path: File path where the ontology will be saved.
inplace: If ``True``, overwrite the file the ontology was loaded from.
document_format: Desired serialisation format (see tables above).
Defaults to ``"rdfxml"`` when not specified.
Raises:
ValueError: If an unsupported format string is provided.
"""
# convert it into str.
if isinstance(path, IRI):
path = path.as_str()
# Sanity checking
if inplace is False:
assert isinstance(path, str), (
f"path must be string if inplace is set to False. "
f"Current path is {type(path)}"
)
fmt_key = document_format.strip().lower() if document_format is not None else "rdfxml"
# Owlready2-native formats
_OWLREADY2_NATIVE = {"rdfxml": "rdfxml", "ntriples": "ntriples", "nt": "ntriples"}
owlready2_fmt = _OWLREADY2_NATIVE.get(fmt_key)
# Get the current ontology defined in the world.
ont_x: owlready2.namespace.Ontology
ont_x = self._world.get_ontology(self.get_ontology_id().get_ontology_iri().as_str())
if inplace:
save_path = self._iri.as_str() if os.path.exists(self._iri.as_str()) else "demo.owl"
logger.info(f"Saving {self} inplace to {save_path}...")
if owlready2_fmt is not None:
ont_x.save(file=save_path, format=owlready2_fmt)
else:
self._save_via_rdflib_owlready2(ont_x, save_path, fmt_key)
else:
logger.info(f"Saving {path}..")
if owlready2_fmt is not None:
ont_x.save(file=path, format=owlready2_fmt)
else:
self._save_via_rdflib_owlready2(ont_x, path, fmt_key)
def _save_via_rdflib_owlready2(self, ont_x: "owlready2.namespace.Ontology", path: str, fmt_key: str) -> None:
"""Save *ont_x* (an owlready2 Ontology) to *path* using rdflib.
Strategy:
1. Dump to a temporary RDF/XML file via owlready2.
2. Parse with rdflib.
3. Re-serialise in the requested format to *path*.
4. Delete the temporary file.
Args:
ont_x: The owlready2 ontology to save.
path: Destination file path.
fmt_key: Normalised (lower-case) format key from ``_RDFLIB_FORMATS``.
Raises:
ValueError: If *fmt_key* is not in ``_RDFLIB_FORMATS``.
"""
import tempfile
# "turtle"/"ttl" are not in _RDFLIB_FORMATS (SyncOntology routes them to the OWL API), but
# this legacy class has no OWL API and serialises Turtle through rdflib.
rdflib_format = {"turtle": "turtle", "ttl": "turtle"}.get(fmt_key) or _RDFLIB_FORMATS.get(fmt_key)
if rdflib_format is None:
all_supported = sorted(
{"rdfxml", "ntriples", "nt", "turtle", "ttl"} | set(_RDFLIB_FORMATS.keys())
)
raise ValueError(
f"Unsupported document format '{fmt_key}'. "
f"Supported formats: {', '.join(all_supported)}"
)
# Ensure parent directory exists
parent = os.path.dirname(os.path.abspath(path))
os.makedirs(parent, exist_ok=True)
# Step 1 – dump to a temp RDF/XML file
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".owl", prefix="_owlapy_tmp_")
os.close(tmp_fd)
try:
logger.info(f" (converting to '{rdflib_format}' via rdflib)")
ont_x.save(file=tmp_path, format="rdfxml")
# Step 2 – parse into the appropriate graph type
if rdflib_format in _RDFLIB_CONJUNCTIVE_FORMATS:
g = rdflib.ConjunctiveGraph()
else:
g = rdflib.Graph()
g.parse(tmp_path, format="xml")
# Step 3 – re-serialise
g.serialize(destination=path, format=rdflib_format)
finally:
# Step 4 – always remove the temp file
if os.path.exists(tmp_path):
os.remove(tmp_path)
[docs]
def get_original_iri(self):
"""Get the IRI argument that was used to create this ontology."""
return self._iri
[docs]
def __eq__(self, other):
if type(other) is type(self):
return self._onto.loaded == other._onto.loaded and self._onto.base_iri == other._onto.base_iri
return NotImplemented
[docs]
def __hash__(self):
return hash(("Ontology", self._onto.base_iri))
[docs]
def __repr__(self):
return f'Ontology({self._onto.base_iri}, loaded:{self._onto.loaded})'
[docs]
class SyncOntology(AbstractOWLOntology):
def __init__(self, path: Union[IRI, str], load: bool = True):
if not jpype.isJVMStarted():
startJVM()
# noinspection PyUnresolvedReferences
# noinspection PyUnresolvedReferences
from java.io import File
# noinspection PyUnresolvedReferences
from java.util.stream import Stream
from org.semanticweb.owlapi.apibinding import OWLManager
# noinspection PyUnresolvedReferences
from org.semanticweb.owlapi.model import IRI as owlapi_IRI
from owlapy.owlapi_mapper import OWLAPIMapper
self.owlapi_manager = OWLManager.createOWLOntologyManager()
self.path = path
self.load = load
if isinstance(path, IRI):
file_path = path.str
else:
file_path = path
if not load: # create new ontology
if isinstance(path, IRI):
self.owlapi_ontology = self.owlapi_manager.createOntology(Stream.empty(), owlapi_IRI.create(path.str))
else:
try:
self.owlapi_ontology = self.owlapi_manager.createOntology(Stream.empty(),
owlapi_IRI.create(path))
except Exception as e:
logger.error(f"Error creating ontology from path: {e}")
raise NotImplementedError("Cant initialize a new ontology using path. Use IRI instead")
else: # means we are loading an existing ontology
self.owlapi_ontology = self.owlapi_manager.loadOntologyFromOntologyDocument(File(file_path))
self.mapper = OWLAPIMapper()
[docs]
def __eq__(self, other):
if isinstance(other, SyncOntology):
try:
return self.owlapi_ontology.getOntologyID().equals(other.owlapi_ontology.getOntologyID())
except Exception:
# If JVM is not running, fall back to comparing IRI paths
return self.path == other.path
return False
[docs]
def __hash__(self):
return int(self.owlapi_ontology.getOntologyID().hashCode())
[docs]
def __repr__(self):
return f'SyncOntology({self.path}, loaded:{self.load})'
[docs]
def __len__(self):
return len([i for i in self.get_abox_axioms()] + [i for i in self.get_abox_axioms()])
[docs]
def classes_in_signature(self) -> Iterable[OWLClass]:
return self.mapper.map_(self.owlapi_ontology.getClassesInSignature())
[docs]
def data_properties_in_signature(self) -> Iterable[OWLDataProperty]:
return self.mapper.map_(self.owlapi_ontology.getDataPropertiesInSignature())
[docs]
def object_properties_in_signature(self) -> Iterable[OWLObjectProperty]:
return self.mapper.map_(self.owlapi_ontology.getObjectPropertiesInSignature())
[docs]
def individuals_in_signature(self) -> Iterable[OWLNamedIndividual]:
return self.mapper.map_(self.owlapi_ontology.getIndividualsInSignature())
def _to_owlapi_iri(self, entity: Union[IRI, str, OWLClass, OWLObjectProperty, OWLDataProperty,
OWLAnnotationProperty, OWLDatatype, OWLNamedIndividual]):
"""Coerce an IRI, IRI string, or named entity into an OWL API ``IRI`` object."""
if isinstance(entity, str):
entity = IRI.create(entity)
if not isinstance(entity, IRI):
entity = entity.iri
return self.mapper.map_(entity)
[docs]
def contains_class_in_signature(self, entity: Union[IRI, str, OWLClass],
include_imports_closure: bool = True) -> bool:
"""Check whether a class with this IRI is in the signature of this ontology.
Args:
entity: The class (or its IRI) to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if a class with this IRI is in the signature, False otherwise.
"""
return bool(self.owlapi_ontology.containsClassInSignature(
self._to_owlapi_iri(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def contains_object_property_in_signature(self, entity: Union[IRI, str, OWLObjectProperty],
include_imports_closure: bool = True) -> bool:
"""Check whether an object property with this IRI is in the signature of this ontology.
Args:
entity: The object property (or its IRI) to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if an object property with this IRI is in the signature, False otherwise.
"""
return bool(self.owlapi_ontology.containsObjectPropertyInSignature(
self._to_owlapi_iri(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def contains_data_property_in_signature(self, entity: Union[IRI, str, OWLDataProperty],
include_imports_closure: bool = True) -> bool:
"""Check whether a data property with this IRI is in the signature of this ontology.
Args:
entity: The data property (or its IRI) to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if a data property with this IRI is in the signature, False otherwise.
"""
return bool(self.owlapi_ontology.containsDataPropertyInSignature(
self._to_owlapi_iri(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def contains_annotation_property_in_signature(self, entity: Union[IRI, str, OWLAnnotationProperty],
include_imports_closure: bool = True) -> bool:
"""Check whether an annotation property with this IRI is in the signature of this ontology.
Args:
entity: The annotation property (or its IRI) to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if an annotation property with this IRI is in the signature, False otherwise.
"""
return bool(self.owlapi_ontology.containsAnnotationPropertyInSignature(
self._to_owlapi_iri(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def contains_individual_in_signature(self, entity: Union[IRI, str, OWLNamedIndividual],
include_imports_closure: bool = True) -> bool:
"""Check whether a named individual with this IRI is in the signature of this ontology.
Args:
entity: The named individual (or its IRI) to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if a named individual with this IRI is in the signature, False otherwise.
"""
return bool(self.owlapi_ontology.containsIndividualInSignature(
self._to_owlapi_iri(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def is_declared(self, entity: Union[OWLClass, OWLObjectProperty, OWLDataProperty,
OWLAnnotationProperty, OWLDatatype, OWLNamedIndividual],
include_imports_closure: bool = True) -> bool:
"""Check whether the given entity has a declaration axiom in this ontology.
Args:
entity: The class, object/data/annotation property, datatype, or named individual to check.
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
True if the entity is declared, False otherwise.
"""
return bool(self.owlapi_ontology.isDeclared(
self.mapper.map_(entity), self._get_imports_enum(include_imports_closure)))
[docs]
def equivalent_classes_axioms(self, c: OWLClass) -> Iterable[OWLEquivalentClassesAxiom]:
return self.mapper.map_(self.owlapi_ontology.getEquivalentClassesAxioms(self.mapper.map_(c)))
[docs]
def general_class_axioms(self) -> Iterable[OWLClassAxiom]:
return self.mapper.map_(self.owlapi_ontology.getGeneralClassAxioms())
[docs]
def data_property_domain_axioms(self, property: OWLDataProperty) -> Iterable[OWLDataPropertyDomainAxiom]:
return self.mapper.map_(self.owlapi_ontology.getDataPropertyDomainAxioms(self.mapper.map_(property)))
[docs]
def data_property_range_axioms(self, property: OWLDataProperty) -> Iterable[OWLDataPropertyRangeAxiom]:
return self.mapper.map_(self.owlapi_ontology.getDataPropertyRangeAxioms(self.mapper.map_(property)))
[docs]
def object_property_domain_axioms(self, property: OWLObjectProperty) -> Iterable[OWLObjectPropertyDomainAxiom]:
return self.mapper.map_(self.owlapi_ontology.getObjectPropertyDomainAxioms(self.mapper.map_(property)))
[docs]
def object_property_range_axioms(self, property: OWLObjectProperty) -> Iterable[OWLObjectPropertyRangeAxiom]:
return self.mapper.map_(self.owlapi_ontology.getObjectPropertyRangeAxioms(self.mapper.map_(property)))
[docs]
def annotation_assertion_axioms(self, entity: Union[OWLEntity, IRI]) -> Iterable[OWLAnnotationAssertionAxiom]:
"""Gets the annotation assertion axioms about the given entity in this ontology, e.g. its
`rdfs:label`/`rdfs:comment` or any other `owl:AnnotationProperty` assertion.
Args:
entity: The entity (or its IRI directly) to get annotation assertions for.
Returns:
Annotation assertion axioms whose subject is the given entity's IRI.
"""
subject_iri = entity if isinstance(entity, IRI) else entity.iri
return self.mapper.map_(self.owlapi_ontology.getAnnotationAssertionAxioms(self.mapper.map_(subject_iri)))
def _get_imports_enum(self, include_imports_closure: bool):
# noinspection PyUnresolvedReferences
from org.semanticweb.owlapi.model.parameters import Imports
if include_imports_closure:
imports = Imports.INCLUDED
else:
imports = Imports.EXCLUDED
return imports
[docs]
def get_signature(self, include_imports_closure: bool = True):
"""Gets the entities that are in the signature of this ontology.
Args:
include_imports_closure: Whether to include/exclude imports from searches.
Returns:
Entities in signature.
"""
return self.mapper.map_(self.owlapi_ontology.getSignature(self._get_imports_enum(include_imports_closure)))
[docs]
def get_punned_iris(self, include_imports_closure: bool = True) -> Iterable[IRI]:
"""Get the IRIs that are used for more than one entity type in this ontology (punning), e.g. an
IRI that is used both as a class and as a named individual.
Args:
include_imports_closure: Whether to include/exclude imports from the search.
Returns:
The punned IRIs.
"""
return self.mapper.map_(self.owlapi_ontology.getPunnedIRIs(self._get_imports_enum(include_imports_closure)))
[docs]
def get_axioms(self, include_imports_closure: bool = True) -> Iterable[OWLAxiom]:
"""Get all axioms in this ontology.
Args:
include_imports_closure: Whether to include/exclude imports from searches.
Returns:
All axioms in this ontology.
"""
return self.mapper.map_(self.owlapi_ontology.getAxioms(self._get_imports_enum(include_imports_closure)))
[docs]
def contains_axiom(self, axiom: OWLAxiom) -> bool:
"""Check whether this ontology contains the given axiom (imports excluded).
Args:
axiom: The axiom to check for.
Returns:
True if this ontology contains the axiom, False otherwise.
"""
return bool(self.owlapi_ontology.containsAxiom(self.mapper.map_(axiom)))
[docs]
def get_abox_axioms(self, include_imports_closure: bool = True) -> Iterable[OWLAxiom]:
"""Get all ABox axioms.
Args:
include_imports_closure: Whether to include/exclude imports from searches.
Returns:
ABox axioms.
"""
return self.mapper.map_(self.owlapi_ontology.getABoxAxioms(self._get_imports_enum(include_imports_closure)))
[docs]
def get_rbox_axioms(self, include_imports_closure: bool = True) -> Iterable[OWLAxiom]:
"""Get all RBox axioms.
Args:
include_imports_closure: Whether to include/exclude imports from searches.
Returns:
RBox axioms.
"""
return self.mapper.map_(self.owlapi_ontology.getRBoxAxioms(self._get_imports_enum(include_imports_closure)))
[docs]
def get_tbox_axioms(self, include_imports_closure: bool = True) -> Iterable[OWLAxiom]:
"""Get all TBox axioms.
Args:
include_imports_closure: Whether to include/exclude imports from searches.
Returns:
TBox axioms.
"""
return self.mapper.map_(self.owlapi_ontology.getTBoxAxioms(self._get_imports_enum(include_imports_closure)))
[docs]
def get_owlapi_ontology(self):
return self.owlapi_ontology
[docs]
def get_dl_expressivity(self) -> str:
"""Compute the DL expressivity name of this ontology, e.g. "ALCHN(D)"."""
from owlapy.expressivity import get_dl_expressivity
return get_dl_expressivity(self)
def _get_prefix_format(self):
"""Return the OWL API document format currently associated with this ontology.
This is the live format object tracked by the OWL API manager, so mutating it
(via ``setPrefix``/``setPrefixManager``) persists across calls and is picked up
by ``save()`` whenever no explicit ``document_format`` override is requested.
"""
return self.owlapi_manager.getOntologyFormat(self.owlapi_ontology)
[docs]
def get_prefixes(self) -> Dict[str, str]:
"""Get the prefix -> namespace IRI mappings currently registered for this ontology.
These prefixes are honoured by :meth:`save` both for the OWL API–backed formats
that support them (RDF/XML, OWL/XML, Turtle, Functional Syntax, Manchester Syntax)
and for the rdflib-backed formats (Turtle, N3, TriG, JSON-LD).
Returns:
Mapping of prefix name (without a trailing colon) to namespace IRI.
"""
fmt = self._get_prefix_format()
if not hasattr(fmt, "getPrefixName2PrefixMap"):
return {}
return {
(str(name)[:-1] if str(name).endswith(":") else str(name)): str(ns)
for name, ns in fmt.getPrefixName2PrefixMap().items()
}
[docs]
def set_prefix(self, prefix: str, namespace: str) -> None:
"""Declare or update a prefix -> namespace IRI mapping for this ontology.
Args:
prefix: Short prefix name, e.g. ``"foaf"`` (no trailing colon).
namespace: Namespace IRI the prefix expands to, e.g.
``"http://xmlns.com/foaf/0.1/"``.
Raises:
ValueError: If the ontology's current document format does not support
prefixes (e.g. LaTeX, DL Syntax, KRSS2, OBO).
"""
fmt = self._get_prefix_format()
if not hasattr(fmt, "setPrefix"):
raise ValueError(f"The ontology's current document format ({fmt}) does not support prefixes.")
fmt.setPrefix(f"{prefix}:", namespace)
[docs]
def remove_prefix(self, prefix: str) -> None:
"""Remove a previously declared prefix mapping, if present.
Args:
prefix: Short prefix name to remove (no trailing colon).
Raises:
ValueError: If the ontology's current document format does not support
prefixes (e.g. LaTeX, DL Syntax, KRSS2, OBO).
"""
# noinspection PyUnresolvedReferences
from org.semanticweb.owlapi.util import DefaultPrefixManager
fmt = self._get_prefix_format()
if not hasattr(fmt, "setPrefixManager"):
raise ValueError(f"The ontology's current document format ({fmt}) does not support prefixes.")
key = f"{prefix}:"
remaining = DefaultPrefixManager()
for name, ns in fmt.getPrefixName2PrefixMap().items():
if str(name) != key:
remaining.setPrefix(str(name), str(ns))
fmt.setPrefixManager(remaining)
[docs]
def get_ontology_id(self) -> OWLOntologyID:
return self.mapper.map_(self.owlapi_ontology.getOntologyID())
[docs]
def add_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
if isinstance(axiom, OWLAxiom):
self.owlapi_ontology.addAxiom(self.mapper.map_(axiom))
else:
self.owlapi_ontology.addAxioms(self.mapper.map_(axiom))
[docs]
def remove_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
if isinstance(axiom, OWLAxiom):
self.owlapi_ontology.removeAxiom(self.mapper.map_(axiom))
else:
self.owlapi_ontology.removeAxioms(self.mapper.map_(axiom))
[docs]
def save(self, path: str = None, document_iri: Optional[IRI] = None,
document_format: Optional[str] = None):
"""Save the ontology to a file, optionally specifying the output format.
**OWL API–backed formats** (case-insensitive):
========================= ==========================================
Format string(s) Serialisation
========================= ==========================================
``"rdf/xml"`` / ``"rdfxml"`` RDF/XML
``"owl/xml"`` / ``"owlxml"`` OWL/XML
``"turtle"`` / ``"ttl"`` Turtle (OWL API)
``"functional"`` / ``"fss"`` OWL Functional Syntax
``"manchester"`` / ``"ms"`` Manchester OWL Syntax
``"latex"`` LaTeX
``"dlsyntax"`` / ``"dl"`` DL Syntax
``"krss2"`` KRSS2
``"obo"`` OBO
========================= ==========================================
**rdflib-backed formats** – the ontology is first saved as RDF/XML by
OWL API to a temporary file, loaded by rdflib, re-serialised in the
requested format, then the temporary file is deleted:
============================= ==========================================
Format string(s) Serialisation
============================= ==========================================
``"turtle2"`` Turtle (rdflib)
``"json-ld"`` / ``"jsonld"`` JSON-LD
``"ntriples"`` / ``"nt"`` N-Triples
``"nt11"`` N-Triples 1.1
``"n3"`` Notation3
``"trig"`` TriG
``"trix"`` TriX
``"nquads"`` / ``"nq"`` N-Quads
============================= ==========================================
If *document_format* is ``None`` the ontology's current format is kept.
Args:
path: File path where the ontology will be saved.
document_iri: Reserved for future use; must be ``None``.
document_format: Desired serialisation format (see tables above).
Raises:
ValueError: If an unsupported format string is provided.
"""
assert isinstance(path, str), "Path must be a string"
if document_iri is not None:
raise NotImplementedError("document_iri must be None for the time being")
# Ensure parent directory exists
parent = os.path.dirname(os.path.abspath(path))
os.makedirs(parent, exist_ok=True)
fmt_key = document_format.strip().lower() if document_format is not None else None
# ── rdflib-backed path ──────────────────────────────────────────────
if fmt_key is not None and fmt_key in _RDFLIB_FORMATS:
self._save_via_rdflib(path, _RDFLIB_FORMATS[fmt_key])
return
# ── OWL API path ────────────────────────────────────────────────────
# noinspection PyUnresolvedReferences
# noinspection PyUnresolvedReferences
import org.semanticweb.owlapi.formats
from java.io import File, FileOutputStream
if fmt_key is not None:
fmt_class_name = _DOCUMENT_FORMATS.get(fmt_key)
if fmt_class_name is None:
all_supported = sorted(
set(_DOCUMENT_FORMATS.keys()) | set(_RDFLIB_FORMATS.keys())
)
raise ValueError(
f"Unsupported document format '{document_format}'. "
f"Supported formats: {', '.join(all_supported)}"
)
fmt_class = getattr(org.semanticweb.owlapi.formats, fmt_class_name)
owlapi_format = fmt_class()
# Carry over any custom prefixes registered via set_prefix()/remove_prefix()
# onto the freshly created format instance.
current_format = self.owlapi_manager.getOntologyFormat(self.owlapi_ontology)
if hasattr(owlapi_format, "copyPrefixesFrom") and hasattr(current_format, "getPrefixName2PrefixMap"):
owlapi_format.copyPrefixesFrom(current_format)
else:
owlapi_format = self.owlapi_manager.getOntologyFormat(self.owlapi_ontology)
logger.info(f"Saving Ontology into {path}")
# Always use FileOutputStream so that every format (including non-RDF
# text-based storers like DL Syntax, KRSS2, LaTeX, Manchester) reliably
# writes to the requested file rather than stdout.
with FileOutputStream(File(path)) as fos:
self.owlapi_manager.saveOntology(self.owlapi_ontology, owlapi_format, fos)
def _save_via_rdflib(self, path: str, rdflib_format: str) -> None:
"""Save using rdflib as the serialiser.
Strategy:
1. Save the ontology to a *temporary* RDF/XML file via OWL API.
2. Parse that file with rdflib.
3. Serialise with rdflib in the requested format to *path*.
4. Delete the temporary file.
Formats that require a context-aware store (N-Quads, TriX) are
loaded into a ``ConjunctiveGraph``; all others use a plain ``Graph``.
Args:
path: Destination file path.
rdflib_format: rdflib format name (e.g. ``"ntriples"``, ``"trig"``).
"""
import tempfile
import rdflib
# noinspection PyUnresolvedReferences
from java.io import File, FileOutputStream
# noinspection PyUnresolvedReferences
from org.semanticweb.owlapi.formats import RDFXMLDocumentFormat
# Step 1 – dump to a temp RDF/XML file
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".owl", prefix="_owlapy_tmp_")
os.close(tmp_fd)
try:
logger.info(f"Saving Ontology into {path} (via rdflib '{rdflib_format}')")
with FileOutputStream(File(tmp_path)) as fos:
self.owlapi_manager.saveOntology(
self.owlapi_ontology, RDFXMLDocumentFormat(), fos
)
# Step 2 – parse into the appropriate graph type
if rdflib_format in _RDFLIB_CONJUNCTIVE_FORMATS:
g = rdflib.ConjunctiveGraph()
else:
g = rdflib.Graph()
g.parse(tmp_path, format="xml")
# Bind any custom prefixes registered via set_prefix() so the
# re-serialised output uses them instead of full IRIs / auto-generated ones.
for prefix, namespace in self.get_prefixes().items():
g.bind(prefix, rdflib.Namespace(namespace), override=True)
# Step 3 – re-serialise
g.serialize(destination=path, format=rdflib_format)
finally:
# Step 4 – always remove the temp file
if os.path.exists(tmp_path):
os.remove(tmp_path)
def _rdflib_literal_to_owl_literal(literal: "rdflib.term.Literal") -> OWLLiteral:
"""Convert an rdflib ``Literal`` into an :class:`OWLLiteral`, preserving its datatype where possible."""
try:
return OWLLiteral(literal.toPython())
except NotImplementedError:
# Datatypes rdflib can't map to a native Python type (e.g. xsd:duration, or a
# custom/unrecognized datatype URI, which toPython() returns unchanged) fall back
# to a plain string literal rather than dropping the value.
return OWLLiteral(str(literal), StringOWLDatatype)
def _owl_literal_to_rdflib_literal(literal: OWLLiteral) -> "rdflib.term.Literal":
"""Convert an :class:`OWLLiteral` into an rdflib ``Literal``, the inverse of
:func:`_rdflib_literal_to_owl_literal`."""
return rdflib.Literal(literal.to_python(), datatype=rdflib.URIRef(literal.get_datatype().str))
# owl: vocabulary term for each property-characteristic axiom type RDFLibOntology's write API
# supports. OWLFunctionalObjectPropertyAxiom and OWLFunctionalDataPropertyAxiom map to the same
# owl:FunctionalProperty class -- OWL doesn't distinguish "functional" at the RDF level, the
# property's own declared type (owl:ObjectProperty vs owl:DatatypeProperty) does that.
_PROPERTY_CHARACTERISTIC_TERMS: Final = MappingProxyType({
OWLFunctionalObjectPropertyAxiom: rdflib.OWL.FunctionalProperty,
OWLFunctionalDataPropertyAxiom: rdflib.OWL.FunctionalProperty,
OWLInverseFunctionalObjectPropertyAxiom: rdflib.OWL.InverseFunctionalProperty,
OWLSymmetricObjectPropertyAxiom: rdflib.OWL.SymmetricProperty,
OWLAsymmetricObjectPropertyAxiom: rdflib.OWL.AsymmetricProperty,
OWLTransitiveObjectPropertyAxiom: rdflib.OWL.TransitiveProperty,
OWLReflexiveObjectPropertyAxiom: rdflib.OWL.ReflexiveProperty,
OWLIrreflexiveObjectPropertyAxiom: rdflib.OWL.IrreflexiveProperty,
})
def _declare_entity(ontology, str_iri: str, rdf_type: "rdflib.term.URIRef", cache_list: list) -> None:
ontology.rdflib_graph.add((rdflib.URIRef(str_iri), rdflib.RDF.type, rdf_type))
if str_iri not in cache_list:
cache_list.append(str_iri)
def _undeclare_entity(ontology, str_iri: str, rdf_type: "rdflib.term.URIRef", cache_list: list) -> None:
ontology.rdflib_graph.remove((rdflib.URIRef(str_iri), rdflib.RDF.type, rdf_type))
if str_iri in cache_list:
cache_list.remove(str_iri)
def _declare_class(ontology, cls: OWLClass) -> None:
_declare_entity(ontology, cls.str, rdflib.OWL.Class, ontology.str_owl_classes)
def _declare_individual(ontology, ind: OWLNamedIndividual) -> None:
_declare_entity(ontology, ind.str, rdflib.OWL.NamedIndividual, ontology.str_owl_individuals)
def _declare_object_property(ontology, pe: OWLObjectProperty) -> None:
_declare_entity(ontology, pe.str, rdflib.OWL.ObjectProperty, ontology.str_owl_object_properties)
def _declare_data_property(ontology, pe: OWLDataProperty) -> None:
_declare_entity(ontology, pe.str, rdflib.OWL.DatatypeProperty, ontology.str_owl_data_properties)
def _add_property_characteristic(axiom: OWLAxiom, ontology) -> None:
term = _PROPERTY_CHARACTERISTIC_TERMS.get(type(axiom))
if term is None:
raise NotImplementedError(f"{type(axiom).__name__} is not supported by RDFLibOntology's write API yet.")
pe = axiom.get_property()
if isinstance(pe, OWLObjectProperty):
_declare_object_property(ontology, pe)
elif isinstance(pe, OWLDataProperty):
_declare_data_property(ontology, pe)
else:
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named object/data property as the property "
"of a characteristic axiom; property expressions (e.g. inverses) aren't representable.")
ontology.rdflib_graph.add((rdflib.URIRef(pe.str), rdflib.RDF.type, term))
def _remove_property_characteristic(axiom: OWLAxiom, ontology) -> None:
term = _PROPERTY_CHARACTERISTIC_TERMS.get(type(axiom))
if term is None:
raise NotImplementedError(f"{type(axiom).__name__} is not supported by RDFLibOntology's write API yet.")
ontology.rdflib_graph.remove((rdflib.URIRef(axiom.get_property().str), rdflib.RDF.type, term))
@singledispatch
def _rdflib_add_axiom(axiom: OWLAxiom, ontology) -> None:
raise NotImplementedError(
f"RDFLibOntology.add_axiom() does not support {type(axiom).__name__} yet. Supported: "
f"OWLDeclarationAxiom, OWLClassAssertionAxiom, OWLObjectPropertyAssertionAxiom, "
f"OWLDataPropertyAssertionAxiom, OWLSubClassOfAxiom, OWLEquivalentClassesAxiom, "
f"OWLDisjointClassesAxiom, OWLSubObjectPropertyOfAxiom, OWLSubDataPropertyOfAxiom, "
f"OWL{{Object,Data}}PropertyDomainAxiom, OWL{{Object,Data}}PropertyRangeAxiom, and the "
f"property characteristic axioms (Functional/InverseFunctional/Symmetric/Asymmetric/"
f"Transitive/Reflexive/Irreflexive) -- all between/on *named* entities only.")
@_rdflib_add_axiom.register
def _(axiom: OWLDeclarationAxiom, ontology) -> None:
entity = axiom.get_entity()
if isinstance(entity, OWLClass):
_declare_class(ontology, entity)
elif isinstance(entity, OWLNamedIndividual):
_declare_individual(ontology, entity)
elif isinstance(entity, OWLObjectProperty):
_declare_object_property(ontology, entity)
elif isinstance(entity, OWLDataProperty):
_declare_data_property(ontology, entity)
else:
raise NotImplementedError(
f"RDFLibOntology.add_axiom() cannot declare entities of type {type(entity).__name__}; "
f"supported: OWLClass, OWLNamedIndividual, OWLObjectProperty, OWLDataProperty.")
@_rdflib_add_axiom.register
def _(axiom: OWLClassAssertionAxiom, ontology) -> None:
ind, ce = axiom.get_individual(), axiom.get_class_expression()
if not isinstance(ind, OWLNamedIndividual) or not isinstance(ce, OWLClass):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLClassAssertionAxiom between a named "
"individual and a named class; complex class expressions/anonymous individuals aren't representable.")
_declare_individual(ontology, ind)
_declare_class(ontology, ce)
ontology.rdflib_graph.add((rdflib.URIRef(ind.str), rdflib.RDF.type, rdflib.URIRef(ce.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLObjectPropertyAssertionAxiom, ontology) -> None:
subj, prop, obj = axiom.get_subject(), axiom.get_property(), axiom.get_object()
if not isinstance(subj, OWLNamedIndividual) or not isinstance(prop, OWLObjectProperty) or not isinstance(obj, OWLNamedIndividual):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLObjectPropertyAssertionAxiom between named "
"individuals via a named object property; inverse/anonymous operands aren't representable.")
_declare_individual(ontology, subj)
_declare_individual(ontology, obj)
_declare_object_property(ontology, prop)
ontology.rdflib_graph.add((rdflib.URIRef(subj.str), rdflib.URIRef(prop.str), rdflib.URIRef(obj.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLDataPropertyAssertionAxiom, ontology) -> None:
subj, prop, lit = axiom.get_subject(), axiom.get_property(), axiom.get_object()
if not isinstance(subj, OWLNamedIndividual) or not isinstance(prop, OWLDataProperty):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLDataPropertyAssertionAxiom on a named "
"individual via a named data property.")
_declare_individual(ontology, subj)
_declare_data_property(ontology, prop)
ontology.rdflib_graph.add((rdflib.URIRef(subj.str), rdflib.URIRef(prop.str), _owl_literal_to_rdflib_literal(lit)))
@_rdflib_add_axiom.register
def _(axiom: OWLSubClassOfAxiom, ontology) -> None:
sub, sup = axiom.get_sub_class(), axiom.get_super_class()
if not isinstance(sub, OWLClass) or not isinstance(sup, OWLClass):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLSubClassOfAxiom between two named classes; "
"general class axioms (complex sub-class expressions) aren't representable.")
_declare_class(ontology, sub)
_declare_class(ontology, sup)
ontology.rdflib_graph.add((rdflib.URIRef(sub.str), rdflib.RDFS.subClassOf, rdflib.URIRef(sup.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLEquivalentClassesAxiom, ontology) -> None:
operands = list(axiom.class_expressions())
if len(operands) < 2 or not all(isinstance(c, OWLClass) for c in operands):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLEquivalentClassesAxiom between two or more named classes.")
for c in operands:
_declare_class(ontology, c)
for left, right in zip(operands, operands[1:]):
ontology.rdflib_graph.add((rdflib.URIRef(left.str), rdflib.OWL.equivalentClass, rdflib.URIRef(right.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLDisjointClassesAxiom, ontology) -> None:
operands = list(axiom.class_expressions())
if len(operands) < 2 or not all(isinstance(c, OWLClass) for c in operands):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLDisjointClassesAxiom between two or more named classes.")
for c in operands:
_declare_class(ontology, c)
for i, left in enumerate(operands):
for right in operands[i + 1:]:
ontology.rdflib_graph.add((rdflib.URIRef(left.str), rdflib.OWL.disjointWith, rdflib.URIRef(right.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLSubPropertyAxiom, ontology) -> None:
sub, sup = axiom.get_sub_property(), axiom.get_super_property()
if isinstance(sub, OWLObjectProperty) and isinstance(sup, OWLObjectProperty):
_declare_object_property(ontology, sub)
_declare_object_property(ontology, sup)
elif isinstance(sub, OWLDataProperty) and isinstance(sup, OWLDataProperty):
_declare_data_property(ontology, sub)
_declare_data_property(ontology, sup)
else:
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports OWLSubPropertyAxiom between two named object "
"properties or two named data properties; property chains/inverses aren't representable.")
ontology.rdflib_graph.add((rdflib.URIRef(sub.str), rdflib.RDFS.subPropertyOf, rdflib.URIRef(sup.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLPropertyDomainAxiom, ontology) -> None:
pe, dom = axiom.get_property(), axiom.get_domain()
if not isinstance(dom, OWLClass):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named class as the domain of "
"OWLObjectPropertyDomainAxiom/OWLDataPropertyDomainAxiom.")
if isinstance(pe, OWLObjectProperty):
_declare_object_property(ontology, pe)
elif isinstance(pe, OWLDataProperty):
_declare_data_property(ontology, pe)
else:
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named object/data property as the property of a domain axiom.")
_declare_class(ontology, dom)
ontology.rdflib_graph.add((rdflib.URIRef(pe.str), rdflib.RDFS.domain, rdflib.URIRef(dom.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLPropertyRangeAxiom, ontology) -> None:
pe, rng = axiom.get_property(), axiom.get_range()
if isinstance(pe, OWLObjectProperty):
if not isinstance(rng, OWLClass):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named class as the range of OWLObjectPropertyRangeAxiom.")
_declare_object_property(ontology, pe)
elif isinstance(pe, OWLDataProperty):
if not isinstance(rng, OWLDatatype):
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named datatype as the range of "
"OWLDataPropertyRangeAxiom; complex data ranges aren't representable.")
_declare_data_property(ontology, pe)
else:
raise NotImplementedError(
"RDFLibOntology.add_axiom() only supports a named object/data property as the property of a range axiom.")
ontology.rdflib_graph.add((rdflib.URIRef(pe.str), rdflib.RDFS.range, rdflib.URIRef(rng.str)))
@_rdflib_add_axiom.register
def _(axiom: OWLObjectPropertyCharacteristicAxiom, ontology) -> None:
_add_property_characteristic(axiom, ontology)
@_rdflib_add_axiom.register
def _(axiom: OWLDataPropertyCharacteristicAxiom, ontology) -> None:
_add_property_characteristic(axiom, ontology)
@singledispatch
def _rdflib_remove_axiom(axiom: OWLAxiom, ontology) -> None:
raise NotImplementedError(f"RDFLibOntology.remove_axiom() does not support {type(axiom).__name__} yet.")
@_rdflib_remove_axiom.register
def _(axiom: OWLDeclarationAxiom, ontology) -> None:
entity = axiom.get_entity()
if isinstance(entity, OWLClass):
_undeclare_entity(ontology, entity.str, rdflib.OWL.Class, ontology.str_owl_classes)
elif isinstance(entity, OWLNamedIndividual):
_undeclare_entity(ontology, entity.str, rdflib.OWL.NamedIndividual, ontology.str_owl_individuals)
elif isinstance(entity, OWLObjectProperty):
_undeclare_entity(ontology, entity.str, rdflib.OWL.ObjectProperty, ontology.str_owl_object_properties)
elif isinstance(entity, OWLDataProperty):
_undeclare_entity(ontology, entity.str, rdflib.OWL.DatatypeProperty, ontology.str_owl_data_properties)
else:
raise NotImplementedError(
f"RDFLibOntology.remove_axiom() cannot un-declare entities of type {type(entity).__name__}.")
@_rdflib_remove_axiom.register
def _(axiom: OWLClassAssertionAxiom, ontology) -> None:
ind, ce = axiom.get_individual(), axiom.get_class_expression()
if not isinstance(ind, OWLNamedIndividual) or not isinstance(ce, OWLClass):
raise NotImplementedError(
"RDFLibOntology.remove_axiom() only supports OWLClassAssertionAxiom between a named "
"individual and a named class.")
ontology.rdflib_graph.remove((rdflib.URIRef(ind.str), rdflib.RDF.type, rdflib.URIRef(ce.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLObjectPropertyAssertionAxiom, ontology) -> None:
subj, prop, obj = axiom.get_subject(), axiom.get_property(), axiom.get_object()
ontology.rdflib_graph.remove((rdflib.URIRef(subj.str), rdflib.URIRef(prop.str), rdflib.URIRef(obj.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLDataPropertyAssertionAxiom, ontology) -> None:
subj, prop, lit = axiom.get_subject(), axiom.get_property(), axiom.get_object()
ontology.rdflib_graph.remove((rdflib.URIRef(subj.str), rdflib.URIRef(prop.str), _owl_literal_to_rdflib_literal(lit)))
@_rdflib_remove_axiom.register
def _(axiom: OWLSubClassOfAxiom, ontology) -> None:
sub, sup = axiom.get_sub_class(), axiom.get_super_class()
if not isinstance(sub, OWLClass) or not isinstance(sup, OWLClass):
raise NotImplementedError("RDFLibOntology.remove_axiom() only supports OWLSubClassOfAxiom between two named classes.")
ontology.rdflib_graph.remove((rdflib.URIRef(sub.str), rdflib.RDFS.subClassOf, rdflib.URIRef(sup.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLEquivalentClassesAxiom, ontology) -> None:
operands = list(axiom.class_expressions())
if len(operands) < 2 or not all(isinstance(c, OWLClass) for c in operands):
raise NotImplementedError("RDFLibOntology.remove_axiom() only supports OWLEquivalentClassesAxiom between named classes.")
# Remove both directions defensively: equivalence is symmetric and equivalent_classes_axioms()
# looks both ways, but add_axiom() only asserts a forward chain.
for left, right in zip(operands, operands[1:]):
ontology.rdflib_graph.remove((rdflib.URIRef(left.str), rdflib.OWL.equivalentClass, rdflib.URIRef(right.str)))
ontology.rdflib_graph.remove((rdflib.URIRef(right.str), rdflib.OWL.equivalentClass, rdflib.URIRef(left.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLDisjointClassesAxiom, ontology) -> None:
operands = list(axiom.class_expressions())
if len(operands) < 2 or not all(isinstance(c, OWLClass) for c in operands):
raise NotImplementedError("RDFLibOntology.remove_axiom() only supports OWLDisjointClassesAxiom between named classes.")
for i, left in enumerate(operands):
for right in operands[i + 1:]:
ontology.rdflib_graph.remove((rdflib.URIRef(left.str), rdflib.OWL.disjointWith, rdflib.URIRef(right.str)))
ontology.rdflib_graph.remove((rdflib.URIRef(right.str), rdflib.OWL.disjointWith, rdflib.URIRef(left.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLSubPropertyAxiom, ontology) -> None:
sub, sup = axiom.get_sub_property(), axiom.get_super_property()
ontology.rdflib_graph.remove((rdflib.URIRef(sub.str), rdflib.RDFS.subPropertyOf, rdflib.URIRef(sup.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLPropertyDomainAxiom, ontology) -> None:
pe, dom = axiom.get_property(), axiom.get_domain()
if not isinstance(dom, OWLClass):
raise NotImplementedError("RDFLibOntology.remove_axiom() only supports a named class as a domain axiom's domain.")
ontology.rdflib_graph.remove((rdflib.URIRef(pe.str), rdflib.RDFS.domain, rdflib.URIRef(dom.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLPropertyRangeAxiom, ontology) -> None:
pe, rng = axiom.get_property(), axiom.get_range()
if not isinstance(rng, (OWLClass, OWLDatatype)):
raise NotImplementedError(
"RDFLibOntology.remove_axiom() only supports a named class/datatype as a range axiom's range.")
ontology.rdflib_graph.remove((rdflib.URIRef(pe.str), rdflib.RDFS.range, rdflib.URIRef(rng.str)))
@_rdflib_remove_axiom.register
def _(axiom: OWLObjectPropertyCharacteristicAxiom, ontology) -> None:
_remove_property_characteristic(axiom, ontology)
@_rdflib_remove_axiom.register
def _(axiom: OWLDataPropertyCharacteristicAxiom, ontology) -> None:
_remove_property_characteristic(axiom, ontology)
# Annotation properties every OWL ontology may use without declaring them locally as
# `owl:AnnotationProperty` -- RDFLibOntology.annotation_assertion_axioms() also honours any
# predicate explicitly declared as such in the loaded graph.
_WELL_KNOWN_ANNOTATION_PREDICATES = frozenset((
rdflib.RDFS.label, rdflib.RDFS.comment, rdflib.RDFS.seeAlso, rdflib.RDFS.isDefinedBy,
rdflib.OWL.versionInfo, rdflib.OWL.deprecated, rdflib.OWL.priorVersion,
rdflib.OWL.backwardCompatibleWith, rdflib.OWL.incompatibleWith,
))
_RDFLIB_SCHEMA_TYPES = frozenset((
rdflib.OWL.Class, rdflib.RDFS.Class, rdflib.RDFS.Datatype, rdflib.RDF.Property,
rdflib.OWL.ObjectProperty, rdflib.OWL.DatatypeProperty, rdflib.OWL.AnnotationProperty,
rdflib.OWL.Ontology, rdflib.OWL.Restriction, rdflib.OWL.Axiom, rdflib.OWL.Annotation,
rdflib.OWL.AllDifferent, rdflib.OWL.AllDisjointClasses, rdflib.OWL.AllDisjointProperties,
rdflib.OWL.NegativePropertyAssertion, rdflib.OWL.DataRange, rdflib.RDF.List,
*_PROPERTY_CHARACTERISTIC_TERMS.values(),
))
[docs]
class RDFLibOntology(AbstractOWLOntology):
def __init__(self, path: Union[str, IRI], load: bool = True):
self._path = path.as_str() if isinstance(path, IRI) else path
self.is_modified = False
if load:
assert os.path.exists(self._path)
self.rdflib_graph = rdflib.Graph().parse(self._path)
self.str_owl_classes = [x.n3()[1:-1] for x in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.Class) if not isinstance(x, rdflib.term.BNode)]
self.str_owl_individuals = [x.n3()[1:-1] for x in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.NamedIndividual) if not isinstance(x, rdflib.term.BNode)]
self.str_owl_object_properties = [x.n3()[1:-1] for x in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.ObjectProperty) if not isinstance(x, rdflib.term.BNode)]
self.str_owl_data_properties = [x.n3()[1:-1] for x in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.DatatypeProperty) if not isinstance(x, rdflib.term.BNode)]
self.str_owl_annotation_properties = [x.n3()[1:-1] for x in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.AnnotationProperty) if not isinstance(x, rdflib.term.BNode)]
self._infer_individuals()
else: # create a blank ontology; `path`/`self._path` is treated as the new ontology's IRI
self.rdflib_graph = rdflib.Graph()
self.rdflib_graph.add((rdflib.URIRef(self._path), rdflib.RDF.type, rdflib.OWL.Ontology))
self.str_owl_classes = []
self.str_owl_individuals = []
self.str_owl_object_properties = []
self.str_owl_data_properties = []
self.str_owl_annotation_properties = []
def _annotation_predicates(self):
return _WELL_KNOWN_ANNOTATION_PREDICATES | {rdflib.URIRef(s) for s in self.str_owl_annotation_properties}
def _infer_individuals(self):
graph = self.rdflib_graph
schema_entities = {s for s, t in graph.subject_objects(rdflib.RDF.type) if t in _RDFLIB_SCHEMA_TYPES}
annotations = self._annotation_predicates()
object_properties = {rdflib.URIRef(s) for s in self.str_owl_object_properties}
data_properties = {rdflib.URIRef(s) for s in self.str_owl_data_properties}
individuals = set()
for subject, predicate, obj in graph:
if not isinstance(subject, rdflib.URIRef):
continue
if predicate == rdflib.RDF.type:
if obj not in _RDFLIB_SCHEMA_TYPES:
individuals.add(subject)
elif predicate in annotations:
continue
elif predicate in object_properties:
individuals.add(subject)
if isinstance(obj, rdflib.URIRef):
individuals.add(obj)
elif predicate in data_properties:
individuals.add(subject)
elif (not str(predicate).startswith((str(rdflib.RDF), str(rdflib.RDFS), str(rdflib.OWL)))
and subject not in schema_entities):
if isinstance(obj, rdflib.Literal):
individuals.add(subject)
elif isinstance(obj, rdflib.URIRef) and obj not in schema_entities:
individuals.update((subject, obj))
declared = set(self.str_owl_individuals)
self.str_owl_individuals.extend(str(ind) for ind in sorted(individuals) if str(ind) not in declared)
[docs]
def __len__(self) -> int:
return len(self.rdflib_graph)
[docs]
def get_tbox_axioms(self) -> Iterable[OWLAxiom]:
"""Get all TBox (schema-level) axioms this loader can represent: class declarations,
SubClassOf, EquivalentClasses and DisjointClasses axioms between *named* classes.
Predicates this loader doesn't understand as TBox structure (e.g. `rdfs:label`,
`rdfs:comment`, or any other annotation on a class) are silently skipped rather than
raising -- they aren't TBox axioms, and virtually every real-world ontology has them.
"""
results = []
for owl_class in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.Class):
if not isinstance(owl_class, rdflib.term.URIRef):
continue
cls = OWLClass(owl_class.n3()[1:-1])
results.append(OWLDeclarationAxiom(cls))
for super_cls in self.rdflib_graph.objects(owl_class, rdflib.RDFS.subClassOf):
if isinstance(super_cls, rdflib.term.URIRef):
results.append(OWLSubClassOfAxiom(sub_class=cls, super_class=OWLClass(super_cls.n3()[1:-1])))
for equiv_cls in self.rdflib_graph.objects(owl_class, rdflib.OWL.equivalentClass):
if isinstance(equiv_cls, rdflib.term.URIRef):
results.append(OWLEquivalentClassesAxiom([cls, OWLClass(equiv_cls.n3()[1:-1])]))
for disjoint_cls in self.rdflib_graph.objects(owl_class, rdflib.OWL.disjointWith):
if isinstance(disjoint_cls, rdflib.term.URIRef):
results.append(OWLDisjointClassesAxiom([cls, OWLClass(disjoint_cls.n3()[1:-1])]))
return results
[docs]
def get_abox_axioms(self) -> Iterable:
"""Get all ABox (assertional) axioms this loader can represent: class assertions,
object-property assertions, and data-property assertions for every named individual.
Unlike earlier versions of this method, a class-assertion target does not need to be
independently declared `rdf:type owl:Class` (OWL doesn't require that for `i rdf:type C`
to be valid), and an object-property target does not need to be independently declared
`rdf:type owl:NamedIndividual` -- neither omission is an error, so neither raises anymore.
"""
results = []
annotations = self._annotation_predicates()
for str_iri in self.str_owl_individuals:
owl_individual = rdflib.URIRef(str_iri)
subject = OWLNamedIndividual(str_iri)
for (_, p, o) in self.rdflib_graph.triples(triple=(owl_individual, None, None)):
if isinstance(o, rdflib.term.BNode) or p in annotations:
continue
if p == rdflib.RDF.type:
if o == rdflib.OWL.NamedIndividual or o in _RDFLIB_SCHEMA_TYPES:
continue
if isinstance(o, rdflib.term.URIRef):
results.append(OWLClassAssertionAxiom(subject, OWLClass(o.n3()[1:-1])))
elif isinstance(o, rdflib.term.Literal):
results.append(OWLDataPropertyAssertionAxiom(subject, OWLDataProperty(p.n3()[1:-1]),
_rdflib_literal_to_owl_literal(o)))
elif isinstance(o, rdflib.term.URIRef):
results.append(OWLObjectPropertyAssertionAxiom(subject, OWLObjectProperty(p.n3()[1:-1]),
OWLNamedIndividual(o.n3()[1:-1])))
return results
[docs]
def classes_in_signature(self) -> Iterable[OWLClass]:
for str_iri in self.str_owl_classes:
yield OWLClass(str_iri)
[docs]
def data_properties_in_signature(self) -> Iterable[OWLDataProperty]:
for str_iri in self.str_owl_data_properties:
yield OWLDataProperty(str_iri)
[docs]
def object_properties_in_signature(self) -> Iterable[OWLObjectProperty]:
for str_iri in self.str_owl_object_properties:
yield OWLObjectProperty(str_iri)
[docs]
def properties_in_signature(self) -> Iterable[OWLProperty]:
yield from self.object_properties_in_signature()
yield from self.data_properties_in_signature()
[docs]
def individuals_in_signature(self) -> Iterable[OWLNamedIndividual]:
for str_iri in self.str_owl_individuals:
yield OWLNamedIndividual(str_iri)
[docs]
def get_abox_axioms_between_individuals(self) -> Iterable[OWLObjectPropertyAssertionAxiom]:
return [axiom for axiom in self.get_abox_axioms() if isinstance(axiom, OWLObjectPropertyAssertionAxiom)]
[docs]
def get_abox_axioms_between_individuals_and_classes(self) -> Iterable[OWLClassAssertionAxiom]:
return [axiom for axiom in self.get_abox_axioms() if isinstance(axiom, OWLClassAssertionAxiom)]
[docs]
def annotation_assertion_axioms(self, entity: Union[OWLEntity, IRI]) -> Iterable[OWLAnnotationAssertionAxiom]:
"""Gets the annotation assertion axioms about the given entity in this ontology, e.g. its
`rdfs:label`/`rdfs:comment`.
Recognizes the well-known annotation predicates (`rdfs:label`, `rdfs:comment`, `rdfs:seeAlso`,
`rdfs:isDefinedBy`, `owl:versionInfo`, `owl:deprecated`, `owl:priorVersion`,
`owl:backwardCompatibleWith`, `owl:incompatibleWith`) plus any predicate explicitly declared
`rdf:type owl:AnnotationProperty` in the loaded graph. A blank-node annotation value (an
anonymous individual) is skipped, matching how blank nodes are handled elsewhere in this class.
Args:
entity: The entity (or its IRI directly) to get annotation assertions for.
Returns:
Annotation assertion axioms whose subject is the given entity's IRI.
"""
subject_iri = entity if isinstance(entity, IRI) else entity.iri
subject_ref = rdflib.URIRef(subject_iri.str)
annotation_predicates = self._annotation_predicates()
results = []
for (_, p, o) in self.rdflib_graph.triples((subject_ref, None, None)):
if p not in annotation_predicates:
continue
if isinstance(o, rdflib.term.Literal):
value = _rdflib_literal_to_owl_literal(o)
elif isinstance(o, rdflib.term.URIRef):
value = IRI.create(o.n3()[1:-1])
else:
continue
results.append(OWLAnnotationAssertionAxiom(subject_iri, OWLAnnotation(OWLAnnotationProperty(p.n3()[1:-1]), value)))
return results
[docs]
def equivalent_classes_axioms(self, c: OWLClass) -> Iterable[OWLEquivalentClassesAxiom]:
c_uri = rdflib.URIRef(c.str)
seen = set()
for o in self.rdflib_graph.objects(c_uri, rdflib.OWL.equivalentClass):
if isinstance(o, rdflib.term.URIRef) and o not in seen:
seen.add(o)
yield OWLEquivalentClassesAxiom([c, OWLClass(o.n3()[1:-1])])
for s in self.rdflib_graph.subjects(rdflib.OWL.equivalentClass, c_uri):
if isinstance(s, rdflib.term.URIRef) and s not in seen:
seen.add(s)
yield OWLEquivalentClassesAxiom([c, OWLClass(s.n3()[1:-1])])
[docs]
def general_class_axioms(self) -> Iterable[OWLClassAxiom]:
# General class axioms are SubClassOf/EquivalentClasses/DisjointClasses axioms with a
# complex (non-atomic) class expression as an operand. Recognizing them requires
# reconstructing an OWLClassExpression tree from blank-node RDF structures
# (owl:Restriction, owl:intersectionOf, owl:unionOf, ...), which this triple-based loader
# does not do -- see get_tbox_axioms(), which likewise skips every blank-node operand and
# only understands axioms between two named classes.
raise NotImplementedError("RDFLibOntology does not support parsing complex (blank-node) "
"class expressions, so general class axioms cannot be retrieved.")
[docs]
def data_property_domain_axioms(self, pe: OWLDataProperty) -> Iterable[OWLDataPropertyDomainAxiom]:
pe_uri = rdflib.URIRef(pe.str)
for dom in self.rdflib_graph.objects(pe_uri, rdflib.RDFS.domain):
if isinstance(dom, rdflib.term.URIRef):
yield OWLDataPropertyDomainAxiom(pe, OWLClass(dom.n3()[1:-1]))
else:
logger.warning("Complex domain expression for %s is not supported by RDFLibOntology", pe)
[docs]
def data_property_range_axioms(self, pe: OWLDataProperty) -> Iterable[OWLDataPropertyRangeAxiom]:
pe_uri = rdflib.URIRef(pe.str)
for rng in self.rdflib_graph.objects(pe_uri, rdflib.RDFS.range):
if isinstance(rng, rdflib.term.URIRef):
yield OWLDataPropertyRangeAxiom(pe, OWLDatatype(rng.n3()[1:-1]))
else:
logger.warning("Complex range expression for %s is not supported by RDFLibOntology", pe)
[docs]
def object_property_domain_axioms(self, pe: OWLObjectProperty) -> Iterable[OWLObjectPropertyDomainAxiom]:
pe_uri = rdflib.URIRef(pe.str)
for dom in self.rdflib_graph.objects(pe_uri, rdflib.RDFS.domain):
if isinstance(dom, rdflib.term.URIRef):
yield OWLObjectPropertyDomainAxiom(pe, OWLClass(dom.n3()[1:-1]))
else:
logger.warning("Complex domain expression for %s is not supported by RDFLibOntology", pe)
[docs]
def object_property_range_axioms(self, pe: OWLObjectProperty) -> Iterable[OWLObjectPropertyRangeAxiom]:
pe_uri = rdflib.URIRef(pe.str)
for rng in self.rdflib_graph.objects(pe_uri, rdflib.RDFS.range):
if isinstance(rng, rdflib.term.URIRef):
yield OWLObjectPropertyRangeAxiom(pe, OWLClass(rng.n3()[1:-1]))
else:
logger.warning("Complex range expression for %s is not supported by RDFLibOntology", pe)
[docs]
def add_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
"""Add the given axiom(s) to this ontology, translating them into rdflib triples.
Only axioms between/on *named* entities are supported (matching the read API's scope --
see `get_tbox_axioms()`/`get_abox_axioms()`); axioms involving complex class/property
expressions raise `NotImplementedError`. Adding an axiom that references an entity not yet
declared in this ontology (e.g. an `OWLSubClassOfAxiom` between two classes that were never
independently declared `owl:Class`) implicitly declares those entities too, so the axiom
stays visible to `get_tbox_axioms()`/`classes_in_signature()`/etc. immediately afterwards.
"""
self.is_modified = True
if isinstance(axiom, OWLAxiom):
_rdflib_add_axiom(axiom, self)
else:
for ax in axiom:
_rdflib_add_axiom(ax, self)
[docs]
def remove_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
"""Remove the given axiom(s) from this ontology.
Removing an axiom only removes the triple(s) it directly represents -- it does not
cascade-remove entity declarations that other axioms might still depend on. Use
`remove_axiom(OWLDeclarationAxiom(entity))` explicitly to un-declare an entity.
"""
self.is_modified = True
if isinstance(axiom, OWLAxiom):
_rdflib_remove_axiom(axiom, self)
else:
for ax in axiom:
_rdflib_remove_axiom(ax, self)
[docs]
def save(self, path: Union[str, IRI] = None, inplace: bool = False, document_format: Optional[str] = None):
"""Save this ontology to a file via rdflib's own serializer.
Args:
path: Destination file path. Required unless `inplace` is True.
inplace: If True, overwrite the file this ontology was loaded from (or, for a
blank ontology created with `load=False`, the IRI it was constructed with --
which must itself be a valid filesystem path for that to work).
document_format: Serialization format. Accepts rdflib's own format names (`"xml"`,
`"turtle"`, `"n3"`, `"nt"`, `"nt11"`, `"trig"`, `"trix"`, `"nquads"`, `"json-ld"`)
plus the OWL-API-style aliases `Ontology`/`SyncOntology`'s `save()` accept
(`"rdfxml"`, `"ttl"`, `"ntriples"`, `"jsonld"`). Defaults to RDF/XML when omitted.
Raises:
ValueError: If an unsupported format string is provided.
"""
if isinstance(path, IRI):
path = path.as_str()
if inplace is False:
assert isinstance(path, str), f"path must be string if inplace is set to False. Current path is {type(path)}"
target = self._path if inplace else path
fmt_key = document_format.strip().lower() if document_format is not None else "rdfxml"
rdflib_format = _RDFLIB_NATIVE_FORMATS.get(fmt_key)
if rdflib_format is None:
raise ValueError(
f"Unsupported document format '{document_format}'. "
f"Supported formats: {', '.join(sorted(_RDFLIB_NATIVE_FORMATS))}")
parent = os.path.dirname(os.path.abspath(target))
os.makedirs(parent, exist_ok=True)
logger.info(f"Saving {self} to {target} (format={rdflib_format})...")
self.rdflib_graph.serialize(destination=target, format=rdflib_format)
[docs]
def get_ontology_id(self) -> OWLOntologyID:
onto_iri = None
version_iri = None
for s in self.rdflib_graph.subjects(rdflib.RDF.type, rdflib.OWL.Ontology):
if isinstance(s, rdflib.term.URIRef):
onto_iri = s.n3()[1:-1]
version = self.rdflib_graph.value(s, rdflib.OWL.versionIRI)
if isinstance(version, rdflib.term.URIRef):
version_iri = version.n3()[1:-1]
break
return OWLOntologyID(IRI.create(onto_iri) if onto_iri is not None else None,
IRI.create(version_iri) if version_iri is not None else None)
[docs]
def __eq__(self, other):
if type(other) is type(self):
return self._path == other._path
return NotImplemented
[docs]
def __hash__(self):
return hash(("RDFLibOntology", self._path))
[docs]
def __repr__(self):
return f'RDFLibOntology({self._path})'
OWLREADY2_FACET_KEYS = MappingProxyType({
OWLFacet.MIN_INCLUSIVE: "min_inclusive",
OWLFacet.MIN_EXCLUSIVE: "min_exclusive",
OWLFacet.MAX_INCLUSIVE: "max_inclusive",
OWLFacet.MAX_EXCLUSIVE: "max_exclusive",
OWLFacet.LENGTH: "length",
OWLFacet.MIN_LENGTH: "min_length",
OWLFacet.MAX_LENGTH: "max_length",
OWLFacet.PATTERN: "pattern",
OWLFacet.TOTAL_DIGITS: "total_digits",
OWLFacet.FRACTION_DIGITS: "fraction_digits"
})
######################### Below classes must be outside of this script #########################
[docs]
class ToOwlready2:
__slots__ = '_world'
_world: owlready2.World
def __init__(self, world: owlready2.World):
"""Map owlapy model classes to owlready2.
Args:
world: Owlready2 World to use for mapping.
"""
self._world = world
[docs]
@singledispatchmethod
def map_object(self, o: OWLObject):
"""Map owlapy object classes."""
raise NotImplementedError(f'don\'t know how to map {o}')
@map_object.register
def _(self, ce: OWLClassExpression) -> Union[owlready2.ClassConstruct, owlready2.ThingClass]:
return self.map_concept(ce)
@map_object.register
def _(self, ont: AbstractOWLOntology) -> owlready2.namespace.Ontology:
return self._world.get_ontology(
ont.get_ontology_id().get_ontology_iri().as_str()
)
@map_object.register
def _(self, ap: OWLAnnotationProperty) -> owlready2.annotation.AnnotationPropertyClass:
return self._world[ap.str]
# @TODO CD: map_object is buggy. and it can return None
# single dispatch is still not implemented in mypy, see https://github.com/python/mypy/issues/2904
[docs]
@singledispatchmethod
def map_concept(self, o: OWLClassExpression) \
-> Union[owlready2.ClassConstruct, owlready2.ThingClass]:
"""Map owlapy concept classes."""
raise NotImplementedError(o)
@singledispatchmethod
def _to_owlready2_property(self, p: OWLPropertyExpression) -> owlready2.Property:
raise NotImplementedError(p)
@_to_owlready2_property.register
def _(self, p: OWLObjectInverseOf):
p_x = self._to_owlready2_property(p.get_named_property())
return owlready2.Inverse(p_x)
@_to_owlready2_property.register
def _(self, p: OWLObjectProperty) -> owlready2.prop.ObjectPropertyClass:
return self._world[p.str]
@_to_owlready2_property.register
def _(self, p: OWLDataProperty) -> owlready2.prop.DataPropertyClass:
return self._world[p.str]
@singledispatchmethod
def _to_owlready2_individual(self, i: OWLIndividual) -> owlready2.Thing:
raise NotImplementedError(i)
@_to_owlready2_individual.register
def _(self, i: OWLNamedIndividual):
return self._world[i.str]
@map_concept.register
def _(self, c: OWLClass) -> owlready2.ThingClass:
x = self._world[c.str]
try:
assert x is not None
except AssertionError:
logger.warning(f"The world attribute{self._world} maps {c} into None")
return x
@map_concept.register
def _(self, c: OWLObjectComplementOf) -> owlready2.class_construct.Not:
return owlready2.Not(self.map_concept(c.get_operand()))
@map_concept.register
def _(self, ce: OWLObjectUnionOf) -> owlready2.class_construct.Or:
return owlready2.Or(map(self.map_concept, ce.operands()))
@map_concept.register
def _(self, ce: OWLObjectIntersectionOf) -> owlready2.class_construct.And:
return owlready2.And(map(self.map_concept, ce.operands()))
@map_concept.register
def _(self, ce: OWLObjectSomeValuesFrom) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
assert isinstance(ce.get_filler(),
OWLClassExpression), f"{ce.get_filler()} is not an OWL Class expression and cannot be serialized at the moment"
return prop.some(self.map_concept(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLObjectAllValuesFrom) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.only(self.map_concept(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLObjectOneOf) -> owlready2.class_construct.OneOf:
return owlready2.OneOf(list(map(self._to_owlready2_individual, ce.individuals())))
@map_concept.register
def _(self, ce: OWLObjectExactCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.exactly(ce.get_cardinality(), self.map_concept(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLObjectMaxCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.max(ce.get_cardinality(), self.map_concept(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLObjectMinCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.min(ce.get_cardinality(), self.map_concept(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLObjectHasValue) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.value(self._to_owlready2_individual(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataSomeValuesFrom) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.some(self.map_datarange(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataAllValuesFrom) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.only(self.map_datarange(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataExactCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.exactly(ce.get_cardinality(), self.map_datarange(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataMaxCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.max(ce.get_cardinality(), self.map_datarange(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataMinCardinality) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.min(ce.get_cardinality(), self.map_datarange(ce.get_filler()))
@map_concept.register
def _(self, ce: OWLDataHasValue) -> owlready2.class_construct.Restriction:
prop = self._to_owlready2_property(ce.get_property())
return prop.value(ce.get_filler().to_python())
[docs]
@singledispatchmethod
def map_datarange(self, p: OWLDataRange) -> Union[owlready2.ClassConstruct, type]:
"""Map owlapy data range classes."""
raise NotImplementedError(p)
@map_datarange.register
def _(self, p: OWLDataComplementOf) -> owlready2.class_construct.Not:
return owlready2.Not(self.map_datarange(p.get_data_range()))
@map_datarange.register
def _(self, p: OWLDataUnionOf) -> owlready2.class_construct.Or:
return owlready2.Or(map(self.map_datarange, p.operands()))
@map_datarange.register
def _(self, p: OWLDataIntersectionOf) -> owlready2.class_construct.And:
return owlready2.And(map(self.map_datarange, p.operands()))
@map_datarange.register
def _(self, p: OWLDataOneOf) -> owlready2.class_construct.OneOf:
return owlready2.OneOf([lit.to_python() for lit in p.operands()])
@map_datarange.register
def _(self, p: OWLDatatypeRestriction) -> owlready2.class_construct.ConstrainedDatatype:
facet_args = dict()
for facet_res in p.get_facet_restrictions():
value = facet_res.get_facet_value().to_python()
facet_key = OWLREADY2_FACET_KEYS[facet_res.get_facet()]
facet_args[facet_key] = value
return owlready2.ConstrainedDatatype(self.map_datarange(p.get_datatype()), **facet_args)
@map_datarange.register
def _(self, type_: OWLDatatype) -> type:
if type_ == BooleanOWLDatatype:
return bool
elif type_ == DoubleOWLDatatype:
return float
elif type_ == IntegerOWLDatatype:
return int
elif type_ == StringOWLDatatype:
return str
elif type_ == DateOWLDatatype:
return date
elif type_ == DateTimeOWLDatatype:
return datetime
elif type_ == DurationOWLDatatype:
return Timedelta
else:
raise ValueError(type_)
[docs]
class FromOwlready2:
"""Map owlready2 classes to owlapy model classes."""
__slots__ = ()
[docs]
@singledispatchmethod
def map_concept(self, c: Union[owlready2.ClassConstruct, owlready2.ThingClass]) -> OWLClassExpression:
"""Map concept classes."""
raise NotImplementedError(c)
@singledispatchmethod
def _from_owlready2_property(self, c: Union[owlready2.PropertyClass, owlready2.Inverse]) -> OWLPropertyExpression:
raise NotImplementedError(c)
@_from_owlready2_property.register
def _(self, p: owlready2.ObjectPropertyClass) -> OWLObjectProperty:
return OWLObjectProperty(IRI.create(p.iri))
@_from_owlready2_property.register
def _(self, p: owlready2.DataPropertyClass) -> OWLDataProperty:
return OWLDataProperty(IRI.create(p.iri))
@_from_owlready2_property.register
def _(self, i: owlready2.Inverse) -> OWLObjectInverseOf:
return OWLObjectInverseOf(self._from_owlready2_property(i.property))
@map_concept.register
def _(self, c: owlready2.ThingClass) -> OWLClass:
return OWLClass(IRI.create(c.iri))
@map_concept.register
def _(self, c: owlready2.Not) -> OWLObjectComplementOf:
return OWLObjectComplementOf(self.map_concept(c.Class))
@map_concept.register
def _(self, c: owlready2.And) -> OWLObjectIntersectionOf:
return OWLObjectIntersectionOf(map(self.map_concept, c.Classes))
@map_concept.register
def _(self, c: owlready2.Or) -> OWLObjectUnionOf:
return OWLObjectUnionOf(map(self.map_concept, c.Classes))
@map_concept.register
def _(self, c: owlready2.OneOf) -> OWLObjectOneOf:
return OWLObjectOneOf([OWLNamedIndividual(IRI.create(ind.iri)) for ind in c.instances])
@map_concept.register
def _(self, c: owlready2.Restriction) -> OWLRestriction:
if isinstance(c.property, owlready2.ObjectPropertyClass):
return self._to_object_property(c)
elif isinstance(c.property, owlready2.DataPropertyClass):
return self._to_data_property(c)
else:
raise NotImplementedError(c)
def _to_object_property(self, c: owlready2.Restriction) -> OWLObjectRestriction:
p = self._from_owlready2_property(c.property)
assert isinstance(p, OWLObjectPropertyExpression)
if c.type == owlready2.VALUE:
ind = OWLNamedIndividual(IRI.create(c.value.iri))
return OWLObjectHasValue(p, ind)
else:
f = self.map_concept(c.value)
if c.type == owlready2.SOME:
return OWLObjectSomeValuesFrom(p, f)
elif c.type == owlready2.ONLY:
return OWLObjectAllValuesFrom(p, f)
elif c.type == owlready2.EXACTLY:
return OWLObjectExactCardinality(c.cardinality, p, f)
elif c.type == owlready2.MIN:
return OWLObjectMinCardinality(c.cardinality, p, f)
elif c.type == owlready2.MAX:
return OWLObjectMaxCardinality(c.cardinality, p, f)
else:
raise NotImplementedError(c)
def _to_data_property(self, c: owlready2.Restriction) -> OWLDataRestriction:
p = self._from_owlready2_property(c.property)
assert isinstance(p, OWLDataPropertyExpression)
if c.type == owlready2.VALUE:
return OWLDataHasValue(p, OWLLiteral(c.value))
else:
f = self.map_datarange(c.value)
if c.type == owlready2.SOME:
return OWLDataSomeValuesFrom(p, f)
elif c.type == owlready2.ONLY:
return OWLDataAllValuesFrom(p, f)
elif c.type == owlready2.EXACTLY:
return OWLDataExactCardinality(c.cardinality, p, f)
elif c.type == owlready2.MIN:
return OWLDataMinCardinality(c.cardinality, p, f)
elif c.type == owlready2.MAX:
return OWLDataMaxCardinality(c.cardinality, p, f)
else:
raise NotImplementedError(c)
[docs]
@singledispatchmethod
def map_datarange(self, p: owlready2.ClassConstruct) -> OWLDataRange:
"""Map data range classes."""
raise NotImplementedError(p)
@map_datarange.register
def _(self, p: owlready2.Not) -> OWLDataComplementOf:
return OWLDataComplementOf(self.map_datarange(p.Class))
@map_datarange.register
def _(self, p: owlready2.Or) -> OWLDataUnionOf:
return OWLDataUnionOf(map(self.map_datarange, p.Classes))
@map_datarange.register
def _(self, p: owlready2.And) -> OWLDataIntersectionOf:
return OWLDataIntersectionOf(map(self.map_datarange, p.Classes))
@map_datarange.register
def _(self, p: owlready2.OneOf) -> OWLDataOneOf:
return OWLDataOneOf([OWLLiteral(i) for i in p.instances])
@map_datarange.register
def _(self, p: owlready2.ConstrainedDatatype) -> OWLDatatypeRestriction:
restrictions = []
for facet in OWLFacet:
value = getattr(p, OWLREADY2_FACET_KEYS[facet], None)
if value is not None:
restrictions.append(OWLFacetRestriction(facet, OWLLiteral(value)))
return OWLDatatypeRestriction(self.map_datarange(p.base_datatype), restrictions)
@map_datarange.register
def _(self, type_: type) -> OWLDatatype:
if type_ is bool:
return BooleanOWLDatatype
elif type_ is float:
return DoubleOWLDatatype
elif type_ is int:
return IntegerOWLDatatype
elif type_ is str:
return StringOWLDatatype
elif type_ is date:
return DateOWLDatatype
elif type_ is datetime:
return DateTimeOWLDatatype
elif type_ is Timedelta:
return DurationOWLDatatype
else:
# Some ontologies contain invalid data-range fillers (e.g. owl:Thing used as the
# filler of a data property restriction). Degrade gracefully to the top datatype
# instead of raising, so a single malformed axiom doesn't abort hierarchy traversal.
logger.warning("Unrecognized data range filler %s, falling back to rdfs:Literal (top datatype)", type_)
return TopOWLDatatype
[docs]
def is_valid_entity(text_input: str):
return True if "/" in text_input else False
[docs]
class NeuralOntology(AbstractOWLOntology):
STR_IRI_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
STR_IRI_OWL_CLASS = "http://www.w3.org/2002/07/owl#Class"
STR_IRI_OBJECT_PROPERTY = "http://www.w3.org/2002/07/owl#ObjectProperty"
STR_IRI_DATA_PROPERTY = "http://www.w3.org/2002/07/owl#DatatypeProperty"
def __init__(self, path_neural_embedding: str, train_if_not_exists: bool = False,
training_params: Optional[Union[Dict[str, Any], str]] = None, batch_size: int = 1024, device: str = "gpu",
gamma: float = 0.5):
try:
# installing dicee will also install torch
import torch
from dicee.knowledge_graph_embeddings import KGE
except ImportError:
raise ImportError("The 'dicee' package is required to use NeuralOntology. "
"Please install it via 'pip install dicee'."
"For the CPU-only version (recommended): pip install dicee --extra-index-url https://download.pytorch.org/whl/cpu")
"""
Initialize a Neural Ontology from a KGE model.
Args:
path_neural_embedding: Path to a pretrained KGE model, a directory containing a train.txt file or a .owl file
train_if_not_exists: If True, train a new model if no pretrained model is found at the given path
training_params: Optional dictionary of training parameters to override defaults
"""
super().__init__()
self.gamma = gamma
path = Path(path_neural_embedding)
if os.path.isdir(path) and os.path.exists(os.path.join(path, "configuration.json")):
# Path leads to a folder containing pretrained model
self.model = KGE(path=path_neural_embedding)
elif train_if_not_exists:
# Train a new model
self._train_model(path_neural_embedding, training_params)
else:
raise ValueError(
f"No pretrained KGE model found at {path_neural_embedding} and train_if_not_exists is False")
self.batch_size = batch_size
if device == "gpu" and torch.cuda.is_available():
self.model.to("cuda")
logger.info("EBR inference on GPU")
elif device == "cpu":
self.model.to("cpu")
logger.info("EBR inference on CPU")
else:
logger.warning(f"Device {device} not supported, EBR will use CPU")
def _train_model(self, path: str, training_params: Optional[Union[Dict[str, Any], str]] = None):
"""
Train a new KGE model with the given parameters.
Args:
path: Path to a directory containing train.txt or to the dataset directory
training_params: Optional dictionary of training parameters or path to a JSON file.
"""
from dicee.config import Namespace
from dicee.executer import Execute
from dicee.knowledge_graph_embeddings import KGE
args = Namespace()
# Set default parameters
args.model = 'Keci'
args.scoring_technique = "AllvsAll"
if os.path.isdir(path):
args.dataset_dir = path
args.path_single_kg = None
else:
args.path_single_kg = path
if path.endswith(".owl"):
args.backend = "rdflib"
path = os.path.dirname(path)
args.trainer = "PL"
# Always include the file name in the path
file_name = os.path.basename(args.path_single_kg).replace(".owl", "")
args.path_to_store_single_run = os.path.join(path, f"{file_name}_trained_model")
args.num_epochs = 500
args.embedding_dim = 512
args.batch_size = 64
# Override with user-provided parameters if any
if training_params is None:
json_path = os.path.join(path, "NeuralOntology.json")
if os.path.exists(json_path):
with open(json_path, "r") as f:
training_params = json.load(f)
elif isinstance(training_params, str):
with open(training_params, "r") as f:
training_params = json.load(f)
if training_params is not None:
for key, value in training_params.items():
setattr(args, key, value)
# Check if model already exists before training
if os.path.isdir(args.path_to_store_single_run) and \
os.path.exists(os.path.join(args.path_to_store_single_run, "configuration.json")):
# Load existing pretrained model
logger.info(f"Loading existing model from {args.path_to_store_single_run}")
self.model = KGE(path=args.path_to_store_single_run)
else:
# Train the model
logger.info(f"Training new model, will be saved to {args.path_to_store_single_run}")
Execute(args).start()
# Load the trained model
self.model = KGE(path=args.path_to_store_single_run)
[docs]
def predict(self, h: List[str] = None, r: List[str] = None, t: List[str] = None) -> List[Tuple[str, float]]:
if r is None:
topk = len(self.model.relation_to_idx)
else:
topk = len(self.model.entity_to_idx)
# Filter out unknown relations; return empty list (0 scores) if none remain
if r is not None:
if isinstance(r, str):
if r not in self.model.relation_to_idx:
return []
else:
r = [ri for ri in r if ri in self.model.relation_to_idx]
if not r:
return []
# Filter out unknown head entities; return empty list (0 scores) if none remain
if h is not None:
if isinstance(h, str):
if h not in self.model.entity_to_idx:
return []
else:
h = [hi for hi in h if hi in self.model.entity_to_idx]
if not h:
return []
# Filter out unknown tail entities; return empty list (0 scores) if none remain
if t is not None:
if isinstance(t, str):
if t not in self.model.entity_to_idx:
return []
else:
t = [ti for ti in t if ti in self.model.entity_to_idx]
if not t:
return []
return [(top_entity, score) for row in
self.model.predict_topk(h=h, r=r, t=t, topk=topk, batch_size=self.batch_size) for top_entity, score in
row if score >= self.gamma and is_valid_entity(top_entity)]
[docs]
def classes_in_signature(self) -> List[OWLClass]:
return [OWLClass(top_entity) for top_entity, score in self.predict(h=None,
r=self.STR_IRI_TYPE,
t=self.STR_IRI_OWL_CLASS)]
[docs]
def individuals_in_signature(self) -> List[OWLNamedIndividual]:
set_str_entities = set()
for top_entity, score in self.predict(h=None,
r=self.STR_IRI_TYPE,
t=[owl_class.iri.str for owl_class in self.classes_in_signature()]):
set_str_entities.add(top_entity)
return [OWLNamedIndividual(entity) for entity in set_str_entities]
[docs]
def data_properties_in_signature(self) -> List[OWLDataProperty]:
return [OWLDataProperty(top_entity) for top_entity, score in self.predict(h=None,
r=self.STR_IRI_TYPE,
t=self.STR_IRI_DATA_PROPERTY)]
[docs]
def object_properties_in_signature(self) -> List[OWLObjectProperty]:
return [OWLObjectProperty(top_entity) for top_entity, score in self.predict(h=None,
r=self.STR_IRI_TYPE,
t=self.STR_IRI_OBJECT_PROPERTY)]
[docs]
def equivalent_classes_axioms(self, c: OWLClass) -> Iterable[OWLEquivalentClassesAxiom]:
raise NotImplementedError("Not implemented")
[docs]
def general_class_axioms(self) -> Iterable[OWLClassAxiom]:
raise NotImplementedError("Not implemented")
[docs]
def data_property_domain_axioms(self, property: OWLDataProperty) -> Iterable[OWLDataPropertyDomainAxiom]:
raise NotImplementedError("Not implemented")
[docs]
def data_property_range_axioms(self, property: OWLDataProperty) -> Iterable[OWLDataPropertyRangeAxiom]:
raise NotImplementedError("Not implemented")
[docs]
def object_property_domain_axioms(self, property: OWLObjectProperty) -> Iterable[OWLObjectPropertyDomainAxiom]:
raise NotImplementedError("Not implemented")
[docs]
def get_ontology_id(self) -> _OI:
raise NotImplementedError("Not implemented")
# LF: here we would need to retrain? -> ie we would need to store / have the path for the underlying ontology?
[docs]
def add_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
raise NotImplementedError("Not implemented")
[docs]
def remove_axiom(self, axiom: Union[OWLAxiom, Iterable[OWLAxiom]]):
raise NotImplementedError("Not implemented")
[docs]
def save(self, document_iri: Optional[IRI] = None):
raise NotImplementedError("Not implemented")
[docs]
def __eq__(self, other):
return self.model == other.model
[docs]
def __hash__(self):
return hash(self.model)
[docs]
def __repr__(self):
return f"NeuralOntology(model={self.model})"
[docs]
def object_property_range_axioms(self, property: OWLObjectProperty) -> Iterable[OWLObjectPropertyRangeAxiom]:
raise NotImplementedError("Not implemented")
_parse_concept_to_owlapy = FromOwlready2().map_concept
_parse_datarange_to_owlapy = FromOwlready2().map_datarange