"""Format converter."""
from collections import defaultdict
from contextlib import contextmanager
from functools import singledispatchmethod
from json import dumps
from types import MappingProxyType
from typing import Callable, Dict, Iterable, List, Optional, Set
from rdflib.namespace import XSD
from rdflib.plugins.sparql.parser import parseQuery
from owlapy.class_expression import (
OWLClass,
OWLClassExpression,
OWLDataAllValuesFrom,
OWLDataCardinalityRestriction,
OWLDataExactCardinality,
OWLDataHasValue,
OWLDataMaxCardinality,
OWLDataMinCardinality,
OWLDataOneOf,
OWLDataSomeValuesFrom,
OWLDatatypeRestriction,
OWLObjectAllValuesFrom,
OWLObjectCardinalityRestriction,
OWLObjectComplementOf,
OWLObjectExactCardinality,
OWLObjectHasSelf,
OWLObjectHasValue,
OWLObjectIntersectionOf,
OWLObjectMaxCardinality,
OWLObjectMinCardinality,
OWLObjectOneOf,
OWLObjectSomeValuesFrom,
OWLObjectUnionOf,
)
from owlapy.owl_data_ranges import OWLDataComplementOf, OWLDataIntersectionOf, OWLDataUnionOf
from owlapy.owl_datatype import OWLDatatype
from owlapy.owl_individual import OWLNamedIndividual
from owlapy.owl_literal import OWLLiteral, TopOWLDatatype
from owlapy.owl_object import OWLEntity
from owlapy.owl_property import OWLDataProperty, OWLObjectProperty
from owlapy.vocab import OWLFacet, OWLRDFVocabulary
_Variable_facet_comp = MappingProxyType({
OWLFacet.MIN_INCLUSIVE: ">=",
OWLFacet.MIN_EXCLUSIVE: ">",
OWLFacet.MAX_INCLUSIVE: "<=",
OWLFacet.MAX_EXCLUSIVE: "<"
})
_INTEGER_BOUNDS = MappingProxyType({
"integer": (None, None), "nonNegativeInteger": (0, None), "positiveInteger": (1, None),
"nonPositiveInteger": (None, 0), "negativeInteger": (None, -1),
"long": (-2**63, 2**63 - 1), "int": (-2**31, 2**31 - 1),
"short": (-2**15, 2**15 - 1), "byte": (-2**7, 2**7 - 1),
"unsignedLong": (0, 2**64 - 1), "unsignedInt": (0, 2**32 - 1),
"unsignedShort": (0, 2**16 - 1), "unsignedByte": (0, 2**8 - 1),
})
_STRING_DATATYPES = frozenset(str(XSD[name]) for name in (
"string", "normalizedString", "token", "language", "Name", "NCName", "NMTOKEN",
))
[docs]
def peek(x):
"""Peek the last element of an array.
Returns:
The last element arr[-1].
"""
return x[-1]
[docs]
class VariablesMapping:
"""Helper class for owl-to-sparql conversion."""
__slots__ = 'class_cnt', 'prop_cnt', 'ind_cnt', 'dict'
def __init__(self):
self.class_cnt = 0
self.prop_cnt = 0
self.ind_cnt = 0
self.dict = dict()
[docs]
def get_variable(self, e: OWLEntity) -> str:
if e in self.dict:
return self.dict[e]
if isinstance(e, OWLClass):
self.class_cnt += 1
var = f"?cls_{self.class_cnt}"
elif isinstance(e, OWLObjectProperty) or isinstance(e, OWLDataProperty):
self.prop_cnt += 1
var = f"?p_{self.prop_cnt}"
elif isinstance(e, OWLNamedIndividual):
self.ind_cnt += 1
var = f"?ind_{self.ind_cnt}"
else:
raise ValueError(e)
self.dict[e] = var
return var
[docs]
def new_individual_variable(self) -> str:
self.ind_cnt += 1
return f"?s_{self.ind_cnt}"
[docs]
def new_property_variable(self) -> str:
self.prop_cnt += 1
return f"?p_{self.prop_cnt}"
[docs]
def __contains__(self, item: OWLEntity) -> bool:
return item in self.dict
[docs]
def __getitem__(self, item: OWLEntity) -> str:
return self.dict[item]
[docs]
class Owl2SparqlConverter:
"""Convert OWL class expressions to SPARQL queries.
This class converts OWL class expressions from the owlapy model into equivalent SPARQL queries.
It maintains internal state for variable management, parent-child relationships, and query building.
The converter uses a recursive approach to traverse the class expression tree and generates
SPARQL patterns for each type of class expression (e.g., OWLClass, OWLObjectSomeValuesFrom,
OWLObjectIntersectionOf, etc.).
Attributes:
ce: The root OWL class expression being converted
sparql: List of SPARQL query patterns being constructed
variables: List of variable names used in the query
parent: Stack of parent class expressions during traversal
parent_var: Stack of parent variable names during traversal
variable_entities: Set of OWL entities associated with variables
properties: Mapping from expression indices to their properties
_intersection: Mapping tracking whether expressions are in intersections
mapping: Variable mapping manager for De Morgan transformations
grouping_vars: Variables used in GROUP BY clauses for cardinality restrictions
having_conditions: HAVING clause conditions for cardinality restrictions
cnt: Counter for generating unique variable names
for_all_de_morgan: Whether to apply De Morgan's laws for universal quantification
named_individuals: Whether to restrict results to named individuals only
"""
__slots__ = 'ce', 'sparql', 'variables', 'parent', 'parent_var', 'properties', 'variable_entities', 'cnt', \
'mapping', 'grouping_vars', 'having_conditions', 'for_all_de_morgan', 'named_individuals', \
'_intersection', 'negation_default', 'sub_property_resolver', 'inverse_property_resolver', \
'subclass_resolver'
ce: OWLClassExpression
sparql: List[str]
variables: List[str]
parent: List[OWLClassExpression]
parent_var: List[str]
variable_entities: Set[OWLEntity]
properties: Dict[int, List[OWLEntity]]
_intersection: Dict[int, bool]
mapping: VariablesMapping
grouping_vars: Dict[OWLClassExpression, Set[str]]
having_conditions: Dict[OWLClassExpression, Set[str]]
cnt: int
for_all_de_morgan: bool
named_individuals: bool
negation_default: bool
sub_property_resolver: Optional[Callable]
inverse_property_resolver: Optional[Callable]
subclass_resolver: Optional[Callable]
[docs]
def convert(self, root_variable: str,
ce: OWLClassExpression,
for_all_de_morgan: bool = True,
named_individuals: bool = False,
negation_default: bool = True,
sub_property_resolver: Optional[Callable] = None,
inverse_property_resolver: Optional[Callable] = None,
subclass_resolver: Optional[Callable] = None):
"""Used to convert owl class expression to SPARQL syntax.
Args:
root_variable (str): Root variable name that will be used in SPARQL query.
ce (OWLClassExpression): The owl class expression to convert.
named_individuals (bool): If 'True' return only entities that are instances of owl:NamedIndividual.
negation_default (bool): If 'True' (closed world, the default), OWLObjectComplementOf is translated
to a `FILTER NOT EXISTS` pattern. If 'False' (open world), OWLObjectComplementOf always
contributes no results, since under OWA nothing can be inferred to NOT be a member of a class
without an explicit negative assertion.
sub_property_resolver: Optional callable mapping an OWLObjectProperty/OWLDataProperty to its
sub-properties. When provided, every rendered property predicate is expanded into a SPARQL
property-path alternation of itself and all of its sub-properties, so that queries also match
individuals connected via a (semantically entailing) sub-property.
inverse_property_resolver: Optional callable mapping an OWLObjectProperty to its declared
owl:inverseOf partner(s). When provided, every rendered object property predicate is
additionally expanded with a reverse (`^`) property-path alternative for each declared
inverse, so that a property with no physically-asserted triples of its own (but a declared
inverse that does) is still correctly entailed.
subclass_resolver: Optional callable mapping an OWLClass to its subclasses. When provided,
every named-class membership triple (`?x a <cls>`) is expanded to also match any subclass
of `cls`, via a `VALUES` clause, so that individuals typed only at a leaf subclass are not
silently missed.
Returns:
list[str]: The SPARQL query.
"""
self.ce = ce
self.sparql = []
self.variables = []
self.parent = []
self.parent_var = []
self.properties = defaultdict(list)
self.variable_entities = set()
self._intersection = defaultdict(bool)
self.cnt = 0
self.mapping = VariablesMapping()
self.grouping_vars = defaultdict(set)
self.having_conditions = defaultdict(set)
self.for_all_de_morgan = for_all_de_morgan
self.named_individuals = named_individuals
self.negation_default = negation_default
self.sub_property_resolver = sub_property_resolver
self.inverse_property_resolver = inverse_property_resolver
self.subclass_resolver = subclass_resolver
# # if named_individuals is True, we return only entities that are instances of owl:NamedIndividual
if named_individuals:
self.append_triple(root_variable, 'a', f"<{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}>")
with self.stack_variable(root_variable):
with self.stack_parent(ce):
self.process(ce)
return self.sparql
@property
def modal_depth(self):
return len(self.variables)
[docs]
@singledispatchmethod
def render(self, e):
raise NotImplementedError(e)
@render.register
def _(self, lit: OWLLiteral):
return f'{dumps(lit.get_literal(), ensure_ascii=False)}^^<{lit.get_datatype().to_string_id()}>'
@render.register
def _(self, e: OWLEntity):
if e in self.variable_entities:
s = self.mapping.get_variable(e)
elif (self.sub_property_resolver is not None or self.inverse_property_resolver is not None) \
and isinstance(e, (OWLObjectProperty, OWLDataProperty)):
# Expand into a SPARQL property-path alternation of the property itself, all of its
# sub-properties (matches individuals connected via a semantically entailing
# sub-property), and -- for object properties -- a reverse (`^`) alternative for each
# declared owl:inverseOf partner (matches when the inverse holds in the other
# direction, even if this property itself has no physically-asserted triples). Works
# uniformly regardless of OWLObjectInverseOf direction, since inversion is otherwise
# handled by triple position.
alternatives = [f"<{e.to_string_id()}>"]
subs = list(self.sub_property_resolver(e)) if self.sub_property_resolver is not None else []
alternatives.extend(f"<{sub.to_string_id()}>" for sub in subs)
if self.inverse_property_resolver is not None and isinstance(e, OWLObjectProperty):
invs = list(self.inverse_property_resolver(e))
alternatives.extend(f"^<{inv.to_string_id()}>" for inv in invs)
# A sub-property of a declared inverse of e, reversed, also entails e (e.g. e's
# inverse is q, and p is a sub-property of q -- p reversed entails q reversed
# entails e). Similarly, the inverse of a sub-property of e, reversed, entails e.
if self.sub_property_resolver is not None:
for inv in invs:
alternatives.extend(f"^<{sub.to_string_id()}>" for sub in self.sub_property_resolver(inv))
for sub in subs:
alternatives.extend(f"^<{inv.to_string_id()}>" for inv in self.inverse_property_resolver(sub))
s = "(" + "|".join(alternatives) + ")"
else:
s = f"<{e.to_string_id()}>"
if isinstance(e, OWLObjectProperty):
self.properties[self.modal_depth].append(e)
return s
def _maybe_quote(self, e):
assert isinstance(e, str)
if e.startswith("?"):
return e
else:
return f"<{e}>"
def _maybe_quote_p(self, p):
if isinstance(p, str):
if p.startswith("?") or p == "a" or p.startswith("<"):
return p
else:
return f"<{p}>"
else:
return self.render(p)
def _maybe_render(self, o):
if isinstance(o, str):
return o
else:
return self.render(o)
[docs]
@contextmanager
def stack_variable(self, var):
self.variables.append(var)
try:
yield
finally:
self.variables.pop()
[docs]
@contextmanager
def stack_parent(self, parent: OWLClassExpression):
self.parent.append(parent)
self.parent_var.append(self.current_variable)
try:
yield
finally:
self.parent.pop()
self.parent_var.pop()
@property
def current_variable(self):
return peek(self.variables)
# this method is responsible for translating class expressions to SPARQL queries
# the decorator "@singledispatchmethod" denotes that the method is overload
# each overload of the method is responsible for processing a different type of class expressions (e.g., ⊔ or ⊓)
[docs]
@singledispatchmethod
def process(self, ce: OWLClassExpression):
raise NotImplementedError(f"We cannot create SPARQL query based on the following owl class {ce}")
# an overload of process function
# this overload is responsible for handling single concepts (e.g., Brother)
# general case: C
# this is the final step of the recursion
@process.register
def _(self, ce: OWLClass):
if self.ce == ce or not ce.is_owl_thing():
if self.subclass_resolver is not None and not ce.is_owl_thing():
# Expand to also match individuals typed only at a subclass of ce, via a bounded
# VALUES list (cheaper than a UNION of one BGP per subclass).
type_var = self.mapping.new_individual_variable()
classes = [ce] + list(self.subclass_resolver(ce))
values = " ".join(f"<{c.to_string_id()}>" for c in classes)
self.append_triple(self.current_variable, "a", type_var)
self.append(f"VALUES {type_var} {{ {values} }} ")
else:
self.append_triple(self.current_variable, "a", self.render(ce))
# old_var = self.current_variable
# new_var = self.mapping.new_individual_variable()
# with self.stack_variable(new_var):
# self.append_triple(old_var, "a", new_var)
# self.append_triple(new_var, "<http://www.w3.org/2000/01/rdf-schema#subClassOf>*", self.render(ce))
# else: ce is owl:Thing used as a nested filler (not the root expression). owl:Thing
# tautologically matches every resource, and the variable is already bound by the
# surrounding pattern that introduced it (e.g. the property triple in
# OWLObjectSomeValuesFrom), so no additional triple is needed. Emitting `?x a owl:Thing`
# here would wrongly require individuals to be explicitly asserted rdf:type owl:Thing,
# which real ontologies essentially never do.
# an overload of process function
# this overload is responsible for handling intersections of concepts (e.g., Brother ⊓ Father)
# general case: C1 ⊓ ... ⊓ Cn
@process.register
def _(self, ce: OWLObjectIntersectionOf):
# we iterate over the concepts that appear in the intersection
for op in ce.operands():
self.process(op)
# an overload of process function
# this overload is responsible for handling unions of concepts (e.g., Brother ⊔ Sister)
# general case: C1 ⊔ ... ⊔ Cn
@process.register
def _(self, ce: OWLObjectUnionOf):
first = True
# we iterate over the concepts that appear in the union
for op in ce.operands():
# SPARQL's UNION comes after the first concept
if first:
first = False
else:
self.append(" UNION ")
self.append("{ ")
with self.stack_parent(op):
self.process(op)
self.append(" }")
# an overload of process function
# this overload is responsible for handling complements of concepts (e.g., ¬Brother)
# general case: ¬C
@process.register
def _(self, ce: OWLObjectComplementOf):
if not self.negation_default:
# Open-world semantics: nothing can be inferred to NOT be a member of a class
# without an explicit negative assertion (not modeled here), so a complement never
# contributes any results, however deeply nested.
# Note: "FILTER(1=0)" rather than "FILTER(false)" -- rdflib's SPARQL engine does not
# actually filter out solutions for a bare boolean-literal FILTER(false)/FILTER(0),
# but a always-false comparison is evaluated and applied correctly.
self.append("FILTER(1=0)")
return
subject = self.current_variable
# the conversion was trying here to optimize the query
# but the proposed optimization alters the semantics of some queries
# example: ( A ⊓ ( B ⊔ ( ¬C ) ) )
# with the proposed optimization, the group graph pattern for (¬C) will be { FILTER NOT EXISTS { ?x a C } }
# however, the expected pattern is { ?x ?p ?o . FILTER NOT EXISTS { ?x a C } }
# the exclusion of "?x ?p ?o" results in the group graph pattern to just return true or false (not bindings)
# as a result, we need to comment out the if-clause of the following line
# if not self.in_intersection and self.modal_depth == 1:
# However, if the complement is directly inside an intersection at the top level,
# the intersection already provides bindings, so we don't need the extra triple pattern
if not (len(self.parent) > 0 and isinstance(self.parent[-1], OWLObjectIntersectionOf) and self.modal_depth == 1):
self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
self.append("FILTER NOT EXISTS { ")
# process the concept after the ¬
self.process(ce.get_operand())
self.append(" }")
# an overload of process function
# this overload is responsible for handling the exists operator (e.g., ∃hasChild.Male)
# general case: ∃r.C
@process.register
def _(self, ce: OWLObjectSomeValuesFrom):
object_variable = self.mapping.new_individual_variable()
# property expression holds the role of the class expression (hasChild in our example)
property_expression = ce.get_property()
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(object_variable, property_expression.get_named_property(), self.current_variable)
else:
self.append_triple(self.current_variable, property_expression.get_named_property(), object_variable)
# filler holds the concept of the expression (Male in our example) and is processed recursively
filler = ce.get_filler()
with self.stack_variable(object_variable):
self.process(filler)
@process.register
def _(self, ce: OWLObjectAllValuesFrom):
if self.for_all_de_morgan is True:
self.forAllDeMorgan(ce)
else:
self.forAll(ce)
# an overload of process function
# this overload is responsible for handling the forAll operator (e.g., ∀hasChild.Male)
# general case: ∀r.C
[docs]
def forAll(self, ce: OWLObjectAllValuesFrom):
subject = self.current_variable
object_variable = self.mapping.new_individual_variable()
# property expression holds the role of the class expression (hasChild in our example)
property_expression = ce.get_property()
predicate = property_expression.get_named_property()
# filler holds the concept of the expression (Male in our example) and is processed recursively
filler = ce.get_filler()
self.append("{")
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(object_variable, predicate, self.current_variable)
else:
self.append_triple(self.current_variable, predicate, object_variable)
# restrict filler
var = self.mapping.new_individual_variable()
cnt_var1 = self.new_count_var()
# the count needs to use distinct
self.append(f"{{ SELECT {subject} ( COUNT( DISTINCT {var} ) AS {cnt_var1} ) WHERE {{ ")
self.append_triple(subject, predicate, var)
# here, we recursively process the filler (Male in our example)
with self.stack_variable(var):
self.process(filler)
self.append(f" }} GROUP BY {subject} }}")
var = self.mapping.new_individual_variable()
cnt_var2 = self.new_count_var()
# the count needs to use distinct
self.append(f"{{ SELECT {subject} ( COUNT( DISTINCT {var} ) AS {cnt_var2} ) WHERE {{ ")
self.append_triple(subject, predicate, var)
self.append(f" }} GROUP BY {subject} }}")
self.append(f" FILTER( {cnt_var1} = {cnt_var2} )")
self.append("} UNION { ")
# here, the second group graph pattern starts
# the second group graph pattern returns all those entities that do not appear in a triple with the property
self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
self.append("FILTER NOT EXISTS { ")
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(self.mapping.new_individual_variable(), predicate, self.current_variable)
else:
self.append_triple(self.current_variable, predicate, self.mapping.new_individual_variable())
self.append(" } }")
# an overload of process function
# this overload is responsible for handling the forAll operator but as if the DeMorgan law was applied
# (e.g., ∀hasChild.Male == ¬(∃hasChild.¬Male))
# general case: ∀r.C == ¬(∃r.¬C)
[docs]
def forAllDeMorgan(self, ce: OWLObjectAllValuesFrom):
subject = self.current_variable
# here, we need to apply the complement rule twice
# the first filter not exists covers the outer ¬
self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
self.append("FILTER NOT EXISTS { ")
object_variable = self.mapping.new_individual_variable()
# property expression holds the role of the class expression (hasChild in our example)
property_expression = ce.get_property()
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(object_variable, property_expression.get_named_property(), self.current_variable)
else:
self.append_triple(self.current_variable, property_expression.get_named_property(), object_variable)
# the second filter not exists covers the inner ¬
# filler holds the concept of the expression (Male in our example) and is processed recursively
self.append("FILTER NOT EXISTS { ")
filler = ce.get_filler()
with self.stack_variable(object_variable):
self.process(filler)
self.append(" }")
self.append(" }")
# an overload of process function
# this overload is responsible for handling the exists operator combined with an individual (e.g., ∃hasChild.{john})
# general case: ∃r.{a}
@process.register
def _(self, ce: OWLObjectHasValue):
property_expression = ce.get_property()
value = ce.get_filler()
# we ensure that the value is an individual
assert isinstance(value, OWLNamedIndividual)
if property_expression.is_anonymous():
self.append_triple(value.to_string_id(), property_expression.get_named_property(), self.current_variable)
else:
self.append_triple(self.current_variable, property_expression.get_named_property(), value)
# an overload of process function
# this overload is responsible for handling the exists operator combined with an individual(e.g., >=3 hasChild.Male)
# general case: \theta n r.C
@process.register
def _(self, ce: OWLObjectCardinalityRestriction):
subject_variable = self.current_variable
object_variable = self.mapping.new_individual_variable()
property_expression = ce.get_property()
cardinality = ce.get_cardinality()
if isinstance(ce, OWLObjectMinCardinality):
comparator = ">="
elif isinstance(ce, OWLObjectMaxCardinality):
comparator = "<="
elif isinstance(ce, OWLObjectExactCardinality):
comparator = "="
else:
raise ValueError(ce)
# if the comparator is ≤ or the cardinality is 0, we need an additional group graph pattern
# the additional group graph pattern will take care the cases where an individual is not associated with the
# property expression
if comparator == "<=" or cardinality == 0:
self.append("{")
self.append(f"{{ SELECT {subject_variable} WHERE {{ ")
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(object_variable, property_expression.get_named_property(), subject_variable)
else:
self.append_triple(subject_variable, property_expression.get_named_property(), object_variable)
filler = ce.get_filler()
with self.stack_variable(object_variable):
self.process(filler)
self.append(f" }} GROUP BY {subject_variable}"
f" HAVING ( COUNT ( DISTINCT {object_variable} ) {comparator} {cardinality} ) }}")
# here, the second group graph pattern starts
if comparator == "<=" or cardinality == 0:
self.append("} UNION {")
# Bind subject_variable to any resource that has at least one rdf:type assertion
# (i.e. any individual), rather than "any triple with any predicate" -- the latter
# forces the query engine to scan every triple in the graph grouped by subject, which
# is catastrophically expensive on non-trivial ontologies for what is otherwise a cheap
# "does this individual exist" check.
self.append_triple(subject_variable, "a", self.mapping.new_individual_variable())
# OPTIONAL + !BOUND() rather than FILTER NOT EXISTS: the latter is evaluated as a
# correlated existence check per subject_variable candidate, which is catastrophically
# slow on non-trivial ontologies (rdflib's SPARQL engine has no index for it); OPTIONAL
# is evaluated as a single left join and is dramatically cheaper for the same result.
# (No extra `{ SELECT ... }` subquery wrapper here -- nested subqueries make rdflib's
# query planner fall back to a much slower evaluation strategy.)
self.append(" OPTIONAL { ")
object_variable = self.mapping.new_individual_variable()
if property_expression.is_anonymous():
# property expression is inverse of a property
self.append_triple(object_variable, property_expression.get_named_property(), subject_variable)
else:
self.append_triple(subject_variable, property_expression.get_named_property(), object_variable)
with self.stack_variable(object_variable):
self.process(filler)
self.append(f" }} FILTER ( !BOUND ( {object_variable} ) ) }}")
@process.register
def _(self, ce: OWLDataCardinalityRestriction):
subject_variable = self.current_variable
object_variable = self.mapping.new_individual_variable()
property_expression = ce.get_property()
assert isinstance(property_expression, OWLDataProperty)
cardinality = ce.get_cardinality()
if isinstance(ce, OWLDataMinCardinality):
comparator = ">="
elif isinstance(ce, OWLDataMaxCardinality):
comparator = "<="
elif isinstance(ce, OWLDataExactCardinality):
comparator = "="
else:
raise ValueError(ce)
if comparator == "<=" or cardinality == 0:
self.append("{")
self.append(f"{{ SELECT {subject_variable} WHERE {{ ")
self.append_triple(subject_variable, property_expression, object_variable)
filler = ce.get_filler()
with self.stack_variable(object_variable):
self.process(filler)
self.append(f" }} GROUP BY {subject_variable}"
f" HAVING ( COUNT ( DISTINCT {object_variable} ) {comparator} {cardinality} ) }}")
if comparator == "<=" or cardinality == 0:
self.append("} UNION {")
self.append_triple(subject_variable, "a", self.mapping.new_individual_variable())
self.append(" OPTIONAL { ")
object_variable = self.mapping.new_individual_variable()
self.append_triple(subject_variable, property_expression, object_variable)
with self.stack_variable(object_variable):
self.process(filler)
self.append(f" }} FILTER ( !BOUND ( {object_variable} ) ) }}")
# an overload of process function
# this overload is responsible for handling the exists operator combined with SELF
# general case: ∃r.SELF
@process.register
def _(self, ce: OWLObjectHasSelf):
subject = self.current_variable
property = ce.get_property()
self.append_triple(subject, property.get_named_property(), subject)
# an overload of process function
# this overload is responsible for handling the one of case (e.g., { john, jane }
# general case: { a1, ..., an }
@process.register
def _(self, ce: OWLObjectOneOf):
subject = self.current_variable
if self.modal_depth == 1:
self.append_triple(subject, "?p", "?o")
self.append(f" FILTER ( {subject} IN ( ")
first = True
for ind in ce.individuals():
if first:
first = False
else:
self.append(",")
assert isinstance(ind, OWLNamedIndividual)
self.append(f"<{ind.to_string_id()}>")
self.append(" ) )")
@process.register
def _(self, ce: OWLDataSomeValuesFrom):
object_variable = self.mapping.new_individual_variable()
property_expression = ce.get_property()
assert isinstance(property_expression, OWLDataProperty)
self.append_triple(self.current_variable, property_expression, object_variable)
filler = ce.get_filler()
with self.stack_variable(object_variable):
self.process(filler)
@process.register
def _(self, ce: OWLDataAllValuesFrom):
subject = self.current_variable
object_variable = self.mapping.new_individual_variable()
property_expression = ce.get_property()
assert isinstance(property_expression, OWLDataProperty)
filler = ce.get_filler()
if self.modal_depth == 1 and not self.named_individuals:
self.append_triple(subject, self.mapping.new_individual_variable(), self.mapping.new_individual_variable())
self.append("FILTER NOT EXISTS { ")
self.append_triple(subject, property_expression, object_variable)
with self.stack_variable(object_variable):
self.process(OWLDataComplementOf(filler))
self.append(" }")
@process.register
def _(self, ce: OWLDataHasValue):
property_expression = ce.get_property()
value = ce.get_filler()
assert isinstance(value, OWLLiteral)
self.append_triple(self.current_variable, property_expression, value)
@process.register(OWLDatatype)
@process.register(OWLDataOneOf)
@process.register(OWLDatatypeRestriction)
@process.register(OWLDataIntersectionOf)
@process.register(OWLDataUnionOf)
@process.register(OWLDataComplementOf)
def _process_data_range(self, node):
if isinstance(node, OWLDataOneOf) and self.modal_depth == 1:
self.append_triple(self.current_variable, "?p", "?o")
self.append(f" FILTER ( {self._data_range_condition(node)} ) ")
def _datatype_condition(self, node: OWLDatatype):
datatype = node.to_string_id()
name = datatype.removeprefix(str(XSD))
if datatype.startswith(str(XSD)) and (name == "decimal" or name in _INTEGER_BOUNDS):
variable = self.current_variable
datatypes = ", ".join(f"<{XSD[n]}>" for n in ("decimal", *_INTEGER_BOUNDS))
condition = f"DATATYPE ( {variable} ) IN ( {datatypes} )"
if name in _INTEGER_BOUNDS:
condition += f" && {variable} = FLOOR ( {variable} )"
lower, upper = _INTEGER_BOUNDS[name]
if lower is not None:
condition += f" && {variable} >= {lower}"
if upper is not None:
condition += f" && {variable} <= {upper}"
return condition
elif datatype == str(XSD.string):
datatypes = ", ".join(f"<{dt}>" for dt in sorted(_STRING_DATATYPES))
return f"DATATYPE ( {self.current_variable} ) IN ( {datatypes} )"
elif node != TopOWLDatatype:
return f"DATATYPE ( {self.current_variable} ) = <{datatype}>"
else:
return f"isLiteral ( {self.current_variable} )"
def _data_range_condition(self, node):
variable = self.current_variable
if isinstance(node, OWLDatatype):
return self._datatype_condition(node)
if isinstance(node, OWLDataOneOf):
groups = defaultdict(list)
for value in node.values():
groups[value.get_datatype()].append(self.render(value))
return " || ".join(
f"( ( {self._datatype_condition(datatype)} ) && ( {' || '.join(f'{variable} = {value}' for value in values)} ) )"
for datatype, values in groups.items()
) or "1=0"
if isinstance(node, OWLDataComplementOf):
return f"!( {self._data_range_condition(node.get_data_range())} )"
if isinstance(node, (OWLDataIntersectionOf, OWLDataUnionOf)):
operator = " && " if isinstance(node, OWLDataIntersectionOf) else " || "
return operator.join(f"( {self._data_range_condition(op)} )" for op in node.operands())
if isinstance(node, OWLDatatypeRestriction):
conditions = [self._datatype_condition(node.get_datatype())]
for fr in node.get_facet_restrictions():
facet = fr.get_facet()
value = self.render(fr.get_facet_value())
if facet in _Variable_facet_comp:
conditions.append(f"{variable} {_Variable_facet_comp[facet]} {value}")
elif facet in (OWLFacet.LENGTH, OWLFacet.MIN_LENGTH, OWLFacet.MAX_LENGTH):
if node.get_datatype().to_string_id() not in _STRING_DATATYPES | {str(XSD.anyURI)}:
raise NotImplementedError(f"SPARQL conversion does not support {facet.symbolic_form} for {node.get_datatype()}")
comparator = {OWLFacet.LENGTH: "=", OWLFacet.MIN_LENGTH: ">=", OWLFacet.MAX_LENGTH: "<="}[facet]
conditions.append(f"STRLEN ( STR ( {variable} ) ) {comparator} {value}")
else:
raise NotImplementedError(f"SPARQL conversion does not support the {facet.symbolic_form} facet")
return " && ".join(f"( {condition} )" for condition in conditions)
raise NotImplementedError(f"SPARQL conversion does not support the data range {node}")
[docs]
def new_count_var(self) -> str:
self.cnt += 1
return f"?cnt_{self.cnt}"
[docs]
def append_triple(self, subject, predicate, object_):
self.append(self.triple(subject, predicate, object_))
[docs]
def append(self, frag):
self.sparql.append(frag)
[docs]
def triple(self, subject, predicate, object_):
return f"{self._maybe_quote(subject)} {self._maybe_quote_p(predicate)} {self._maybe_render(object_)} . "
[docs]
def as_query(self,
root_variable: str,
ce: OWLClassExpression,
for_all_de_morgan: bool = True,
count: bool = False,
values: Optional[Iterable[OWLNamedIndividual]] = None,
named_individuals: bool = False,
validate: bool = False,
negation_default: bool = True,
sub_property_resolver: Optional[Callable] = None,
inverse_property_resolver: Optional[Callable] = None,
subclass_resolver: Optional[Callable] = None) -> str:
assert isinstance(ce,OWLClassExpression), f"ce must be an instance of OWLClassExpression. Currently {type(ce)}"
# root variable: the variable that will be projected
# ce: the class expression to be transformed to a SPARQL query
# for_all_de_morgan: true -> ¬(∃r.¬C), false -> (∀r.C)
# count: True, counts the results ; False, projects the individuals
# values: positive or negative examples from a class expression problem
# named_individuals: if set to True, the generated SPARQL query will return only entities that are instances
# of owl:NamedIndividual
# validate: if set to True, validates the generated SPARQL query using rdflib.parseQuery (slower but safer)
qs = ["SELECT"]
tp = self.convert(root_variable, ce, for_all_de_morgan=for_all_de_morgan, named_individuals=named_individuals,
negation_default=negation_default, sub_property_resolver=sub_property_resolver,
inverse_property_resolver=inverse_property_resolver, subclass_resolver=subclass_resolver)
if count:
qs.append(f" ( COUNT ( DISTINCT {root_variable} ) AS ?cnt ) WHERE {{ ")
else:
qs.append(f" DISTINCT {root_variable} WHERE {{ ")
if values is not None and root_variable.startswith("?"):
q = [f"VALUES {root_variable} {{ "]
for x in values:
q.append(f"<{x.to_string_id()}>")
q.append("} . ")
qs.extend(q)
if named_individuals:
qs.append(f"{root_variable} a <{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}> . ")
qs.extend(tp)
qs.append(" }")
query = "\n".join(qs)
if validate:
parseQuery(query)
return query
[docs]
def as_confusion_matrix_query(self,
root_variable: str,
ce: OWLClassExpression,
positive_examples: Iterable[OWLNamedIndividual],
negative_examples: Iterable[OWLNamedIndividual],
for_all_de_morgan: bool = True,
named_individuals: bool = False,
validate: bool = False) -> str:
# get the graph pattern corresponding to the provided class expression (ce)
tp = self.convert(root_variable, ce, for_all_de_morgan=for_all_de_morgan, named_individuals=named_individuals)
if named_individuals:
named_individual_triple = f"{root_variable} a <{OWLRDFVocabulary.OWL_NAMED_INDIVIDUAL.as_str()}> . "
graph_pattern_str = named_individual_triple + "".join(tp)
else:
graph_pattern_str = "".join(tp)
# preparation for the final query
# required to compute false negatives
number_of_positive_examples = 0
# required to compute true negatives
number_of_negative_examples = 0
# string representation of the positive examples (to be passed to the first VALUES clause)
positive_examples_as_str = ""
# iterate over the positive examples
for positive_example in positive_examples:
number_of_positive_examples += 1
positive_examples_as_str += f"<{positive_example.to_string_id()}> "
assert (len(positive_examples_as_str) > 0)
# string representation of the positive examples (to be passed to the first VALUES clause)
negative_examples_as_str = ""
# iterate over the negative examples
for negative_example in negative_examples:
number_of_negative_examples += 1
negative_examples_as_str += f"<{negative_example.to_string_id()}> "
assert(len(negative_examples_as_str) > 0)
# create the sparql query
sparql_str= f"""
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
SELECT * WHERE {{
{{
SELECT (COUNT(DISTINCT {root_variable}) as ?tp) (({number_of_positive_examples} - COUNT(DISTINCT {root_variable})) as ?fn) WHERE {{
VALUES {root_variable} {{ {positive_examples_as_str} }}
{graph_pattern_str}
}}
}}
{{
SELECT (COUNT(DISTINCT {root_variable}) as ?fp) (({number_of_negative_examples} - COUNT(DISTINCT {root_variable})) as ?tn) WHERE {{
VALUES {root_variable} {{ {negative_examples_as_str} }}
{graph_pattern_str}
}}
}}
}}
"""
if validate:
parseQuery(sparql_str)
return sparql_str
converter = Owl2SparqlConverter()
[docs]
def owl_expression_to_sparql(expression: OWLClassExpression = None,
root_variable: str = "?x",
values: Optional[Iterable[OWLNamedIndividual]] = None,
for_all_de_morgan: bool = True,
named_individuals: bool = False,
validate: bool = False,
negation_default: bool = True,
sub_property_resolver: Optional[Callable] = None,
inverse_property_resolver: Optional[Callable] = None,
subclass_resolver: Optional[Callable] = None) -> str:
"""Convert an OWL Class Expression (https://www.w3.org/TR/owl2-syntax/#Class_Expressions) into a SPARQL query
root variable: the variable that will be projected
expression: the class expression to be transformed to a SPARQL query
values: positive or negative examples from a class expression problem. Unclear
for_all_de_morgan: if set to True, the SPARQL mapping will use the mapping containing the nested FILTER NOT EXISTS
patterns for the universal quantifier (¬(∃r.¬C)), instead of the counting query
named_individuals: if set to True, the generated SPARQL query will return only entities
that are instances of owl:NamedIndividual
validate: if set to True, validates the generated SPARQL query using rdflib.parseQuery (slower but safer)
negation_default: if True (closed world, the default), OWLObjectComplementOf is translated via
FILTER NOT EXISTS; if False (open world), OWLObjectComplementOf always contributes no results.
sub_property_resolver: optional callable mapping a property to its sub-properties, used to expand
every rendered property into a SPARQL property-path alternation of itself and its sub-properties.
inverse_property_resolver: optional callable mapping an object property to its declared owl:inverseOf
partner(s), used to expand every rendered object property with a reverse (`^`) path alternative.
subclass_resolver: optional callable mapping a class to its subclasses, used to expand every named
class membership triple to also match individuals typed only at a subclass.
"""
assert expression is not None, "expression cannot be None"
return converter.as_query(root_variable, expression, count=False, values=values,
named_individuals=named_individuals, for_all_de_morgan=for_all_de_morgan,
validate=validate, negation_default=negation_default,
sub_property_resolver=sub_property_resolver,
inverse_property_resolver=inverse_property_resolver,
subclass_resolver=subclass_resolver)
[docs]
def owl_expression_to_sparql_with_confusion_matrix(expression: OWLClassExpression,
positive_examples: Optional[Iterable[OWLNamedIndividual]],
negative_examples: Optional[Iterable[OWLNamedIndividual]],
root_variable: str = "?x",
for_all_de_morgan: bool = True,
named_individuals: bool = False,
validate: bool = False) -> str:
"""Convert an OWL Class Expression (https://www.w3.org/TR/owl2-syntax/#Class_Expressions) into a SPARQL query
root variable: the variable that will be projected
expression: the class expression to be transformed to a SPARQL query
positive_examples: positive examples from a class expression problem
negative_examples: positive examples from a class expression problem
for_all_de_morgan: if set to True, the SPARQL mapping will use the mapping containing the nested FILTER NOT EXISTS
patterns for the universal quantifier (¬(∃r.¬C)), instead of the counting query
named_individuals: if set to True, the generated SPARQL query will return only entities
that are instances of owl:NamedIndividual
validate: if set to True, validates the generated SPARQL query using rdflib.parseQuery (slower but safer)
"""
assert expression is not None, "expression cannot be None"
assert positive_examples is not None, "positive examples cannot be None"
assert negative_examples is not None, "negative examples cannot be None"
return converter.as_confusion_matrix_query(root_variable,
expression,
positive_examples=positive_examples,
negative_examples=negative_examples,
named_individuals=named_individuals,
for_all_de_morgan=for_all_de_morgan,
validate=validate)