mirror of
https://github.com/home-assistant/core
synced 2024-08-28 03:36:46 +02:00
Improve flume typing (#107444)
This commit is contained in:
parent
cecb12a93c
commit
cd8adfc84e
@ -1,4 +1,6 @@
|
||||
"""The flume integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pyflume import FlumeAuth, FlumeDeviceList
|
||||
from requests import Session
|
||||
from requests.exceptions import RequestException
|
||||
@ -41,7 +43,9 @@ LIST_NOTIFICATIONS_SERVICE_SCHEMA = vol.All(
|
||||
)
|
||||
|
||||
|
||||
def _setup_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
def _setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry
|
||||
) -> tuple[FlumeAuth, FlumeDeviceList, Session]:
|
||||
"""Config entry set up in executor."""
|
||||
config = entry.data
|
||||
|
||||
|
@ -1,4 +1,6 @@
|
||||
"""Config flow for flume integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
import logging
|
||||
import os
|
||||
@ -36,7 +38,9 @@ DATA_SCHEMA = vol.Schema(
|
||||
)
|
||||
|
||||
|
||||
def _validate_input(hass: core.HomeAssistant, data: dict, clear_token_file: bool):
|
||||
def _validate_input(
|
||||
hass: core.HomeAssistant, data: dict[str, Any], clear_token_file: bool
|
||||
) -> FlumeDeviceList:
|
||||
"""Validate in the executor."""
|
||||
flume_token_full_path = hass.config.path(
|
||||
f"{BASE_TOKEN_FILENAME}-{data[CONF_USERNAME]}"
|
||||
@ -56,8 +60,8 @@ def _validate_input(hass: core.HomeAssistant, data: dict, clear_token_file: bool
|
||||
|
||||
|
||||
async def validate_input(
|
||||
hass: core.HomeAssistant, data: dict, clear_token_file: bool = False
|
||||
):
|
||||
hass: core.HomeAssistant, data: dict[str, Any], clear_token_file: bool = False
|
||||
) -> dict[str, Any]:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
@ -85,11 +89,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Init flume config flow."""
|
||||
self._reauth_unique_id = None
|
||||
self._reauth_unique_id: str | None = None
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
await self.async_set_unique_id(user_input[CONF_USERNAME])
|
||||
self._abort_if_unique_id_configured()
|
||||
@ -111,10 +117,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
self._reauth_unique_id = self.context["unique_id"]
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(self, user_input=None):
|
||||
async def async_step_reauth_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle reauth input."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
existing_entry = await self.async_set_unique_id(self._reauth_unique_id)
|
||||
assert existing_entry
|
||||
if user_input is not None:
|
||||
new_data = {**existing_entry.data, CONF_PASSWORD: user_input[CONF_PASSWORD]}
|
||||
try:
|
||||
|
@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
from typing import Any
|
||||
|
||||
import pyflume
|
||||
from pyflume import FlumeDeviceList
|
||||
from pyflume import FlumeAuth, FlumeData, FlumeDeviceList
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
@ -21,7 +21,7 @@ from .const import (
|
||||
class FlumeDeviceDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
"""Data update coordinator for an individual flume device."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, flume_device) -> None:
|
||||
def __init__(self, hass: HomeAssistant, flume_device: FlumeData) -> None:
|
||||
"""Initialize the Coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
@ -79,7 +79,7 @@ class FlumeDeviceConnectionUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
class FlumeNotificationDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
"""Data update coordinator for flume notifications."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, auth) -> None:
|
||||
def __init__(self, hass: HomeAssistant, auth: FlumeAuth) -> None:
|
||||
"""Initialize the Coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
@ -88,15 +88,15 @@ class FlumeNotificationDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
update_interval=NOTIFICATION_SCAN_INTERVAL,
|
||||
)
|
||||
self.auth = auth
|
||||
self.active_notifications_by_device: dict = {}
|
||||
self.notifications: list[dict[str, Any]]
|
||||
self.active_notifications_by_device: dict[str, set[str]] = {}
|
||||
self.notifications: list[dict[str, Any]] = []
|
||||
|
||||
def _update_lists(self):
|
||||
def _update_lists(self) -> None:
|
||||
"""Query flume for notification list."""
|
||||
# Get notifications (read or unread).
|
||||
# The related binary sensors (leak detected, high flow, low battery)
|
||||
# will be active until the notification is deleted in the Flume app.
|
||||
self.notifications: list[dict[str, Any]] = pyflume.FlumeNotificationList(
|
||||
self.notifications = pyflume.FlumeNotificationList(
|
||||
self.auth, read=None
|
||||
).notification_list
|
||||
_LOGGER.debug("Notifications %s", self.notifications)
|
||||
|
@ -58,7 +58,7 @@ class FlumeEntity(CoordinatorEntity[_FlumeCoordinatorT]):
|
||||
configuration_url="https://portal.flumewater.com",
|
||||
)
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Request an update when added."""
|
||||
await super().async_added_to_hass()
|
||||
# We do not ask for an update with async_add_entities()
|
||||
|
@ -12,6 +12,7 @@ from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import UnitOfVolume
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
|
||||
from .const import (
|
||||
DEVICE_SCAN_INTERVAL,
|
||||
@ -139,7 +140,7 @@ class FlumeSensor(FlumeEntity[FlumeDeviceDataUpdateCoordinator], SensorEntity):
|
||||
"""Representation of the Flume sensor."""
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
def native_value(self) -> StateType:
|
||||
"""Return the state of the sensor."""
|
||||
sensor_key = self.entity_description.key
|
||||
if sensor_key not in self.coordinator.flume_device.values:
|
||||
|
Loading…
Reference in New Issue
Block a user