MQTT

Apps can exchange messages with an MQTT server, for example to read Zigbee2MQTT devices directly or to talk to devices Home Assistant does not know. Stewart keeps one connection to the server, and apps ask for Mqtt in their constructor. This app copies temperatures from MQTT into Home Assistant, and publishes the living room thermostat’s state back:

apps/ThermostatBridge.php

<?php

declare(strict_types=1);

namespace App;

use Stewart\Contracts\App;
use Stewart\Contracts\Automation;
use Stewart\Contracts\HaContext;
use Stewart\Contracts\Mqtt\Mqtt;
use Stewart\Contracts\Mqtt\MqttMessage;
use Stewart\Contracts\Mqtt\MqttQos;
use Stewart\Contracts\State\StateChange;

#[Automation(id: 'thermostat-bridge')]
final readonly class ThermostatBridge implements App
{
    public function __construct(
        private HaContext $ha,
        private Mqtt $mqtt,
    ) {}

    public function initialize(): void
    {
        $this->mqtt->watchMessages('zigbee2mqtt/+/temperature')->subscribe(function (MqttMessage $message): void {
            $this->ha->callService('input_number', 'set_value', ['value' => (float) $message->payload]);
        });

        $this->ha->watchStateChanges('climate.living_room')->subscribe(function (StateChange $change): void {
            $this->mqtt->publish('home/living_room/climate', ['state' => $change->to?->state], MqttQos::AtLeastOnce, retain: true);
        });
    }

    public function dispose(): void {}
}

Setting it up

Set the server in .env:

STEWART_MQTT__URL=mqtt://user:password@mosquitto:1883

Use mqtts://…:8883 for TLS. Without a URL, an app that asks for Mqtt does not start. Projects from the skeleton already have stewart-php/mqtt installed; other projects run composer require stewart-php/mqtt.

Optional settings in stewart.yaml:

mqtt:
  client_id: stewart-home        # default: stewart-<hostname>
  keepalive: 30s
  will: { topic: stewart/status, payload: offline, qos: 1, retain: true }

The will is a message the server publishes for Stewart when its connection dies.

Receiving

watchMessages('zigbee2mqtt/+/temperature') returns a stream of MqttMessages with topic, payload (text), qos and retain. decodeJsonPayload() reads a JSON payload.

Topic filters use MQTT’s wildcards: + matches one level, # the rest. A filter that starts with a wildcard does not match $SYS/… topics.

Publishing

publish($topic, $payload, $qos, retain: …) sends a message. A string payload is sent as it is, and an array as JSON.

QoS 0 and 1 are supported. A QoS 1 message published while the server is unreachable waits, up to mqtt.outbound_buffer messages, and is sent after the reconnect.

Testing

$harness->mqtt stands in for the server. pushMessage() sends a message to the app, and published lists what the app sent:

tests/ThermostatBridgeTest.php

<?php

declare(strict_types=1);

namespace App\Tests;

use App\ThermostatBridge;
use PHPUnit\Framework\TestCase;
use Stewart\Contracts\Mqtt\MqttMessage;
use Stewart\Contracts\Mqtt\MqttQos;
use Stewart\Testing\App\AppHarness;

final class ThermostatBridgeTest extends TestCase
{
    private AppHarness $harness;

    protected function setUp(): void
    {
        $this->harness = new AppHarness();
        $this->harness->ha->seedState('climate.living_room', 'off');
        $this->harness->startApp(ThermostatBridge::class);
    }

    public function testTemperatureFromMqttSetsTheInputNumber(): void
    {
        $this->harness->mqtt->pushMessage(new MqttMessage('zigbee2mqtt/kitchen/temperature', '21.5'));

        $call = $this->harness->ha->getLastCall();
        self::assertSame('input_number.set_value', $call->getServiceName());
        self::assertSame(['value' => 21.5], $call->data);
    }

    public function testClimateStateIsPublishedRetained(): void
    {
        $this->harness->ha->pushState('climate.living_room', 'heat');

        $message = $this->harness->mqtt->published->getLast();
        self::assertNotNull($message);
        self::assertSame('home/living_room/climate', $message->topic);
        self::assertSame(['state' => 'heat'], $message->decodeJsonPayload());
        self::assertSame(MqttQos::AtLeastOnce, $message->qos);
        self::assertTrue($message->retain);
    }
}

Common problems

MQTT is not configured. Set STEWART_MQTT__URL in .env.

Nothing arrives. Check the topic with an MQTT client, such as mosquitto_sub -h host -t '#' -v. Topics are case-sensitive.