jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-15 17:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve MQTT type hints for dict get assignments (#83634)

Browse Source 
* Improve type hints for dict get assignments

* Move _expire_after as class attr

* Remove extra check
This commit is contained in:
Jan Bouwhuis 2022-12-09 16:22:36 +01:00 committed by GitHub
parent a1758a71d2
commit e15388a2a0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 22 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
homeassistant/components/mqtt/__init__.py
View File

@ -375,10 +375,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_publish_service(call: ServiceCall) -> None:
"""Handle MQTT publish service calls."""
msg_topic = call.data.get(ATTR_TOPIC)
msg_topic_template = call.data.get(ATTR_TOPIC_TEMPLATE)
payload = call.data.get(ATTR_PAYLOAD)
payload_template = call.data.get(ATTR_PAYLOAD_TEMPLATE)
msg_topic: str | None = call.data.get(ATTR_TOPIC)
msg_topic_template: str | None = call.data.get(ATTR_TOPIC_TEMPLATE)
payload: PublishPayloadType = call.data.get(ATTR_PAYLOAD)
payload_template: str | None = call.data.get(ATTR_PAYLOAD_TEMPLATE)
qos: int = call.data[ATTR_QOS]
retain: bool = call.data[ATTR_RETAIN]
if msg_topic_template is not None:

21
homeassistant/components/mqtt/binary_sensor.py
View File

@ -106,6 +106,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
_entity_id_format = binary_sensor.ENTITY_ID_FORMAT
_expired: bool | None
_expire_after: int | None
def __init__(
self,
@ -123,8 +124,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
async def mqtt_async_added_to_hass(self) -> None:
"""Restore state for entities with expire_after set."""
if (
(expire_after := self._config.get(CONF_EXPIRE_AFTER)) is not None
and expire_after > 0
self._expire_after
and (last_state := await self.async_get_last_state()) is not None
and last_state.state not in [STATE_UNKNOWN, STATE_UNAVAILABLE]
# We might have set up a trigger already after subscribing from
@ -132,7 +132,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
and not self._expiration_trigger
):
expiration_at: datetime = last_state.last_changed + timedelta(
seconds=expire_after
seconds=self._expire_after
)
if expiration_at < (time_now := dt_util.utcnow()):
# Skip reactivating the binary_sensor
@ -167,8 +167,8 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
def _setup_from_config(self, config: ConfigType) -> None:
"""(Re)Setup the entity."""
expire_after: int | None = config.get(CONF_EXPIRE_AFTER)
if expire_after is not None and expire_after > 0:
self._expire_after = config.get(CONF_EXPIRE_AFTER)
if self._expire_after:
self._expired = True
else:
self._expired = None
@ -195,9 +195,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
def state_message_received(msg: ReceiveMessage) -> None:
"""Handle a new received MQTT state message."""
# auto-expire enabled?
expire_after: int | None = self._config.get(CONF_EXPIRE_AFTER)
if expire_after is not None and expire_after > 0:
if self._expire_after:
# When expire_after is set, and we receive a message, assume device is
# not expired since it has to be to receive the message
@ -208,7 +206,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
self._expiration_trigger()
# Set new trigger
expiration_at = dt_util.utcnow() + timedelta(seconds=expire_after)
expiration_at = dt_util.utcnow() + timedelta(seconds=self._expire_after)
self._expiration_trigger = async_track_point_in_utc_time(
self.hass, self._value_is_expired, expiration_at
@ -248,7 +246,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
self._delay_listener()
self._delay_listener = None
off_delay = self._config.get(CONF_OFF_DELAY)
off_delay: int | None = self._config.get(CONF_OFF_DELAY)
if self._attr_is_on and off_delay is not None:
self._delay_listener = evt.async_call_later(
self.hass, off_delay, off_delay_listener
@ -284,8 +282,7 @@ class MqttBinarySensor(MqttEntity, BinarySensorEntity, RestoreEntity):
@property
def available(self) -> bool:
"""Return true if the device is available and value has not expired."""
expire_after: int | None = self._config.get(CONF_EXPIRE_AFTER)
# mypy doesn't know about fget: https://github.com/python/mypy/issues/6185
return MqttAvailability.available.fget(self) and ( # type: ignore[attr-defined]
expire_after is None or not self._expired
self._expire_after is None or not self._expired
)

8
homeassistant/components/mqtt/button.py
View File

@ -6,7 +6,7 @@ import functools
import voluptuous as vol
from homeassistant.components import button
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_CLASS, CONF_NAME
from homeassistant.core import HomeAssistant
@ -104,6 +104,7 @@ class MqttButton(MqttEntity, ButtonEntity):
self._command_template = MqttCommandTemplate(
config.get(CONF_COMMAND_TEMPLATE), entity=self
).async_render
self._attr_device_class = self._config.get(CONF_DEVICE_CLASS)
def _prepare_subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
@ -111,11 +112,6 @@ class MqttButton(MqttEntity, ButtonEntity):
async def _subscribe_topics(self) -> None:
"""(Re)Subscribe to topics."""
@property
def device_class(self) -> ButtonDeviceClass | None:
"""Return the device class of the sensor."""
return self._config.get(CONF_DEVICE_CLASS)
async def async_press(self) -> None:
"""Turn the device on.

4
homeassistant/components/mqtt/client.py
View File

@ -303,8 +303,8 @@ class MqttClientSetup:
# Enable logging
self._client.enable_logger()
username = config.get(CONF_USERNAME)
password = config.get(CONF_PASSWORD)
username: str | None = config.get(CONF_USERNAME)
password: str | None = config.get(CONF_PASSWORD)
if username is not None:
self._client.username_pw_set(username, password)

1
homeassistant/components/mqtt/climate.py
View File

@ -749,6 +749,7 @@ class MqttClimate(MqttEntity, ClimateEntity):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
operation_mode: HVACMode | None
if (operation_mode := kwargs.get(ATTR_HVAC_MODE)) is not None:
await self.async_set_hvac_mode(operation_mode)

4
homeassistant/components/mqtt/siren.py
View File

@ -180,10 +180,10 @@ class MqttSiren(MqttEntity, SirenEntity):
def _setup_from_config(self, config: ConfigType) -> None:
"""(Re)Setup the entity."""
state_on = config.get(CONF_STATE_ON)
state_on: str | None = config.get(CONF_STATE_ON)
self._state_on = state_on if state_on else config[CONF_PAYLOAD_ON]
state_off = config.get(CONF_STATE_OFF)
state_off: str | None = config.get(CONF_STATE_OFF)
self._state_off = state_off if state_off else config[CONF_PAYLOAD_OFF]
self._attr_extra_state_attributes = {}

1
homeassistant/components/mqtt/vacuum/schema_legacy.py
View File

@ -220,6 +220,7 @@ class MqttVacuum(MqttEntity, VacuumEntity):
_entity_id_format = ENTITY_ID_FORMAT
_attributes_extra_blocked = MQTT_LEGACY_VACUUM_ATTRIBUTES_BLOCKED
_command_topic: str | None
_encoding: str | None
_qos: bool
_retain: bool

1
homeassistant/components/mqtt/vacuum/schema_state.py
View File

@ -180,6 +180,7 @@ class MqttStateVacuum(MqttEntity, StateVacuumEntity):
_entity_id_format = ENTITY_ID_FORMAT
_attributes_extra_blocked = MQTT_VACUUM_ATTRIBUTES_BLOCKED
_command_topic: str | None
_set_fan_speed_topic: str | None
_send_command_topic: str | None
_payloads: dict[str, str | None]