Open Fdd concept ideas of putting Python code into the Graph

concept.md

🧠 Executable Brick: Embedding Python into RDF for Automatic Optimization

Future Vision for Open-FDD

---

🚀 Concept

Today, Brick describes what equipment exists in a building.

But what if Brick could also describe:

✔ what software should run ✔ what FDD rules apply ✔ what optimization programs deploy ✔ what control strategies activate

Based on:

brick:Chiller_Plant
brick:Air_Handler
brick:VAV
brick:Boiler

---

Then your workflow becomes:

Scan BACnet → Generate Brick → Run SPARQL
→ Optimization auto-deploys itself

No GUI. No manual engineering. No Niagara programming.

---

🧬 Architecture Idea

We introduce:

ofdd:EmbeddedPythonProgram

A Python program stored inside RDF and linked to:

brick equipment classes

So:

:SimmerResetLogic
    ofdd:appliesTo brick:Chiller_Plant .

Now:

Any building tagged as a ChillerPlant automatically gets chiller reset optimization.

---

📁 File Layout

Your tutorial demo folder:

bake_a_py/
│
├── chiller_program.py
├── import_py_to_graph.py
├── run_the_py_from_ttl.py
└── my_model.ttl

---

📦 1. Concept Idea Optimization Program

chiller_program.py

This is the control logic that gets embedded into Brick.

def run_logic(ctx):
    # 1. Fetch live telemetry provided by the engine
    req = ctx.get('telemetry_req', 0)
    current_target = ctx.get('telemetry_target', 100.0)
    equipment_name = ctx.get('equipment_name', 'Unknown_Chiller')

    # 2. Control Logic
    IGNORE_THRESHOLD = 2
    effective_requests = max(0, req - IGNORE_THRESHOLD)
    
    # Calculate new target based on requests
    new_target = 100.0 + (effective_requests * 3.0)
    
    # 3. Output results
    print(f"[{equipment_name}] Requests: {req} | Old Target: {current_target}% | New Target: {new_target}%")

Source:

---

🧱 2. Bake Python into RDF

import_py_to_graph.py

This script:

✔ Reads Python source ✔ Serializes it ✔ Stores it in RDF ✔ Links it to equipment

import os
from rdflib import Graph, Namespace, Literal
from rdflib.namespace import RDF

EX = Namespace("http://example.org/")
BRICK = Namespace("https://brickschema.org/schema/Brick#")

def serialize_to_graph(py_filepath, program_uri_name, ttl_filepath="my_model.ttl"):
    if not os.path.exists(py_filepath):
        print(f"Error: Could not find {py_filepath}")
        return
        
    with open(py_filepath, "r") as f:
        raw_code = f.read()

    escaped_source = raw_code.strip().replace('\\', '\\\\').replace('\n', '\\n').replace('"', '\\"')

    g = Graph()
    g.bind("ex", EX)
    g.bind("brick", BRICK)
    
    if os.path.exists(ttl_filepath):
        g.parse(ttl_filepath, format="turtle")

    prog_uri = EX[program_uri_name]
    equip_uri = EX["Chiller_Plant_01"]

    g.remove((prog_uri, None, None))

    g.add((equip_uri, RDF.type, BRICK.Chiller))
    g.add((equip_uri, EX.hasControlProgram, prog_uri))

    g.add((prog_uri, RDF.type, EX.Program))
    g.add((prog_uri, EX.language, Literal("python")))
    g.add((prog_uri, EX.entrypoint, Literal("run_logic")))
    g.add((prog_uri, EX.source, Literal(escaped_source)))

    g.serialize(destination=ttl_filepath, format="turtle")
    print(f"Success! Program saved to {ttl_filepath}.")

Source:

---

📡 3. SPARQL → Run Program Automatically

run_the_py_from_ttl.py

This script:

✔ Queries RDF ✔ Finds equipment ✔ Finds linked Python ✔ Extracts code ✔ Executes logic

import random
from rdflib import Graph

def run_from_config():
    g = Graph()
    g.parse("my_model.ttl", format="turtle")

    query = """
    PREFIX ex: <http://example.org/>
    PREFIX brick: <https://brickschema.org/schema/Brick#>
    PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
    
    SELECT ?equip ?prog ?src ?entry WHERE {
        ?equip ex:hasControlProgram ?prog .
        ?prog rdf:type ex:Program .
        ?prog ex:source ?src .
        ?prog ex:entrypoint ?entry .
    }
    """
    
    results = g.query(query)
    print(f"--- SPARQL found {len(results)} equipment/program pairs ---")

    for row in results:
        equipment_uri = str(row.equip)
        entry_fn = str(row.entry)
        
        source_string = str(row.src).replace('\\n', '\n').replace('\\"', '"').replace('\\\\', '\\')

        fake_live_requests = random.randint(0, 10)
        fake_live_reset = random.choice([100.0, 115.0, 120.0, 95.0])

        ctx = {
            "equipment_name": equipment_uri.split('/')[-1],
            "telemetry_req": fake_live_requests, 
            "telemetry_target": fake_live_reset
        }

        exec_env = {}
        exec(source_string, exec_env)
        exec_env[entry_fn](ctx)

Source:

---

🧾 4. Resulting RDF Model

my_model.ttl

After running:

python import_py_to_graph.py

Your graph now contains a serialized Python script:

@prefix brick: <https://brickschema.org/schema/Brick#> .
@prefix ex: <http://example.org/> .

ex:Chiller_Plant_01 a brick:Chiller ;
    ex:hasControlProgram ex:SimmerResetLogic .

ex:SimmerResetLogic a ex:Program ;
    ex:entrypoint "run_logic" ;
    ex:language "python" ;
    ex:source "def run_logic(ctx):\\n    # 1. Fetch live telemetry provided by the engine\\n    req = ctx.get('telemetry_req', 0)\\n    current_target = ctx.get('telemetry_target', 100.0)\\n    equipment_name = ctx.get('equipment_name', 'Unknown_Chiller')\\n\\n    # 2. Control Logic\\n    IGNORE_THRESHOLD = 2\\n    effective_requests = max(0, req - IGNORE_THRESHOLD)\\n    \\n    # Calculate new target based on requests\\n    new_target = 100.0 + (effective_requests * 3.0)\\n    \\n    # 3. Output results\\n    print(f\\\"[{equipment_name}] Requests: {req} | Old Target: {current_target}% | New Target: {new_target}%\\\")" .

---

▶️ 5. Run Optimization from RDF

Now simply:

python run_the_py_from_ttl.py

Output:

--- SPARQL found 1 equipment/program pairs ---
[Chiller_Plant_01] Requests: 4 | Old Target: 100.0% | New Target: 106.0%

Run again:

[Chiller_Plant_01] Requests: 6 | Old Target: 95.0% | New Target: 112.0%

---

🧠 What Just Happened?

We did NOT:

❌ deploy logic manually ❌ configure optimization ❌ write site-specific code ❌ program Niagara blocks

Instead:

✔ tagged equipment ✔ queried graph ✔ pulled program ✔ executed automatically

---

🔮 Future Open-FDD Vision

Imagine:

Brick Tag Exists	Graph Pulls
AHU	Static Reset
Chiller	Staging Optimization
Boiler	HW Reset
VAV System	Reheat DP Reset
RTU	Economizer FDD
Heat Pump	Optimal Start

---

🌐 Brick Becomes Executable

Instead of:

Graph = Description

You now have:

Graph = Description + Behavior

Your building ontology becomes:

A deployable runtime system.

---

Welcome to:

Ontology-Driven Building Optimization 🧠🏢