Metadata-Version: 2.4
Name: ebus-mqtt-client
Version: 0.5.0
Summary: Standalone MQTT client wrapper around paho-mqtt with TLS, reconnection, subscription recovery, and topic pattern matching
Author: Clark Communications Corporation
License-Expression: MIT
Project-URL: Homepage, https://ebus.energy
Project-URL: Repository, https://github.com/electrification-bus/ebus-mqtt-client
Project-URL: Issues, https://github.com/electrification-bus/ebus-mqtt-client/issues
Keywords: mqtt,paho,iot,tls
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: paho-mqtt>=1.5.0
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: ruff>=0.15.0; extra == "dev"
Requires-Dist: mypy>=1.8; extra == "dev"
Dynamic: license-file

# ebus-mqtt-client

[![PyPI version](https://img.shields.io/pypi/v/ebus-mqtt-client.svg)](https://pypi.org/project/ebus-mqtt-client/)
[![Python versions](https://img.shields.io/pypi/pyversions/ebus-mqtt-client.svg)](https://pypi.org/project/ebus-mqtt-client/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![CI](https://github.com/electrification-bus/ebus-mqtt-client/actions/workflows/lint.yml/badge.svg)](https://github.com/electrification-bus/ebus-mqtt-client/actions/workflows/lint.yml)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

Standalone MQTT client wrapper around [paho-mqtt](https://pypi.org/project/paho-mqtt/).

## Features

- TLS support (secure with CA verification, insecure, or plaintext)
- Resilient connect: construction never blocks or raises on a down broker (`connect_async`); the connection is established and retried on the network loop started by `start()`
- QoS 0 retained publishes issued before the link is up are held (newest value per topic, bounded) and flushed on connect, rather than being discarded by paho and lost
- Automatic reconnection with configurable backoff
- Subscription recovery on reconnect
- Topic pattern matching via paho's `MQTTMatcher`
- Last Will and Testament (LWT)
- MQTTv3 and MQTTv5 protocol support
- Factory method for dict-based configuration
- Bounded, broker-independent shutdown: `stop(timeout=...)` returns promptly even against a dead broker
- Bounded publish flush: `publish_and_flush(...)` lands a final message before a clean disconnect, no fixed sleep
- Optional loop-native driving: `AsyncioMqttDriver` runs the network loop on your asyncio event loop instead of a background thread, for hosts that already own a loop

## Install

```bash
pip install ebus-mqtt-client
```

## Quick start

```python
from ebus_mqtt_client import MqttClient

client = MqttClient(
    client_id="my-client",
    endpoint="broker.example.com",
    port=1883,
)
client.start()

client.subscribe("sensors/#", callback_param)
client.publish("sensors/temp", "22.5")

client.stop()
```

### Graceful shutdown

Publish a final retained message and flush it (bounded) before disconnecting, then stop within a time bound even when the broker is unreachable:

```python
# Land a final state update, waiting up to 1s for it to actually be sent.
# Returns True on flush; False (without blocking or raising) if not connected,
# the publish fails, or the flush exceeds the timeout.
client.publish_and_flush(
    "devices/my-client/state", "disconnected", retain=True, timeout=1.0
)

# Returns within ~timeout seconds even if the broker is gone.
client.stop(timeout=2.0)
```

`publish()` also returns paho's `MQTTMessageInfo` (or `None` if there is no client), so you can wait for a single message yourself: `client.publish(topic, data).wait_for_publish(1.0)`.

### Publishing before the connection is up

Construction registers the broker with `connect_async` and returns; CONNACK does not arrive until the network loop started by `start()` receives it, so paho refuses anything published in between with `MQTT_ERR_NO_CONN`. That is an expected transient of startup, not a fault, so it no longer logs a warning. What becomes of the message depends on its QoS, because paho only discards some of them:

- **QoS 1 and 2** (`publish()` defaults to QoS 1): paho stores the message in its own out-queue *before* returning `MQTT_ERR_NO_CONN`, keeps it across reconnects, and re-sends it itself once CONNACK arrives. Nothing is held here, because a second copy would put every such message on the wire twice per connect.
- **QoS 0, retained**: paho keeps nothing, so the message would be lost. It is held here and flushed on connect, before the subscription recovery and before your `on_connect_callback`. The hold keeps the newest value per topic rather than replaying every attempt, because retained state is last-value-wins: a queue that replayed attempts in order could write a stale value on top of a newer one published after the connection came up. It is bounded by `client.pending_limit` (default 512 topics, evicting the oldest) so a client that never connects cannot grow without limit.
- **QoS 0, not retained**: dropped, with a debug-level line. These are events, and delivering one after an arbitrary delay announces something that was true once.

Any other publish failure still logs a warning, with the paho result code. A publish refused this way still returns paho's `MQTTMessageInfo` with `rc == MQTT_ERR_NO_CONN` whichever branch it took, so a caller that inspects the result sees exactly what paho said. `stop()` discards anything still held.

Two things this does not change, both of them paho's behavior rather than this wrapper's. paho's own out-queue is unbounded unless you set `client.mqttc.max_queued_messages_set(...)`, so `pending_limit` bounds only the QoS 0 hold. And paho replays that queue *after* `on_connect` returns, in the order the publishes were attempted, so at QoS 1 and 2 a value revised while disconnected is re-sent stale-first: if you republish inside `on_connect_callback`, paho's older copy can land after yours. Publish state you care about at QoS 0, or republish after the connection is established.

`publish_and_flush()` is unaffected: it checks `is_connected()` and returns `False` immediately rather than holding, since its whole purpose is to confirm a message reached the broker now.

### From a config dict

```python
cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_cert": "/path/to/ca.pem",
    "authentication": {
        "type": "USER_PASS",
        "username": "user",
        "password": "secret",
    },
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()
```

### mTLS (client-certificate authentication)

When the broker authenticates the client via the TLS handshake (no username/password), supply a client cert and key. File-path form:

```python
cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_cert": "/path/to/ca.pem",
    "tls_client_cert": "/path/to/client.crt",
    "tls_client_key": "/path/to/client.key",
    # "tls_client_key_password": "...",  # only if the key is encrypted
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()
```

In-memory form — useful when the cert/key are fetched from a secret store rather than the filesystem. If both the path and `*_data` forms are supplied for the same item, the `*_data` form wins and a warning is logged:

```python
cfg = {
    "host": "broker.example.com",
    "port": 8883,
    "use_tls": True,
    "tls_insecure": False,
    "tls_ca_data": ca_pem_str,
    "tls_client_cert_data": client_cert_pem_str,
    "tls_client_key_data": client_key_pem_str,
}

client = MqttClient.from_config(cfg, client_id="my-client")
client.start()
```

### Loop-native driving (asyncio)

By default `start()` runs paho's network loop on a background thread. If your program already owns an asyncio event loop (for example a Home Assistant integration), you can drive the same client on that loop with no extra thread, via the optional `AsyncioMqttDriver`:

```python
import asyncio
from ebus_mqtt_client import AsyncioMqttDriver, MqttClient

async def main():
    client = MqttClient.from_config(cfg, client_id="my-client")
    driver = client.asyncio_driver()      # or: AsyncioMqttDriver(client, loop=my_loop)
    await driver.start()                   # instead of client.start()
    client.subscribe("sensors/#", callback_param)
    # ... all MQTT I/O now runs on this event loop ...
    await driver.stop()

asyncio.run(main())
```

Thread mode (`client.start()`) and the driver are mutually exclusive per client: pick one. The driver module is imported lazily (only when you reference `AsyncioMqttDriver` or call `asyncio_driver()`), so a thread-only consumer never loads the asyncio machinery.

If you inject the client into `ebus_sdk.Controller(mqttc=client)` as a bring-your-own transport, wire `Controller.resync` onto the on-connect callback (`client.on_connect_callback = controller.resync`) so the retained tree re-walks after a reconnect; the SDK does that automatically only for a client it creates itself.

## Releasing

The version lives in exactly one place: `__version__` in `src/ebus_mqtt_client/__init__.py`. `pyproject.toml` reads it dynamically, the `setup.py` legacy shim reads it by regex, and the publish workflow refuses to release a tag that disagrees with it. To cut a release:

1. Bump `__version__` in `src/ebus_mqtt_client/__init__.py` (the only place).
2. Move the CHANGELOG's `[Unreleased]` entries under a new version heading.
3. Commit: `git commit -am "Release X.Y.Z"`.
4. Tag it to match, `v`-prefixed: `git tag vX.Y.Z`.
5. Push the tag: `git push --tags` (a plain `git push` does not trigger a release).

Pushing a `v*` tag runs the publish workflow, which verifies the tag equals `v$__version__` (a mismatch fails before anything is built), builds the sdist and wheel, and publishes to PyPI via Trusted Publishing (OIDC, no stored token).

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md) for how to file Discussions, Issues, and pull requests. The library is intentionally a thin MQTT-only layer — Homie / eBus features belong in [`ebus-sdk`](https://github.com/electrification-bus/python-sdk).

## License

[MIT License](LICENSE) — Copyright (c) 2026 Clark Communications Corporation
