Skip to content
Pipelet Developers

Onboard a Charging Fleet

This guide walks through onboarding a fleet of charging stations to Pipelet — from the moment the hardware ships to the first invoice. We’ll use the Headless CPMS API and the Python SDK for the heavy lifting.

The full source for the workflow is in cpms-headless/examples/08_full_workflow.py — this guide narrates and contextualizes it.

What you’ll automate

Section titled “What you’ll automate”
  1. Verifying gateway health
  2. Discovering newly-connected stations
  3. Reading current OCPP configuration
  4. Starting a charging session
  5. Monitoring the session in flight
  6. Stopping the session and capturing the meter delta

Prerequisites

Section titled “Prerequisites”
  • A running Pipelet stack
  • Stations physically connected and reachable to your gateway over OCPP WebSocket
  • A Headless CPMS API key
Terminal window
export CPMS_API_KEY="hcpms_live_..."
export CPMS_URL="http://localhost:8080"

  1. Verify the gateway. A 30-second health check before doing anything else.

    import os, requests, json
    

    API_KEY = os.environ["CPMS_API_KEY"]
    BASE_URL = os.environ["CPMS_URL"]
    headers = {"X-API-Key": API_KEY, "Content-Type": "application/json"}
    

    def api_get(path, params=None):
        r = requests.get(f"{BASE_URL}{path}", headers=headers, params=params)
        return r.status_code, r.json()
    

    def api_post(path, body=None):
        r = requests.post(f"{BASE_URL}{path}", headers=headers, json=body or {})
        return r.status_code, r.json()
    

    status, data = api_get("/api/v1/system/health")
    assert status == 200, f"gateway unhealthy: {data}"

  2. Discover connected stations. As stations boot, they appear here.

    status, data = api_get("/api/v1/stations/compact")
    for s in data:
        print(f"  {s['station_id']:20} {s.get('vendor', '?'):20} {s.get('connector_count', 0)} connectors")

    compact returns a lighter response than /stations — perfect for monitoring loops.

  3. Read OCPP configuration. Some keys you almost always want to verify on a fresh station: HeartbeatInterval, MeterValueSampleInterval, LocalAuthListEnabled.

    STATION = "WALLBOX_001"
    status, data = api_post(
        "/api/v1/configuration/get",
        {"station_id": STATION, "key": ["HeartbeatInterval", "MeterValueSampleInterval"]},
    )
    for kv in data.get("configurationKey", []):
        print(f"  {kv['key']} = {kv['value']}")

    If the values are wrong, write them with /api/v1/configuration/set — see the API reference for the body schema.

  4. Start a charging session. Use a known-good idTag from your customer database.

    status, data = api_post(
        "/api/v1/transactions/start",
        {"station_id": STATION, "connectorId": 1, "idTag": "RFID_DEMO"},
    )
    if data.get("status") != "Accepted":
        raise RuntimeError(f"Remote start rejected: {data}")
    transaction_id = data["transactionId"]

  5. Monitor the session. In production, you’d subscribe to webhooks. For a quick interactive check, poll /api/v1/transactions:

    import time
    for _ in range(3):
        time.sleep(5)
        _, txns = api_get("/api/v1/transactions")
        for tx in txns.get("transactions", []):
            print(f"  Station={tx['chargepoint_id']} Energy={tx.get('energy_wh', 0)} Wh")

  6. Stop the session.

    status, data = api_post(
        "/api/v1/transactions/stop-connector",
        {"station_id": STATION, "connectorId": 1},
    )

With the Python SDK

Section titled “With the Python SDK”

The same workflow, but cleaner with wallecpms:

from wallecpms import HeadlessCPMS


with HeadlessCPMS(BASE_URL, api_key=API_KEY) as cpms:
    assert cpms.system.health()["status"] == "ok"


    for station in cpms.stations.list_compact():
        print(station["station_id"])


    cpms.configuration.get(station_id="WALLBOX_001", keys=["HeartbeatInterval"])


    result = cpms.transactions.start("WALLBOX_001", connector_id=1, id_tag="RFID_DEMO")
    txn_id = result["transactionId"]


    # ...later
    cpms.transactions.stop_connector("WALLBOX_001", connector_id=1)

Productionizing the flow

Section titled “Productionizing the flow”

A real onboarding flow does more than this script:

  • Webhooks instead of polling. Subscribe to StartTransaction, StopTransaction, and StatusNotification so your billing system updates without polling. See Webhooks Overview.
  • Bulk operations. When onboarding 50+ stations at once, fan out the configuration writes in parallel. The gateway is concurrency-safe.
  • Pre-flight validation. Before accepting a station into production, verify firmware version, supported OCPP version, and a successful BootNotification round-trip.
  • Failure modes. Stations can be physically connected but mis-configured — handle Rejected responses and Faulted connector status gracefully.
  • Audit trail. Log every API call with the operator who triggered it. Pipelet’s gateway logs include the API key name, but you’ll want app-level audit too.

Next

Section titled “Next”