#!/usr/bin/env python3
from itertools import count
from owlrl import DeductiveClosure, RDFS_Semantics
from rdflib import RDFS, Graph, Namespace, URIRef
from fosf.config import FOSF_BASE_URI
from fosf.syntax import Term
from fosf.syntax.base import (
DisjunctiveSort,
FrozenDisjunctiveSort,
Sort,
Tag,
URIFeature,
URISort,
)
from fosf.syntax.taxonomy import SortTaxonomy
from fosf.syntax.terms import DisjunctiveTerm
from fosf.syntax.theory import OsfTheory
[docs]
def term_to_query(
term: Term | DisjunctiveTerm,
vars: list | None = None,
prefixes: dict | None = None,
base: Namespace | None = None,
include_prefixes: bool = True,
include_subclassof: bool = False,
):
all_tags = term.tag_counts()
tag_counter = count(0)
def _new_tag():
while (tag := Tag(f"type{next(tag_counter)}")) in all_tags:
pass
return tag
g = Graph()
if vars is None:
vars = [term.X]
sparql_prefixes = []
if prefixes is None:
prefixes = {}
for prefix, uri in prefixes.items():
g.bind(prefix, Namespace(uri))
sparql_prefixes.append(f"prefix {prefix}: <{uri}>")
if base is None:
base = FOSF_BASE_URI
# TODO: what if base is not necessary?
g.bind("", base)
sparql_prefixes.append(f"prefix : <{base}>")
sparql_vars = " ".join(["?" + str(var) for var in vars])
sparql_prefix = "\n".join(sparql_prefixes)
def urify(symbol):
if isinstance(symbol, (URISort, URIFeature)):
return symbol
if isinstance(symbol, Sort):
return URISort(base[symbol.value])
return URIFeature(base[symbol])
rdf_type = "rdf:type/rdfs:subClassOf*" if include_subclassof else "rdf:type"
def visit(term, query, spaces, tag=None):
X = term.X
if isinstance(term, DisjunctiveTerm):
for i, disjunct in enumerate(term.terms):
if i > 0:
query += " " * spaces + "UNION\n"
query += " " * spaces + "{\n"
query = visit(disjunct, query, spaces + 2, X)
query += " " * spaces + "}\n"
return query
# Else, normal term
# if tag is not None and tag != X: # TODO: bind or filter
# query += " " * spaces + f"?{tag} = ?{X} .\n"
if term.s is not None:
s = term.s
if isinstance(s, (DisjunctiveSort, FrozenDisjunctiveSort)):
type_tag = _new_tag()
query += (
" " * spaces + f"?{X} {rdf_type} ?{type_tag} .\n" + " " * spaces
)
urified_sorts = (urify(ds).n3(g.namespace_manager) for ds in s)
query += f"VALUES ?{type_tag} {{ {' '.join(urified_sorts)} }}\n"
else:
s = urify(s).n3(g.namespace_manager)
query += " " * spaces + f"?{X} {rdf_type} {s} .\n"
for f, subterm in term.iter_subterms():
Y = subterm.X
f = urify(f).n3(g.namespace_manager)
query += " " * spaces + f"?{X} {f} ?{Y} .\n"
query = visit(subterm, query, spaces)
# if tag is not None and tag != X: # TODO: bind or filter
# query += " " * spaces + f"BIND(?{X} AS ?{tag})\n"
return query
sparql_query = f"{sparql_prefix}\n" if include_prefixes else ""
sparql_query += f"SELECT DISTINCT {sparql_vars} WHERE {{\n"
sparql_query = visit(term, sparql_query, 2) + "}"
return sparql_query
[docs]
def expand_rdf_graph(
graph: Graph,
semantics=RDFS_Semantics,
taxonomy: SortTaxonomy | None = None,
theory: OsfTheory | None = None,
):
if theory is not None and taxonomy is not None:
print(
"Warning: both a theory and a taxonomy are provided. The theory will take precedence."
)
def urify(symbol):
if isinstance(symbol, URISort):
return URIRef(symbol.value)
if isinstance(symbol, URIFeature):
return URIRef(str(symbol))
if isinstance(symbol, Sort):
return URIRef(FOSF_BASE_URI[symbol.value])
return URIRef(FOSF_BASE_URI[str(symbol)])
if theory is not None:
taxonomy = theory.taxonomy
for f, s in theory.domains.items():
graph.add((urify(f), RDFS.domain, urify(s)))
for f, s in theory.ranges.items():
graph.add((urify(f), RDFS.range, urify(s)))
if taxonomy is not None:
bot, top = taxonomy.bot, taxonomy.top
for u, v in taxonomy.graph.edges():
if u == bot or v == top:
continue
graph.add((urify(u), RDFS.subClassOf, urify(v)))
DeductiveClosure(semantics).expand(graph)
return graph