def hierarchy(
path_str_or_graph: Union[Path, str, Graph],
graph_iri: Optional[Union[str, URIRef]] = None,
use_names: bool = False,
) -> None:
"""Print the class, property and concept hierarchies in an RDF graph.
``path_str_or_graph`` may be an RDF file, serialized RDF, or an RDFLib
graph. ``graph_iri`` selects one named graph and is only valid for a remote
URL or a ``.trig``/``.jsonld`` file. Without it, the source is parsed as a
context-less graph. Resources are displayed as namespace-qualified names
when possible. If ``use_names`` is true, names are selected in order from
``skos:prefLabel``, ``dcterms:title``, ``schema:name`` and ``rdfs:label``,
with IRIs used as a fallback. Separate hierarchy roots (and separate
hierarchy kinds) are divided by a blank line.
"""
is_remote = isinstance(path_str_or_graph, str) and path_str_or_graph.startswith(
"http"
)
is_named_graph_file = isinstance(path_str_or_graph, Path) and (
path_str_or_graph.suffix.lower() in {".trig", ".jsonld"}
)
if graph_iri is not None:
if not (is_remote or is_named_graph_file):
raise ValueError(
"graph_iri is only allowed for a remote HTTP source or a "
".trig/.jsonld file"
)
if not isinstance(graph_iri, URIRef):
graph_iri = URIRef(graph_iri)
dataset = _parse_dataset(path_str_or_graph)
graph = dataset.graph(graph_iri)
elif isinstance(path_str_or_graph, Graph):
graph = Graph()
for prefix, namespace in path_str_or_graph.namespaces():
graph.bind(prefix, namespace)
for triple in path_str_or_graph:
graph.add(triple)
elif isinstance(path_str_or_graph, Path):
graph = Graph().parse(path_str_or_graph)
else:
graph = load_graph(path_str_or_graph)
class_types = {OWL.Class, RDFS.Class}
property_types = {
RDF.Property,
URIRef(f"{RDFS}Property"),
OWL.ObjectProperty,
OWL.DatatypeProperty,
OWL.AnnotationProperty,
OWL.FunctionalProperty,
OWL.InverseFunctionalProperty,
OWL.SymmetricProperty,
OWL.TransitiveProperty,
}
def typed_resources(types: set[URIRef]) -> set:
return {
subject
for rdf_type in types
for subject in graph.subjects(RDF.type, rdf_type)
}
def display_name(resource) -> str:
if use_names:
name_predicates = (
SKOS.prefLabel,
DCTERMS.title,
URIRef("https://schema.org/name"),
RDFS.label,
)
for predicate in name_predicates:
values = sorted(graph.objects(resource, predicate), key=str)
if values:
return str(values[0])
if isinstance(resource, URIRef):
try:
return graph.namespace_manager.normalizeUri(resource)
except Exception: # RDFLib may reject an IRI it cannot compact.
return f"<{resource}>"
return resource.n3(graph.namespace_manager)
def forests(
nodes: set,
predicates: tuple[URIRef, ...],
inverse=(),
root_links: tuple[URIRef, ...] = (),
inverse_root_links: tuple[URIRef, ...] = (),
membership_predicate: Optional[URIRef] = None,
container_roots: set = frozenset(),
) -> list[str]:
children = {node: set() for node in nodes}
parents = {node: set() for node in nodes}
def add_edge(parent, child) -> None:
if parent in nodes and child in nodes:
children[parent].add(child)
parents[child].add(parent)
for predicate in predicates:
for child, parent in graph.subject_objects(predicate):
add_edge(parent, child)
for predicate in inverse:
for parent, child in graph.subject_objects(predicate):
add_edge(parent, child)
for predicate in root_links:
for child, parent in graph.subject_objects(predicate):
add_edge(parent, child)
for predicate in inverse_root_links:
for parent, child in graph.subject_objects(predicate):
add_edge(parent, child)
if membership_predicate is not None:
for child, parent in graph.subject_objects(membership_predicate):
# inScheme expresses membership, not a direct hierarchy edge.
# Attach only concepts that do not already have a broader
# concept or an explicit top-concept relationship.
if child in nodes and not parents[child]:
add_edge(parent, child)
# Ontologies do not have a standard predicate linking them to every
# declared class or property. Likewise, some SKOS sources omit
# inScheme/top-concept links. When there is one unambiguous container,
# place all otherwise top-level resources beneath it.
if len(container_roots) == 1:
container = next(iter(container_roots))
for node in sorted(nodes - container_roots, key=display_name):
if not parents[node]:
add_edge(container, node)
visited = set()
active = []
active_set = set()
def check_for_cycle(node) -> None:
if node in active_set:
cycle_start = active.index(node)
cycle = active[cycle_start:] + [node]
raise ValueError(
"Cycle detected in hierarchy: "
+ " -> ".join(display_name(item) for item in cycle)
)
if node in visited:
return
active.append(node)
active_set.add(node)
for child in sorted(children[node], key=display_name):
check_for_cycle(child)
active.pop()
active_set.remove(node)
visited.add(node)
for node in sorted(nodes, key=display_name):
check_for_cycle(node)
connected = {node for node in nodes if children[node] or parents[node]}
roots = sorted(
(node for node in connected if not parents[node]), key=display_name
)
covered = set()
output = []
def render(node, prefix="", connector=""):
output.append(f"{prefix}{connector}{display_name(node)}")
covered.add(node)
descendants = sorted(children[node], key=display_name)
for index, child in enumerate(descendants):
last = index == len(descendants) - 1
render(
child,
prefix
+ (" " if connector == "└── " else "│ " if connector else ""),
"└── " if last else "├── ",
)
for root in roots:
if output:
output.append("")
render(root)
# This also covers a component reached from an already-rendered node in
# a hierarchy where a resource has more than one parent.
for node in sorted(connected - covered, key=display_name):
if node in covered:
continue
if output:
output.append("")
render(node)
return output
sections = []
ontologies = typed_resources({OWL.Ontology})
class_lines = forests(
typed_resources(class_types) | ontologies,
(RDFS.subClassOf,),
container_roots=ontologies,
)
if class_lines:
sections.append("\n".join(class_lines))
property_lines = forests(
typed_resources(property_types) | ontologies,
(RDFS.subPropertyOf,),
container_roots=ontologies,
)
if property_lines:
sections.append("\n".join(property_lines))
concepts = typed_resources({SKOS.Concept})
concept_schemes = typed_resources({SKOS.ConceptScheme})
concept_lines = forests(
concepts | concept_schemes,
(SKOS.broader,),
(SKOS.narrower,),
root_links=(SKOS.topConceptOf,),
inverse_root_links=(SKOS.hasTopConcept,),
membership_predicate=SKOS.inScheme,
container_roots=concept_schemes,
)
if concept_lines:
sections.append("\n".join(concept_lines))
if sections:
print("\n\n".join(sections))