wire-band-edge 0.1.0

Lightweight Wire.Band client for IoT gateway hardware — Raspberry Pi, NVIDIA Jetson, industrial PCs

wire-band-edge

Lightweight Wire.Band client for IoT gateway hardware — Raspberry Pi, NVIDIA Jetson, industrial PCs, any Linux SBC.

Runs close to your devices: subscribes to an MQTT broker, classifies each message into a typed semantic frame, buffers locally during connectivity loss, and forwards compressed event batches to a Wire.Band backend over HTTP.

Features

• Zero private dependencies — installs anywhere with pip install wire-band-edge
• MQTT bridge — subscribes to topic patterns, classifies + compresses each message
• Local ring buffer — survives network interruptions; events are retried with exponential backoff
• Delta filtering — suppress sensor readings that haven't meaningfully changed (configurable threshold)
• Semantic classification — maps MQTT topic paths and payload keys to typed IoT symbols (sensor, actuator, GPIO, edge lifecycle, metrics, geographic)
• Async throughout — built on asyncio + httpx; non-blocking on constrained hardware

Installation

# Core client (HTTP flush only)
pip install wire-band-edge

# With MQTT broker support
pip install "wire-band-edge[mqtt]"

Python 3.10+ required.

Quick Start

CLI (after install)

# Basic — bridge all topics to a local backend
wire-band-edge --broker mqtt://localhost:1883 --backend http://localhost:8000

# With device ID and delta filtering
wire-band-edge \
    --broker mqtt://broker.example.com:1883 \
    --topics "sensors/#" "machines/#" \
    --device-id factory-rpi4 \
    --delta-threshold 0.02 \
    --backend http://my-wireband-server:8000

# Or with python -m
python -m wire_band_edge --help

Full MQTT gateway

import asyncio
from wire_band_edge import WireBandEdgeClient, MQTTConnector

async def main():
    client = WireBandEdgeClient(
        backend_url="http://your-wireband-server:8000",
        device_id="factory-floor-rpi4",
        delta_threshold=0.02,   # suppress readings with <2% change
    )
    connector = MQTTConnector("mqtt://localhost:1883")
    await client.run_mqtt(connector, topics=["sensors/#", "machines/#"])

asyncio.run(main())

Programmatic ingest (serial drivers, custom sensors)

import asyncio
from wire_band_edge import WireBandEdgeClient

async def main():
    async with WireBandEdgeClient(
        backend_url="http://your-wireband-server:8000",
        device_id="serial-node-01",
    ) as client:
        await client.ingest({"temp": 72.3, "humidity": 45.1}, topic="env/zone-a")
        await client.ingest({"rpm": 1420, "current": 3.2},    topic="motor/spindle")

asyncio.run(main())

TLS broker with authentication

connector = MQTTConnector(
    broker_url="mqtts://broker.example.com:8883",
    username="gateway",
    password="secret",
    delta_threshold=0.01,
)

Configuration

WireBandEdgeClient

Parameter	Default	Description
backend_url	http://localhost:8000	Wire.Band backend URL
device_id	edge-node	Unique gateway identifier
api_key	None	Bearer token for backend auth
buffer_size	50000	Local ring buffer capacity (events)
flush_interval	1.0	Flush attempt interval (seconds)
flush_batch	200	Max events per HTTP request
max_retries	3	Retry attempts before backoff
backoff_base	2.0	Exponential backoff base (seconds)
delta_threshold	0.0	Numeric delta filter threshold (0 = forward all)

MQTTConnector

Parameter	Default	Description
broker_url	mqtt://localhost:1883	Broker URL (mqtt:// or mqtts://)
client_id	wire-band-connector	MQTT client identifier
username	None	Broker username
password	None	Broker password
keepalive	60	MQTT keepalive interval (seconds)
max_queue_size	10000	Max buffered events before drop
delta_threshold	0.0	Numeric delta filter threshold

Semantic Classification

Each MQTT message is automatically classified to a typed symbol based on the topic path and payload keys. Classifications are prioritised in order:

1. Exact topic overrides (register_topic_symbol)
2. Topic path keyword matching (rightmost segment first)
3. Payload key heuristics
4. Fallback: generic sensor poll

Topic keyword examples:

Topic contains	Symbol type
temp, temperature	Hardware Sensor — Temperature
humidity, pressure, voltage, current	Hardware Sensor
motor, servo, relay, led, buzzer	Hardware Actuator
gpio, pin, pwm, adc, dac	Hardware GPIO
ota, update, watchdog, twin, shadow	Edge Lifecycle
metrics, telemetry, counter, event	Metrics

Register custom mappings at runtime:

connector.register_topic_symbol("factory/line-1/emergency-stop", 0xFC6D)  # SENSOR_ALERT
connector.register_keyword_symbol("spindle", 0xFC90)                       # ACTUATOR_MOTOR_START

Stats

Both WireBandEdgeClient and MQTTConnector expose a .stats() method:

print(client.stats())
# {
#   "events_ingested": 14823,
#   "events_flushed": 14800,
#   "events_dropped": 0,
#   "flush_errors": 0,
#   "bytes_sent": 187432,
#   "buffer_depth": 23,
#   "buffer_capacity": 50000,
#   "uptime_seconds": 3612.4,
#   "device_id": "factory-floor-rpi4",
#   "backend_url": "http://your-wireband-server:8000"
# }

print(connector.stats())
# {
#   "messages_received": 14823,
#   "messages_compressed": 14823,
#   "messages_delta_filtered": 412,
#   "bytes_in": 1923441,
#   "bytes_out": 187432,
#   "compression_ratio": 10.27,
#   "connected": true,
#   "uptime_seconds": 3612.4
# }

Links

• Homepage: https://wire.band
• Repository: https://github.com/maco144/wireband
• Issues: https://github.com/maco144/wireband/issues