Pulsebloom

Every device blooms in Python.

What if talking to an ESP32 felt as easy as defining a Django model?

Pulsebloom is a Python framework that turns a class definition into a complete IoT stack: MQTT bridge, REST API, WebSocket streams, admin dashboard, and a firmware library. One pip install, one class, one firmware flash — live telemetry in a browser.

from pulsebloom import Pulsebloom, Device, telemetry

class Thermostat(Device):
    temperature: float = telemetry(unit="°C", freq="5s")
    humidity: float    = telemetry(unit="%",   freq="5s")

app = Pulsebloom()
app.register(Thermostat)
app.run()

Run that. Open http://localhost:8000/admin. Flash the firmware to an ESP32. Done.

---

Why Pulsebloom

Every IoT project starts the same: you want a temperature sensor and a dashboard. Two hours later you're writing MQTT reconnection logic, designing a JSON schema, building a WebSocket bridge, and gluing three libraries together.

Pulsebloom handles all of that. You describe the device; it handles the wiring.

---

Quickstart (5 minutes)

1. Install

pip install pulsebloom

You also need a running MQTT broker. The easiest option is Mosquitto.

2. Define your device + run the server

# thermostat.py
from pulsebloom import Pulsebloom, Device, telemetry

class Thermostat(Device):
    temperature: float = telemetry(unit="°C", freq="5s")
    humidity: float    = telemetry(unit="%",   freq="5s")

app = Pulsebloom()
app.register(Thermostat)
app.run()

python thermostat.py

3. Flash the firmware

Open firmware/arduino/PulsebloomDevice/examples/Thermostat/Thermostat.ino, fill in your WiFi credentials and server IP, and flash to an ESP32.

4. Watch it live

Open http://localhost:8000/admin. The device appears and updates in real time.

No real hardware? Test with a single command:

mosquitto_pub -t "pulsebloom/v1/thermostat-001/t" \
  -m '{"dev":"thermostat-001","seq":1,"ts":1713268800.0,"type":"t","data":{"temperature":22.4,"humidity":45.2}}'

---

What you get

Defining a Device class gives you, for free:

Feature	Endpoint
Device list + last values	GET /devices
Per-device detail	GET /devices/{id}
Telemetry history	GET /devices/{id}/history?field=temperature
Latest field values	GET /devices/{id}/latest
Live WebSocket stream	ws://.../ws/telemetry
HTTP ingest endpoint	POST /ingest
Admin dashboard	http://localhost:8000/admin
Swagger / OpenAPI docs	http://localhost:8000/docs

---

Swappable Backends

v0.2 introduces a fully pluggable architecture. Every infrastructure layer is an ABC — swap it out without changing your device classes.

Zero-config (defaults — same as v0.1)

app = Pulsebloom()   # MQTT + SQLite + Console notifier + built-in dashboard

Fully configured (production)

from pulsebloom import Pulsebloom, Device, telemetry
from pulsebloom.transports.mqtt import MQTTTransport
from pulsebloom.transports.http import HTTPTransport
from pulsebloom.storage.sqlite import SQLiteStorage
from pulsebloom.notify.webhook import WebhookNotifier
from pulsebloom.notify.console import ConsoleNotifier

app = Pulsebloom(
    transports=[
        MQTTTransport(broker="emqx.prod", port=1883),
        HTTPTransport(),
    ],
    storage=SQLiteStorage(path="./data.db"),
    notifiers=[
        WebhookNotifier(url="https://hooks.slack.com/services/..."),
        ConsoleNotifier(),
    ],
)

URI shorthand

app = Pulsebloom(
    transports="mqtt://emqx.prod:1883",
    storage="sqlite:///data.db",
)

Testing (pure in-memory, zero dependencies)

from pulsebloom import Pulsebloom, Device, telemetry, Frame
from pulsebloom.transports.memory import MemoryTransport
from pulsebloom.storage.memory import MemoryStorage

transport = MemoryTransport()
storage = MemoryStorage()

app = Pulsebloom(transports=[transport], storage=storage, dashboard=None)
await app.start()

# Simulate a device sending a frame:
await transport.inject(Frame(device_id="t-001", type="t", data={"temperature": 22.5}))

# Assert the full pipeline ran:
points = await storage.query("t-001", "temperature", since=0)
assert points[0].value == 22.5

No broker. No SQLite file. No ports. Fast.

---

Architecture

pulsebloom/
├── core/          PURE PYTHON — Device, Frame, Dispatcher, Registry
├── transports/    MQTTTransport, HTTPTransport, MemoryTransport
├── storage/       SQLiteStorage, MemoryStorage
├── notify/        ConsoleNotifier, WebhookNotifier
├── dashboard/     BuiltinDashboard (Alpine.js + Chart.js)
└── api/           FastAPI REST + WebSocket

Iron rule: core/ has zero imports from any infrastructure layer.

---

Core concepts

Device

A Device is a Python class that describes a physical thing:

class WeatherStation(Device):
    temperature: float = telemetry(unit="°C")
    humidity:    float = telemetry(unit="%")
    pressure:    float = telemetry(unit="hPa")
    wind_speed:  float = telemetry(unit="m/s")

telemetry()

Marks a field as a measurement the device publishes.

Parameter	Purpose
unit	Display unit (shown in dashboard)
freq	Suggested publish interval ("1s", "5s", "1m")

@command (v0.3+)

Marks a Device method as a server-to-device RPC. The server sends a command frame; the device executes it and replies with an ack.

from pulsebloom import Pulsebloom, Device, command, telemetry

class Thermostat(Device):
    temperature: float = telemetry(unit="°C", freq="5s")

    @command
    async def set_led(self, state: bool) -> bool:
        """Turn the onboard LED on or off."""
        ...

app = Pulsebloom()
app.register(Thermostat)

From the REST API:

curl -X POST http://localhost:8000/devices/thermostat-001/commands/set_led \
     -H 'Content-Type: application/json' \
     -d '{"params": {"state": true}}'
# {"status": "ok", "result": {"ok": true}}

From Python:

result = await app.commands.send_command(
    device_id="thermostat-001",
    command_name="set_led",
    params={"state": True},
)

The dashboard shows a button for each @command on the device detail page.

On the Arduino side, register a handler for each command:

device.onCommand("set_led", [](JsonObject& params) -> bool {
    bool state = params["state"] | false;
    digitalWrite(LED_BUILTIN, state ? HIGH : LOW);
    return true;
});

The library subscribes to pulsebloom/v1/{device_id}/c, calls the handler, and publishes an ack to pulsebloom/v1/{device_id}/ca automatically.

Frame

The protocol-agnostic internal message format. Every transport produces Frames; every storage backend consumes them.

Frame(
    device_id="thermostat-001",
    type="t",          # "t"=telemetry, "c"=command, "e"=event, "h"=heartbeat
    seq=42,
    timestamp=1713268800.123,
    data={"temperature": 22.4, "humidity": 45.2},
)

Wire protocol

Every MQTT message is a JSON frame:

{
  "dev":  "thermostat-001",
  "seq":  42,
  "ts":   1713268800.123,
  "type": "t",
  "data": { "temperature": 22.4, "humidity": 45.2 }
}

Topic: pulsebloom/v1/{device_id}/t. Full spec in proto/SPEC.md.

---

CLI (v0.4+)

# Start the server
pulsebloom run                        # loads app.py automatically
pulsebloom run --app myproject.py --port 9000

# Simulate devices (no hardware needed)
pulsebloom simulate Thermostat --count 5 --broker mqtt://localhost:1883

# List connected devices
pulsebloom devices
pulsebloom devices --server http://myserver:8000

The simulate command reads your Device class from app.py and spins up N virtual devices publishing realistic random-walk telemetry to the MQTT broker.

---

Firmware codegen (v0.7+)

Generate complete, runnable firmware from your Python Device class:

# Arduino / ESP32
pulsebloom codegen --target arduino --device Thermostat --output firmware/
# Creates firmware/Thermostat.h and firmware/Thermostat.ino

# MicroPython
pulsebloom codegen --target micropython --device Thermostat --output firmware/
# Creates firmware/thermostat.py

The generated code includes:

• WiFi + MQTT setup with reconnect logic
• Telemetry publish loop at the correct frequency
• onCommand() stubs for each @command method
• State variable declarations with defaults (synced on reconnect via MQTT retained messages)
• TODO placeholders where sensor-reading code goes

You only fill in the sensor reads. Everything else is wired up.

---

Supported hardware

The firmware library targets ESP32 (Arduino framework). It should also work on ESP8266 without changes. Any device that can publish MQTT JSON is compatible — the wire protocol is 40 lines of MQTT + JSON.

---

Examples

Example	What it shows
examples/thermostat/	Complete walkthrough: server + firmware + wiring

---

Roadmap

v0.1 — shipped

• Device class with @telemetry
• JSON wire protocol (MQTT, QoS 1)
• SQLite storage (WAL mode)
• FastAPI REST + WebSocket
• Admin dashboard (Alpine.js + Chart.js)
• Arduino firmware library

v0.2 — shipped

• Pluggable Transport ABC (MQTTTransport, HTTPTransport, MemoryTransport)
• Pluggable Storage ABC (SQLiteStorage, MemoryStorage)
• Pluggable Notifier ABC (ConsoleNotifier, WebhookNotifier)
• Pluggable Dashboard ABC (BuiltinDashboard)
• Frame as protocol-agnostic internal message format
• URI shorthand for backends
• In-memory backends for zero-dependency testing

v0.3 — next

• @command decorator (server → device RPC)
• CBOR wire encoding (~30% smaller than JSON)
• HMAC-SHA256 frame authentication
• TLS support

See ROADMAP.md and POST_MVP.md for the full plan.

---

FAQ

How is this different from Home Assistant? Home Assistant is an app — you configure it. Pulsebloom is a framework — you build with it.

How is this different from AWS IoT / Azure IoT Hub? Those are cloud services. Pulsebloom runs on your laptop or Pi — no vendor, no bill, no internet.

Is it production-ready? v0.2 is suitable for prototypes and homelab. v0.3 adds auth — at that point, small production deployments are reasonable.

---

Contributing

MIT license. Issues and PRs welcome at github.com/pulsebloom/pulsebloom.

---

License

MIT © Meraj Safari