Source code for fosf.parsers.rdf

#!/usr/bin/env python3

from lark import Lark, Transformer
from rdflib import XSD, Graph, Namespace

from fosf.config import FOSF_GRAMMAR
from fosf.syntax.base import LiteralSort, URISort


class _RDFTransformer(Transformer):
    def __init__(self):
        super().__init__()
        self.prefix: dict
        self.triples: set[tuple]

    def prefix_dec(self, tree):
        if len(tree) == 1:
            prefix = ""
            uriref = URISort(tree[0])
        else:
            prefix = tree[0].value
            uriref = URISort(tree[1])
        self.prefix[prefix] = uriref

    def literal(self, tree):
        return tree[0]

    def decimal_literal(self, tree):
        if len(tree) == 2:
            value = tree[0].value + tree[1].value.strip('"')
        else:
            value = tree[0].value.strip('"')
        return LiteralSort(value, datatype=XSD.decimal)

    def integer_literal(self, tree):
        if len(tree) == 2:
            value = tree[0].value + tree[1].value.strip('"')
        else:
            value = tree[0].value.strip('"')
        return LiteralSort(value, datatype=XSD.integer)

    def simple_literal(self, tree):
        return LiteralSort(tree[0].value.strip('"'))

    def language_literal(self, tree):
        return LiteralSort(tree[0].value.strip('"'), lang=tree[1].value)

    def datatype_literal(self, tree):
        return LiteralSort(tree[0].value.strip('"'), datatype=tree[1])

    def uriref(self, tree):
        return URISort(tree[0].strip("<|>"))

    def qname(self, tree):
        prefix = tree[0].value if tree[0] is not None else ""
        if prefix not in self.prefix:
            raise KeyError(f"Prefix '{prefix}' not defined")
        prefix = self.prefix[prefix]
        local_name = tree[1].value
        uri = URISort(prefix + local_name)
        return uri

    def triple(self, tree):
        s, p, o = tree[0], tree[1], tree[2]
        self._rdf_graph.add((s, p, o))

    def rdf_graph(self, tree):
        return self._rdf_graph

    def subject(self, tree):
        return tree[0]

    def predicate(self, tree):
        return tree[0]

    def object(self, tree):
        o = tree[0]
        if isinstance(o, LiteralSort):
            return o
        return o

    def base_dec(self, tree):
        if self.base is None:
            self.base = Namespace(tree[0].value.strip("<|>"))
        else:
            new_base = Namespace(tree[0].value.strip("<|>"))
            if self.base != new_base:
                msg = (
                    "At most one base URI must be specified. "
                    f"Current base: {self.base}. Found: {new_base}"
                )
                raise RuntimeError(msg)

    def transform(self, tree):
        self.base: Namespace | None = None
        self.prefix = dict()
        self.triples = set()
        self._rdf_graph = Graph()
        return super().transform(tree)




class RDFParser:
    def __init__(self):
        self.parser = Lark.open_from_package("fosf", FOSF_GRAMMAR, start="rdf_graph")
        self.transformer = _RDFTransformer()



    def parse(self, expression: str, **kwargs):
        parse_tree = self.parser.parse(expression)
        return self.transformer.transform(parse_tree, **kwargs)