Skip to content

get_sensors.py§

get_sensors(state, config) §

Retrieve sensor information for the building structure.

Parameters:

Name Type Description Default
state State

The current state.

required
config dict

Configuration dictionary containing the language model.

required

Returns: dict: A dictionary containing sensor UUIDs mapped to their locations.

Source code in brickllm/nodes/get_sensors.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_sensors(state: State, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Retrieve sensor information for the building structure.

    Args:
        state (State): The current state.
        config (dict): Configuration dictionary containing the language model.
    Returns:
        dict: A dictionary containing sensor UUIDs mapped to their locations.
    """
    custom_logger.eurac("📡 Getting sensors information")

    user_prompt = state["user_prompt"]
    graph = state["graph"]

    g = Graph()
    g.parse(ontology_path, format="turtle")

    if graph is not None:
        for triple in graph:
            g.add(triple)

    # Create a dictionary to store the sensor informations
    sensor_dict = {}

    # Identify the Point entities in the graph
    query = """
    PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
    PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
    PREFIX brick: <https://brickschema.org/schema/Brick#>

    SELECT ?point ?type
    WHERE {
        ?point rdf:type/rdfs:subClassOf* brick:Point .
        ?point rdf:type ?type .
    }
    """

    res_query = g.query(query)
    point_type_list = []
    for row in res_query:
        point_type_list.append(str(row['type']).split("#")[1])
        sensor_dict[str(row['point']).split("#")[1]] = {
            "name": str(row['point']).split("#")[1],
            "type": str(row['type']).split("#")[1],
            "id": None,
            "unit": None
        }

    point_type_list = list(set(point_type_list))
    uom_dict = get_uom_dict(point_type_list)

    sensor_structure_json = json.dumps(sensor_dict, indent=2)
    uom_dict_json = json.dumps(uom_dict, indent=2)
    # Get the model name from the config
    llm = config.get("configurable", {}).get("llm_model")

    # Enforce structured output
    structured_llm = llm.with_structured_output(SensorSchema)
    # System message
    system_message = get_sensors_instructions.format(
        prompt=user_prompt, sensor_structure=sensor_structure_json, unit_of_measures=uom_dict_json
    )

    # Generate question
    answer = structured_llm.invoke(
        [SystemMessage(content=system_message)]
    )

    # Encode the sensor information in the graph
    sensors = answer.sensors

    BLDG = Namespace("urn:Building#")
    UNIT = Namespace("http://qudt.org/vocab/unit/")
    BRICK = Namespace("https://brickschema.org/schema/Brick#")
    REF = Namespace("https://brickschema.org/schema/Brick/ref#")
    RDF = Namespace("http://www.w3.org/1999/02/22-rdf-syntax-ns#")

    graph.bind("bldg", BLDG)
    graph.bind("brick", BRICK)
    graph.bind("unit", UNIT)
    graph.bind("ref", REF)
    graph.bind("rdf", RDF)

    for sensor in sensors:
        timeseries_ref = BNode()
        if sensor.unit is not None and sensor.unit != "None":
            try:
                graph.add((BLDG[sensor.name], BRICK.hasUnit, UNIT[sensor.unit]))
            except:
                custom_logger.warning(f"Unit {sensor.unit} not found in QUDT")

        if sensor.id is not None:
            graph.add((BLDG[sensor.name], REF.hasExternalReference, timeseries_ref))
            graph.add((timeseries_ref, RDF.type, REF.TimeseriesReference))
            graph.add((timeseries_ref, REF.hasTimeseriesId, Literal(sensor.id, datatype=XSD.string)))

    return {"graph": graph, "sensor_dict": sensors}