MQTT

Interaction with the MQTT broker

Interaction with the MQTT broker is done through the self.mqtt object in the rule or through the MqttItem. When receiving a topic for the first time a new MqttItem will automatically be created.

Rule Interface

class mqtt

publish(topic: str, payload: typing.Any[, qos: int = None, retain: bool = None]) → int

Publish a value under a certain topic.

Parameters:

topic – MQTT topic
payload – MQTT Payload
qos (int) – QoS, can be 0, 1 or 2. If not specified value from configuration file will be used.
retain (bool) – retain message. If not specified value from configuration file will be used.

Returns:

0 if successful

subscribe(self, topic: str[, qos: int = None]) → int

Subscribe to a MQTT topic. Please note that subscriptions made this way are volatile, and will only remain until the next disconnect. For persistent subscriptions use the corresponding entry in the configuration file. By default HABApp listens to all topics so the topics can be used in listen_event.

Parameters:

topic – MQTT topic to subscribe to
qos – QoS, can be 0, 1 or 2. If not specified value from configuration file will be used.

Returns:

0 if successful

unsubscribe(self, topic: str) → int

Unsubscribe from a MQTT topic

Parameters:: topic – MQTT topic
Returns:: 0 if successful

Mqtt item types

Mqtt items have an additional publish method which make interaction with the mqtt broker easier.

from HABApp.mqtt.items import MqttItem
from HABApp.core.events import ValueChangeEvent

# Messages with a retain flag will automatically create a corresponding item in HABApp.
# All other items have to be created manually
my_mqtt_item = MqttItem.get_create_item('test/topic')

# easy to publish values
my_mqtt_item.publish('new_value')

# comparing the item to get the state works, too
if my_mqtt_item == 'test':
    pass # do something

MqttItem

class MqttItem()

A simple item that represents a topic and a value

classmethod get_create_item(name, initial_value=None)

Creates a new item in HABApp and returns it or returns the already existing one with the given name

Parameters:

name (str) – item name
initial_value – state the item will have if it gets created

Return type:

MqttItem

Returns:

item

classmethod get_item(name)

Returns an already existing item. If it does not exist or has a different item type an exception will occur.

Parameters:: name (str) – Name of the item
Return type:: Self

command_value(value)

Send a command to the topic, the same as publish

Parameters:: value (Any) – value to be sent
Return type:: None

get_value(default_value=None)

Return the value of the item. This is a helper function that returns a default in case the item value is None.

Parameters:: default_value – Return this value if the item value is None
Return type:: Any
Returns:: value of the item

listen_event(callback, event_filter=None)

Register an event listener which listens to all event that the item receives

Parameters:

callback (Callable[[Any], Any]) – callback that accepts one parameter which will contain the event
event_filter (EventFilterBase | None) – Event filter. This is typically ValueUpdateEventFilter or ValueChangeEventFilter which will also trigger on changes/update from openhab or mqtt. Additionally, it can be an instance of EventFilter which additionally filters on the values of the event. It is also possible to group filters logically with, e.g. AndFilterGroup and OrFilterGroup

Return type:

EventBusListener

post_value(new_value)

Set a new value and post appropriate events on the HABApp event bus (ValueUpdateEvent, ValueChangeEvent)

Parameters:: new_value (Any) – new value of the item
Return type:: bool
Returns:: True if state has changed

post_value_if(new_value, *, equal=<Missing>, eq=<Missing>, not_equal=<Missing>, ne=<Missing>, lower_than=<Missing>, lt=<Missing>, lower_equal=<Missing>, le=<Missing>, greater_than=<Missing>, gt=<Missing>, greater_equal=<Missing>, ge=<Missing>, is_=<Missing>, is_not=<Missing>)

Post a value depending on the current state of the item. If one of the comparisons is true the new state will be posted.

Parameters:

new_value – new value to post
equal – item state has to be equal to the passed value
eq – item state has to be equal to the passed value
not_equal – item state has to be not equal to the passed value
ne – item state has to be not equal to the passed value
lower_than – item state has to be lower than the passed value
lt – item state has to be lower than the passed value
lower_equal – item state has to be lower equal the passed value
le – item state has to be lower equal the passed value
greater_than – item state has to be greater than the passed value
gt – item state has to be greater than the passed value
greater_equal – item state has to be greater equal the passed value
ge – item state has to be greater equal the passed value
is – item state has to be the same object as the passt value (e.g. None)
is_not – item state has to be not the same object as the passt value (e.g. None)

Return type:

bool

Returns:

True if the new value was posted else False

publish(payload, qos=None, retain=None)

Publish the payload under the topic from the item.

Parameters:

payload – MQTT Payload
qos (int | None) – QoS, can be 0, 1 or 2. If not specified value from configuration file will be used.
retain (bool | None) – retain message. If not specified value from configuration file will be used.

set_value(new_value)

Set a new value without creating events on the event bus

Parameters:: new_value (Any) – new value of the item
Return type:: bool
Returns:: True if state has changed

watch_change(secs)

Generate an event if the item does not change for a certain period of time. Has to be called from inside a rule function.

Parameters:: secs (timedelta | TimeDelta | int | float | str) – secs after which the event will occur, max 1 decimal digit for floats
Return type:: ItemNoChangeWatch
Returns:: The watch obj which can be used to cancel the watch

watch_update(secs)

Generate an event if the item does not receive and update for a certain period of time. Has to be called from inside a rule function.

Parameters:: secs (timedelta | TimeDelta | int | float | str) – secs after which the event will occur, max 1 decimal digit for floats
Return type:: ItemNoUpdateWatch
Returns:: The watch obj which can be used to cancel the watch

property last_change: InstantView

Returns:: Timestamp of the last time when the item has been changed (read only)

property last_update: InstantView

Returns:: Timestamp of the last time when the item has been updated (read only)

property name: str

Returns:: Name of the item (read only)

MqttPairItem

An item that consolidates a topic that reports states from a device and a topic that is used to write to a device. It is created on the topic that reports the state from the device.

from HABApp.mqtt.items import MqttPairItem

# MqttPairItem works out of the box with zigbee2mqtt
mqtt = MqttPairItem.get_create_item("zigbee2mqtt/my_bulb/brightness")
mqtt.publish("255")  # <-- will use the write topic

# equivalent to
mqtt = MqttPairItem.get_create_item("zigbee2mqtt/my_bulb/brightness", write_topic="zigbee2mqtt/my_bulb/set/brightness")

Inheritance diagram of HABApp.mqtt.items.MqttPairItem

class MqttPairItem()

An item that represents both a topic that is used to read and a corresponding topic that is used to write values

classmethod get_create_item(name, write_topic=None, initial_value=None)

Creates a new item in HABApp and returns it or returns the already existing one with the given name. HABApp tries to automatically derive the write topic from the item name. In cases where this does not work it can be specified manually.

Parameters:

name (str) – item name (topic that reports the state)
write_topic (str | None) – topic that is used to write values or None (default) to build it automatically
initial_value – state the item will have if it gets created

Return type:

MqttPairItem

Returns:

item

classmethod get_item(name)

Returns an already existing item. If it does not exist or has a different item type an exception will occur.

Parameters:: name (str) – Name of the item
Return type:: Self

command_value(value)

Send a ValueCommandEvent for the item to the HABApp event bus. A ValueCommandEvent is typically used to indicate that the item should change the value. E.g. a command “ON” to a dimmer might result in a brightness value of 100%.

Parameters:: value (Any) – the commanded value
Return type:: None

get_value(default_value=None)

Return the value of the item. This is a helper function that returns a default in case the item value is None.

Parameters:: default_value – Return this value if the item value is None
Return type:: Any
Returns:: value of the item

listen_event(callback, event_filter=None)

Register an event listener which listens to all event that the item receives

Parameters:

callback (Callable[[Any], Any]) – callback that accepts one parameter which will contain the event
event_filter (EventFilterBase | None) – Event filter. This is typically ValueUpdateEventFilter or ValueChangeEventFilter which will also trigger on changes/update from openhab or mqtt. Additionally, it can be an instance of EventFilter which additionally filters on the values of the event. It is also possible to group filters logically with, e.g. AndFilterGroup and OrFilterGroup

Return type:

EventBusListener

post_value(new_value)

Set a new value and post appropriate events on the HABApp event bus (ValueUpdateEvent, ValueChangeEvent)

Parameters:: new_value (Any) – new value of the item
Return type:: bool
Returns:: True if state has changed

post_value_if(new_value, *, equal=<Missing>, eq=<Missing>, not_equal=<Missing>, ne=<Missing>, lower_than=<Missing>, lt=<Missing>, lower_equal=<Missing>, le=<Missing>, greater_than=<Missing>, gt=<Missing>, greater_equal=<Missing>, ge=<Missing>, is_=<Missing>, is_not=<Missing>)

Post a value depending on the current state of the item. If one of the comparisons is true the new state will be posted.

Parameters:

new_value – new value to post
equal – item state has to be equal to the passed value
eq – item state has to be equal to the passed value
not_equal – item state has to be not equal to the passed value
ne – item state has to be not equal to the passed value
lower_than – item state has to be lower than the passed value
lt – item state has to be lower than the passed value
lower_equal – item state has to be lower equal the passed value
le – item state has to be lower equal the passed value
greater_than – item state has to be greater than the passed value
gt – item state has to be greater than the passed value
greater_equal – item state has to be greater equal the passed value
ge – item state has to be greater equal the passed value
is – item state has to be the same object as the passt value (e.g. None)
is_not – item state has to be not the same object as the passt value (e.g. None)

Return type:

bool

Returns:

True if the new value was posted else False

publish(payload, qos=None, retain=None)

Publish the payload under the write topic from the item.

Parameters:

payload – MQTT Payload
qos (int | None) – QoS, can be 0, 1 or 2. If not specified value from configuration file will be used.
retain (bool | None) – retain message. If not specified value from configuration file will be used.

Returns:

0 if successful

set_value(new_value)

Set a new value without creating events on the event bus

Parameters:: new_value (Any) – new value of the item
Return type:: bool
Returns:: True if state has changed

watch_change(secs)

Generate an event if the item does not change for a certain period of time. Has to be called from inside a rule function.

Parameters:: secs (timedelta | TimeDelta | int | float | str) – secs after which the event will occur, max 1 decimal digit for floats
Return type:: ItemNoChangeWatch
Returns:: The watch obj which can be used to cancel the watch

watch_update(secs)

Generate an event if the item does not receive and update for a certain period of time. Has to be called from inside a rule function.

Parameters:: secs (timedelta | TimeDelta | int | float | str) – secs after which the event will occur, max 1 decimal digit for floats
Return type:: ItemNoUpdateWatch
Returns:: The watch obj which can be used to cancel the watch

property last_change: InstantView

Returns:: Timestamp of the last time when the item has been changed (read only)

property last_update: InstantView

Returns:: Timestamp of the last time when the item has been updated (read only)

property name: str

Returns:: Name of the item (read only)

Mqtt event types

MqttValueUpdateEvent

Since this event inherits from ValueUpdateEvent you can listen to ValueUpdateEvent and it will also trigger for MqttValueUpdateEvent.

class MqttValueUpdateEvent(name, value)

MqttValueChangeEvent

Since this event inherits from ValueChangeEvent you can listen to ValueChangeEvent and it will also trigger for MqttValueChangeEvent.

class MqttValueChangeEvent(name, value, old_value)

Mqtt util

MqttPublishOptions

class MqttPublishOptions(topic, qos=None, retain=None)

Allows to store the topic, qos and retain settings for a topic. These values can then be used to publish

publish(payload)

Publish a payload

Parameters:: payload (Any) – MQTT Payload
Return type:: None

replace(topic=None, qos=None, retain=None)

Replace the topic, qos and retain with the given values and return a new object.

Parameters:

topic (str | None) – New topic (if provided)
qos (int | None) – New qos (if provided)
retain (bool | None) – New retain (if provided)

Return type:

Self

Returns:

New object

property qos: int | None: QOS

property retain: bool | None: Retain

property topic: str: The topic

from HABApp.mqtt.util import MqttPublishOptions

topic = MqttPublishOptions('my/output/only/topic')
topic.publish('new_value')

topic_qos = MqttPublishOptions('my/output/only/topic', qos=2)
topic_qos.publish('new_value')

# create new through replace command wich will use qos=2 and retain=True
topic_qos_retain = topic_qos.replace(retain=True)
topic_qos_retain.publish('new_value')

Example MQTT rule

import datetime
import random

import HABApp
from HABApp.core.events import ValueUpdateEvent, ValueUpdateEventFilter
from HABApp.mqtt.items import MqttItem


class ExampleMqttTestRule(HABApp.Rule):
    def __init__(self) -> None:
        super().__init__()

        self.run.every(
            start_time=datetime.timedelta(seconds=10),
            interval=datetime.timedelta(seconds=20),
            callback=self.publish_rand_value
        )

        self.my_mqtt_item = MqttItem.get_create_item('test/test')

        self.listen_event('test/test', self.topic_updated, ValueUpdateEventFilter())

    def publish_rand_value(self) -> None:
        print('test mqtt_publish')
        self.my_mqtt_item.publish(str(random.randint(0, 1000)))

    def topic_updated(self, event) -> None:
        assert isinstance(event, ValueUpdateEvent), type(event)
        print( f'mqtt topic "test/test" updated to {event.value}')


ExampleMqttTestRule()