Python Provisioning Starter
This page shows a provisioning script that ties together the workflow from the Power Users overview with the scenarios from Common Provisioning Scenarios.
1 What the starter script shows
It shows how to:
- establish a connection to a given unit
- read its current configuration
- apply a baseline configuration
- set unit-specific values
- update the firmware
- wait for the unit to restart
- verify the resulting configuration
2 Documentation-in-code
The script contains comments explaining the different parts of the code and the reasoning behind the approach, sometimes linking to the online documentation for further information.
The end result of running the script will look something like this:
-- Step 1: Discovering controllers on the network --
E0:AE:B2:09:0E:DC -> 192.168.123.123
-- Step 2: Initial device report --
192.168.123.123 (E0:AE:B2:09:0E:DC)
Firmware: 5.33.5-21148
Identity: STATION-FCEABD
Serial:
MAC: e0:ae:b2:09:0e:dc
Parameter Value
------------------------------- -----------------------------------------------
ChargeBoxIdentity_custom STATION-FCEABD
ChargePointUUID_ocpp 1c868472-4189-4238-b035-232810770d9a
FreeChargingMode_ocpp 4
FreeCharging_vehicleif Off
ManufacturerPwd_custom TWZyLTg5RENGNEY3
MasterSlaveMode_ms Off
OperatorPwd_custom_default T3BlcmF0b3ItMjUzQUYy
ResetPasswordPUK_custom AE427281778F
SerialNumberManufacturer_custom MFR-SN-DA0F4F
UserInstallerPwdCheck_custom 0
-- Step 3: Baseline configuration --
Skipping baseline upload — no files found under C:\Users\<username>\cc-baseline
-- Step 4: Apply configuration --
192.168.123.123: ChargePointIdentity=STATION-237888, Serial=MFR-SN-4AD782
192.168.123.123: Manufacturer=Mfr-89A8E19C, Operator=Operator-AC7C03, PUK=CCF8F1D91BF2
Parameter Value
------------------------------- ----------------
ChargeBoxIdentity_custom STATION-237888
SerialNumberManufacturer_custom MFR-SN-4AD782
FreeCharging_vehicleif Off
FreeChargingMode_ocpp 4
ResetPasswordPUK_custom CCF8F1D91BF2
ManufacturerPwd (base64-encoded)
OperatorPwd (base64-encoded)
-- Step 5: Firmware upgrade --
192.168.123.123: 5.33.5-21148 -> 5.38.2-21917
Uploading <your-firmware-file>.deb...
Uploading: 26.8/26.8 MB (100%)
Installing via opkg...
Cleaning up...
192.168.123.123: waiting for shutdown...
192.168.123.123: confirmed offline
192.168.123.123: waiting for boot...
192.168.123.123: probably still rebooting... retrying in 10s
...
192.168.123.123: back online
-- Step 6: Final device report --
192.168.123.123 (E0:AE:B2:09:0E:DC)
Firmware: 5.38.2-21917
Identity: STATION-237888
Serial:
MAC: e0:ae:b2:09:0e:dc
Parameter Value
------------------------------- -----------------------------------------------
ChargeBoxIdentity_custom STATION-237888
ChargePointUUID_ocpp 1c868472-4189-4238-b035-232810770d9a
FreeChargingMode_ocpp 4
FreeCharging_vehicleif Off
InstallerPwd_custom T3BlcmF0b3ItQUM3QzAz
ManufacturerPwd_custom TWZyLTg5QThFMTlD
MasterSlaveMode_ms Off
OperatorPwd_custom_default T3BlcmF0b3ItQUM3QzAz
ResetPasswordPUK_custom CCF8F1D91BF2
SerialNumberManufacturer_custom MFR-SN-4AD782
SysTime_mon 1842
UserInstallerPwdCheck_custom 0
Credentials:
--------------------- ---------------
Manufacturer password Mfr-89A8E19C
Operator password Operator-AC7C03
PUK CCF8F1D91BF2
ChargePointIdentity STATION-237888
Serial MFR-SN-4AD782
--------------------- ---------------
3 The script
illustrative-provisioning-example.py
#!/usr/bin/env python3
"""
Illustrative provisioning script for Charge Controllers.
This is not production-ready code and only for educational purposes.
Requirements:
pip install fabric scp tabulate
Documentation:
Provisioning overview: https://bender.de/docs/charge-controller/Provisioning
SSH provisioning: https://bender.de/docs/charge-controller/Provisioning/SSH
Resets and defaults: https://bender.de/docs/charge-controller/Provisioning/Resets-and-Defaults
RED 3.3 compliance: https://bender.de/docs/charge-controller/Security/red-compliance
"""
from __future__ import annotations
import base64
import re
import subprocess
import time
import logging
import uuid
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
import urllib.request
import urllib.error
from fabric import Connection
from scp import SCPClient
from tabulate import tabulate
# Suppress paramiko's noisy internal logging (e.g. "Error reading SSH protocol banner"
# tracebacks that appear during reboot polling.
logging.getLogger("paramiko").setLevel(logging.CRITICAL)
# All controller configuration lives in flat files under this directory.
# Each file is named after the parameter key (e.g. "ChargePointIdentity_custom")
# and contains the parameter value as plain text (or base64 for passwords).
# See: https://bender.de/docs/charge-controller/Provisioning/SSH
PERSISTENCY_DIR = "/home/charge/persistency"
RETRY_INTERVAL = 10 # seconds between connection retries
# Typical firmware upgrade reboot times by product line:
# ICC1324: ~7-15 min
# CC613: ~12-20 min
# Depending on the unit, the controller may take longer to boot after a firmware upgrade.
# To prevent false positives for failing upgrades, you may wait >>> 35 minutes <<< before giving up the wait.
@dataclass
class Target:
mac: str
user: str = "charge"
password: str = "orange_zone" # SSH password for the charge user = manufacturer password
port: int = 22
# IP is resolved at runtime from the MAC address — not known upfront
host: str = field(default="", init=False)
_conn: Connection | None = field(default=None, repr=False, init=False)
def resolve_ip_from_mac(self, mac: str) -> str:
"""
Resolve a controller's IP address from its MAC address using the local ARP table.
This example assumes the provisioning network is configured so the controller
appears in the local ARP table. Network setup and device discovery vary between
environments and must be handled by your technical staff.
"""
mac_lower = mac.lower().replace(":", "-")
result = subprocess.run(["arp", "-a"], capture_output=True, text=True)
for line in result.stdout.splitlines():
if mac_lower in line.lower():
ip_match = re.search(r"(\d+\.\d+\.\d+\.\d+)", line)
if ip_match:
return ip_match.group(1)
raise RuntimeError(
f"Could not find IP for MAC {mac}. "
"Check that the controller is reachable from the provisioning network and powered on."
)
def discover(self) -> None:
"""Resolve the controller's IP address from its MAC via the local ARP table."""
self.host = self.resolve_ip_from_mac(self.mac)
print(f" Discovered {self.mac} -> {self.host}")
@property
def conn(self) -> Connection:
if not self.host:
raise RuntimeError("Call discover() first to resolve the IP address")
if self._conn is None:
self._conn = Connection(
self.host,
user=self.user,
port=self.port,
connect_kwargs={
"password": self.password,
"timeout": 10,
"banner_timeout": 10,
"auth_timeout": 10,
},
)
return self._conn
def run(self, cmd: str, **kwargs) -> str:
try:
result = self.conn.run(cmd, hide=True, warn=True, **kwargs)
return result.stdout.strip()
except Exception as e:
raise RuntimeError(f"SSH command failed on {self.host}: {cmd!r} — {e}") from e
def put(self, local: str | Path, remote: str) -> None:
"""Upload a file via SCP. The controller runs BusyBox and does not
support SFTP, so we use the SCP protocol over the existing SSH connection."""
def progress(filename: bytes, size: int, sent: int) -> None:
pct = sent * 100 // size
mb_sent = sent / 1024 / 1024
mb_total = size / 1024 / 1024
print(f"\r Uploading: {mb_sent:.1f}/{mb_total:.1f} MB ({pct}%)", end="", flush=True)
try:
self.conn.open() # ensure the SSH connection is active
with SCPClient(self.conn.client.get_transport(), progress=progress) as scp:
scp.put(str(local), remote)
print() # newline after progress
except Exception as e:
raise RuntimeError(f"SCP upload failed to {self.host}: {remote} — {e}") from e
def close(self) -> None:
if self._conn:
self._conn.close()
self._conn = None
def reboot(self) -> None:
"""
Hard reboot ("Save & Restart") — the entire Linux system restarts.
Use this after changes that require a full boot cycle, such as master/slave
topology, network-related settings, or firmware upgrades.
The controller takes ~60-90s to come back.
"""
self.run("sync && reboot")
self.close()
def reboot_soft(self) -> None:
"""
Soft restart ("Save & Soft Restart") — restarts only the charge controller
application, not the OS. Faster than a hard reboot.
Use for settings that just need the application to re-read its configuration,
such as DLM or backend/OCPP configuration.
This sends the same HTTP request as clicking "Save & Soft Restart" on the
manufacturer configuration page in the web interface.
"""
import urllib.request
url = f"http://{self.host}/legacy/manufacturer/manufacturer"
credentials = base64.b64encode(f"manufacturer:{self.password}".encode()).decode()
data = b"SUBMITTYPE=6d"
req = urllib.request.Request(url, data=data, method="POST")
req.add_header("Authorization", f"Basic {credentials}")
req.add_header("Content-Type", "application/x-www-form-urlencoded")
try:
with urllib.request.urlopen(req, timeout=10) as resp:
print(f" {self.host}: soft restart triggered (HTTP {resp.status})")
except urllib.error.HTTPError as e:
print(f" {self.host}: soft restart returned HTTP {e.code}")
def wait_until_reachable(self, timeout: int | None = None) -> None:
"""
After a reboot, wait for the controller to go down, then wait for it to
come back up. Two phases:
1. Wait until SSH becomes unreachable (confirms the reboot started)
2. Wait until SSH becomes reachable again (controller is back online)
Args:
timeout: max seconds to wait for the controller to come back (phase 2).
None (default) = wait forever. Pass an explicit value when you
know the upper bound (e.g. 35*60 for firmware upgrades).
"""
# Phase 1: wait for the controller to go down
print(f" {self.host}: waiting for shutdown...")
while True:
try:
self.close()
self.run("echo ok")
# Still up — wait and retry
time.sleep(RETRY_INTERVAL)
except Exception:
print(f" {self.host}: confirmed offline")
self.close()
break
# Phase 2: wait for it to come back
print(f" {self.host}: waiting for boot...")
elapsed = 0
while True:
try:
self.close()
self.run("echo ok")
print(f" {self.host}: back online")
return
except Exception:
print(f" {self.host}: probably still rebooting... retrying in {RETRY_INTERVAL}s")
time.sleep(RETRY_INTERVAL)
elapsed += RETRY_INTERVAL
if timeout is not None and elapsed >= timeout:
raise TimeoutError(
f"{self.host} did not come back within {timeout // 60} min."
)
def read_firmware_version(self) -> str:
"""
Read the firmware version via opkg.
See: https://bender.de/docs/charge-controller/Provisioning/SSH/cookbook#device-identity-recipes
"""
return self.run("opkg list-installed | cut -d' ' -f3")
def read_identity(self) -> str:
"""
Read the ChargePoint identity (OCPP ChargeBoxIdentity).
See: https://bender.de/docs/charge-controller/OCPP/Reference/ocpp-parameter-table
"""
return self.run(f"head -n1 {PERSISTENCY_DIR}/ChargeBoxIdentity_custom")
def read_serial(self) -> str:
"""
Read the controller serial number.
See: https://bender.de/docs/charge-controller/Provisioning/SSH/cookbook#device-identity-recipes
"""
return self.run(f"head -n1 {PERSISTENCY_DIR}/SerialNumber_custom")
def read_mac(self) -> str:
"""
Read the MAC address of the controller's Ethernet interface (eth0).
See: https://bender.de/docs/charge-controller/Provisioning/SSH/cookbook#device-identity-recipes
"""
return self.run("head -n1 /sys/class/net/eth0/address")
def read_info(self) -> dict[str, str]:
"""Collect key device info for reporting."""
return {
"firmware": self.read_firmware_version(),
"identity": self.read_identity(),
"serial": self.read_serial(),
"mac": self.read_mac(),
}
def read_config(self) -> list[tuple[str, str]]:
"""
Read all persistency files and return the filename and first line (value).
This gives a snapshot of every configuration parameter on the controller.
See: https://bender.de/docs/charge-controller/Provisioning/SSH/cookbook#configuration-recipes
"""
output = self.run(f"for f in {PERSISTENCY_DIR}/*; do echo \"$(basename $f)|$(head -n1 $f)\"; done")
entries = []
for line in output.splitlines():
if "|" in line:
name, _, value = line.partition("|")
entries.append((name, value))
return entries
class Role(Enum):
OPERATOR = "operator"
INSTALLER = "installer"
MANUFACTURER = "manufacturer"
def write_param(target: Target, key: str, value: str) -> None:
"""
Write a single parameter to the controller's persistency directory.
The file is named after the parameter key and contains the value as plain text.
After writing, call target.run("sync") to flush to disk before rebooting.
"""
target.run(f"echo -n {value!r} > {PERSISTENCY_DIR}/{key}")
def apply_configuration(target: Target, local_dir: str, overrides: dict[str, str] | None = None) -> None:
"""
Upload a full set of configuration files from a local directory to the controller.
Use 'overrides' for values that differ per unit (e.g. ChargePointIdentity).
Typical workflow:
1. Configure one "golden" unit manually through the Config UI
2. Pull its persistency directory to your machine
(scp -r charge@<ip>:/home/charge/persistency/ ./cc-baseline/)
3. Use that directory as the baseline for all other units
Important: only include the config files you actually want to push. Remove or override
any unit-specific values (like ChargePointIdentity or SerialNumberManufacturer)
from the baseline — those should be set per-unit via write_param() instead.
"""
baseline = Path(local_dir).expanduser().resolve()
if not baseline.is_dir():
print(f" Skipping baseline upload — no files found under {baseline}")
return
files = sorted(p for p in baseline.iterdir() if p.is_file())
if not files:
print(f" Skipping baseline upload — no files found under {baseline}")
return
for f in files:
target.put(f, f"{PERSISTENCY_DIR}/{f.name}")
for key, value in (overrides or {}).items():
write_param(target, key, value)
target.run("sync")
# ── Set passwords ──
#
# Password file naming convention:
# See: https://bender.de/docs/charge-controller/Security/red-compliance
#
# Operator and Installer use "_default" suffix (e.g. OperatorPwd_custom_default).
# This means the value acts as a *default* password. On first login, the user is
# prompted to change it — the firmware enforces this automatically. This is the
# intended RED 3.3 compliant flow: ship with a known default, force the end user
# to set their own.
#
# Manufacturer uses NO "_default" suffix (ManufacturerPwd_custom).
# This is the actual password — there is no "default + force change" flow for
# the manufacturer role. The manufacturer sets a strong unique password per device
# during production and stores it securely.
#
# All passwords are stored as base64-encoded strings. The controller firmware
# decodes them internally when checking credentials.
PASSWORD_FILES: dict[Role, list[str]] = {
Role.OPERATOR: ["OperatorPwd_custom_default"],
Role.INSTALLER: ["InstallerPwd_custom_default"],
Role.MANUFACTURER: ["ManufacturerPwd_custom"],
}
def set_password(target: Target, role: Role, password: str) -> None:
"""Set a base64-encoded password for the given role. This follows the RFC 4648 base64 encoding standard, which asks for padding with '=' characters if necessary. Make sure your tools/libraries do this."""
b64 = base64.b64encode(password.encode()).decode()
for filename in PASSWORD_FILES[role]:
write_param(target, filename, b64)
target.run("sync")
# ── Firmware update ──
def update_firmware(target: Target, firmware_path: str) -> None:
"""
Upload a .deb firmware package via scp and install it via opkg.
The controller needs a hard reboot afterwards to boot into the new firmware.
After reboot, call read_firmware_version() (opkg list-installed) to confirm the new version.
See: https://bender.de/docs/charge-controller/Provisioning/SSH/cookbook#firmware-update-recipes
"""
local = Path(firmware_path).expanduser().resolve()
if not local.is_file():
raise FileNotFoundError(f"Firmware not found: {local}")
remote = "/home/charge/sw_update.deb"
print(f" Uploading {local.name}...")
target.put(local, remote)
print(f" Installing via opkg... (this may take a while)")
target.run(f"opkg install {remote}")
print(f" Cleaning up...")
target.run(f"rm {remote}")
# ── Example scenario ──
#
# This scenario provisions a single charge controller.
# We know the MAC address and resolve the IP address at runtime.
#
# Step 1: Discover the controller's IP address from its MAC using the local ARP table.
# This example assumes the provisioning network is configured so the
# controller appears in that table. Network setup and device discovery
# vary between environments and must be handled by your technical staff.
#
# Step 2: Initial device report — read back device info and dump the full
# persistency config table (filename + first line of each file) so you
# have a "before" snapshot.
#
# Step 3: Upload baseline configuration from a local directory (./cc-baseline/).
# If the directory doesn't exist yet, this step is skipped gracefully.
# This is how you apply a "golden unit" config to other controllers.
# Only include the config files you actually want — remove unit-specific
# values (like ChargePointIdentity or SerialNumberManufacturer) from the
# baseline and set those per-unit in Step 4 instead.
# See: https://bender.de/docs/charge-controller/Provisioning/Resets-and-Defaults
#
# Step 4: Apply configuration — write individual settings, set passwords
# (base64-encoded, per RED 3.3), and restart the controller.
#
# Step 5: Firmware upgrade via opkg (currently no-op) + hard reboot.
# Install the firmware package, then hard reboot so the controller boots
# into the new version. After reboot we verify by reading the version file.
#
# Step 6: Final device report — same as Step 2, so you can compare before/after.
#
# Parameter names and values can be found by:
# 1. Exporting the config from a "golden" unit configured through the Config UI
# (scp -r charge@<ip>:/home/charge/persistency/ ./cc-baseline/)
# 2. The OCPP parameter reference table:
# https://www.bender.de/docs/charge-controller/OCPP/Reference/ocpp-parameter-table
# ── Provisioning steps ──
def discover_controllers(units: list[Target]) -> None:
"""Resolve each controller's IP from its MAC address via ARP.
For simplicity, we use the USB provisioning IP directly. This assumes a single unit connected to your computer through USB. In a batch setup on an Ethernet switch, use unit.discover() with your own implementation instead."""
print("\n-- Step 1: Discovering controllers on the network --")
for unit in units:
unit.host = "192.168.123.123"
print(f" {unit.mac} -> {unit.host}")
def print_device_report(units: list[Target], label: str) -> None:
"""Read back device info and dump the full persistency config table.
This gives a snapshot of every parameter on the controller — useful
for before/after comparison."""
print(f"\n-- {label} --")
for unit in units:
info = unit.read_info()
config = unit.read_config() # reads first line of each file in /home/charge/persistency/
print(f"\n {unit.host} ({unit.mac})")
print(f" Firmware: {info['firmware']}")
print(f" Identity: {info['identity']}")
print(f" Serial: {info['serial']}")
print(f" MAC: {info['mac']}")
print(tabulate(config, headers=["Parameter", "Value"], tablefmt="simple"))
def upload_baseline_configuration(units: list[Target], baseline_dir: str) -> None:
"""Upload a golden-unit config baseline. Skipped if the directory doesn't exist."""
print("\n-- Step 3: Baseline configuration --")
for unit in units:
apply_configuration(unit, baseline_dir)
def apply_per_unit_configuration(units: list[Target]) -> dict:
"""Set unique identity, serial, passwords, PUK, and charging mode per unit.
Returns the generated credentials for later reporting.
In production, store these in your secure database. They cannot be
recovered from the controller later."""
print("\n-- Step 4: Apply configuration --")
creds = {}
for unit in units:
cp_id = f"STATION-{uuid.uuid4().hex[:6].upper()}"
serial = f"MFR-SN-{uuid.uuid4().hex[:6].upper()}"
manufacturer_pwd = f"Mfr-{uuid.uuid4().hex[:8].upper()}"
operator_pwd = f"Operator-{uuid.uuid4().hex[:6].upper()}"
puk = uuid.uuid4().hex[:12].upper()
print(f" {unit.host}: ChargePointIdentity={cp_id}, Serial={serial}")
print(f" {unit.host}: Manufacturer={manufacturer_pwd}, Operator={operator_pwd}, PUK={puk}")
config_table = [
["ChargeBoxIdentity_custom", cp_id],
["SerialNumberManufacturer_custom", serial],
["FreeCharging_vehicleif", "Off"],
["FreeChargingMode_ocpp", "4"],
["ResetPasswordPUK_custom", puk],
["ManufacturerPwd", "(base64-encoded)"],
["OperatorPwd", "(base64-encoded)"],
]
print(tabulate(config_table, headers=["Parameter", "Value"], tablefmt="simple"))
write_param(unit, "ChargeBoxIdentity_custom", cp_id)
write_param(unit, "SerialNumberManufacturer_custom", serial)
write_param(unit, "FreeCharging_vehicleif", "Off")
write_param(unit, "FreeChargingMode_ocpp", "4") # 0 = No OCPP
write_param(unit, "ResetPasswordPUK_custom", puk) # plain text, up to 128 chars
set_password(unit, Role.MANUFACTURER, manufacturer_pwd) # base64-encoded
set_password(unit, Role.OPERATOR, operator_pwd) # base64-encoded, _default suffix → forced change on first login for the Operator to set their own password
unit.run("sync") # persist all changes to disk before reboot.
creds[unit.mac] = {
"cp_id": cp_id, "serial": serial,
"manufacturer_pwd": manufacturer_pwd,
"operator_pwd": operator_pwd, "puk": puk,
}
return creds
def upgrade_firmware_or_restart(units: list[Target], firmware_path: str, creds: dict) -> None:
"""Upload and install firmware if needed, or soft-restart to apply config.
After opkg returns, the controller auto-reboots via stacked timers:
~5s — normal reboot once the app detects the update
2 min — backup reboot if the app doesn't reboot on its own
30 min — failsafe reboot if everything hangs"""
print("\n-- Step 5: Firmware upgrade --")
fw_name = Path(firmware_path).name
fw_match = re.search(r"(\d+\.\d+\.\d+-\d+)", fw_name)
target_fw = fw_match.group(1) if fw_match else ""
for unit in units:
current_fw = unit.read_firmware_version()
if target_fw and current_fw.startswith(target_fw):
print(f" {unit.host}: already on {current_fw}, skipping firmware upgrade")
# No firmware change needed, but config from Step 4 still needs a restart.
unit.reboot_soft()
else:
print(f" {unit.host}: {current_fw} -> {target_fw}")
update_firmware(unit, firmware_path)
# No manual reboot needed — the controller handles it automatically.
for unit in units:
# After reboot, the SSH password = new manufacturer password from persistency.
unit.password = creds[unit.mac]["manufacturer_pwd"]
unit.wait_until_reachable(timeout=35 * 60) # 35 min — covers the 30-min failsafe timer
def print_credentials(units: list[Target], creds: dict) -> None:
"""Print all generated credentials for record-keeping. In production you'd persist these in a database or similar."""
print("\nCredentials:")
for unit in units:
c = creds[unit.mac]
print(tabulate([
["Manufacturer password", c["manufacturer_pwd"]],
["Operator password", c["operator_pwd"]],
["PUK", c["puk"]],
["ChargePointIdentity", c["cp_id"]],
["Serial", c["serial"]],
], tablefmt="simple"))
print()
if __name__ == "__main__":
UNITS = [
Target(mac="E0:AE:B2:09:0E:DC"),
]
FIRMWARE = r"C:\Users\<your-username>\Downloads\<your-firmware-file>.deb"
# ── Run all steps ──
discover_controllers(UNITS)
print_device_report(UNITS, "Step 2: Initial device report")
upload_baseline_configuration(UNITS, "./cc-baseline")
credentials = apply_per_unit_configuration(UNITS)
upgrade_firmware_or_restart(UNITS, FIRMWARE, credentials)
print_device_report(UNITS, "Step 6: Final device report")
print_credentials(UNITS, credentials)