How to use owlrl - 7 common examples

G.add( (BLDG.MAFS1, BRICK.hasTag, TAG.Flow) )
G.add( (BLDG.MAFS1, BRICK.hasTag, TAG.Setpoint) )
G.add( (BLDG.MAFS1, BRICK.hasTag, TAG.Limit) )
G.add( (BLDG.MAFS1, BRICK.hasTag, TAG.Max) )

G.add( (BLDG.AFS1, A, BRICK.Air_Flow_Sensor) )

G.add( (BLDG.co2s1, A, BRICK.CO2_Level_Sensor) )

G.add( (BLDG.standalone, A, BRICK.Temperature_Sensor) )

# Apply reasoner
import time
t1 = time.time()
import owlrl
owlrl.DeductiveClosure(owlrl.OWLRL_Semantics).expand(G)

G.bind('rdf', RDF)
G.bind('owl', OWL)
G.bind('rdfs', RDFS)
G.bind('skos', SKOS)
G.bind('brick', BRICK)
G.bind('tag', TAG)
G.bind('bldg', BLDG)

t2 = time.time()
print("Reasoning took {0}".format(t2-t1))
s = G.serialize(format='ttl')
print('expanded:', len(G))

with open('Brick_expanded.ttl','wb') as f:
    f.write(s)

def reason_rdfs(g):
    """
    Applies RDFS reasoning. Takes a few seconds.

    Lets us use:
    - class hierarchy (rdf:type, not rdf:type/rdfs:subClassOf*)
    """
    start_time = time.time()
    owlrl.DeductiveClosure(owlrl.RDFS_Semantics).expand(g)
    end_time = time.time()
    print('owlrl reasoning took {0} seconds.'.format(int(end_time - start_time)))

def reason_owlrl(g):
    """
    Applies full OWL RL Reasoning. WARNING: takes a few hours.

    Lets us use:
    - tags <--> classes
    - measures properties
    - transitive properties
    - inverse properties
    - class hierarchy (rdf:type, not rdf:type/rdfs:subClassOf*)
    """
    start_time = time.time()
    owlrl.DeductiveClosure(owlrl.OWLRL_Semantics).expand(g)
    end_time = time.time()
    print('owlrl reasoning took {0} seconds.'.format(int(end_time - start_time)))

G.add( (DEP[spbc], XBOS.usesResource, DEP.uPMU_0_L3_resource) )
    G.add( (DEP[spbc], XBOS.usesResource, DEP.uPMU_0_C1_resource) )
    G.add( (DEP[spbc], XBOS.usesResource, DEP.uPMU_0_C2_resource) )
    G.add( (DEP[spbc], XBOS.usesResource, DEP.uPMU_0_C3_resource) )

    # a more interesting way: dynamic binding based on a query.
    # still have to know the PMU by name though
    upmu_0_resources = G.query(f"SELECT ?res WHERE {{ <{DEP.uPMU_0}> xbos:hasResource ?res }}")
    for res in upmu_0_resources:
        G.add( (DEP[spbc], XBOS.usesResource, res[0]) )

    # TODO: even more interesting: find uPMU through a query

# apply reasoner
Q = deepcopy(G)
owlrl.DeductiveClosure(owlrl.OWLRL_Semantics).expand(Q)

# infer the entities
all_processes = Q.query("""SELECT ?proc ?label WHERE { ?proc rdf:type xbos:Process . ?proc rdfs:label ?label}""")
for proc, label in all_processes:
    ent = DEP[f"{label}_entity"]
    G.add( (ent, A, XBOS.Entity) )
    G.add( (ent, RDFS.label, Literal(label)) )
    G.add( (proc, XBOS.hasEntity, ent) )

with open('test.ttl','wb') as f:
    f.write(G.serialize(format='turtle'))

def expand(self, graph):
        for triple in graph:
            self.g.add(triple)
        owlrl.DeductiveClosure(owlrl.OWLRL_Semantics).expand(self.g.g)
        return self.g

def run(self, closure_bool = False, axiomatic_bool = False, datatype_axiom_bool = False):
        app = flask.current_app # set app to be current flask app 
        new_graph = rdflib.ConjunctiveGraph() # create new empty graph to run reasoner
        new_graph += app.db # add current graph to new graph
        owlrl.DeductiveClosure(owlrl.OWLRL_Extension, rdfs_closure = closure_bool, axiomatic_triples = axiomatic_bool, datatype_axioms = datatype_axiom_bool).expand(new_graph) # run reasoner using specified options
        new_graph -= app.db # remove content from initial graph from the new graph
        # output graph (for testing purposes mostly)
        new_graph.serialize("inferred_graph.ttl", format="turtle")

def expand(self, graph):
        for triple in graph:
            self.g.add(triple)
        owlrl.DeductiveClosure(owlrl.RDFS_Semantics).expand(self.g.g)
        return self.g