Rule-Based Model Builder

This repository implements an experimental rule-based approach to creating Brick models (or really any RDF graph) from variably-structured metadata sources. The APIs and interfaces in this repository may change drastically with little notice, and documentation is sparse! That said, this is fast emerging as my favorite way for creating Brick models from arbitrary sources.

The key idea is to write small functions that generate part of a Brick graph when certain conditions are met on the input data. The input data is structured as a sequence of key-value documents, called rows. It is usually straightforward to turn some source data into a sequence of key-value documents, so this is a natural "intermediate representation" (IR) for arbitrary metadata sources.

The framework defined in this repository provides a number of simple "rule conditions" for triggering these functions which produce parts of the Brick graph. These rule conditions are designed to make it easier to deal with heterogeneously-structured input data.

Rule Conditions

Each condition is implemented as a Python decorator. Decorated functions should take a single argument which is the "row" (a Python dict) that matches the conditions defined by the decorator

tags: triggers the rule when all of the keys are present

@tags("AHU")
def add_ahu(row):
    ahu_name = row.get("AHU")
    G.add((BLDG[ahu_name], A, BRICK.AHU))
    G.add((BLDG[ahu_name], RDFS.label, Literal(row.get("AHU"))))

oneof: triggers the rule when at least one of the provided keys is present

point_types = {
    "alarm variable": BRICK.Alarm,
    "air volume": BRICK.Supply_Air_Flow_Sensor,
    "differential pressure status sensor": BRICK.Differential_Pressure_Sensor,
    "discharge air pressure sensor": BRICK.Discharge_Air_Static_Pressure_Sensor,
    "fan speed reset": BRICK.Speed_Reset_Command,
    "zone temp": BRICK.Zone_Air_Temperature_Sensor,
    "zone temp setpoint": BRICK.Zone_Air_Temperature_Setpoint,
}
@oneof(*point_types.keys())
def define_point(row):
    for tag, brickclass in point_types.items():
        if tag in row:
            sensor = row.get(tag)
            G.add((BLDG[sensor], A, brickclass))

values: triggers the rule when the row has the given key-value pairs

@values({"Category": "Level", "ModelObject": "IfcBuildingStorey"})
def add_floors(row):
    floor_id = EX[row["ModelID"]]
    G.add((floor_id, RDFS.label, Literal(row["Name"])))
    G.add((floor_id, A, BRICK.Floor))
    G.add((floor_id, BRICK.elevation, [
        (BRICK.value, Literal(float(row["Elevation"]))),
        (BRICK.unit, units["linear"]),
    ]))

_and_: triggers the rule when all of the given conditions are met. Sorry for the funky syntax; ideally the decorators compose easily but this will take some work to figure out how to do

@_and_([tags, "AHU", "uuid"], [oneof, *point_types.keys()])
def ahu_points(row):
    ahu_name = row.get('AHU')
    for cls, brickclass in point_types.items():
        if cls in row:
            point = row.get(cls)
            G.add((BLDG[point], A, brickclass))
            G.add((BLDG[point], BRICK.timeseries, [
                (BRICK.hasTimeseriesId, Literal(row.get("uuid")))
            ]))
            G.add((BLDG[ahu_name], BRICK.hasPoint, BLDG[point]))

fixedpoint: executes the rule if any new triples were added to the graph since it was last executed. This is helpful to avoid any dependency constraints between rules (e.g. making sure the floors exist so the biulding can be linked to them). The argument to the fixedpoint decorator is the string name of the graph object you are appending to (should be G in most cases)

The triples generated by this rule should be consistent so that RDFlib can effectively deduplicate. Generating a blank node with a random identifier can lead this rule to execute forever. Use with caution! This should always be the decorator immediately above the function definition.

@tags("AHU")
@fixedpoint("G")
def ahu_feeds(row):
    ahu = row.get("AHU")
    idnum = ahu[-1]
    vavs = G.subjects(predicate=A, object=BRICK.VAV)
    for vav in vavs:
        label = list(G.objects(subject=vav, predicate=RDFS.label))[0]
        if label.startswith(idnum):
            G.add((BLDG[ahu], BRICK.feeds, vav))

Driving

The stream of documents are done through a Stream object. A Stream object implements an optional setup method and a next method. A CSVFileStream and JSONFileStream implementation are provided. Some files (e.g. examples/smc/smc.py will define their own extensions to these streams to handle tasks like column renaming, data cleaning and transformations.

Call the drive function on 1 or more Stream objects to create the graph (by default, stored in a global G variable), then optionally validate (G.validate()) and serialize.