flexbit-sdk 1.0.2

FlexBIT SDK client for Python

FlexBIT Python Client

Official FlexBIT SDK client for Python. It publishes and consumes FlexBIT telemetry and control messages over Kafka.

Installation

uv add flexbit-sdk
# or
pip install flexbit-sdk

Configuration

Pass connection settings when creating a connector:

options = {
    "siteId": "44a7a2f2-1cc9-464d-96f4-b767045f8206",
    "clientId": "my-flexbit-client",
    "host": "app.cetp-flexbit.eu:9092",
    "api": {
        "id": "your_api_id",
        "secret": "your_api_secret",
    },
}

The SDK also accepts Python-style option names:

from flexbit import CreateConnectorOptions

options = CreateConnectorOptions(
    site_id="44a7a2f2-1cc9-464d-96f4-b767045f8206",
    client_id="my-flexbit-client",
    host="app.cetp-flexbit.eu:9092",
    api={"id": "your_api_id", "secret": "your_api_secret"},
)

The runnable samples can read these values from environment variables:

FLEXBIT_HOST=app.cetp-flexbit.eu:9092
FLEXBIT_API_ID=your_api_id
FLEXBIT_API_SECRET=your_api_secret

If host is omitted, the client defaults to 127.0.0.1:9092.

Usage

Device Connector

Use a device connector from an on-site integration. It sends telemetry to FlexBIT and subscribes to control commands for the same site.

import os
import threading

from flexbit import create_device_connector

site_id = "44a7a2f2-1cc9-464d-96f4-b767045f8206"
asset_id = "86d33428-2fb7-40f1-8bb9-01b5a476bb29"

device = create_device_connector(
    {
        "siteId": site_id,
        "clientId": "site-device-client",
        "host": os.environ["FLEXBIT_HOST"],
        "api": {
            "id": os.environ["FLEXBIT_API_ID"],
            "secret": os.environ["FLEXBIT_API_SECRET"],
        },
    }
)

threading.Thread(
    target=lambda: device.subscribe(
        lambda command: print(
            "Control command:", command["type"], command["meta_asset_id"]
        )
    ),
    daemon=True,
).start()

device.ingest(
    "bess",
    asset_id,
    {
        "meta_asset_id": asset_id,
        "bess_storage_soc": 82.5,
        "bess_inv_power_active": -15.3,
        "bess_storage_status": "charge",
    },
)

Module Connector

Use a module connector from a platform-side module. It subscribes to telemetry for the configured site and sends control commands back to assets at that site.

import os

from flexbit import create_module_connector

site_id = "44a7a2f2-1cc9-464d-96f4-b767045f8206"

module = create_module_connector(
    {
        "siteId": site_id,
        "clientId": "platform-module-client",
        "host": os.environ["FLEXBIT_HOST"],
        "api": {
            "id": os.environ["FLEXBIT_API_ID"],
            "secret": os.environ["FLEXBIT_API_SECRET"],
        },
    }
)


def handle_telemetry(telemetry):
    if telemetry["type"] != "bess" or "bess_storage_soc" not in telemetry:
        return

    if telemetry["bess_storage_soc"] > 90:
        module.control(
            "bess",
            telemetry["meta_asset_id"],
            {
                "meta_site_id": telemetry["meta_site_id"],
                "meta_asset_id": telemetry["meta_asset_id"],
                "bess_inv_set_enable": False,
            },
        )


module.subscribe(handle_telemetry)

API

create_device_connector(options)

Returns:

• ingest(type, asset_id, content, meta=None) - publish telemetry to ingestion.{siteId}.
• subscribe(callback) - consume control messages from control.{siteId}.

create_module_connector(options)

Returns:

• control(type, asset_id, content, meta=None) - publish control commands to control.{siteId}.
• subscribe(callback) - consume telemetry messages from ingestion.{siteId}.

The SDK injects type, meta_site_id, meta_asset_id, and meta_timestamp into every produced message. meta_timestamp defaults to the current time when not provided.

Supported Asset Types

Telemetry types:

bess, ev_charger, pv, caes, community, thermal_cold, thermal_heat, energy_demand_consumer, energy_demand_prosumer, hydrogen_local_plant, grid_meter

Control types:

bess, ev_charger, pv, community, thermal_cold, thermal_heat, hydrogen_local_plant, grid_meter, energy_demand_prosumer

Topics

Direction	Topic
Telemetry (device to module)	ingestion.{siteId}
Control (module to device)	control.{siteId}

Messages are JSON payloads. The asset kind is carried in the message type field.

Types

The package exports generated payload dataclasses and message aliases:

from flexbit.interfaces import BessTelemetry, BessControl
from flexbit.interfaces.export import ControlMessage, IngestionMessage

Payload shapes are generated from the FlexBIT OpenAPI specification under src/interfaces.

Development

uv sync
python3 -m py_compile src/lib.py src/__init__.py src/interfaces/__init__.py

Authentication

Generate an API key from your site settings in the FlexBIT Platform. The API ID and secret are used as SASL/SCRAM-SHA-256 credentials for Kafka.

License

MIT