wire-band-edge

Lightweight Wire.Band client — semantic data middleware for any domain. Connects any data source to a Wire.Band backend: IoT sensors, AI/ML pipelines, DeFi activity, legal records, geospatial streams, supply chain events, and more.

Runs anywhere Python runs: subscribe to an MQTT broker, ingest directly from code, or bridge any structured data source. Classifies each message into a typed semantic frame, buffers locally during connectivity loss, and forwards event batches to a Wire.Band backend over HTTP.

Features

• Zero private dependencies — installs anywhere with pip install wire-band-edge
• Universal domain coverage — 650+ symbol types across IoT, AI/ML, DeFi, legal, geographic, temporal, security, storage, networking, and more
• MQTT bridge — subscribes to topic patterns, classifies + frames each message
• Local ring buffer — survives network interruptions; events are retried with exponential backoff
• Delta filtering — suppress readings that haven't meaningfully changed (configurable threshold)
• Semantic classification — maps topic paths and payload keys to typed symbols across all supported domains
• 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

MQTTConnector

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:

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
# }

WebSocket Streaming (v0.4.0+)

For high-frequency sensors or compliance-critical streams where the 1-second HTTP batch flush interval adds too much latency, use WireBandStreamClient to deliver each event immediately over a persistent WebSocket connection.

pip install "wire-band-edge[stream]"

import asyncio
from wire_band_edge import WireBandStreamClient

async def main():
    async with WireBandStreamClient(
        backend_url="http://your-wireband-server:8000",
        device_id="compliance-node-01",
    ) as client:
        # Each send() delivers immediately — no buffering, no batch interval
        await client.send({"value": 72.3, "unit": "C"}, topic="sensors/temp")
        await client.send({"event": "access", "user": "admin"}, topic="audit/access")

        print(client.stats())
        # {
        #   "events_sent": 2,
        #   "send_errors": 0,
        #   "reconnects": 0,
        #   "connected": true,
        #   "uptime_seconds": 0.1,
        #   "device_id": "compliance-node-01"
        # }

asyncio.run(main())

WireBandStreamClient automatically reconnects with exponential backoff if the connection drops.

Choosing between HTTP batch and WebSocket streaming

Links

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