Use shorthand attributes for MQTT device tracker entity (#142671)

homeassistant/components/mqtt/device_tracker.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import voluptuous as vol
 
@@ -28,8 +28,8 @@ from homeassistant.helpers.typing import ConfigType, VolSchemaType
 
 from . import subscription
 from .config import MQTT_BASE_SCHEMA
-from .const import CONF_PAYLOAD_RESET, CONF_STATE_TOPIC
+from .const import CONF_JSON_ATTRS_TOPIC, CONF_PAYLOAD_RESET, CONF_STATE_TOPIC
-from .entity import CONF_JSON_ATTRS_TOPIC, MqttEntity, async_setup_entity_entry_helper
+from .entity import MqttEntity, async_setup_entity_entry_helper
 from .models import MqttValueTemplate, ReceiveMessage
 from .schemas import MQTT_ENTITY_COMMON_SCHEMA
 from .util import valid_subscribe_topic
@@ -111,6 +111,7 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
         self._value_template = MqttValueTemplate(
             config.get(CONF_VALUE_TEMPLATE), entity=self
         ).async_render_with_possible_json_value
+        self._attr_source_type = self._config[CONF_SOURCE_TYPE]
 
     @callback
     def _tracker_message_received(self, msg: ReceiveMessage) -> None:
@@ -124,72 +125,44 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
             )
             return
         if payload == self._config[CONF_PAYLOAD_HOME]:
-            self._location_name = STATE_HOME
+            self._attr_location_name = STATE_HOME
         elif payload == self._config[CONF_PAYLOAD_NOT_HOME]:
-            self._location_name = STATE_NOT_HOME
+            self._attr_location_name = STATE_NOT_HOME
         elif payload == self._config[CONF_PAYLOAD_RESET]:
-            self._location_name = None
+            self._attr_location_name = None
         else:
             if TYPE_CHECKING:
                 assert isinstance(msg.payload, str)
-            self._location_name = msg.payload
+            self._attr_location_name = msg.payload
 
     @callback
     def _prepare_subscribe_topics(self) -> None:
         """(Re)Subscribe to topics."""
         self.add_subscription(
-            CONF_STATE_TOPIC, self._tracker_message_received, {"_location_name"}
+            CONF_STATE_TOPIC, self._tracker_message_received, {"_attr_location_name"}
         )
 
-    @property
-    def force_update(self) -> bool:
-        """Do not force updates if the state is the same."""
-        return False
-
     async def _subscribe_topics(self) -> None:
         """(Re)Subscribe to topics."""
         subscription.async_subscribe_topics_internal(self.hass, self._sub_state)
 
-    @property
+    @callback
-    def latitude(self) -> float | None:
+    def _process_update_extra_state_attributes(
-        """Return latitude if provided in extra_state_attributes or None."""
+        self, extra_state_attributes: dict[str, Any]
+    ) -> None:
+        """Extract the location from the extra state attributes."""
+        self._attr_latitude = extra_state_attributes.get(ATTR_LATITUDE)
+        self._attr_longitude = extra_state_attributes.get(ATTR_LONGITUDE)
         if (
-            self.extra_state_attributes is not None
+            ATTR_LATITUDE in extra_state_attributes
-            and ATTR_LATITUDE in self.extra_state_attributes
+            or ATTR_LONGITUDE in extra_state_attributes
         ):
-            latitude: float = self.extra_state_attributes[ATTR_LATITUDE]
+            # Reset manual set location
-            return latitude
+            self._attr_location_name = None
-        return None
 
-    @property
+        self._attr_location_accuracy = extra_state_attributes.get(ATTR_GPS_ACCURACY, 0)
-    def location_accuracy(self) -> int:
+        self._attr_extra_state_attributes = {
-        """Return location accuracy if provided in extra_state_attributes or None."""
+            attribute: value
-        if (
+            for attribute, value in extra_state_attributes.items()
-            self.extra_state_attributes is not None
+            if attribute not in {ATTR_GPS_ACCURACY, ATTR_LATITUDE, ATTR_LONGITUDE}
-            and ATTR_GPS_ACCURACY in self.extra_state_attributes
+        }
-        ):
-            accuracy: int = self.extra_state_attributes[ATTR_GPS_ACCURACY]
-            return accuracy
-        return 0
-
-    @property
-    def longitude(self) -> float | None:
-        """Return longitude if provided in extra_state_attributes or None."""
-        if (
-            self.extra_state_attributes is not None
-            and ATTR_LONGITUDE in self.extra_state_attributes
-        ):
-            longitude: float = self.extra_state_attributes[ATTR_LONGITUDE]
-            return longitude
-        return None
-
-    @property
-    def location_name(self) -> str | None:
-        """Return a location name for the current location of the device."""
-        return self._location_name
-
-    @property
-    def source_type(self) -> SourceType:
-        """Return the source type, eg gps or router, of the device."""
-        source_type: SourceType = self._config[CONF_SOURCE_TYPE]
-        return source_type

homeassistant/components/mqtt/entity.py

@@ -402,6 +402,7 @@ class MqttAttributesMixin(Entity):
     _message_callback: Callable[
         [MessageCallbackType, set[str] | None, ReceiveMessage], None
     ]
+    _process_update_extra_state_attributes: Callable[[dict[str, Any]], None]
 
     def __init__(self, config: ConfigType) -> None:
         """Initialize the JSON attributes mixin."""
@@ -438,7 +439,13 @@ class MqttAttributesMixin(Entity):
                     "msg_callback": partial(
                         self._message_callback,
                         self._attributes_message_received,
-                        {"_attr_extra_state_attributes"},
+                        {
+                            "_attr_extra_state_attributes",
+                            "_attr_gps_accuracy",
+                            "_attr_latitude",
+                            "_attr_location_name",
+                            "_attr_longitude",
+                        },
                     ),
                     "entity_id": self.entity_id,
                     "qos": self._attributes_config.get(CONF_QOS),
@@ -477,7 +484,11 @@ class MqttAttributesMixin(Entity):
                     if k not in MQTT_ATTRIBUTES_BLOCKED
                     and k not in self._attributes_extra_blocked
                 }
-                self._attr_extra_state_attributes = filtered_dict
+                if hasattr(self, "_process_update_extra_state_attributes"):
+                    self._process_update_extra_state_attributes(filtered_dict)
+                else:
+                    self._attr_extra_state_attributes = filtered_dict
+
             else:
                 _LOGGER.warning("JSON result was not a dictionary")
 

tests/components/mqtt/test_device_tracker.py

@@ -454,12 +454,10 @@ async def test_setting_device_tracker_location_via_lat_lon_message(
 
     async_fire_mqtt_message(hass, "attributes-topic", '{"longitude": -117.22743}')
     state = hass.states.get("device_tracker.test")
-    assert state.attributes["longitude"] == -117.22743
     assert state.state == STATE_UNKNOWN
 
     async_fire_mqtt_message(hass, "attributes-topic", '{"latitude":32.87336}')
     state = hass.states.get("device_tracker.test")
-    assert state.attributes["latitude"] == 32.87336
     assert state.state == STATE_UNKNOWN