diy-edge-lambda-agents

The diy-edge-lambda-agents is the developer-side toolkit for building AWS-style Python “Lambda” bundles that run on the DIY Edge Lambda Manager next to real BACnet systems. Each agent lives in its own folder under agents/, containing a lambda_function.py, a config.json(.example), and an optional requirements.txt, following a workflow similar to AWS Lambda development. Under the hood, pip install -t is used to vendor dependencies, and the project is packaged into a self-contained bundle in dist/. That ZIP is then uploaded to the IoT edge (typically over VPN into the building OT LAN) using the Edge Lambda Manager Swagger API, where it runs directly on the device as its own isolated Python subprocess.

Be sure to check out these related projects that are designed to work hand-in-hand as a smart-building edge microservice ecosystem 👇

diy-bacnet-server — a lightweight FastAPI + bacpypes3 BACnet/IP server that exposes a JSON-RPC API for reading, writing, and supervising BACnet devices at the edge.
diy-edge-lambda-agents — a collection of edge “Lambda-style” HVAC optimization agents (optimal start, Guideline 36, FDD tools, testing agents, etc.) packaged as deployable ZIP workloads.
diy-edge-lambda-manager — a local “AWS Lambda-like” runtime for the edge that lets you upload, run, stop, and monitor agents via a clean FastAPI + Swagger UI, using real Linux subprocess execution under the hood.

Together, these projects form a modular, production-ready edge automation platform, ideal for real-world smart-building IoT deployments — supporting safe validation workflows today while scaling to advanced supervisory logic like Guideline 36, Optimal Start, and future analytics-driven control strategies.

📦 Prerequisites

Docker Desktop (Windows / macOS / Linux)
Docker Engine ≥ 20.x

Verify Docker Engine:

docker version

Verify Docker Compose (V2):

docker compose version

Note: This project uses the modern docker compose command (with a space).

🐳 Docker Best Practices for Building Agents (x86 Dev → ARM Edge)

Recommended: At the time of writing, for computer architecture compatibility purposes, the best practice is to package Python agent builds using Docker on the same base image that runs on the edge device. Ideally, build agents directly on the edge IoT device, or on a dedicated test bench system with the same architecture (for example, an ARM device if your production edge is ARM), to ensure compatibility and prevent architecture drift.

# From IoT edge device or same device on test bench
cd hvac-edge-lambda-agents

docker run --rm -it \
  -v "$(pwd)":/app \
  -w /app \
  python:3.13-slim \
  bash -lc "pip install --upgrade pip && python pack_agent.py agent_device_1 && python pack_agent.py agent_device_2"

Uses python:3.13-slim, the same base image your edge stack uses.
pack_agent.py runs inside that container.
dist/agent_device_1.zip is built in a controlled, reproducible environment.

✅ Agent Development Requirements

To build + run agents correctly, you need:

Environment

Python 3.10+
pip
Running DIY BACnet Server
Running DIY Edge Lambda Manager
Ability to reach the edge (LAN/VPN)

Agent Folder Structure

Every agent lives under:

agents/<agent_name>/

Must contain:

lambda_function.py      ← main script (runs forever loop)
config.json             ← device + behavior config
requirements.txt        ← ONLY if you need Python deps

Optional:

config.json.example     ← helpful template

Packaging

Agents are packaged using the provided builder:

python pack_agent.py <agent_name>

This does:

copies code
installs Python dependencies into the bundle (pip install -t)
creates dist/<agent_name>.zip
ready to upload to Edge Manager

Upload + Run

Upload via Swagger:

POST /agents/upload/<agent_id>
POST /agents/start/<agent_id>
GET /agents/logs/<agent_id>
POST /agents/stop/<agent_id>

Agents run as isolated OS processes, not threads. Linux schedules them like microservices.

✅ Final `config.json`

Use this in BOTH dueling agents:

{
  "bacnet_base_url": "http://192.168.204.12:8080",

  "ahu": {
    "device_instance": 3456789
  },

  "vav": {
    "device_instance": 3456790,
    "object_identifier": "analog-value,1",
    "priority_level": 12,
    "cooling_sp_value": 70.0
  },

  "interval_seconds": 15.0
}

📌 Meaning

Key	Purpose
`bacnet_base_url`	URL to DIY BACnet JSON-RPC Server
`ahu.device_instance`	Fake AHU device instance used for RPM read
`vav.device_instance`	Fake VAV device instance
`vav.object_identifier`	ZoneCoolingSpt point
`priority_level`	Priority array slot to duel over
`cooling_sp_value`	Value to write if priority slot empty
`interval_seconds`	Loop interval

✅ `requirements.txt`

The bot only needs:

requests

Nothing else required 🎯 No BACnet library. No FastAPI. No nonsense. Everything is done via JSON-RPC HTTP → 💪 lightweight edge microservice.

🤖 Testing Tutorial for Dueling BACnet Agents

For testing purposes, two agents are created to demonstrate BACnet Read Multiple (RPM) functionality on an AHU device and to perform a safe, non-invasive test on a VAV box setpoint to avoid disturbing occupants. Both agents share the same code base. Each agent first performs an RPM operation on the AHU device to validate successful RPM communication. Then, the agent executes a priority-request on a point named ZoneCoolingSpt. If data already exists at priority 12, the agent issues a BACnet release by writing "null" at priority 12. If data does not exist at priority 12, the agent writes a value at priority 12 that is intentionally very close to the actual HVAC control value, but not close enough to cause a noticeable change in system operation. This approach allows validation of agent behavior and platform robustness before implementing real operational strategies.

1️⃣ AHU — READ_MULTIPLE (RPM-style)

Read the following BACnet points:

Object	Description
AI2	Supply Air Temperature (SA-T)
AI3	Mixed Air Temperature (MA-T)
AI4	Return Air Temperature (RA-T)
AI6	Outdoor Air Temperature (OA-T)
AO1	Supply Fan Output (SF-O)

2️⃣ VAV — Priority Array Check (ZoneCoolingSpt)

Check priority[12] on ZoneCoolingSpt:

IF priority[12] exists:
    WRITE "null" @ priority 12   (release override)
ELSE:
    WRITE 70.0 @ priority 12     (apply setpoint)

3️⃣ Loop

Sleep for defined interval
Repeat steps 1–2

1️⃣ Folder layout (two “Dueling agents)

In diy-edge-lambda-agents/agents/:

agents/
  ├─ dueling_agent_1/
  │   ├─ lambda_function.py
  │   ├─ config.json.example
  │   └─ requirements.txt
  └─ dueling_agent_2/
      ├─ lambda_function.py
      ├─ config.json.example
      └─ requirements.txt

For now, both folders contain identical code + config. If you ever want them slightly different (intervals, setpoints, etc.), just tweak each config.json.

2️⃣ `lambda_function.py` (same for both agents)

Drop this file into both:

#!/usr/bin/env python3
"""
lambda_function.py

Dueling BACnet Agents for DIY BACnet JSON-RPC server.

Each agent loop does:

1) AHU READ_MULTIPLE (RPM-style):
   - AI2  SA-T
   - AI3  MA-T
   - AI4  RA-T
   - AI6  OA-T
   - AO1  SF-O

2) VAV priority-array check on ZoneCoolingSpt:
   IF priority[12] exists
       WRITE "null" @ 12  (release)
   ELSE
       WRITE 70.0 @ 12     (apply setpoint)

3) Sleep, then repeat.

Run TWO copies of this agent (with the same config) and they will
continuously undo each other at Priority 12 – perfect for tcpdump
and BACnet discovery tool demos.

Stop each agent via /agents/stop/{agent_id} on the edge-lambda-manager.
"""

import json
import os
import time
from typing import Any, Dict, List, Optional

import requests


def load_config() -> Dict[str, Any]:
    """Load config.json from the same folder as this file."""
    here = os.path.dirname(os.path.abspath(__file__))
    cfg_path = os.path.join(here, "config.json")
    with open(cfg_path, "r") as f:
        return json.load(f)


def rpc_call(base_url: str, method: str, params: Dict[str, Any], id_: str = "0") -> Dict[str, Any]:
    """Generic helper to call a JSON-RPC endpoint on the DIY BACnet server."""
    payload = {
        "jsonrpc": "2.0",
        "id": id_,
        "method": method,
        "params": params,
    }
    url = f"{base_url}/{method}"
    resp = requests.post(url, json=payload, timeout=10)
    resp.raise_for_status()
    return resp.json()


# -----------------------------
# AHU READ_MULTIPLE (RPM demo)
# -----------------------------

def ahu_read_multiple(base_url: str, ahu_cfg: Dict[str, Any]) -> None:
    """
    Call /client_read_multiple on the AHU and print results.

    We read:
      AI2  SA-T
      AI3  MA-T
      AI4  RA-T
      AI6  OA-T
      AO1  SF-O
    """
    device_instance = ahu_cfg["device_instance"]

    # Hard-coded RPM request list (you can also move this into config.json)
    requests_list: List[Dict[str, Any]] = [
        {"object_identifier": "analog-input,2", "property_identifier": "present-value"},  # SA-T
        {"object_identifier": "analog-input,3", "property_identifier": "present-value"},  # MA-T
        {"object_identifier": "analog-input,4", "property_identifier": "present-value"},  # RA-T
        {"object_identifier": "analog-input,6", "property_identifier": "present-value"},  # OA-T
        {"object_identifier": "analog-output,1", "property_identifier": "present-value"}, # SF-O
    ]

    params = {
        "request": {
            "device_instance": device_instance,
            "requests": requests_list,
        }
    }

    print("\n[AHU] Calling client_read_multiple...")
    data = rpc_call(base_url, "client_read_multiple", params)
    result = data.get("result", {})

    # Just dump raw JSON so you can see the structure and values in the logs
    print("[AHU READ_MULTIPLE] Raw result:")
    print(json.dumps(result, indent=2))


# ---------------------------------------
# VAV Priority Array "Dueling" bot
# ---------------------------------------

def vav_check_and_duel(base_url: str, vav_cfg: Dict[str, Any]) -> None:
    """
    For the configured VAV ZoneCoolingSpt point:

    - Call /client_read_point_priority_array
    - If priority[priority_level] exists and is not null:
          write "null" @ that priority (release)
      ELSE:
          write cooling_sp_value @ that priority

    This is what makes two identical agents "duel" – they keep
    flipping the Priority 12 slot between 70.0 and null.
    """
    device_instance = vav_cfg["device_instance"]
    object_identifier = vav_cfg["object_identifier"]
    priority_level = int(vav_cfg.get("priority_level", 12))
    cooling_sp_value = float(vav_cfg.get("cooling_sp_value", 70.0))

    # --- Read priority array ---
    params = {
        "request": {
            "device_instance": device_instance,
            "object_identifier": object_identifier,
        }
    }

    print("\n[VAV] Calling client_read_point_priority_array...")
    data = rpc_call(base_url, "client_read_point_priority_array", params)
    result = data.get("result", {})

    print("[VAV PRIORITY ARRAY] Raw result:")
    print(json.dumps(result, indent=2))

    priority_array = result.get("priority-array", {})
    slot_key = str(priority_level)
    slot_value = priority_array.get(slot_key)

    # Decide what to do
    if slot_value is not None and str(slot_value).lower() != "null":
        # Slot is occupied → RELEASE
        print(
            f"[VAV] Priority {priority_level} currently has value {slot_value!r}. "
            f"Releasing with value='null'..."
        )
        new_value: Any = "null"
    else:
        # Slot empty or null → APPLY 70.0
        print(
            f"[VAV] Priority {priority_level} is empty/null. "
            f"Writing {cooling_sp_value} at that priority..."
        )
        new_value = cooling_sp_value

    write_request = {
        "device_instance": device_instance,
        "object_identifier": object_identifier,
        "property_identifier": "present-value",
        "value": new_value,
        "priority": priority_level,
    }

    write_params = {"request": write_request}
    write_data = rpc_call(base_url, "client_write_property", write_params)

    print(
        f"[VAV WRITE] device={device_instance}, obj={object_identifier}, "
        f"priority={priority_level}, value={new_value!r}, result={write_data.get('result')}"
    )


# ----------------------
# Main agent loop
# ----------------------

def loop_forever() -> None:
    cfg = load_config()

    base_url = os.environ.get(
        "BACNET_BASE_URL",
        cfg.get("bacnet_base_url", "http://localhost:8080"),
    )

    # AHU + VAV config chunks
    ahu_cfg = cfg["ahu"]
    vav_cfg = cfg["vav"]

    interval = float(cfg.get("interval_seconds", 15.0))

    print("=== Dueling BACnet Bot starting ===")
    print(f"BACnet server base URL: {base_url}")
    print(f"AHU device_instance={ahu_cfg['device_instance']}")
    print(
        f"VAV device_instance={vav_cfg['device_instance']}, "
        f"object_identifier={vav_cfg['object_identifier']}, "
        f"priority_level={vav_cfg.get('priority_level', 12)}, "
        f"cooling_sp_value={vav_cfg.get('cooling_sp_value', 70.0)}"
    )
    print(f"interval_seconds={interval}")

    while True:
        try:
            print("\n========== DUELING TICK ==========")
            # 1) AHU read-multiple
            ahu_read_multiple(base_url, ahu_cfg)

            # 2) VAV priority dueling logic
            vav_check_and_duel(base_url, vav_cfg)

            print(f"[agent] Sleeping {interval} seconds before next cycle...")
        except Exception as e:
            # Don't crash; log and continue
            print(f"[ERROR] {e}")

        time.sleep(interval)


def handler(event=None, context=None):
    """Optional AWS Lambda-style handler."""
    loop_forever()


if __name__ == "__main__":
    loop_forever()

3️⃣ `config.json.example` (same for both agents to start)

Put this in both dueling_agent_1 and dueling_agent_2:

{
  "bacnet_base_url": "http://192.168.204.12:8080",

  "ahu": {
    "device_instance": 3456789
  },

  "vav": {
    "device_instance": 3456790,
    "object_identifier": "analog-value,1",
    "priority_level": 12,
    "cooling_sp_value": 70.0
  },

  "interval_seconds": 15.0
}

Then:

cp agents/dueling_agent_1/config.json.example agents/dueling_agent_1/config.json
cp agents/dueling_agent_2/config.json.example agents/dueling_agent_2/config.json

If you want to get spicy later, you can make one bot use 69°F, the other 71°F, different intervals, etc.

4️⃣ Build and Upload Both Agents

From diy-edge-lambda-agents root:

python pack_agent.py dueling_agent_1
python pack_agent.py dueling_agent_2

You’ll get:

dist/dueling_agent_1.zip
dist/dueling_agent_2.zip

Upload via Edge Lambda Manager Swagger:

POST /agents/upload/1 → dueling_agent_1.zip
POST /agents/upload/2 → dueling_agent_2.zip

Start them:

POST /agents/start/1
POST /agents/start/2

Tail logs:

GET /agents/logs/1
GET /agents/logs/2

Now you’ve got two identical Dueling BACnet Agents constantly:

RPM’ing the AHU sensors
Flipping the VAV ZoneCoolingSpt priority 12 between 70.0 and "null"

…which is perfect for tcpdump, BACnet Discovery Tool, and swagger-side debugging 😈

📜 License

Everything here is MIT Licensed — free, open source, and made for the BAS community.
Use it, remix it, or improve it — just share it forward so others can benefit too. 🥰🌍

【MIT License】

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

Name		Name	Last commit message	Last commit date
Latest commit History 9 Commits
agents		agents
.gitignore		.gitignore
AGENTS.md		AGENTS.md
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
pack_agent.py		pack_agent.py

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Repository files navigation

diy-edge-lambda-agents

📦 Prerequisites

Environment

Agent Folder Structure

Packaging

Upload + Run

✅ Final `config.json`

📌 Meaning

✅ `requirements.txt`

1️⃣ AHU — READ_MULTIPLE (RPM-style)

2️⃣ VAV — Priority Array Check (ZoneCoolingSpt)

3️⃣ Loop

1️⃣ Folder layout (two “Dueling agents)

2️⃣ `lambda_function.py` (same for both agents)

3️⃣ `config.json.example` (same for both agents to start)

4️⃣ Build and Upload Both Agents

📜 License

About

Uh oh!

Releases

Packages

Languages

License

bbartling/diy-edge-lambda-agents

Folders and files

Latest commit

History

Repository files navigation

diy-edge-lambda-agents

📦 Prerequisites

Environment

Agent Folder Structure

Packaging

Upload + Run

✅ Final config.json

📌 Meaning

✅ requirements.txt

1️⃣ AHU — READ_MULTIPLE (RPM-style)

2️⃣ VAV — Priority Array Check (ZoneCoolingSpt)

3️⃣ Loop

1️⃣ Folder layout (two “Dueling agents)

2️⃣ lambda_function.py (same for both agents)

3️⃣ config.json.example (same for both agents to start)

4️⃣ Build and Upload Both Agents

📜 License

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Languages

✅ Final `config.json`

✅ `requirements.txt`

2️⃣ `lambda_function.py` (same for both agents)

3️⃣ `config.json.example` (same for both agents to start)

Packages