Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improved get_unabbreviated_triples() #449

Merged
merged 9 commits into from
Sep 8, 2022
97 changes: 76 additions & 21 deletions ontopy/ontology.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
If desirable some of these additions may be moved back into owlready2.
"""
# pylint: disable=too-many-lines,fixme,arguments-differ,protected-access
from typing import TYPE_CHECKING, Union, Sequence
from typing import TYPE_CHECKING, Optional, Union, Sequence
import os
import itertools
import inspect
Expand Down Expand Up @@ -133,17 +133,9 @@ def get_unabbreviated_triples(

If `blank` is given, it will be used to represent blank nodes.
"""

def _unabbreviate(i):
if isinstance(i, int):
# negative storid corresponds to blank nodes
if i >= 0:
return self._unabbreviate(i)
return BlankNode(self, i) if blank is None else blank
return i

for s, p, o in self.get_triples(subject, predicate, obj):
yield (_unabbreviate(s), _unabbreviate(p), _unabbreviate(o))
return _get_unabbreviated_triples(
self, subject=subject, predicate=predicate, obj=obj, blank=blank
)


class Ontology(owlready2.Ontology): # pylint: disable=too-many-public-methods
Expand Down Expand Up @@ -252,16 +244,21 @@ def __eq__(self, other):
other.get_unabbreviated_triples(blank="_:b")
)

def get_unabbreviated_triples(self, blank=None):
"""Returns all triples unabbreviated.
def get_unabbreviated_triples(
self, subject=None, predicate=None, obj=None, blank=None
):
"""Returns all matching triples unabbreviated.

If `label` is given, it will be used to represent blank nodes.
If `blank` is given, it will be used to represent blank nodes.
"""
return World.get_unabbreviated_triples(self, blank=blank)
# pylint: disable=invalid-name
return _get_unabbreviated_triples(
self, subject=subject, predicate=predicate, obj=obj, blank=blank
)

def get_by_label(
self, label: str, label_annotations: str = None, prefix: str = None
): # pylint: disable=too-many-arguments,too-many-branches
):
"""Returns entity with label annotation `label`.

Args:
Expand All @@ -286,6 +283,7 @@ def get_by_label(
The current implementation also supports "*" as a wildcard
matching any number of characters. This may change in the future.
"""
# pylint: disable=too-many-arguments,too-many-branches
if not isinstance(label, str):
raise TypeError(
f"Invalid label definition, must be a string: {label!r}"
Expand Down Expand Up @@ -392,8 +390,12 @@ def add_label_annotation(self, iri):
if self._label_annotations is None:
self._label_annotations = []
label_annotation = iri if hasattr(iri, "storid") else self.world[iri]
if not label_annotation:
raise ValueError(f"IRI not in ontology: {iri}")
if label_annotation is None:
warnings.warn(f"adding new IRI to ontology: {iri}")
name = iri.rsplit("/")[-1].rsplit("#")[-1]
bases = (owlready2.AnnotationProperty,)
with self:
label_annotation = types.new_class(name, bases)
if label_annotation not in self._label_annotations:
self._label_annotations.append(label_annotation)

Expand Down Expand Up @@ -532,7 +534,7 @@ def _load( # pylint: disable=too-many-arguments,too-many-locals,too-many-branch
"""Help function for load()."""
web_protocol = "http://", "https://", "ftp://"

url = filename if filename else self.base_iri.rstrip("/#")
url = str(filename) if filename else self.base_iri.rstrip("/#")
if url.startswith(web_protocol):
baseurl = os.path.dirname(url)
catalogurl = baseurl + "/" + catalog_file
Expand Down Expand Up @@ -1157,7 +1159,7 @@ class prefLabel(owlready2.label):
counter = 0
while f"{self.base_iri}{name_prefix}{counter}" in self:
counter += 1
obj.name = name_prefix + str(counter)
obj.name = f"{name_prefix}{counter}"
elif name_policy is not None:
raise TypeError(f"invalid name_policy: {name_policy!r}")

Expand Down Expand Up @@ -1616,3 +1618,56 @@ def flatten(items):
yield sub_item
else:
yield item


def _unabbreviate(
onto: Union[World, Ontology],
storid: Union[int, str],
blank: Optional[str] = None,
):
"""Help function returning unabbreviation of `storid`.

The `storid` argument is normally be an integer corresponding to
a store id. If it is not an integer, it is assumed to already
be unabbreviated and returned as is.

If `blank` is given, it will be used to represent blank nodes
(corresponding to a negative store id).
"""
if isinstance(storid, int):
# negative storid corresponds to blank nodes
if storid >= 0:
return onto._unabbreviate(storid)
return BlankNode(onto, storid) if blank is None else blank
return storid


def _get_unabbreviated_triples(
self, subject=None, predicate=None, obj=None, blank=None
):
"""Help function returning all matching triples unabbreviated.

If `blank` is given, it will be used to represent blank nodes.
"""
# pylint: disable=invalid-name
abb = (
None if subject is None else self._abbreviate(subject),
None if predicate is None else self._abbreviate(predicate),
None if obj is None else self._abbreviate(obj),
)
for s, p, o in self._get_obj_triples_spo_spo(*abb):
yield (
_unabbreviate(self, s, blank=blank),
_unabbreviate(self, p, blank=blank),
_unabbreviate(self, o, blank=blank),
)
for s, p, o, d in self._get_data_triples_spod_spod(*abb, d=""):
yield (
_unabbreviate(self, s, blank=blank),
_unabbreviate(self, p, blank=blank),
f'"{o}"{d}'
if isinstance(d, str)
else f'"{o}"^^{_unabbreviate(self, d)}'
if d
else o,
)