Add command template and code_format support for MQTT lock (#85830)

* Add command template for MQTT lock

* Fix tests

homeassistant/components/mqtt/abbreviations.py

@@ -42,6 +42,7 @@ ABBREVIATIONS = {
     "cmd_tpl": "command_template",
     "cod_arm_req": "code_arm_required",
     "cod_dis_req": "code_disarm_required",
+    "cod_form": "code_format",
     "cod_trig_req": "code_trigger_required",
     "curr_hum_t": "current_humidity_topic",
     "curr_hum_tpl": "current_humidity_template",

homeassistant/components/mqtt/lock.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 import functools
+import re
 from typing import Any
 
 import voluptuous as vol
@@ -14,11 +15,12 @@ from homeassistant.const import CONF_NAME, CONF_OPTIMISTIC, CONF_VALUE_TEMPLATE
 from homeassistant.core import HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, TemplateVarsType
 
 from . import subscription
 from .config import MQTT_RW_SCHEMA
 from .const import (
+    CONF_COMMAND_TEMPLATE,
     CONF_COMMAND_TOPIC,
     CONF_ENCODING,
     CONF_QOS,
@@ -32,9 +34,17 @@ from .mixins import (
     async_setup_entry_helper,
     warn_for_legacy_schema,
 )
-from .models import MqttValueTemplate, ReceiveMessage, ReceivePayloadType
+from .models import (
+    MqttCommandTemplate,
+    MqttValueTemplate,
+    PublishPayloadType,
+    ReceiveMessage,
+    ReceivePayloadType,
+)
 from .util import get_mqtt_data
 
+CONF_CODE_FORMAT = "code_format"
+
 CONF_PAYLOAD_LOCK = "payload_lock"
 CONF_PAYLOAD_UNLOCK = "payload_unlock"
 CONF_PAYLOAD_OPEN = "payload_open"
@@ -64,6 +74,8 @@ MQTT_LOCK_ATTRIBUTES_BLOCKED = frozenset(
 
 PLATFORM_SCHEMA_MODERN = MQTT_RW_SCHEMA.extend(
     {
+        vol.Optional(CONF_CODE_FORMAT): cv.is_regex,
+        vol.Optional(CONF_COMMAND_TEMPLATE): cv.template,
         vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
         vol.Optional(CONF_PAYLOAD_LOCK, default=DEFAULT_PAYLOAD_LOCK): cv.string,
         vol.Optional(CONF_PAYLOAD_UNLOCK, default=DEFAULT_PAYLOAD_UNLOCK): cv.string,
@@ -123,8 +135,12 @@ class MqttLock(MqttEntity, LockEntity):
     _entity_id_format = lock.ENTITY_ID_FORMAT
     _attributes_extra_blocked = MQTT_LOCK_ATTRIBUTES_BLOCKED
 
+    _compiled_pattern: re.Pattern[Any] | None
     _optimistic: bool
     _valid_states: list[str]
+    _command_template: Callable[
+        [PublishPayloadType, TemplateVarsType], PublishPayloadType
+    ]
     _value_template: Callable[[ReceivePayloadType], ReceivePayloadType]
 
     def __init__(
@@ -145,7 +161,18 @@ class MqttLock(MqttEntity, LockEntity):
 
     def _setup_from_config(self, config: ConfigType) -> None:
         """(Re)Setup the entity."""
-        self._optimistic = config[CONF_OPTIMISTIC]
+        self._optimistic = (
+            config[CONF_OPTIMISTIC] or self._config.get(CONF_STATE_TOPIC) is None
+        )
+
+        self._compiled_pattern = config.get(CONF_CODE_FORMAT)
+        self._attr_code_format = (
+            self._compiled_pattern.pattern if self._compiled_pattern else None
+        )
+
+        self._command_template = MqttCommandTemplate(
+            config.get(CONF_COMMAND_TEMPLATE), entity=self
+        ).async_render
 
         self._value_template = MqttValueTemplate(
             config.get(CONF_VALUE_TEMPLATE),
@@ -209,9 +236,10 @@ class MqttLock(MqttEntity, LockEntity):
 
         This method is a coroutine.
         """
+        payload = self._command_template(self._config[CONF_PAYLOAD_LOCK], kwargs)
         await self.async_publish(
             self._config[CONF_COMMAND_TOPIC],
-            self._config[CONF_PAYLOAD_LOCK],
+            payload,
             self._config[CONF_QOS],
             self._config[CONF_RETAIN],
             self._config[CONF_ENCODING],
@@ -226,9 +254,10 @@ class MqttLock(MqttEntity, LockEntity):
 
         This method is a coroutine.
         """
+        payload = self._command_template(self._config[CONF_PAYLOAD_UNLOCK], kwargs)
         await self.async_publish(
             self._config[CONF_COMMAND_TOPIC],
-            self._config[CONF_PAYLOAD_UNLOCK],
+            payload,
             self._config[CONF_QOS],
             self._config[CONF_RETAIN],
             self._config[CONF_ENCODING],
@@ -243,9 +272,10 @@ class MqttLock(MqttEntity, LockEntity):
 
         This method is a coroutine.
         """
+        payload = self._command_template(self._config[CONF_PAYLOAD_OPEN], kwargs)
         await self.async_publish(
             self._config[CONF_COMMAND_TOPIC],
-            self._config[CONF_PAYLOAD_OPEN],
+            payload,
             self._config[CONF_QOS],
             self._config[CONF_RETAIN],
             self._config[CONF_ENCODING],

tests/components/mqtt/test_lock.py

@@ -18,6 +18,7 @@ from homeassistant.components.lock import (
 from homeassistant.components.mqtt.lock import MQTT_LOCK_ATTRIBUTES_BLOCKED
 from homeassistant.const import (
     ATTR_ASSUMED_STATE,
+    ATTR_CODE,
     ATTR_ENTITY_ID,
     ATTR_SUPPORTED_FEATURES,
     Platform,
@@ -298,6 +299,67 @@ async def test_sending_mqtt_commands_and_optimistic(
     assert state.attributes.get(ATTR_ASSUMED_STATE)
 
 
+async def test_sending_mqtt_commands_with_template(
+    hass, mqtt_mock_entry_with_yaml_config
+):
+    """Test sending commands with template."""
+    assert await async_setup_component(
+        hass,
+        mqtt.DOMAIN,
+        {
+            mqtt.DOMAIN: {
+                lock.DOMAIN: {
+                    "name": "test",
+                    "code_format": "^\\d{4}$",
+                    "command_topic": "command-topic",
+                    "command_template": '{ "{{ value }}": "{{ code }}" }',
+                    "payload_lock": "LOCK",
+                    "payload_unlock": "UNLOCK",
+                    "payload_open": "OPEN",
+                    "state_locked": "LOCKED",
+                    "state_unlocked": "UNLOCKED",
+                }
+            }
+        },
+    )
+    await hass.async_block_till_done()
+    mqtt_mock = await mqtt_mock_entry_with_yaml_config()
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+    assert state.attributes.get(ATTR_ASSUMED_STATE)
+
+    await hass.services.async_call(
+        lock.DOMAIN,
+        SERVICE_LOCK,
+        {ATTR_ENTITY_ID: "lock.test", ATTR_CODE: "1234"},
+        blocking=True,
+    )
+
+    mqtt_mock.async_publish.assert_called_once_with(
+        "command-topic", '{ "LOCK": "1234" }', 0, False
+    )
+    mqtt_mock.async_publish.reset_mock()
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_LOCKED
+    assert state.attributes.get(ATTR_ASSUMED_STATE)
+
+    await hass.services.async_call(
+        lock.DOMAIN,
+        SERVICE_UNLOCK,
+        {ATTR_ENTITY_ID: "lock.test", ATTR_CODE: "1234"},
+        blocking=True,
+    )
+
+    mqtt_mock.async_publish.assert_called_once_with(
+        "command-topic", '{ "UNLOCK": "1234" }', 0, False
+    )
+    mqtt_mock.async_publish.reset_mock()
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+    assert state.attributes.get(ATTR_ASSUMED_STATE)
+
+
 async def test_sending_mqtt_commands_and_explicit_optimistic(
     hass, mqtt_mock_entry_with_yaml_config
 ):