victron_mqtt

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.

Features

• 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

Installation

pip install victron_mqtt

Quick Start

import 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())

Core Concepts

Hub

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=True and no ssl_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 change
• 0 — deliver on every MQTT message, even if the value is unchanged
• N > 0 — deliver at most one update per N seconds 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. See AUTO_UPDATE_INTERVALS in victron_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 metrics
• hub.installation_id — the Venus OS installation identifier
• hub.connected — whether the MQTT connection is active
• hub.get_metric(unique_id) — look up a metric by its unique ID

Devices

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 None

Device 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 name

This 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.

以上为 README 的开头部分。其余内容请阅读原文。 查看完整 README ↗