connections.py

import rdflib
import itertools
from collections import defaultdict
from rdflib.term import Node
from typing import Dict, Tuple, List, Set

class BrickConnectionMiner:
    brick: rdflib.Graph

    def __init__(self):
        # load in latest copy of Brick ontology
        self.brick = rdflib.Graph()
        self.brick.parse("https://github.com/BrickSchema/Brick/releases/download/nightly/Brick.ttl", format="ttl")

        self.relationships: Dict[Tuple[Node, Node], List[Node]] = defaultdict(list)
        relationship_query = """SELECT ?from ?to ?relationship WHERE {
            ?from a owl:Class ; rdfs:subClassOf* brick:Entity .
            ?to a owl:Class ; rdfs:subClassOf* brick:Entity .
            ?from sh:property ?prop .
            ?prop sh:path ?relationship .
            ?prop sh:or?/rdfs:rest*/rdfs:first?/sh:class ?to  .
            FILTER(?relationship != brick:isReplacedBy)
        }"""
        for row in self.brick.query(relationship_query):
            values = row.asdict()
            key = values['from'],values['to']
            self.relationships[key].append(values['relationship'])

        # brick classes on which relationships are defined/terminated
        self.base_classes = set(itertools.chain.from_iterable((k, v) for k,v in self.relationships.keys()))

    def get_base_classes(self, n: Node) -> Set[Node]:
        candidates = set()
        for bc in self.base_classes:
            res = self.brick.query("ASK { ?n owl:equivalentClass?/rdfs:subClassOf* ?bc }", initBindings={'n': n, 'bc': bc})
            if res.askAnswer:
                candidates.add(bc)
        return candidates

    def get_relationships(self, fromClass: Node, toClass: Node) -> Set[Node]:
        """
        Given 2 nodes, returns all relationships that can exist from the 'fromClass' node to the 'toClass' node .
        Uses the Brick ontology to figure out what the 'base' class of each Node is.
        """
        baseFromClasses = self.get_base_classes(fromClass)
        baseToClasses = self.get_base_classes(toClass)
        return set(itertools.chain.from_iterable(self.relationships[(f, t)] for f,t in itertools.product(baseFromClasses, baseToClasses)))

if __name__ == '__main__':
    miner = BrickConnectionMiner()
    print(miner.get_relationships(rdflib.BRICK['AHU'], rdflib.BRICK['Alarm']))
    print(miner.get_relationships(rdflib.BRICK['AHU'], rdflib.BRICK['Boiler']))
    print(miner.get_relationships(rdflib.BRICK['Room'], rdflib.BRICK['Temperature_Sensor']))
    print(miner.get_relationships(rdflib.BRICK['Temperature_Setpoint'], rdflib.BRICK['Temperature_Sensor']))