direct-method-mqtt-python 1.2.3

MQTT client implementation for Python - part of Direct Method Library

direct-method-mqtt-python

Python MQTT client implementation - part of the Direct Method Library.

Installation

Using pip (recommended)

pip install direct-method-mqtt-python

Development Installation

If you encounter permission issues with system Python, use a virtual environment:

# Create and activate virtual environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install the package
pip install direct-method-mqtt-python
# Or for development: pip install -e .

Developer setup

For contributor-friendly development, we recommend creating a local virtual environment named .venv inside the package directory and installing the project plus dev dependencies there.

From the packages/sdk-python directory run:

# create and activate a local virtualenv
python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

# make sure pip/tools are up-to-date
python -m pip install --upgrade pip setuptools wheel

# install the package for development plus dev dependencies (if defined in pyproject or extras)
# If you maintain a dev requirements file, use: pip install -r requirements-dev.txt
pip install -e .[dev] || pip install -e .

# when done: deactivate
# deactivate

Notes:

• Do NOT commit .venv to the repository; it is already ignored by .gitignore.
• Use pip install -r requirements-dev.txt if you prefer a pinned, shareable developer environment (recommended for teams).
• CI creates its own ephemeral environments; the local .venv is solely for developer ergonomics.

Quick Start

from direct_method_mqtt import DirectMethodMqttClient

# Create client instance
client = DirectMethodMqttClient("localhost", 1883, "test/topic")

# Set up message handler
def on_message(topic, message):
    print(f"Received: {message} on topic: {topic}")

client.on_message_received(on_message)

# Connect and start using the client
try:
    client.connect()
    client.subscribe()
    client.publish("Hello from Python!")
    
    # Keep the application running to receive messages
    # In a real application, you might want to handle this differently
    import time
    time.sleep(5)
    
finally:
    client.disconnect()

Using as Context Manager

from direct_method_mqtt import DirectMethodMqttClient

def handle_message(topic, message):
    print(f"Got message: {message} on {topic}")

# Automatically handles connection and cleanup
with DirectMethodMqttClient("localhost", 1883, "test/topic") as client:
    client.on_message_received(handle_message)
    client.subscribe()
    client.publish("Hello World!")
    
    import time
    time.sleep(2)  # Wait for potential responses

API Reference

Constructor

DirectMethodMqttClient(broker_host, broker_port=1883, topic="", client_id=None)

• broker_host (str): MQTT broker hostname or IP address
• broker_port (int, optional): MQTT broker port (default: 1883)
• topic (str): Topic to subscribe and publish to
• client_id (str, optional): Unique client identifier (auto-generated if not provided)

Methods

connect(timeout=5.0) -> None

Connect to the MQTT broker.

• timeout (float): Connection timeout in seconds

disconnect() -> None

Disconnect from the MQTT broker.

publish(message: str) -> None

Publish a message to the configured topic.

subscribe() -> None

Subscribe to the configured topic.

on_message_received(callback: MessageCallback) -> None

Set a callback function to handle received messages.

• callback: Function that accepts (topic: str, message: str)

Properties

is_connected: bool

Get the current connection status.

topic: str

Get the configured topic.

broker_info: Tuple[str, int]

Get the broker connection details as (host, port).

client_id: str

Get the client ID.

Error Handling

The client includes comprehensive error handling:

from direct_method_mqtt import DirectMethodMqttClient

client = DirectMethodMqttClient("localhost", 1883, "test/topic")

try:
    client.connect(timeout=10.0)
    client.publish("test message")
except ConnectionError as e:
    print(f"Connection failed: {e}")
except RuntimeError as e:
    print(f"MQTT operation failed: {e}")
except ValueError as e:
    print(f"Invalid parameter: {e}")
finally:
    client.disconnect()

Common exceptions:

• ConnectionError: Connection to broker failed
• RuntimeError: MQTT operations when not connected, publish/subscribe failures
• ValueError: Invalid constructor parameters or empty messages

Advanced Usage

Custom Client ID and Logging

import logging
from direct_method_mqtt import DirectMethodMqttClient

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

client = DirectMethodMqttClient(
    "mqtt.example.com", 
    1883, 
    "sensors/temperature",
    "my-unique-client-id"
)

def temperature_handler(topic, message):
    try:
        temp = float(message)
        print(f"Temperature reading: {temp}°C")
    except ValueError:
        print(f"Invalid temperature data: {message}")

client.on_message_received(temperature_handler)

Robust Connection with Retry Logic

import time
from direct_method_mqtt import DirectMethodMqttClient

def connect_with_retry(client, max_retries=3, retry_delay=5.0):
    """Connect to MQTT broker with retry logic."""
    for attempt in range(max_retries):
        try:
            client.connect(timeout=10.0)
            print("Connected successfully!")
            return True
        except ConnectionError as e:
            print(f"Connection attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                print(f"Retrying in {retry_delay} seconds...")
                time.sleep(retry_delay)
    
    print("Failed to connect after all retries")
    return False

client = DirectMethodMqttClient("localhost", 1883, "app/messages")

if connect_with_retry(client):
    try:
        client.subscribe()
        client.publish("Connection established")
        
        # Your application logic here
        time.sleep(10)
        
    finally:
        client.disconnect()

Message Processing with JSON

import json
from direct_method_mqtt import DirectMethodMqttClient

def process_json_message(topic, message):
    """Process JSON messages safely."""
    try:
        data = json.loads(message)
        print(f"Received data: {data}")
        
        # Process specific message types
        if data.get("type") == "sensor_reading":
            print(f"Sensor {data['sensor_id']}: {data['value']} {data['unit']}")
        elif data.get("type") == "alert":
            print(f"ALERT: {data['message']}")
        else:
            print(f"Unknown message type: {data.get('type')}")
            
    except json.JSONDecodeError:
        print(f"Received non-JSON message: {message}")
    except KeyError as e:
        print(f"Missing required field in message: {e}")

client = DirectMethodMqttClient("localhost", 1883, "data/json")
client.on_message_received(process_json_message)

with client:
    client.subscribe()
    
    # Send some test messages
    test_messages = [
        '{"type": "sensor_reading", "sensor_id": "temp01", "value": 23.5, "unit": "°C"}',
        '{"type": "alert", "message": "Temperature too high!"}',
        '{"type": "unknown", "data": "some data"}',
        'not a json message'
    ]
    
    for msg in test_messages:
        client.publish(msg)
        time.sleep(1)
    
    time.sleep(5)  # Wait for message processing

Threading Considerations

The client is thread-safe for most operations, but be aware that:

• Message callbacks are executed in the MQTT client's background thread
• Connection state changes are protected by internal locking
• If you need to publish from within a message callback, it's safe to do so

from direct_method_mqtt import DirectMethodMqttClient
import threading

class MqttHandler:
    def __init__(self, client):
        self.client = client
        self.message_count = 0
        self.lock = threading.Lock()
    
    def handle_message(self, topic, message):
        with self.lock:
            self.message_count += 1
            print(f"Message #{self.message_count}: {message}")
            
            # Safe to publish from callback
            if message.lower() == "ping":
                self.client.publish("pong")

client = DirectMethodMqttClient("localhost", 1883, "test/echo")
handler = MqttHandler(client)

with client:
    client.on_message_received(handler.handle_message)
    client.subscribe()
    client.publish("ping")
    
    time.sleep(2)

Requirements

• Python 3.9 or higher
• paho-mqtt 1.6.0 or higher

Type Hints

This package includes full type hints for better IDE support and type checking:

from typing import Optional
from direct_method_mqtt import DirectMethodMqttClient, MessageCallback

def create_client(host: str, port: int = 1883) -> DirectMethodMqttClient:
    return DirectMethodMqttClient(host, port, "my/topic")

def my_callback(topic: str, message: str) -> None:
    print(f"{topic}: {message}")

client = create_client("localhost")
client.on_message_received(my_callback)

License

MIT License - see the LICENSE file for details.

Contributing

This is part of the Direct Method Library demo project. See the main repository for contribution guidelines.