Add MQTT encoding parameter for all subscribed topics (#62263)

* Add encoding parameter for all subscribable topics

* test setup encoding incoming payload

* remove support for device_tracker and tag+tests

homeassistant/components/mqtt/alarm_control_panel.py

@@ -206,6 +206,7 @@ class MqttAlarm(MqttEntity, alarm.AlarmControlPanelEntity):
                     "topic": self._config[CONF_STATE_TOPIC],
                     "msg_callback": message_received,
                     "qos": self._config[CONF_QOS],
+                    "encoding": self._config[CONF_ENCODING] or None,
                 }
             },
         )

homeassistant/components/mqtt/climate.py

@@ -412,6 +412,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
                     "topic": self._topic[topic],
                     "msg_callback": msg_callback,
                     "qos": qos,
+                    "encoding": self._config[CONF_ENCODING] or None,
                 }
 
         def render_template(msg, template_name):

homeassistant/components/mqtt/cover.py

@@ -440,6 +440,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "topic": self._config.get(CONF_GET_POSITION_TOPIC),
                 "msg_callback": position_message_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         if self._config.get(CONF_STATE_TOPIC):
@@ -447,6 +448,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "topic": self._config.get(CONF_STATE_TOPIC),
                 "msg_callback": state_message_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         if self._config.get(CONF_TILT_STATUS_TOPIC) is not None:
@@ -455,6 +457,7 @@ class MqttCover(MqttEntity, CoverEntity):
                 "topic": self._config.get(CONF_TILT_STATUS_TOPIC),
                 "msg_callback": tilt_message_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         self._sub_state = await subscription.async_subscribe_topics(

homeassistant/components/mqtt/fan.py

@@ -384,6 +384,7 @@ class MqttFan(MqttEntity, FanEntity):
                 "topic": self._topic[CONF_STATE_TOPIC],
                 "msg_callback": state_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         @callback
@@ -428,6 +429,7 @@ class MqttFan(MqttEntity, FanEntity):
                 "topic": self._topic[CONF_PERCENTAGE_STATE_TOPIC],
                 "msg_callback": percentage_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
             self._percentage = None
 
@@ -460,6 +462,7 @@ class MqttFan(MqttEntity, FanEntity):
                 "topic": self._topic[CONF_PRESET_MODE_STATE_TOPIC],
                 "msg_callback": preset_mode_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
             self._preset_mode = None
 
@@ -482,6 +485,7 @@ class MqttFan(MqttEntity, FanEntity):
                 "topic": self._topic[CONF_OSCILLATION_STATE_TOPIC],
                 "msg_callback": oscillation_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
             self._oscillation = False
 

homeassistant/components/mqtt/humidifier.py

@@ -290,6 +290,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
                 "topic": self._topic[CONF_STATE_TOPIC],
                 "msg_callback": state_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         @callback
@@ -335,6 +336,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
                 "topic": self._topic[CONF_TARGET_HUMIDITY_STATE_TOPIC],
                 "msg_callback": target_humidity_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
             self._target_humidity = None
 
@@ -367,6 +369,7 @@ class MqttHumidifier(MqttEntity, HumidifierEntity):
                 "topic": self._topic[CONF_MODE_STATE_TOPIC],
                 "msg_callback": mode_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
             self._mode = None
 

homeassistant/components/mqtt/light/schema_basic.py

@@ -433,6 +433,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
                     "topic": self._topic[topic],
                     "msg_callback": msg_callback,
                     "qos": self._config[CONF_QOS],
+                    "encoding": self._config[CONF_ENCODING] or None,
                 }
 
         def restore_state(attribute, condition_attribute=None):
@@ -465,6 +466,7 @@ class MqttLight(MqttEntity, LightEntity, RestoreEntity):
                 "topic": self._topic[CONF_STATE_TOPIC],
                 "msg_callback": state_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
         elif self._optimistic and last_state:
             self._state = last_state.state == STATE_ON

homeassistant/components/mqtt/light/schema_json.py

@@ -378,6 +378,7 @@ class MqttLightJson(MqttEntity, LightEntity, RestoreEntity):
                         "topic": self._topic[CONF_STATE_TOPIC],
                         "msg_callback": state_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/light/schema_template.py

@@ -254,6 +254,7 @@ class MqttLightTemplate(MqttEntity, LightEntity, RestoreEntity):
                         "topic": self._topics[CONF_STATE_TOPIC],
                         "msg_callback": state_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/lock.py

@@ -150,6 +150,7 @@ class MqttLock(MqttEntity, LockEntity):
                         "topic": self._config.get(CONF_STATE_TOPIC),
                         "msg_callback": message_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/mixins.py

@@ -288,6 +288,7 @@ class MqttAttributes(Entity):
                     "topic": self._attributes_config.get(CONF_JSON_ATTRS_TOPIC),
                     "msg_callback": attributes_message_received,
                     "qos": self._attributes_config.get(CONF_QOS),
+                    "encoding": self._attributes_config[CONF_ENCODING] or None,
                 }
             },
         )

homeassistant/components/mqtt/number.py

@@ -209,6 +209,7 @@ class MqttNumber(MqttEntity, NumberEntity, RestoreEntity):
                         "topic": self._config.get(CONF_STATE_TOPIC),
                         "msg_callback": message_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/select.py

@@ -165,6 +165,7 @@ class MqttSelect(MqttEntity, SelectEntity, RestoreEntity):
                         "topic": self._config.get(CONF_STATE_TOPIC),
                         "msg_callback": message_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/sensor.py

@@ -262,6 +262,7 @@ class MqttSensor(MqttEntity, SensorEntity):
                 "topic": self._config[CONF_LAST_RESET_TOPIC],
                 "msg_callback": last_reset_message_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
 
         self._sub_state = await subscription.async_subscribe_topics(

homeassistant/components/mqtt/switch.py

@@ -159,6 +159,7 @@ class MqttSwitch(MqttEntity, SwitchEntity, RestoreEntity):
                         "topic": self._config.get(CONF_STATE_TOPIC),
                         "msg_callback": state_message_received,
                         "qos": self._config[CONF_QOS],
+                        "encoding": self._config[CONF_ENCODING] or None,
                     }
                 },
             )

homeassistant/components/mqtt/vacuum/schema_legacy.py

@@ -199,7 +199,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
         self._fan_speed_list = config[CONF_FAN_SPEED_LIST]
         self._qos = config[CONF_QOS]
         self._retain = config[CONF_RETAIN]
-        self._encoding = config[CONF_ENCODING]
+        self._encoding = config[CONF_ENCODING] or None
 
         self._command_topic = config.get(CONF_COMMAND_TOPIC)
         self._set_fan_speed_topic = config.get(CONF_SET_FAN_SPEED_TOPIC)
@@ -333,6 +333,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
                     "topic": topic,
                     "msg_callback": message_received,
                     "qos": self._qos,
+                    "encoding": self._encoding,
                 }
                 for i, topic in enumerate(topics_list)
             },

homeassistant/components/mqtt/vacuum/schema_state.py

@@ -214,6 +214,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
                 "topic": self._config.get(CONF_STATE_TOPIC),
                 "msg_callback": state_message_received,
                 "qos": self._config[CONF_QOS],
+                "encoding": self._config[CONF_ENCODING] or None,
             }
         self._sub_state = await subscription.async_subscribe_topics(
             self.hass, self._sub_state, topics

tests/components/mqtt/test_alarm_control_panel.py

@@ -43,6 +43,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -705,6 +706,26 @@ async def test_discovery_broken(hass, mqtt_mock, caplog):
     )
 
 
+@pytest.mark.parametrize(
+    "topic,value",
+    [
+        ("state_topic", "armed_home"),
+        ("state_topic", "disarmed"),
+    ],
+)
+async def test_encoding_subscribable_topics(hass, mqtt_mock, caplog, topic, value):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        alarm_control_panel.DOMAIN,
+        DEFAULT_CONFIG[alarm_control_panel.DOMAIN],
+        topic,
+        value,
+    )
+
+
 async def test_entity_device_info_with_connection(hass, mqtt_mock):
     """Test MQTT alarm control panel device registry integration."""
     await help_test_entity_device_info_with_connection(

tests/components/mqtt/test_binary_sensor.py

@@ -28,6 +28,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -759,6 +760,36 @@ async def test_discovery_update_binary_sensor_template(hass, mqtt_mock, caplog):
     )
 
 
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("json_attributes_topic", '{ "id": 123 }', "id", 123),
+        (
+            "json_attributes_topic",
+            '{ "id": 123, "temperature": 34.0 }',
+            "temperature",
+            34.0,
+        ),
+        ("state_topic", "ON", None, "on"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        binary_sensor.DOMAIN,
+        DEFAULT_CONFIG[binary_sensor.DOMAIN],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )
+
+
 async def test_discovery_update_unchanged_binary_sensor(hass, mqtt_mock, caplog):
     """Test update of discovered binary_sensor."""
     config1 = copy.deepcopy(DEFAULT_CONFIG[binary_sensor.DOMAIN])

tests/components/mqtt/test_climate.py

@@ -9,7 +9,14 @@ import voluptuous as vol
 from homeassistant.components import climate
 from homeassistant.components.climate import DEFAULT_MAX_TEMP, DEFAULT_MIN_TEMP
 from homeassistant.components.climate.const import (
+    ATTR_AUX_HEAT,
+    ATTR_CURRENT_TEMPERATURE,
+    ATTR_FAN_MODE,
     ATTR_HVAC_ACTION,
+    ATTR_PRESET_MODE,
+    ATTR_SWING_MODE,
+    ATTR_TARGET_TEMP_HIGH,
+    ATTR_TARGET_TEMP_LOW,
     CURRENT_HVAC_ACTIONS,
     DOMAIN as CLIMATE_DOMAIN,
     HVAC_MODE_AUTO,
@@ -27,7 +34,7 @@ from homeassistant.components.climate.const import (
     SUPPORT_TARGET_TEMPERATURE_RANGE,
 )
 from homeassistant.components.mqtt.climate import MQTT_CLIMATE_ATTRIBUTES_BLOCKED
-from homeassistant.const import STATE_OFF
+from homeassistant.const import ATTR_TEMPERATURE, STATE_OFF
 from homeassistant.setup import async_setup_component
 
 from .test_common import (
@@ -40,6 +47,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -995,6 +1003,43 @@ async def test_unique_id(hass, mqtt_mock):
     await help_test_unique_id(hass, mqtt_mock, CLIMATE_DOMAIN, config)
 
 
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("action_topic", "heating", ATTR_HVAC_ACTION, "heating"),
+        ("action_topic", "cooling", ATTR_HVAC_ACTION, "cooling"),
+        ("aux_state_topic", "ON", ATTR_AUX_HEAT, "on"),
+        ("away_mode_state_topic", "ON", ATTR_PRESET_MODE, "away"),
+        ("current_temperature_topic", "22.1", ATTR_CURRENT_TEMPERATURE, 22.1),
+        ("fan_mode_state_topic", "low", ATTR_FAN_MODE, "low"),
+        ("hold_state_topic", "mode1", ATTR_PRESET_MODE, "mode1"),
+        ("mode_state_topic", "cool", None, None),
+        ("mode_state_topic", "fan_only", None, None),
+        ("swing_mode_state_topic", "on", ATTR_SWING_MODE, "on"),
+        ("temperature_low_state_topic", "19.1", ATTR_TARGET_TEMP_LOW, 19.1),
+        ("temperature_high_state_topic", "22.9", ATTR_TARGET_TEMP_HIGH, 22.9),
+        ("temperature_state_topic", "19.9", ATTR_TEMPERATURE, 19.9),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[CLIMATE_DOMAIN])
+    config["hold_modes"] = ["mode1", "mode2"]
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        CLIMATE_DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )
+
+
 async def test_discovery_removal_climate(hass, mqtt_mock, caplog):
     """Test removal of discovered climate."""
     data = json.dumps(DEFAULT_CONFIG[CLIMATE_DOMAIN])

tests/components/mqtt/test_common.py

@@ -765,6 +765,138 @@ async def help_test_discovery_broken(hass, mqtt_mock, caplog, domain, data1, dat
     assert state is None
 
 
+async def help_test_encoding_subscribable_topics(
+    hass,
+    mqtt_mock,
+    caplog,
+    domain,
+    config,
+    topic,
+    value,
+    attribute=None,
+    attribute_value=None,
+    init_payload=None,
+    skip_raw_test=False,
+):
+    """Test handling of incoming encoded payload."""
+
+    async def _test_encoding(
+        hass,
+        entity_id,
+        topic,
+        encoded_value,
+        attribute,
+        init_payload_topic,
+        init_payload_value,
+    ):
+        state = hass.states.get(entity_id)
+
+        if init_payload_value:
+            # Sometimes a device needs to have an initialization pay load, e.g. to switch the device on.
+            async_fire_mqtt_message(hass, init_payload_topic, init_payload_value)
+            await hass.async_block_till_done()
+
+        state = hass.states.get(entity_id)
+
+        async_fire_mqtt_message(hass, topic, encoded_value)
+        await hass.async_block_till_done()
+
+        state = hass.states.get(entity_id)
+
+        if attribute:
+            return state.attributes.get(attribute)
+
+        return state.state if state else None
+
+    init_payload_value_utf8 = None
+    init_payload_value_utf16 = None
+    # setup test1 default encoding
+    config1 = copy.deepcopy(config)
+    if domain == "device_tracker":
+        config1["unique_id"] = "test1"
+    else:
+        config1["name"] = "test1"
+    config1[topic] = "topic/test1"
+    # setup test2 alternate encoding
+    config2 = copy.deepcopy(config)
+    if domain == "device_tracker":
+        config2["unique_id"] = "test2"
+    else:
+        config2["name"] = "test2"
+    config2["encoding"] = "utf-16"
+    config2[topic] = "topic/test2"
+    # setup test3 raw encoding
+    config3 = copy.deepcopy(config)
+    if domain == "device_tracker":
+        config3["unique_id"] = "test3"
+    else:
+        config3["name"] = "test3"
+    config3["encoding"] = ""
+    config3[topic] = "topic/test3"
+
+    if init_payload:
+        config1[init_payload[0]] = "topic/init_payload1"
+        config2[init_payload[0]] = "topic/init_payload2"
+        config3[init_payload[0]] = "topic/init_payload3"
+        init_payload_value_utf8 = init_payload[1].encode("utf-8")
+        init_payload_value_utf16 = init_payload[1].encode("utf-16")
+
+    await hass.async_block_till_done()
+
+    assert await async_setup_component(
+        hass, domain, {domain: [config1, config2, config3]}
+    )
+    await hass.async_block_till_done()
+
+    expected_result = attribute_value or value
+
+    # test1 default encoding
+    assert (
+        await _test_encoding(
+            hass,
+            f"{domain}.test1",
+            "topic/test1",
+            value.encode("utf-8"),
+            attribute,
+            "topic/init_payload1",
+            init_payload_value_utf8,
+        )
+        == expected_result
+    )
+
+    # test2 alternate encoding
+    assert (
+        await _test_encoding(
+            hass,
+            f"{domain}.test2",
+            "topic/test2",
+            value.encode("utf-16"),
+            attribute,
+            "topic/init_payload2",
+            init_payload_value_utf16,
+        )
+        == expected_result
+    )
+
+    # test3 raw encoded input
+    if skip_raw_test:
+        return
+
+    try:
+        result = await _test_encoding(
+            hass,
+            f"{domain}.test3",
+            "topic/test3",
+            value.encode("utf-16"),
+            attribute,
+            "topic/init_payload3",
+            init_payload_value_utf16,
+        )
+        assert result != expected_result
+    except (AttributeError, TypeError, ValueError):
+        pass
+
+
 async def help_test_entity_device_info_with_identifier(hass, mqtt_mock, domain, config):
     """Test device registry integration.
 

tests/components/mqtt/test_cover.py

@@ -54,6 +54,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -3165,3 +3166,29 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = cover.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "open", None, None),
+        ("state_topic", "closing", None, None),
+        ("position_topic", "40.0", "current_position", 40.0),
+        ("tilt_status_topic", "60.0", "current_tilt_position", 60.0),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        cover.DOMAIN,
+        DEFAULT_CONFIG[cover.DOMAIN],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )

tests/components/mqtt/test_fan.py

@@ -6,8 +6,22 @@ import pytest
 from voluptuous.error import MultipleInvalid
 
 from homeassistant.components import fan
-from homeassistant.components.fan import NotValidPresetModeError
+from homeassistant.components.fan import (
-from homeassistant.components.mqtt.fan import MQTT_FAN_ATTRIBUTES_BLOCKED
+    ATTR_OSCILLATING,
+    ATTR_PERCENTAGE,
+    ATTR_PRESET_MODE,
+    ATTR_PRESET_MODES,
+    NotValidPresetModeError,
+)
+from homeassistant.components.mqtt.fan import (
+    CONF_OSCILLATION_COMMAND_TOPIC,
+    CONF_OSCILLATION_STATE_TOPIC,
+    CONF_PERCENTAGE_COMMAND_TOPIC,
+    CONF_PERCENTAGE_STATE_TOPIC,
+    CONF_PRESET_MODE_COMMAND_TOPIC,
+    CONF_PRESET_MODE_STATE_TOPIC,
+    MQTT_FAN_ATTRIBUTES_BLOCKED,
+)
 from homeassistant.const import (
     ATTR_ASSUMED_STATE,
     ATTR_SUPPORTED_FEATURES,
@@ -26,6 +40,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -1270,6 +1285,42 @@ async def test_sending_mqtt_commands_and_explicit_optimistic(hass, mqtt_mock, ca
     assert state.attributes.get(ATTR_ASSUMED_STATE)
 
 
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "ON", None, "on"),
+        (CONF_PRESET_MODE_STATE_TOPIC, "auto", ATTR_PRESET_MODE, "auto"),
+        (CONF_PERCENTAGE_STATE_TOPIC, "60", ATTR_PERCENTAGE, 60),
+        (
+            CONF_OSCILLATION_STATE_TOPIC,
+            "oscillate_on",
+            ATTR_OSCILLATING,
+            True,
+        ),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[fan.DOMAIN])
+    config[ATTR_PRESET_MODES] = ["eco", "auto"]
+    config[CONF_PRESET_MODE_COMMAND_TOPIC] = "fan/some_preset_mode_command_topic"
+    config[CONF_PERCENTAGE_COMMAND_TOPIC] = "fan/some_percentage_command_topic"
+    config[CONF_OSCILLATION_COMMAND_TOPIC] = "fan/some_oscillation_command_topic"
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        fan.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )
+
+
 async def test_attributes(hass, mqtt_mock, caplog):
     """Test attributes."""
     assert await async_setup_component(

tests/components/mqtt/test_humidifier.py

@@ -13,7 +13,12 @@ from homeassistant.components.humidifier import (
     SERVICE_SET_HUMIDITY,
     SERVICE_SET_MODE,
 )
-from homeassistant.components.mqtt.humidifier import MQTT_HUMIDIFIER_ATTRIBUTES_BLOCKED
+from homeassistant.components.mqtt.humidifier import (
+    CONF_MODE_COMMAND_TOPIC,
+    CONF_MODE_STATE_TOPIC,
+    CONF_TARGET_HUMIDITY_STATE_TOPIC,
+    MQTT_HUMIDIFIER_ATTRIBUTES_BLOCKED,
+)
 from homeassistant.const import (
     ATTR_ASSUMED_STATE,
     ATTR_ENTITY_ID,
@@ -36,6 +41,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -675,6 +681,34 @@ async def test_sending_mqtt_commands_and_explicit_optimistic(hass, mqtt_mock, ca
     assert state.attributes.get(ATTR_ASSUMED_STATE)
 
 
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "ON", None, "on"),
+        (CONF_MODE_STATE_TOPIC, "auto", ATTR_MODE, "auto"),
+        (CONF_TARGET_HUMIDITY_STATE_TOPIC, "45", ATTR_HUMIDITY, 45),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[humidifier.DOMAIN])
+    config["modes"] = ["eco", "auto"]
+    config[CONF_MODE_COMMAND_TOPIC] = "humidifier/some_mode_command_topic"
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        humidifier.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )
+
+
 async def test_attributes(hass, mqtt_mock, caplog):
     """Test attributes."""
     assert await async_setup_component(

tests/components/mqtt/test_legacy_vacuum.py

@@ -11,6 +11,13 @@ from homeassistant.components.mqtt.vacuum import schema_legacy as mqttvacuum
 from homeassistant.components.mqtt.vacuum.schema import services_to_strings
 from homeassistant.components.mqtt.vacuum.schema_legacy import (
     ALL_SERVICES,
+    CONF_BATTERY_LEVEL_TOPIC,
+    CONF_CHARGING_TOPIC,
+    CONF_CLEANING_TOPIC,
+    CONF_DOCKED_TOPIC,
+    CONF_ERROR_TOPIC,
+    CONF_FAN_SPEED_TOPIC,
+    CONF_SUPPORTED_FEATURES,
     MQTT_LEGACY_VACUUM_ATTRIBUTES_BLOCKED,
     SERVICE_TO_STRING,
 )
@@ -34,6 +41,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -830,3 +838,57 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = vacuum.DOMAIN
     config = DEFAULT_CONFIG
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        (CONF_BATTERY_LEVEL_TOPIC, '{ "battery_level": 60 }', "battery_level", 60),
+        (CONF_CHARGING_TOPIC, '{ "charging": true }', "status", "Stopped"),
+        (CONF_CLEANING_TOPIC, '{ "cleaning": true }', "status", "Cleaning"),
+        (CONF_DOCKED_TOPIC, '{ "docked": true }', "status", "Docked"),
+        (
+            CONF_ERROR_TOPIC,
+            '{ "error": "some error" }',
+            "status",
+            "Error: some error",
+        ),
+        (
+            CONF_FAN_SPEED_TOPIC,
+            '{ "fan_speed": "medium" }',
+            "fan_speed",
+            "medium",
+        ),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    config = deepcopy(DEFAULT_CONFIG)
+    config[CONF_SUPPORTED_FEATURES] = [
+        "turn_on",
+        "turn_off",
+        "pause",
+        "stop",
+        "return_home",
+        "battery",
+        "status",
+        "locate",
+        "clean_spot",
+        "fan_speed",
+        "send_command",
+    ]
+
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        vacuum.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+        skip_raw_test=True,
+    )

tests/components/mqtt/test_light.py

@@ -160,6 +160,16 @@ import pytest
 
 from homeassistant.components import light
 from homeassistant.components.mqtt.light.schema_basic import (
+    CONF_BRIGHTNESS_COMMAND_TOPIC,
+    CONF_COLOR_TEMP_COMMAND_TOPIC,
+    CONF_EFFECT_COMMAND_TOPIC,
+    CONF_EFFECT_LIST,
+    CONF_HS_COMMAND_TOPIC,
+    CONF_RGB_COMMAND_TOPIC,
+    CONF_RGBW_COMMAND_TOPIC,
+    CONF_RGBWW_COMMAND_TOPIC,
+    CONF_WHITE_VALUE_COMMAND_TOPIC,
+    CONF_XY_COMMAND_TOPIC,
     MQTT_LIGHT_ATTRIBUTES_BLOCKED,
 )
 from homeassistant.const import (
@@ -181,6 +191,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -3500,3 +3511,66 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = light.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value,init_payload",
+    [
+        ("state_topic", "ON", None, "on", None),
+        ("brightness_state_topic", "60", "brightness", 60, ("state_topic", "ON")),
+        (
+            "color_mode_state_topic",
+            "200",
+            "color_mode",
+            "200",
+            ("state_topic", "ON"),
+        ),
+        ("color_temp_state_topic", "200", "color_temp", 200, ("state_topic", "ON")),
+        ("effect_state_topic", "random", "effect", "random", ("state_topic", "ON")),
+        ("hs_state_topic", "200,50", "hs_color", (200, 50), ("state_topic", "ON")),
+        (
+            "xy_state_topic",
+            "128,128",
+            "xy_color",
+            (128, 128),
+            ("state_topic", "ON"),
+        ),
+        (
+            "rgb_state_topic",
+            "255,0,240",
+            "rgb_color",
+            (255, 0, 240),
+            ("state_topic", "ON"),
+        ),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value, init_payload
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[light.DOMAIN])
+    config[CONF_EFFECT_COMMAND_TOPIC] = "light/CONF_EFFECT_COMMAND_TOPIC"
+    config[CONF_RGB_COMMAND_TOPIC] = "light/CONF_RGB_COMMAND_TOPIC"
+    config[CONF_BRIGHTNESS_COMMAND_TOPIC] = "light/CONF_BRIGHTNESS_COMMAND_TOPIC"
+    config[CONF_COLOR_TEMP_COMMAND_TOPIC] = "light/CONF_COLOR_TEMP_COMMAND_TOPIC"
+    config[CONF_HS_COMMAND_TOPIC] = "light/CONF_HS_COMMAND_TOPIC"
+    config[CONF_RGB_COMMAND_TOPIC] = "light/CONF_RGB_COMMAND_TOPIC"
+    config[CONF_RGBW_COMMAND_TOPIC] = "light/CONF_RGBW_COMMAND_TOPIC"
+    config[CONF_RGBWW_COMMAND_TOPIC] = "light/CONF_RGBWW_COMMAND_TOPIC"
+    config[CONF_XY_COMMAND_TOPIC] = "light/CONF_XY_COMMAND_TOPIC"
+    config[CONF_EFFECT_LIST] = ["colorloop", "random"]
+    if attribute and attribute == "brightness":
+        config[CONF_WHITE_VALUE_COMMAND_TOPIC] = "light/CONF_WHITE_VALUE_COMMAND_TOPIC"
+
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        light.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+        init_payload,
+    )

tests/components/mqtt/test_light_json.py

@@ -116,6 +116,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -1971,3 +1972,44 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = light.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value,init_payload",
+    [
+        (
+            "state_topic",
+            '{ "state": "ON", "brightness": 200 }',
+            "brightness",
+            200,
+            None,
+        ),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value, init_payload
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[light.DOMAIN])
+    config["color_mode"] = True
+    config["supported_color_modes"] = [
+        "color_temp",
+        "hs",
+        "xy",
+        "rgb",
+        "rgbw",
+        "rgbww",
+    ]
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        light.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+        init_payload,
+        skip_raw_test=True,
+    )

tests/components/mqtt/test_light_template.py

@@ -54,6 +54,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -1154,3 +1155,29 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = light.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value,init_payload",
+    [
+        ("state_topic", "on", None, "on", None),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value, init_payload
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG[light.DOMAIN])
+    config["state_template"] = "{{ value }}"
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        light.DOMAIN,
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+        init_payload,
+    )

tests/components/mqtt/test_lock.py

@@ -30,6 +30,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -638,3 +639,26 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = LOCK_DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "LOCKED", None, "locked"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        LOCK_DOMAIN,
+        DEFAULT_CONFIG[LOCK_DOMAIN],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )

tests/components/mqtt/test_number.py

@@ -36,6 +36,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -689,3 +690,27 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = number.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "10", None, "10"),
+        ("state_topic", "60", None, "60"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        "number",
+        DEFAULT_CONFIG["number"],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )

tests/components/mqtt/test_select.py

@@ -1,4 +1,5 @@
 """The tests for mqtt select component."""
+import copy
 import json
 from unittest.mock import patch
 
@@ -26,6 +27,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -567,3 +569,29 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = select.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "milk", None, "milk"),
+        ("state_topic", "beer", None, "beer"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    config = copy.deepcopy(DEFAULT_CONFIG["select"])
+    config["options"] = ["milk", "beer"]
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        "select",
+        config,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )

tests/components/mqtt/test_sensor.py

@@ -29,6 +29,7 @@ from .test_common import (
     help_test_discovery_update_attr,
     help_test_discovery_update_availability,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_category,
     help_test_entity_debug_info,
     help_test_entity_debug_info_max_messages,
@@ -927,3 +928,27 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = sensor.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "2.21", None, "2.21"),
+        ("state_topic", "beer", None, "beer"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        sensor.DOMAIN,
+        DEFAULT_CONFIG[sensor.DOMAIN],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )

tests/components/mqtt/test_state_vacuum.py

@@ -44,6 +44,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -591,3 +592,38 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = vacuum.DOMAIN
     config = DEFAULT_CONFIG
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        (
+            "state_topic",
+            '{"battery_level": 61, "state": "docked", "fan_speed": "off"}',
+            None,
+            "docked",
+        ),
+        (
+            "state_topic",
+            '{"battery_level": 61, "state": "cleaning", "fan_speed": "medium"}',
+            None,
+            "cleaning",
+        ),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        vacuum.DOMAIN,
+        DEFAULT_CONFIG,
+        topic,
+        value,
+        attribute,
+        attribute_value,
+        skip_raw_test=True,
+    )

tests/components/mqtt/test_switch.py

@@ -25,6 +25,7 @@ from .test_common import (
     help_test_discovery_update,
     help_test_discovery_update_attr,
     help_test_discovery_update_unchanged,
+    help_test_encoding_subscribable_topics,
     help_test_entity_debug_info_message,
     help_test_entity_device_info_remove,
     help_test_entity_device_info_update,
@@ -523,3 +524,26 @@ async def test_reloadable(hass, mqtt_mock, caplog, tmp_path):
     domain = switch.DOMAIN
     config = DEFAULT_CONFIG[domain]
     await help_test_reloadable(hass, mqtt_mock, caplog, tmp_path, domain, config)
+
+
+@pytest.mark.parametrize(
+    "topic,value,attribute,attribute_value",
+    [
+        ("state_topic", "ON", None, "on"),
+    ],
+)
+async def test_encoding_subscribable_topics(
+    hass, mqtt_mock, caplog, topic, value, attribute, attribute_value
+):
+    """Test handling of incoming encoded payload."""
+    await help_test_encoding_subscribable_topics(
+        hass,
+        mqtt_mock,
+        caplog,
+        switch.DOMAIN,
+        DEFAULT_CONFIG[switch.DOMAIN],
+        topic,
+        value,
+        attribute,
+        attribute_value,
+    )