Metadata-Version: 2.5
Name: victron_mqtt
Version: 2026.10.1
Summary: Python library for communicating with Victron Venus OS MQTT interface
Project-URL: Documentation, https://github.com/tomer-w/victron_mqtt#readme
Project-URL: Issues, https://github.com/tomer-w/victron_mqtt/issues
Project-URL: Source, https://github.com/tomer-w/victron_mqtt
Author-email: Tomer <57483589+tomer-w@users.noreply.github.com>
License-Expression: MIT
License-File: LICENSE
Keywords: mqtt,venus os,victron
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Python: >=3.11
Requires-Dist: aiohttp>=3.8
Requires-Dist: paho-mqtt>=2.0
Description-Content-Type: text/markdown

# victron_mqtt

[![PyPI - Version](https://img.shields.io/pypi/v/victron_mqtt.svg)](https://pypi.org/project/victron_mqtt)
[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/victron_mqtt.svg)](https://pypi.org/project/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](https://github.com/tomer-w/ha-victron-mqtt) 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, and time values 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

```bash
pip install victron_mqtt
```

## Quick Start

```python
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.

```python
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:
>
> ```python
> 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.

```python
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`).

### Metrics

Metrics represent individual data points on a device. Each metric has a kind, type, value, and optional unit.

```python
metric = device.get_metric("solarcharger_yield_today")
print(metric.name)              # "Yield today"
print(metric.value)             # 3.45
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`, `BUTTON`, `DEVICE_TRACKER`

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

### Writable Metrics

Switches, numbers, selects, and time metrics can be written back to the device:

```python
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")
```

### Callbacks

Register callbacks to be notified when new devices or metrics are discovered:

```python
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 metric
```

**Notification 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:

```python
metric = device.get_metric("grid_power")
metric.on_update = lambda metric, value: print(f"Grid power: {value}W")
```

### GPS Location

GPS data is combined into a single `GpsLocation` object with `DEVICE_TRACKER` metric kind:

```python
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.

### Enums

Victron enums provide code, id, and human-readable string representations:

```python
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"
```

### Operation Modes

- `OperationMode.FULL` — all metrics, writable controls enabled
- `OperationMode.READ_ONLY` — all writable metrics become read-only sensors
- `OperationMode.EXPERIMENTAL` — includes experimental/unstable metrics

### Device Type Filtering

Exclude specific device types from discovery:

```python
from victron_mqtt import DeviceType

hub = victron_mqtt.Hub(
    "venus.local.", 1883, None, None, False,
    device_type_exclude_filter=[DeviceType.TEMPERATURE, DeviceType.GPS],
)
```

## Supported Device Types

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

## MQTT Token Pairing

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`:

```python
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():
        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.

## Tools

### Metric Viewer

Interactive GUI for browsing devices and metrics with live updates:

```bash
python -m victron_mqtt.utils.view_metrics
```

Features: hierarchical device tree, search/filter, color-coded metric kinds, double-click to view/edit, live value updates, status bar.

### MQTT Dump

Dump the raw MQTT structure from your device:

```bash
python -m victron_mqtt.utils.dump_mqtt --host venus.local. --port 1883
```

### MQTT Listener

Print raw MQTT messages as they arrive from a broker:

```bash
python -m victron_mqtt.utils.dumb_mqtt_listener --host venus.local. --port 1883
```

Use `--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:

```bash
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.

### Topic Definitions

Browse all supported MQTT topic definitions on the [documentation page](https://tomer-w.github.io/victron_mqtt/).

The machine-readable topic definitions JSON is available [here](https://raw.githubusercontent.com/tomer-w/victron_mqtt/refs/heads/main/victron_mqtt.json).
Each topic includes a concise `description`; enum options expose their own `description` without duplicating possible
values in the topic description.

## Contributing

Help extend the library with more topics! See [CONTRIBUTING.md](CONTRIBUTING.md) for instructions.

## Logging Issues

Found a bug? Log issues on [GitHub](https://github.com/tomer-w/victron_mqtt/issues).

Attach the output of `dump_mqtt` to help us debug your setup.

## License

`victron_mqtt` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.

## Acknowledgments

- Thanks to Johan du Plessis <johan@epicwin.co.za> who [started](https://github.com/JohansLab/victronvenusclient) the original library this one is based on.
- Thanks to Victron Energy for their excellent hardware and documentation.
