Add type hints for MQTT tag (#81495)

* Improve typing tag

* Additional type hints for tag platform

* Follow up comment tag

* Folow up comments

* Revert comment removal
pull/81771/head
Jan Bouwhuis 2022-11-08 09:24:16 +01:00 committed by GitHub
parent 88faf33cb8
commit 8c0a7b9d7f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 6 deletions

View File

@ -1,6 +1,7 @@
"""Provides tag scanning for MQTT."""
from __future__ import annotations
from collections.abc import Callable
import functools
import voluptuous as vol
@ -9,11 +10,12 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE, CONF_PLATFORM, CONF_VALUE_TEMPLATE
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import subscription
from .config import MQTT_BASE_SCHEMA
from .const import ATTR_DISCOVERY_HASH, CONF_QOS, CONF_TOPIC
from .discovery import MQTTDiscoveryPayload
from .mixins import (
MQTT_ENTITY_DEVICE_INFO_SCHEMA,
MqttDiscoveryDeviceUpdate,
@ -21,7 +23,7 @@ from .mixins import (
send_discovery_done,
update_device,
)
from .models import MqttValueTemplate, ReceiveMessage
from .models import MqttValueTemplate, ReceiveMessage, ReceivePayloadType
from .subscription import EntitySubscription
from .util import get_mqtt_data, valid_subscribe_topic
@ -87,12 +89,14 @@ def async_has_tags(hass: HomeAssistant, device_id: str) -> bool:
class MQTTTagScanner(MqttDiscoveryDeviceUpdate):
"""MQTT Tag scanner."""
_value_template: Callable[[ReceivePayloadType, str], ReceivePayloadType]
def __init__(
self,
hass: HomeAssistant,
config: ConfigType,
device_id: str | None,
discovery_data: dict,
discovery_data: DiscoveryInfoType,
config_entry: ConfigEntry,
) -> None:
"""Initialize."""
@ -111,10 +115,10 @@ class MQTTTagScanner(MqttDiscoveryDeviceUpdate):
self, hass, discovery_data, device_id, config_entry, LOG_NAME
)
async def async_update(self, discovery_data: dict) -> None:
async def async_update(self, discovery_data: MQTTDiscoveryPayload) -> None:
"""Handle MQTT tag discovery updates."""
# Update tag scanner
config = PLATFORM_SCHEMA(discovery_data)
config: DiscoveryInfoType = PLATFORM_SCHEMA(discovery_data)
self._config = config
self._value_template = MqttValueTemplate(
config.get(CONF_VALUE_TEMPLATE),
@ -127,7 +131,7 @@ class MQTTTagScanner(MqttDiscoveryDeviceUpdate):
"""Subscribe to MQTT topics."""
async def tag_scanned(msg: ReceiveMessage) -> None:
tag_id = self._value_template(msg.payload, "").strip()
tag_id = str(self._value_template(msg.payload, "")).strip()
if not tag_id: # No output from template, ignore
return