An asynchronous Python library for communicating with Victron Energy Venus OS devices (CCGX, Cerbo GX, Ekrano GX) via MQTT. It maps Victron MQTT topics to typed Python objects with live-updating metrics, writable controls, and device hierarchy support.
This package is the backend for the Home Assistant Victron GX integration.
Disclaimer: This is a third-party library and is not affiliated with Victron Energy.
- Asynchronous — non-blocking MQTT communication built on
paho-mqtt - Typed device & metric model — devices, metrics, enums, and values are fully typed Python objects
- Live updates — metrics update in real-time via callbacks as MQTT messages arrive
- Writable metrics — control switches, numbers, selects, time values, and text on your Victron system
- Device hierarchy — parent/child device relationships (e.g., switch → SwitchableOutput sub-devices)
- Formula metrics — computed metrics derived from other metrics (e.g., GPS location, energy calculations)
- Hidden metrics — source metrics consumed by formulas are hidden from public APIs
- Device tracker — combined GPS location with latitude, longitude, altitude, course, speed, and fix handling
- Metric viewer — built-in tkinter GUI for browsing devices and metrics with live updates
pip install victron_mqttimport asyncio
import victron_mqtt
async def main():
# Connect to your Venus OS device
hub = victron_mqtt.Hub("venus.local.", 1883, None, None, False)
await hub.connect()
await hub.wait_for_first_refresh()
# Browse devices and metrics
for device in hub.devices.values():
print(f"Device: {device.name} ({device.device_type})")
for metric in device.metrics:
print(f" {metric.short_id}: {metric.formatted_value}")
await hub.disconnect()
asyncio.run(main())The Hub is the main entry point. It manages the MQTT connection, discovers devices, and routes metric updates.
hub = victron_mqtt.Hub(
host="venus.local.", # MQTT broker hostname or IP
port=1883, # MQTT broker port
username=None, # Optional MQTT username
password=None, # Optional MQTT password
use_ssl=False, # Enable SSL/TLS
installation_id=None, # Auto-discovered if not provided
operation_mode=OperationMode.FULL, # FULL, READ_ONLY, or EXPERIMENTAL
device_type_exclude_filter=None, # Exclude specific device types
update_frequency_seconds=None, # Throttle update frequency
ssl_context=None, # Custom SSL context for certificate verification
)Security note: with
use_ssl=Trueand nossl_context, the TLS connection does not verify the broker's certificate, because GX devices ship with self-signed certificates. To verify, pass a context with your CA loaded:import ssl context = ssl.create_default_context(cafile="/path/to/venus-ca.crt") hub = victron_mqtt.Hub("venus.local.", 8883, None, "password", True, ssl_context=context)
Update frequency: update_frequency_seconds controls how often metric updates are delivered to on_update callbacks. Throttling only applies to numeric values; non-numeric changes are always delivered.
None(default) — deliver on every value change0— deliver on every MQTT message, even if the value is unchangedN > 0— deliver at most one update perNseconds per metric"auto"— per-metric interval chosen by the library based on the metric type: fast-changing metrics users watch live (power, current, location) update every few seconds, while the rest use a longer interval. SeeAUTO_UPDATE_INTERVALSinvictron_mqtt.constants."auto_unthrottled"— like"auto", but fast-changing metrics update on every value change with no time limit.
Key properties:
hub.devices— dict of all devices with visible metricshub.installation_id— the Venus OS installation identifierhub.connected— whether the MQTT connection is activehub.get_metric(unique_id)— look up a metric by its unique ID
Each Victron service (solar charger, battery, inverter, etc.) becomes a Device with properties like name, model, manufacturer, and serial number.
device = hub.devices["solarcharger_288"]
print(device.name) # "SmartSolar MPPT 150|35"
print(device.device_type) # DeviceType.SOLAR_CHARGER
print(device.model) # "SmartSolar MPPT 150|35"
print(device.parent_device) # parent Device or NoneDevice hierarchy: Sub-devices (like SwitchableOutput channels) have a parent_device reference pointing to their parent. Top-level devices are parented to the system device (system_0).
Custom names: Once a device has reported a non-null custom name (including an empty string), you can write a new name:
if device.can_set_custom_name:
device.set_custom_name("House battery")
# device.set_custom_name("") # Clear the custom nameThis also supports SwitchableOutput sub-devices, using their own
Settings/CustomName topic. Names are preserved exactly, without trimming.
device.custom_name and device.name update only when the device reports the
new value over MQTT; successful publishing does not confirm that it was applied.
can_set_custom_name is false in OperationMode.READ_ONLY, without an associated
hub, or before a custom name topic is discovered. It does not indicate connection
status. Unsupported writes raise ValueError, non-string names raise TypeError,
and transport failures propagate as NotConnectedError or PublishError.
Metrics represent individual data points on a device. Each metric has a kind, type, value, and optional unit.
metric = device.get_metric("solarcharger_yield_today")
print(metric.name) # "Yield today"
print(metric.value) # 3.45
print(metric.value_type) # ValueType.FLOAT
print(metric.formatted_value) # "3.45 kWh"
print(metric.metric_kind) # MetricKind.SENSOR
print(metric.metric_type) # MetricType.ENERGY
print(metric.unit_of_measurement) # "kWh"Metric kinds: SENSOR, BINARY_SENSOR, SWITCH, SELECT, NUMBER, TIME, TEXT, BUTTON, DEVICE_TRACKER
Value types: The read-only metric.value_type property returns a ValueType
enum, available via from victron_mqtt import ValueType. Consumers can check
metric.value_type is ValueType.JSON to identify JSON text; its value remains
a string, not a parsed JSON object.
Metric availability: On Venus OS v3.50 or newer, the hub requests a full republish every 180 seconds so constant values stay fresh. Every 60 seconds, it checks for metrics that have not been received for more than 360 seconds and marks them unavailable, retaining their last values. This also detects delivery stalls while the MQTT connection remains active.
Firmware detection is retried at full-publish processing boundaries until a valid version is available. While the version is missing or unparseable, stale detection and periodic full republishes remain enabled as a fail-safe. They are disabled once a version below v3.50 is detected. Older firmware whose version cannot be detected may therefore have constant metrics marked unavailable if it cannot refresh those values.
Switches, numbers, selects, time, and text metrics can be written back to the device:
from victron_mqtt import WritableMetric
metric = device.get_metric("evcharger_charge")
if isinstance(metric, WritableMetric):
metric.set("on") # Turn on
metric.set("off") # Turn off
# Number metrics accept float values
metric = device.get_metric("system_ac_power_setpoint")
if isinstance(metric, WritableMetric):
metric.set(500.0) # Set to 500W
# Select metrics accept enum id strings
metric = device.get_metric("system_ess_mode")
if isinstance(metric, WritableMetric):
metric.set("optimized")
# Text metrics accept strings, including an empty string
if isinstance(metric, WritableMetric) and metric.metric_kind == victron_mqtt.MetricKind.TEXT:
metric.set("New text")
metric.value = "" # The value setter also publishesDefine writable string topics with message_type=MetricKind.TEXT and
value_type=ValueType.STRING in their TopicDescriptor. Text is published as a
JSON string value without trimming whitespace or converting numeric-looking
strings. In OperationMode.READ_ONLY, text topics become read-only sensors.
Read-only mode also applies to formula-backed controls: switches become binary
sensors, and other writable kinds become sensors.
The metric viewer provides a text entry with an Apply button.
WritableMetric.set() (including formula-backed writes), Hub.publish(), and
firmware installation propagate transport failures:
NotConnectedErrorwhen MQTT is disconnected.Hub.publish()also raises it before the installation ID is known.PublishErrorfor other unsuccessful MQTT publish return codes.- Existing validation errors remain
ValueError, including out-of-range numbers and unavailable dropdown options.
Both transport exceptions are exported from victron_mqtt. Successful publication
means only that the MQTT client accepted the message for sending, not that the
device received or applied it. Publishing remains non-blocking with the existing
QoS. Periodic firmware checks log transport failures and retry at their next
scheduled interval rather than stopping.
Hub exceptions remain importable from both victron_mqtt and victron_mqtt.hub.
The unused ProgrammingError exception has been removed.
Register callbacks to be notified when new devices or metrics are discovered:
def on_new_device(hub, device):
print(f"New device: {device.name} (parent: {device.parent_device})")
def on_new_metric(hub, device, metric):
print(f"New metric on {device.name}: {metric.name} = {metric.formatted_value}")
hub.on_new_device = on_new_device # Fires for each new device
hub.on_new_metric = on_new_metric # Fires for each new visible metricNotification order: Devices are notified in topological order — parent devices always before their children. System devices are prioritized at the same depth.
Subscribe to live metric updates:
metric = device.get_metric("grid_power")
metric.on_update = lambda metric, value: print(f"Grid power: {value}W")GPS data is combined into a single GpsLocation object with DEVICE_TRACKER metric kind:
from victron_mqtt import GpsLocation
metric = device.get_metric("gps_location")
if isinstance(metric.value, GpsLocation):
print(f"Lat: {metric.value.latitude}")
print(f"Lon: {metric.value.longitude}")
print(f"Alt: {metric.value.altitude}")
print(f"Course: {metric.value.course}")
print(f"Speed: {metric.value.speed}")The GPS formula returns None when the GPS has no fix (Fix=0), so consumers can handle the "no position" state gracefully.
Victron enums provide code, id, and human-readable string representations:
from victron_mqtt import BatteryState
state = BatteryState.from_code(5)
print(state.code) # 5
print(state.id) # "float"
print(state.string) # "Float"
print(state) # "Float"OperationMode.FULL— all metrics, writable controls enabledOperationMode.READ_ONLY— all writable metrics become read-only sensorsOperationMode.EXPERIMENTAL— includes experimental/unstable metrics
Exclude specific device types from discovery:
from victron_mqtt import DeviceType
hub = victron_mqtt.Hub(
"venus.local.", 1883, None, None, False,
device_type_exclude_filter=[DeviceType.TEMPERATURE, DeviceType.GPS],
)| Device Type | Code | Description |
|---|---|---|
SYSTEM |
system |
Venus OS system |
SOLAR_CHARGER |
solarcharger |
MPPT solar chargers |
INVERTER |
inverter |
VE.Direct inverters |
BATTERY |
battery |
Battery monitors |
GRID |
grid |
Grid meters |
VEBUS |
vebus |
Multi/Quattro inverter-chargers |
EVCHARGER |
evcharger |
EV charging stations |
PVINVERTER |
pvinverter |
PV inverters |
TEMPERATURE |
temperature |
Temperature sensors |
GENERATOR |
generator |
Generator start/stop |
TANK |
tank |
Liquid tank sensors |
GPS |
gps |
GPS receivers |
SWITCH |
switch |
Switches and SwitchableOutputs |
DIGITAL_INPUT |
digitalinput |
Digital inputs |
DC_SYSTEM |
dcsystem |
DC system |
DC_LOAD |
dcload |
DC loads |
ALTERNATOR |
alternator |
Alternator chargers |
CHARGER |
charger |
AC chargers |
HEATPUMP |
heatpump |
Heat pumps |
ACLOAD |
acload |
AC loads |
DCDC |
dcdc |
DC/DC chargers |
TRANSFER_SWITCH |
TransferSwitch |
Transfer switches |
GX devices running Venus OS v3.80 or newer support token-based MQTT authentication. When pairing mode is enabled on the device, you can request credentials programmatically using request_pairing_token. The device_id must be alphanumeric (e.g. ab4c9ab6b98a). The GX device returns a PairingToken with token_name (used as MQTT username) and password:
import asyncio
from victron_mqtt import Hub, PairingError, request_pairing_token
async def main():
host = "venus.local."
try:
credentials = await request_pairing_token(host, "ab4c9ab6b98a")
except PairingError as exc:
print(f"Pairing failed: {exc}")
return
# credentials is a PairingToken(token_name=..., password=...)
print(f"Username: {credentials.token_name}")
print(f"Password: {credentials.password}")
# Use the credentials to connect via MQTT
hub = Hub(host, 8883, credentials.token_name, credentials.password, True)
await hub.connect()
await hub.wait_for_first_refresh()
for device in hub.devices.values():
9306
print(f"Device: {device.name}")
await hub.disconnect()
asyncio.run(main())Note: The GX device must have pairing mode enabled. If MQTT Access was turned off, it will automatically change to "Paired devices only" after a successful pairing.
Interactive GUI for browsing devices and metrics with live updates:
python -m victron_mqtt.utils.view_metricsFeatures: hierarchical device tree, search/filter, color-coded metric kinds, double-click to view/edit, live value updates, status bar.
Dump the raw MQTT structure from your device:
python -m victron_mqtt.utils.dump_mqtt --host venus.local. --port 1883Print raw MQTT messages as they arrive from a broker:
python -m victron_mqtt.utils.dumb_mqtt_listener --host venus.local. --port 1883Use --only_supported_victron to subscribe only to topics recognized by this library. Use --seed_victron to discover the installation ID and, after subscribing, publish a keepalive request that makes the GX device send all its current values:
python -m victron_mqtt.utils.dumb_mqtt_listener --host venus.local. --seed_victron--seed_victron works independently of --only_supported_victron. Both options can also be used together.
Browse all supported MQTT topic definitions on the documentation page.
The machine-readable topic definitions JSON is available here.
Each topic includes a concise description; enum options expose their own description without duplicating possible
values in the topic description.
Help extend the library with more topics! See CONTRIBUTING.md for instructions.
Found a bug? Log issues on GitHub.
Attach the output of dump_mqtt to help us debug your setup.
victron_mqtt is distributed under the terms of the MIT license.
- Thanks to Johan du Plessis johan@epicwin.co.za who started the original library this one is based on.
- Thanks to Victron Energy for their excellent hardware and documentation.