sparql.py

import io
import os
import csv
import platform
import tabulate
import operator
import itertools
import subprocess
import ifcopenshell

from collections import defaultdict

from . import mvdxml_expression


def camel(s):
    """
    Camel case conversion function
    :param s: str
    :return: camel case formatted string
    """

    s = s.title().replace(" ", "")
    if s.endswith("s"):
        s = s[:-1]
    return s[0].lower() + s[1:]


STANDARD_PREFIXES = {
    "rdf": "<http://www.w3.org/1999/02/22-rdf-syntax-ns#>",
    "owl": "<http://www.w3.org/2002/07/owl#>",
    "xsd": "<http://www.w3.org/2001/XMLSchema#>",
    "list": "<https://w3id.org/list#>",
    "ifcowl": "",
    "express": "<https://w3id.org/express#>",
}


def derive_prefix(ttlfn):
    with open(ttlfn, "r") as f:
        for ln in f:
            ln.strip()
            if ln.startswith("@prefix ifcowl"):
                uri = ln.split(":", 1)[1].strip()[:-1].strip()
                STANDARD_PREFIXES["ifcowl"] = uri
                return uri
    raise ValueError(f"No ifcowl prefix found in {ttlfn!r}")


def withschema(fn):
    """
    Decorator that takes a function and adds an IFC latebound schema definition
    in the first parameter. The schema identifier is looked up based on the
    global ifcOwl prefix.

    :param fn: input function
    :return: decorated function
    """

    def _(*args, **kwargs):
        schema_name = STANDARD_PREFIXES["ifcowl"].split("/")[-1][:-2]
        try:
            S = ifcopenshell.ifcopenshell_wrapper.schema_by_name(schema_name)
        except RuntimeError:
            if "_" not in schema_name:
                raise
            S = ifcopenshell.ifcopenshell_wrapper.schema_by_name(schema_name.split("_")[0])
        return fn(S, *args, **kwargs)

    return _


noop = lambda *args: None


class rule_binding(object):
    """
    Object for mapping rules to generated SPARQL variables
    """

    def __init__(self):
        pass


class builder(object):
    """
    A helper class for dealing with SPARQL query statements
    """

    def __init__(self):
        self.prefixes = {"express": "", "ifcowl": ""}
        self.statements = []

    def append(self, *stmt):
        if len(stmt) == 3:
            for i, pos in enumerate(stmt[1:]):
                for po in pos.split("/"):
                    if ":" in pos:
                        a, b = po.split(":")
                        self.prefixes[a] = ""
        self.statements.append(stmt)

    def bind(self, di):
        for k in set(self.prefixes.keys()) & set(di.keys()):
            self.prefixes[k] = di[k]

    def x(self):
        return len(self.statements)

    def __repr__(self):
        def f(s):
            S = " ".join(s)
            if len(s) == 3:
                S += " ."
            return S

        def g(s):
            return "PREFIX %s: %s" % s

        return "\n".join(
            itertools.chain(
                (g(s) for s in self.prefixes.items()), (f(s) for s in self.statements)
            )
        )


class ifcOwl(object):
    """
    Helper class with static function for dealing with ifcOwl attribute names
    """

    @staticmethod
    @withschema
    def supertypes(S, entity):
        """
        Yields ifcOwl subtypes for the supplied entity name

        :param S: schema definition (from decorator)
        :param entity: entity name string
        :return:
        """

        a, b = entity.split("#")
        try:
            en = S.declaration_by_name(b)
            if en.__class__.__name__ == "entity":
                while en.supertype():
                    yield "%s#%s" % (a, en.supertype().name())
                    en = en.supertype()
        except:
            pass

    @staticmethod
    def get_names(e, c):
        """
        Returns inverse or forward attribute names for latebound entity definition

        :param e: latebound entity definition
        :param c: either 'all_attributes' or 'all_inverse_attributes'
        :return: set of attribute names
        """

        return set(map(lambda a: a.name(), getattr(e, c)()))

    @staticmethod
    @withschema
    def is_boxed(S, entity, attribute, predCount=0):
        """
        Returns whether the entity attribute should be boxed in ifcOwl. Which means
        that there is an additional indirection.

        inst:IfcRelDefinesByType_21937
            ifcowl:globalId_IfcRoot      inst:IfcGloballyUniqueId_117684 ;

        inst:IfcGloballyUniqueId_117684
            rdf:type           ifcowl:IfcGloballyUniqueId ;
            express:hasString  "2P9FPkykn0r8rCpmBxZH0w" .

        :param S: schema definition (from decorator)
        :param entity: entity name string
        :param attribute: attribute name string
        :param predCount: numeric identifier to postfix predicate identifier in case of SELECT types
        :return: either a predicate from the express namespace or a variable postfixed with predCount
        """

        en = S.declaration_by_name(entity)
        attr = [a for a in en.all_attributes() if a.name() == attribute][0]
        ty = attr.type_of_attribute()
        is_boxed = False
        while isinstance(ty, ifcopenshell.ifcopenshell_wrapper.named_type):
            ty = ty.declared_type()

        if isinstance(ty, ifcopenshell.ifcopenshell_wrapper.select_type):

            # Just assume there is going to be some boxed type in here.
            # It could be all instance references, but fact is we don't know at this moment.
            # It's likely that mvdXML will only bind to literals?

            return "?pred%d" % predCount

        else:

            while isinstance(ty, ifcopenshell.ifcopenshell_wrapper.type_declaration):
                is_boxed = True
                ty = ty.declared_type()
            if is_boxed and isinstance(
                ty, ifcopenshell.ifcopenshell_wrapper.simple_type
            ):
                ty = ty.declared_type()
                return "express:has%s%s" % (ty[0].upper(), ty[1:])

            return False

    @staticmethod
    @withschema
    def name(S, entity, attribute):
        """
        Names the entity attribute according to ifcOwl

        :param S:
        :param entity:
        :param attribute:
        :return:
        """
        en = S.declaration_by_name(entity)

        while True:
            st = en.supertype()

            attribute_names = ifcOwl.get_names(en, "all_attributes") | ifcOwl.get_names(
                en, "all_inverse_attributes"
            )

            if st:
                attribute_names -= ifcOwl.get_names(
                    st, "all_attributes"
                ) | ifcOwl.get_names(st, "all_inverse_attributes")

            if attribute in attribute_names:
                return (
                    "ifcowl:" + attribute[0].lower() + attribute[1:] + "_" + en.name()
                )

            en = st

            if en is None:
                raise AttributeError("%s not found on %s" % (attribute, entity))

    @staticmethod
    @withschema
    def is_select(S, decl_name):
        """
        Returns True when the declaration is a select type

        :param S:
        :param decl_name:
        :return:
        """
        decl = S.declaration_by_name(decl_name)
        return isinstance(decl, ifcopenshell.ifcopenshell_wrapper.select_type)

    @staticmethod
    @withschema
    def is_inverse(S, entity, attribute):
        """
        When entity attribute is an INVERSE attribute, returns the opposite
        forward entity and attribute name. Otherwise returns (False, False)

        :param S:
        :param entity:
        :param attribute:
        :return:
        """

        en = S.declaration_by_name(entity)
        attrs = [a for a in en.all_inverse_attributes() if a.name() == attribute]
        if not attrs:
            return False, False

        a = attrs[0]
        assert a.type_of_aggregation_string() == "set"
        entity = a.entity_reference().name()
        attr = a.attribute_reference().name()
        return "ifcowl:" + entity, ifcOwl.name(entity, attr)


class convertor(object):

    @staticmethod
    def convert(item, *args, **kwargs):
        return getattr(convertor, item.__class__.__name__)(item, *args, **kwargs)

    @staticmethod
    def concept_or_applicability(concept):
        """
        Convert the Template (SELECT ... WHERE {}) structure and TemplateRule (FILTER)

        :param qtype: 0 or 1, 0 for a general query matching the applicableRootEntity
        :return:
        """

        bld = builder()
        t = concept.template()
        bld.append("# %s" % camel(concept.root_name))
        convertor.template(t, bld, concept.root_entity)
        bld.bind(STANDARD_PREFIXES)

        return bld

    @staticmethod
    def root(rootEntity):
        args = ["URI", "GlobalId"]

        b = builder()
        b.args = args
        b.append("SELECT " + " ".join("?" + a for a in args) + " WHERE {")
        b.append("?URI", "rdf:type", "ifcowl:%s" % rootEntity)
        b.append("?URI", "ifcowl:globalId_IfcRoot/express:hasString", "?GlobalId")
        b.append("}")
        b.bind(STANDARD_PREFIXES)

        return b

    @staticmethod
    def template(template, bld=None, rootEntity=None):
        if bld is None:
            bld = builder()

        if rootEntity is None:
            rootEntity = template.entity

        args = ["URI", "GlobalId"]

        def enumerate(rule, **kwargs):
            if rule.bind:
                args.append(rule.bind)

        template.traverse(enumerate)

        bld.args = args

        bld.append("SELECT " + " ".join("?" + a for a in args) + " WHERE {")

        args = set(args)

        bld.append("?URI", "rdf:type", "ifcowl:%s" % rootEntity)
        bld.append("?URI", "ifcowl:globalId_IfcRoot/express:hasString", "?GlobalId")

        nm = "?URI"
        ROOT = type("_", (), {"attribute": rootEntity})()
        # rule_stack = [ROOT]
        # name_stack = [nm]
        # callback_stack = [noop]
        # G = type('_', (object,), dict(indent = 0, nm = nm, next_nm=None, first=True))

        rule_mapping = defaultdict(rule_binding)
        rule_mapping[ROOT].name = nm

        def build(rule, parents):
            # print "AAA", rule.tag, parent.tag if parent else parent
            # print map(id, rule_stack)
            # print name_stack
            INDENT = " " * (len(parents) * 2)
            return_value = None

            if rule.optional:
                bld.append(INDENT + "OPTIONAL {")
                return_value = lambda: bld.append(INDENT + "}")

            if rule.tag == "EntityRule":
                siblings = [
                    child
                    for child in getattr(parents[-1], "nodes", ())
                    if child.tag == "EntityRule"
                ]
                if len(siblings) > 1:
                    bld.append(INDENT + "{")

                    def close_alternative():
                        bld.append(INDENT + ("} UNION" if rule is not siblings[-1] else "}"))

                    return_value = close_alternative
                # G.nm = G.next_nm

                if not ifcOwl.is_select(rule.attribute):
                    # SELECT types should never be qualified as they cannot be inferred
                    bld.append(
                        INDENT + rule_mapping[parents[-1]].name,
                        "rdf:type",
                        "ifcowl:" + rule.attribute,
                    )

                # propagate binding name
                rule_mapping[rule].name = rule_mapping[parents[-1]].name
            else:

                # if rule_stack[-1] is parent:
                #     # print "sl", id(parent), id(rule)
                #     # same level
                #     pass
                # elif parent in rule_stack:
                #     # print "up", id(parent), id(rule)
                #     while rule_stack[-1] is not parent:
                #        #  rule_stack.pop()
                #         name_stack.pop()
                #         callback_stack.pop()()
                # else:
                #     # print "dn", id(parent), id(rule)
                #     pass

                indirect = False

                if rule.bind:
                    if len(rule.nodes) == 1:
                        indirect = ifcOwl.is_boxed(
                            parents[-1].attribute, rule.attribute, predCount=bld.x()
                        )

                if rule.bind and not indirect:
                    next_nm = "?" + rule.bind
                else:
                    next_nm = "?var%d" % bld.x()

                rule_mapping[rule].name = next_nm

                inventy, invattr = ifcOwl.is_inverse(
                    parents[-1].attribute, rule.attribute
                )
                if invattr:
                    # This seems not to be necessary, because the entity name is also stated in mvdXML
                    # q.append(
                    #     next_nm,
                    #     "rdf:type",
                    #     inventy
                    # )
                    bld.append(
                        INDENT + next_nm, invattr, rule_mapping[parents[-1]].name
                    )
                else:
                    bld.append(
                        INDENT + rule_mapping[parents[-1]].name,
                        ifcOwl.name(parents[-1].attribute, rule.attribute),
                        next_nm,
                    )

                if rule.bind and indirect:
                    # For boxed literals, only strings atm
                    bld.append(INDENT + next_nm, indirect, "?" + rule.bind)

            # rule_stack.append(rule)
            # name_stack.append(next_nm)
            # if rule.optional:
            #     callback_stack.append(lambda: q.append(INDENT+"}"))
            # else:
            #     callback_stack.append(noop)

            # print(q.statements[-1])

            return return_value

        template.traverse(build, root=ROOT, with_parents=True)

        # while callback_stack:
        #     callback_stack.pop()()

        if template.constraints:
            bld.append(convertor.build_filter(template))

        bld.append("}")

        return bld

    @staticmethod
    def build_filter(self):
        def v(p):
            if isinstance(p, mvdxml_expression.node):
                if p.b == "Value":
                    if p.c.lower() in {"true", "false"}:
                        yield "(%s?%s)" % ("!" if p.c.lower() == "false" else "", p.a)
                    else:
                        yield "(?%s = %s)" % (p.a, p.c)
                elif p.b == "Exists":
                    yield "(!isBLANK(?%s))" % p.a
                else:
                    raise Exception("Unsupported " + p.b)
            elif isinstance(p, str):
                yield {"and": "&&", "or": "||", "not": "&& !"}[p.lower()]
            else:
                yield "("
                for q in p:
                    yield from v(q)
                yield ")"

        return "FILTER(%s)" % " ".join(v(self.constraints))


def infer_subtypes(ttlfn):
    # Disabled currently
    return ttlfn

    if not os.path.exists(ttlfn + ".subclass.nt"):

        import hashlib
        import rdflib

        a = rdflib.namespace.RDF.type

        # Hardly possible on Windows
        # graph = rdflib.Graph("Sleepycat")
        # graph.open("store", create=True)
        # graph.parse(ttlfn)

        from sqlalchemy import create_engine
        from rdflib_sqlalchemy.store import SQLAlchemy

        if os.path.exists("db.sqlite"):
            os.unlink("db.sqlite")

        uri = rdflib.Literal("sqlite:///%(here)s/db.sqlite" % {"here": os.getcwd()})
        ident = rdflib.URIRef(hashlib.sha1(ttlfn.encode()).hexdigest())
        engine = create_engine(uri)
        store = SQLAlchemy(
            identifier=ident,
            engine=engine,
        )
        graph = rdflib.Graph(
            store,
            identifier=ident,
        )
        graph.open(uri, create=True)
        graph.parse(ttlfn, format="ttl")

        # print out all the triples in the graph
        def _():
            for subject, predicate, object in graph:
                if predicate == a:
                    for sup in ifcOwl.supertypes(object):
                        yield subject, a, rdflib.URIRef(sup)

        for stmt in list(_()):
            graph.add(stmt)

        graph.serialize(destination=ttlfn + ".subclass.nt", format="nt")

    ttlfn += ".subclass.nt"


if platform.system() == "Windows" and (jena_home := os.environ.get("JENA_HOME")):
    JENA_SPARQL = os.path.join(jena_home, "bat", "sparql.bat")
else:
    JENA_SPARQL = "sparql"


class executor(object):
    @staticmethod
    def run(CR, fn, ttlfn):
        """
        Generates SPARQL queries for the parsed MVD and executes on the building model

        :param CR: A parsed concept root
        :param fn: A filename used as the prefix to store generate SPARQL queries to disk
        :param ttlfn: A filename for the LD representation of an IFC model
        :return:
        """

        report = io.StringIO()

        def dict_to_list(headers):
            return lambda di: [di[h] for h in headers]

        def execute(query, *args):
            sparqlfn = ".".join(itertools.chain([fn], map(str, args))) + ".sparql"
            with open(sparqlfn, "w") as f:
                print(query, file=f)

            proc = subprocess.Popen(
                [
                    JENA_SPARQL,
                    "--data=" + ttlfn,
                    "--query=" + sparqlfn,
                    "--results=CSV",
                ],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            stdout, stderr = proc.communicate()

            csvf = io.StringIO(stdout.decode("utf-8"))

            return list(csv.DictReader(csvf))

        root_query = convertor.root(CR.entity)
        roots = execute(root_query, 0)

        print(
            "\nFile contains %d elements of type %s" % (len(roots), CR.entity),
            file=report,
        )

        passing_all = {}

        # for summary below
        num_columns = 0

        try:
            # Full MVD with multiple concepts
            is_template = False
            concept_enumerator = list(
                itertools.chain([CR.applicability()], CR.concepts())
            )
        except:
            is_template = True
            concept_enumerator = [CR]

        for ci, C in enumerate(concept_enumerator):

            num_columns += 1

            if is_template or ci > 1:
                print("\n%s" % C.name, file=report)
            else:
                print("\nApplicability", file=report)

            query = convertor.convert(C)

            print("\nSPARQL query", file=report)
            print("============", file=report)
            print(query, file=report)

            passing = execute(query, ci, 1)
            passing_guids = set(r["GlobalId"] for r in passing)

            print("\nElements passing", file=report)
            print(
                tabulate.tabulate(
                    list(map(dict_to_list(query.args), passing)),
                    query.args,
                    tablefmt="grid",
                ),
                file=report,
            )

            print("\nElements failing concept", file=report)
            hd = ["URI", "GlobalId"]
            print(
                tabulate.tabulate(
                    list(
                        map(
                            dict_to_list(hd),
                            [r for r in roots if r["GlobalId"] not in passing_guids],
                        )
                    ),
                    hd,
                    tablefmt="grid",
                ),
                file=report,
            )

            passing_all[ci] = passing_guids

        print("\nSummary", file=report)

        for ci, C in enumerate(concept_enumerator):
            print("(%d) %s" % (ci + (0 if is_template else 0), C.name), file=report)

        def get_stats(guid):
            v = lambda i: guid in passing_all[i]
            st = [guid] + ["x" if v(i) else "" for i in range(num_columns)]
            if not is_template:
                st += [
                    "x" if not v(0) or all(v(i) for i in range(1, num_columns)) else " "
                ]
            return st

        hd = ["GlobalId"] + list(map(str, range(num_columns)))
        if not is_template:
            hd += ["Valid"]

        print(
            tabulate.tabulate(
                list(map(get_stats, map(operator.itemgetter("GlobalId"), roots))),
                hd,
                tablefmt="grid",
            ),
            file=report,
        )
        return report.getvalue()