home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Enable strict type checks of cloud (#92576)

Browse Source 
* Enable strict type checks of cloud

* Fix circular import

* Address review comments

* Apply suggestions from code review
This commit is contained in:
Erik Montnemery
2023-05-24 11:46:11 +02:00
committed by GitHub
parent a8ce4d3c66
commit 937ebb374c
20 changed files with 324 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -87,6 +87,7 @@ homeassistant.components.camera.*
homeassistant.components.canary.*
homeassistant.components.clickatell.*
homeassistant.components.clicksend.*
homeassistant.components.cloud.*
homeassistant.components.configurator.*
homeassistant.components.cover.*
homeassistant.components.cpuspeed.*

12
homeassistant/components/alexa/config.py
View File

@ -1,4 +1,6 @@
"""Config helpers for Alexa."""
from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
import logging
@ -17,15 +19,15 @@ _LOGGER = logging.getLogger(__name__)
class AbstractConfig(ABC):
"""Hold the configuration for Alexa."""
_store: AlexaConfigStore
_unsub_proactive_report: CALLBACK_TYPE | None = None
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize abstract config."""
self.hass = hass
self._enable_proactive_mode_lock = asyncio.Lock()
self._store = None
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Perform async initialization of config."""
self._store = AlexaConfigStore(self.hass)
await self._store.async_load()
@ -65,7 +67,7 @@ class AbstractConfig(ABC):
def user_identifier(self):
"""Return an identifier for the user that represents this config."""
async def async_enable_proactive_mode(self):
async def async_enable_proactive_mode(self) -> None:
"""Enable proactive mode."""
_LOGGER.debug("Enable proactive mode")
async with self._enable_proactive_mode_lock:
@ -75,7 +77,7 @@ class AbstractConfig(ABC):
self.hass, self
)
async def async_disable_proactive_mode(self):
async def async_disable_proactive_mode(self) -> None:
"""Disable proactive mode."""
_LOGGER.debug("Disable proactive mode")
if unsub_func := self._unsub_proactive_report:
@ -105,7 +107,7 @@ class AbstractConfig(ABC):
"""Return authorization status."""
return self._store.authorized
async def set_authorized(self, authorized):
async def set_authorized(self, authorized) -> None:
"""Set authorization status.
- Set when an incoming message is received from Alexa.

13
homeassistant/components/alexa/state_report.py
View File

@ -5,7 +5,7 @@ import asyncio
from http import HTTPStatus
import json
import logging
from typing import cast
from typing import TYPE_CHECKING, cast
import aiohttp
import async_timeout
@ -23,6 +23,9 @@ from .entities import ENTITY_ADAPTERS, AlexaEntity, generate_alexa_id
from .errors import NoTokenAvailable, RequireRelink
from .messages import AlexaResponse
if TYPE_CHECKING:
from .config import AbstractConfig
_LOGGER = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 10
@ -188,7 +191,9 @@ async def async_send_changereport_message(
)
async def async_send_add_or_update_message(hass, config, entity_ids):
async def async_send_add_or_update_message(
hass: HomeAssistant, config: AbstractConfig, entity_ids: list[str]
) -> aiohttp.ClientResponse:
"""Send an AddOrUpdateReport message for entities.
https://developer.amazon.com/docs/device-apis/alexa-discovery.html#add-or-update-report
@ -223,7 +228,9 @@ async def async_send_add_or_update_message(hass, config, entity_ids):
)
async def async_send_delete_message(hass, config, entity_ids):
async def async_send_delete_message(
hass: HomeAssistant, config: AbstractConfig, entity_ids: list[str]
) -> aiohttp.ClientResponse:
"""Send an DeleteReport message for entities.
https://developer.amazon.com/docs/device-apis/alexa-discovery.html#deletereport-event

33
homeassistant/components/cloud/__init__.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from datetime import timedelta
from datetime import datetime, timedelta
from enum import Enum
from hass_nabucasa import Cloud
@ -18,7 +18,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HassJob, HomeAssistant, ServiceCall, callback
from homeassistant.core import Event, HassJob, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entityfilter
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -31,7 +31,6 @@ from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.service import async_register_admin_service
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from homeassistant.util.aiohttp import MockRequest
from . import account_link, http_api
from .client import CloudClient
@ -184,8 +183,10 @@ async def async_create_cloudhook(hass: HomeAssistant, webhook_id: str) -> str:
if not async_is_logged_in(hass):
raise CloudNotAvailable
hook = await hass.data[DOMAIN].cloudhooks.async_create(webhook_id, True)
return hook["cloudhook_url"]
cloud: Cloud[CloudClient] = hass.data[DOMAIN]
hook = await cloud.cloudhooks.async_create(webhook_id, True)
cloudhook_url: str = hook["cloudhook_url"]
return cloudhook_url
@bind_hass
@ -213,14 +214,6 @@ def async_remote_ui_url(hass: HomeAssistant) -> str:
return f"https://{remote_domain}"
def is_cloudhook_request(request):
"""Test if a request came from a cloudhook.
Async friendly.
"""
return isinstance(request, MockRequest)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Initialize the Home Assistant cloud."""
# Process configs
@ -243,7 +236,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
cloud = hass.data[DOMAIN] = Cloud(client, **kwargs)
cloud.iot.register_on_connect(client.on_cloud_connected)
async def _shutdown(event):
async def _shutdown(event: Event) -> None:
"""Shutdown event."""
await cloud.stop()
@ -263,7 +256,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass, DOMAIN, SERVICE_REMOTE_DISCONNECT, _service_handler
)
async def async_startup_repairs(_=None) -> None:
async def async_startup_repairs(_: datetime) -> None:
"""Create repair issues after startup."""
if not cloud.is_logged_in:
return
@ -273,7 +266,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
loaded = False
async def _on_start():
async def _on_start() -> None:
"""Discover platforms."""
nonlocal loaded
@ -292,19 +285,19 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
await async_load_platform(hass, Platform.TTS, DOMAIN, tts_info, config)
await asyncio.gather(stt_platform_loaded.wait(), tts_platform_loaded.wait())
async def _on_connect():
async def _on_connect() -> None:
"""Handle cloud connect."""
async_dispatcher_send(
hass, SIGNAL_CLOUD_CONNECTION_STATE, CloudConnectionState.CLOUD_CONNECTED
)
async def _on_disconnect():
async def _on_disconnect() -> None:
"""Handle cloud disconnect."""
async_dispatcher_send(
hass, SIGNAL_CLOUD_CONNECTION_STATE, CloudConnectionState.CLOUD_DISCONNECTED
)
async def _on_initialized():
async def _on_initialized() -> None:
"""Update preferences."""
await prefs.async_update(remote_domain=cloud.remote.instance_domain)
@ -330,7 +323,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@callback
def _remote_handle_prefs_updated(cloud: Cloud) -> None:
def _remote_handle_prefs_updated(cloud: Cloud[CloudClient]) -> None:
"""Handle remote preferences updated."""
cur_pref = cloud.client.prefs.remote_enabled
lock = asyncio.Lock()

22
homeassistant/components/cloud/account_link.py
View File

@ -1,5 +1,8 @@
"""Account linking via the cloud."""
from __future__ import annotations
import asyncio
from datetime import datetime
import logging
from typing import Any
@ -24,14 +27,16 @@ CURRENT_PLAIN_VERSION = AwesomeVersion(
@callback
def async_setup(hass: HomeAssistant):
def async_setup(hass: HomeAssistant) -> None:
"""Set up cloud account link."""
config_entry_oauth2_flow.async_add_implementation_provider(
hass, DOMAIN, async_provide_implementation
)
async def async_provide_implementation(hass: HomeAssistant, domain: str):
async def async_provide_implementation(
hass: HomeAssistant, domain: str
) -> list[config_entry_oauth2_flow.AbstractOAuth2Implementation]:
"""Provide an implementation for a domain."""
services = await _get_services(hass)
@ -55,9 +60,11 @@ async def async_provide_implementation(hass: HomeAssistant, domain: str):
return []
async def _get_services(hass):
async def _get_services(hass: HomeAssistant) -> list[dict[str, Any]]:
"""Get the available services."""
if (services := hass.data.get(DATA_SERVICES)) is not None:
services: list[dict[str, Any]]
if DATA_SERVICES in hass.data:
services = hass.data[DATA_SERVICES]
return services
try:
@ -68,7 +75,7 @@ async def _get_services(hass):
hass.data[DATA_SERVICES] = services
@callback
def clear_services(_now):
def clear_services(_now: datetime) -> None:
"""Clear services cache."""
hass.data.pop(DATA_SERVICES, None)
@ -102,7 +109,7 @@ class CloudOAuth2Implementation(config_entry_oauth2_flow.AbstractOAuth2Implement
)
authorize_url = await helper.async_get_authorize_url()
async def await_tokens():
async def await_tokens() -> None:
"""Wait for tokens and pass them on when received."""
try:
tokens = await helper.async_get_tokens()
@ -125,7 +132,8 @@ class CloudOAuth2Implementation(config_entry_oauth2_flow.AbstractOAuth2Implement
async def async_resolve_external_data(self, external_data: Any) -> dict:
"""Resolve external data to tokens."""
# We already passed in tokens
return external_data
dict_data: dict = external_data
return dict_data
async def _async_refresh_token(self, token: dict) -> dict:
"""Refresh a token."""

60
homeassistant/components/cloud/alexa_config.py
View File

@ -4,9 +4,10 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable
from contextlib import suppress
from datetime import timedelta
from datetime import datetime, timedelta
from http import HTTPStatus
import logging
from typing import TYPE_CHECKING, Any
import aiohttp
import async_timeout
@ -29,10 +30,11 @@ from homeassistant.components.homeassistant.exposed_entities import (
)
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import CLOUD_NEVER_EXPOSED_ENTITIES
from homeassistant.core import HomeAssistant, callback, split_entity_id
from homeassistant.core import Event, HomeAssistant, callback, split_entity_id
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er, start
from homeassistant.helpers.entity import get_device_class
from homeassistant.helpers.entityfilter import EntityFilter
from homeassistant.helpers.event import async_call_later
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
@ -47,6 +49,9 @@ from .const import (
)
from .prefs import ALEXA_SETTINGS_VERSION, CloudPreferences
if TYPE_CHECKING:
from .client import CloudClient
_LOGGER = logging.getLogger(__name__)
CLOUD_ALEXA = f"{CLOUD_DOMAIN}.{ALEXA_DOMAIN}"
@ -132,7 +137,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
config: dict,
cloud_user: str,
prefs: CloudPreferences,
cloud: Cloud,
cloud: Cloud[CloudClient],
) -> None:
"""Initialize the Alexa config."""
super().__init__(hass)
@ -141,13 +146,13 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
self._prefs = prefs
self._cloud = cloud
self._token = None
self._token_valid = None
self._token_valid: datetime | None = None
self._cur_entity_prefs = async_get_assistant_settings(hass, CLOUD_ALEXA)
self._alexa_sync_unsub: Callable[[], None] | None = None
self._endpoint = None
self._endpoint: Any = None
@property
def enabled(self):
def enabled(self) -> bool:
"""Return if Alexa is enabled."""
return (
self._cloud.is_logged_in
@ -156,12 +161,12 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
)
@property
def supports_auth(self):
def supports_auth(self) -> bool:
"""Return if config supports auth."""
return True
@property
def should_report_state(self):
def should_report_state(self) -> bool:
"""Return if states should be proactively reported."""
return (
self._prefs.alexa_enabled
@ -170,7 +175,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
)
@property
def endpoint(self):
def endpoint(self) -> Any | None:
"""Endpoint for report state."""
if self._endpoint is None:
raise ValueError("No endpoint available. Fetch access token first")
@ -178,22 +183,22 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
return self._endpoint
@property
def locale(self):
def locale(self) -> str:
"""Return config locale."""
# Not clear how to determine locale atm.
return "en-US"
@property
def entity_config(self):
def entity_config(self) -> dict[str, Any]:
"""Return entity config."""
return self._config.get(CONF_ENTITY_CONFIG) or {}
@callback
def user_identifier(self):
def user_identifier(self) -> str:
"""Return an identifier for the user that represents this config."""
return self._cloud_user
def _migrate_alexa_entity_settings_v1(self):
def _migrate_alexa_entity_settings_v1(self) -> None:
"""Migrate alexa entity settings to entity registry options."""
if not self._config[CONF_FILTER].empty_filter:
# Don't migrate if there's a YAML config
@ -210,11 +215,11 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
self._should_expose_legacy(entity_id),
)
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Initialize the Alexa config."""
await super().async_initialize()
async def on_hass_started(hass):
async def on_hass_started(hass: HomeAssistant) -> None:
if self._prefs.alexa_settings_version != ALEXA_SETTINGS_VERSION:
if self._prefs.alexa_settings_version < 2 or (
# Recover from a bug we had in 2023.5.0 where entities didn't get exposed
@ -235,7 +240,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
self.hass, CLOUD_ALEXA, self._async_exposed_entities_updated
)
async def on_hass_start(hass):
async def on_hass_start(hass: HomeAssistant) -> None:
if self.enabled and ALEXA_DOMAIN not in self.hass.config.components:
await async_setup_component(self.hass, ALEXA_DOMAIN, {})
@ -248,14 +253,14 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
self._handle_entity_registry_updated,
)
def _should_expose_legacy(self, entity_id):
def _should_expose_legacy(self, entity_id: str) -> bool:
"""If an entity should be exposed."""
if entity_id in CLOUD_NEVER_EXPOSED_ENTITIES:
return False
entity_configs = self._prefs.alexa_entity_configs
entity_config = entity_configs.get(entity_id, {})
entity_expose = entity_config.get(PREF_SHOULD_EXPOSE)
entity_expose: bool | None = entity_config.get(PREF_SHOULD_EXPOSE)
if entity_expose is not None:
return entity_expose
@ -279,21 +284,22 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
)
@callback
def should_expose(self, entity_id):
def should_expose(self, entity_id: str) -> bool:
"""If an entity should be exposed."""
if not self._config[CONF_FILTER].empty_filter:
entity_filter: EntityFilter = self._config[CONF_FILTER]
if not entity_filter.empty_filter:
if entity_id in CLOUD_NEVER_EXPOSED_ENTITIES:
return False
return self._config[CONF_FILTER](entity_id)
return entity_filter(entity_id)
return async_should_expose(self.hass, CLOUD_ALEXA, entity_id)
@callback
def async_invalidate_access_token(self):
def async_invalidate_access_token(self) -> None:
"""Invalidate access token."""
self._token_valid = None
async def async_get_access_token(self):
async def async_get_access_token(self) -> Any:
"""Get an access token."""
if self._token_valid is not None and self._token_valid > utcnow():
return self._token
@ -380,7 +386,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
self.hass, SYNC_DELAY, self._sync_prefs
)
async def _sync_prefs(self, _now):
async def _sync_prefs(self, _now: datetime) -> None:
"""Sync the updated preferences to Alexa."""
self._alexa_sync_unsub = None
old_prefs = self._cur_entity_prefs
@ -432,7 +438,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
if await self._sync_helper(to_update, to_remove):
self._cur_entity_prefs = new_prefs
async def async_sync_entities(self):
async def async_sync_entities(self) -> bool:
"""Sync all entities to Alexa."""
# Remove any pending sync
if self._alexa_sync_unsub:
@ -452,7 +458,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
return await self._sync_helper(to_update, to_remove)
async def _sync_helper(self, to_update, to_remove) -> bool:
async def _sync_helper(self, to_update: list[str], to_remove: list[str]) -> bool:
"""Sync entities to Alexa.
Return boolean if it was successful.
@ -497,7 +503,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
_LOGGER.warning("Error trying to sync entities to Alexa: %s", err)
return False
async def _handle_entity_registry_updated(self, event):
async def _handle_entity_registry_updated(self, event: Event) -> None:
"""Handle when entity registry updated."""
if not self.enabled or not self._cloud.is_logged_in:
return

11
homeassistant/components/cloud/binary_sensor.py
View File

@ -2,6 +2,10 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
from typing import Any
from hass_nabucasa import Cloud
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -13,6 +17,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .client import CloudClient
from .const import DISPATCHER_REMOTE_UPDATE, DOMAIN
WAIT_UNTIL_CHANGE = 3
@ -41,10 +46,10 @@ class CloudRemoteBinary(BinarySensorEntity):
_attr_unique_id = "cloud-remote-ui-connectivity"
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, cloud):
def __init__(self, cloud: Cloud[CloudClient]) -> None:
"""Initialize the binary sensor."""
self.cloud = cloud
self._unsub_dispatcher = None
self._unsub_dispatcher: Callable[[], None] | None = None
@property
def is_on(self) -> bool:
@ -59,7 +64,7 @@ class CloudRemoteBinary(BinarySensorEntity):
async def async_added_to_hass(self) -> None:
"""Register update dispatcher."""
async def async_state_update(data):
async def async_state_update(data: Any) -> None:
"""Update callback."""
await asyncio.sleep(WAIT_UNTIL_CHANGE)
self.async_write_ha_state()

17
homeassistant/components/cloud/client.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from datetime import datetime
from http import HTTPStatus
import logging
from pathlib import Path
@ -95,7 +96,9 @@ class CloudClient(Interface):
if self._alexa_config is None:
async with self._alexa_config_init_lock:
if self._alexa_config is not None:
return self._alexa_config
# This is reachable if the config was set while we waited
# for the lock
return self._alexa_config # type: ignore[unreachable]
cloud_user = await self._prefs.get_cloud_user()
@ -136,7 +139,7 @@ class CloudClient(Interface):
"""When cloud is connected."""
is_new_user = await self.prefs.async_set_username(self.cloud.username)
async def enable_alexa(_):
async def enable_alexa(_: Any) -> None:
"""Enable Alexa."""
aconf = await self.get_alexa_config()
try:
@ -156,7 +159,7 @@ class CloudClient(Interface):
enable_alexa_job = HassJob(enable_alexa, cancel_on_shutdown=True)
async def enable_google(_):
async def enable_google(_: datetime) -> None:
"""Enable Google."""
gconf = await self.get_google_config()
@ -210,7 +213,7 @@ class CloudClient(Interface):
"""Process cloud alexa message to client."""
cloud_user = await self._prefs.get_cloud_user()
aconfig = await self.get_alexa_config()
return await alexa_smart_home.async_handle_message(
return await alexa_smart_home.async_handle_message( # type: ignore[no-any-return, no-untyped-call]
self._hass,
aconfig,
payload,
@ -223,9 +226,11 @@ class CloudClient(Interface):
gconf = await self.get_google_config()
if not self._prefs.google_enabled:
return ga.api_disabled_response(payload, gconf.agent_user_id)
return ga.api_disabled_response( # type: ignore[no-any-return, no-untyped-call]
payload, gconf.agent_user_id
)
return await ga.async_handle_message(
return await ga.async_handle_message( # type: ignore[no-any-return, no-untyped-call]
self._hass, gconf, gconf.cloud_user, payload, google_assistant.SOURCE_CLOUD
)

58
homeassistant/components/cloud/google_config.py
View File

@ -1,8 +1,10 @@
"""Google config for Cloud."""
from __future__ import annotations
import asyncio
from http import HTTPStatus
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
from hass_nabucasa import Cloud, cloud_api
from hass_nabucasa.google_report_state import ErrorResponse
@ -24,12 +26,14 @@ from homeassistant.core import (
CoreState,
Event,
HomeAssistant,
State,
callback,
split_entity_id,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er, start
from homeassistant.helpers.entity import get_device_class
from homeassistant.helpers.entityfilter import EntityFilter
from homeassistant.setup import async_setup_component
from .const import (
@ -42,6 +46,9 @@ from .const import (
)
from .prefs import GOOGLE_SETTINGS_VERSION, CloudPreferences
if TYPE_CHECKING:
from .client import CloudClient
_LOGGER = logging.getLogger(__name__)
CLOUD_GOOGLE = f"{CLOUD_DOMAIN}.{GOOGLE_DOMAIN}"
@ -123,7 +130,7 @@ class CloudGoogleConfig(AbstractConfig):
config: dict[str, Any],
cloud_user: str,
prefs: CloudPreferences,
cloud: Cloud,
cloud: Cloud[CloudClient],
) -> None:
"""Initialize the Google config."""
super().__init__(hass)
@ -134,7 +141,7 @@ class CloudGoogleConfig(AbstractConfig):
self._sync_entities_lock = asyncio.Lock()
@property
def enabled(self):
def enabled(self) -> bool:
"""Return if Google is enabled."""
return (
self._cloud.is_logged_in
@ -143,34 +150,34 @@ class CloudGoogleConfig(AbstractConfig):
)
@property
def entity_config(self):
def entity_config(self) -> dict[str, Any]:
"""Return entity config."""
return self._config.get(CONF_ENTITY_CONFIG) or {}
@property
def secure_devices_pin(self):
def secure_devices_pin(self) -> str | None:
"""Return entity config."""
return self._prefs.google_secure_devices_pin
@property
def should_report_state(self):
def should_report_state(self) -> bool:
"""Return if states should be proactively reported."""
return self.enabled and self._prefs.google_report_state
def get_local_webhook_id(self, agent_user_id):
def get_local_webhook_id(self, agent_user_id: Any) -> str:
"""Return the webhook ID to be used for actions for a given agent user id via the local SDK."""
return self._prefs.google_local_webhook_id
def get_local_agent_user_id(self, webhook_id):
def get_local_agent_user_id(self, webhook_id: Any) -> str:
"""Return the user ID to be used for actions received via the local SDK."""
return self._user
@property
def cloud_user(self):
def cloud_user(self) -> str:
"""Return Cloud User account."""
return self._user
def _migrate_google_entity_settings_v1(self):
def _migrate_google_entity_settings_v1(self) -> None:
"""Migrate Google entity settings to entity registry options."""
if not self._config[CONF_FILTER].empty_filter:
# Don't migrate if there's a YAML config
@ -195,7 +202,7 @@ class CloudGoogleConfig(AbstractConfig):
_2fa_disabled,
)
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Perform async initialization of config."""
await super().async_initialize()
@ -246,18 +253,18 @@ class CloudGoogleConfig(AbstractConfig):
self._handle_device_registry_updated,
)
def should_expose(self, state):
def should_expose(self, state: State) -> bool:
"""If a state object should be exposed."""
return self._should_expose_entity_id(state.entity_id)
def _should_expose_legacy(self, entity_id):
def _should_expose_legacy(self, entity_id: str) -> bool:
"""If an entity ID should be exposed."""
if entity_id in CLOUD_NEVER_EXPOSED_ENTITIES:
return False
entity_configs = self._prefs.google_entity_configs
entity_config = entity_configs.get(entity_id, {})
entity_expose = entity_config.get(PREF_SHOULD_EXPOSE)
entity_expose: bool | None = entity_config.get(PREF_SHOULD_EXPOSE)
if entity_expose is not None:
return entity_expose
@ -282,36 +289,37 @@ class CloudGoogleConfig(AbstractConfig):
and _supported_legacy(self.hass, entity_id)
)
def _should_expose_entity_id(self, entity_id):
def _should_expose_entity_id(self, entity_id: str) -> bool:
"""If an entity should be exposed."""
if not self._config[CONF_FILTER].empty_filter:
entity_filter: EntityFilter = self._config[CONF_FILTER]
if not entity_filter.empty_filter:
if entity_id in CLOUD_NEVER_EXPOSED_ENTITIES:
return False
return self._config[CONF_FILTER](entity_id)
return entity_filter(entity_id)
return async_should_expose(self.hass, CLOUD_GOOGLE, entity_id)
@property
def agent_user_id(self):
def agent_user_id(self) -> str:
"""Return Agent User Id to use for query responses."""
return self._cloud.username
@property
def has_registered_user_agent(self):
def has_registered_user_agent(self) -> bool:
"""Return if we have a Agent User Id registered."""
return len(self._store.agent_user_ids) > 0
def get_agent_user_id(self, context):
def get_agent_user_id(self, context: Any) -> str:
"""Get agent user ID making request."""
return self.agent_user_id
def _2fa_disabled_legacy(self, entity_id):
def _2fa_disabled_legacy(self, entity_id: str) -> bool | None:
"""If an entity should be checked for 2FA."""
entity_configs = self._prefs.google_entity_configs
entity_config = entity_configs.get(entity_id, {})
return entity_config.get(PREF_DISABLE_2FA)
def should_2fa(self, state):
def should_2fa(self, state: State) -> bool:
"""If an entity should be checked for 2FA."""
try:
settings = async_get_entity_settings(self.hass, state.entity_id)
@ -322,14 +330,14 @@ class CloudGoogleConfig(AbstractConfig):
assistant_options = settings.get(CLOUD_GOOGLE, {})
return not assistant_options.get(PREF_DISABLE_2FA, DEFAULT_DISABLE_2FA)
async def async_report_state(self, message, agent_user_id: str):
async def async_report_state(self, message: Any, agent_user_id: str) -> None:
"""Send a state report to Google."""
try:
await self._cloud.google_report_state.async_send_message(message)
except ErrorResponse as err:
_LOGGER.warning("Error reporting state - %s: %s", err.code, err.message)
async def _async_request_sync_devices(self, agent_user_id: str):
async def _async_request_sync_devices(self, agent_user_id: str) -> int:
"""Trigger a sync with Google."""
if self._sync_entities_lock.locked():
return HTTPStatus.OK
@ -338,7 +346,7 @@ class CloudGoogleConfig(AbstractConfig):
resp = await cloud_api.async_google_actions_request_sync(self._cloud)
return resp.status
async def _async_prefs_updated(self, prefs):
async def _async_prefs_updated(self, prefs: CloudPreferences) -> None:
"""Handle updated preferences."""
if not self._cloud.is_logged_in:
if self.is_reporting_state:

77
homeassistant/components/cloud/http_api.py
View File

@ -1,14 +1,15 @@
"""The HTTP api to control the cloud integration."""
import asyncio
from collections.abc import Mapping
from collections.abc import Awaitable, Callable, Coroutine, Mapping
from contextlib import suppress
import dataclasses
from functools import wraps
from http import HTTPStatus
import logging
from typing import Any
from typing import Any, Concatenate, ParamSpec, TypeVar
import aiohttp
from aiohttp import web
import async_timeout
import attr
from hass_nabucasa import Cloud, auth, thingtalk
@ -32,6 +33,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.util.location import async_detect_location_info
from .alexa_config import entity_supported as entity_supported_by_alexa
from .client import CloudClient
from .const import (
DOMAIN,
PREF_ALEXA_REPORT_STATE,
@ -50,7 +52,7 @@ from .subscription import async_subscription_info
_LOGGER = logging.getLogger(__name__)
_CLOUD_ERRORS = {
_CLOUD_ERRORS: dict[type[Exception], tuple[HTTPStatus, str]] = {
asyncio.TimeoutError: (
HTTPStatus.BAD_GATEWAY,
"Unable to reach the Home Assistant cloud.",
@ -62,7 +64,7 @@ _CLOUD_ERRORS = {
}
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> None:
"""Initialize the HTTP API."""
websocket_api.async_register_command(hass, websocket_cloud_status)
websocket_api.async_register_command(hass, websocket_subscription)
@ -107,11 +109,21 @@ async def async_setup(hass):
)
def _handle_cloud_errors(handler):
_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView)
_P = ParamSpec("_P")
def _handle_cloud_errors(
handler: Callable[Concatenate[_HassViewT, web.Request, _P], Awaitable[web.Response]]
) -> Callable[
Concatenate[_HassViewT, web.Request, _P], Coroutine[Any, Any, web.Response]
]:
"""Webview decorator to handle auth errors."""
@wraps(handler)
async def error_handler(view, request, *args, **kwargs):
async def error_handler(
view: _HassViewT, request: web.Request, *args: _P.args, **kwargs: _P.kwargs
) -> web.Response:
"""Handle exceptions that raise from the wrapped request handler."""
try:
result = await handler(view, request, *args, **kwargs)
@ -126,25 +138,37 @@ def _handle_cloud_errors(handler):
return error_handler
def _ws_handle_cloud_errors(handler):
def _ws_handle_cloud_errors(
handler: Callable[
[HomeAssistant, websocket_api.ActiveConnection, dict[str, Any]],
Coroutine[None, None, None],
]
) -> Callable[
[HomeAssistant, websocket_api.ActiveConnection, dict[str, Any]],
Coroutine[None, None, None],
]:
"""Websocket decorator to handle auth errors."""
@wraps(handler)
async def error_handler(hass, connection, msg):
async def error_handler(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Handle exceptions that raise from the wrapped handler."""
try:
return await handler(hass, connection, msg)
except Exception as err: # pylint: disable=broad-except
err_status, err_msg = _process_cloud_exception(err, msg["type"])
connection.send_error(msg["id"], err_status, err_msg)
connection.send_error(msg["id"], str(err_status), err_msg)
return error_handler
def _process_cloud_exception(exc, where):
def _process_cloud_exception(exc: Exception, where: str) -> tuple[HTTPStatus, str]:
"""Process a cloud exception."""
err_info = None
err_info: tuple[HTTPStatus, str] | None = None
for err, value_info in _CLOUD_ERRORS.items():
if isinstance(exc, err):
@ -165,10 +189,10 @@ class GoogleActionsSyncView(HomeAssistantView):
name = "api:cloud:google_actions/sync"
@_handle_cloud_errors
async def post(self, request):
async def post(self, request: web.Request) -> web.Response:
"""Trigger a Google Actions sync."""
hass = request.app["hass"]
cloud: Cloud = hass.data[DOMAIN]
cloud: Cloud[CloudClient] = hass.data[DOMAIN]
gconf = await cloud.client.get_google_config()
status = await gconf.async_sync_entities(gconf.agent_user_id)
return self.json({}, status_code=status)
@ -184,7 +208,7 @@ class CloudLoginView(HomeAssistantView):
@RequestDataValidator(
vol.Schema({vol.Required("email"): str, vol.Required("password"): str})
)
async def post(self, request, data):
async def post(self, request: web.Request, data: dict[str, Any]) -> web.Response:
"""Handle login request."""
def cloud_assist_pipeline(hass: HomeAssistant) -> str | None:
@ -221,7 +245,7 @@ class CloudLogoutView(HomeAssistantView):
name = "api:cloud:logout"
@_handle_cloud_errors
async def post(self, request):
async def post(self, request: web.Request) -> web.Response:
"""Handle logout request."""
hass = request.app["hass"]
cloud = hass.data[DOMAIN]
@ -247,7 +271,7 @@ class CloudRegisterView(HomeAssistantView):
}
)
)
async def post(self, request, data):
async def post(self, request: web.Request, data: dict[str, Any]) -> web.Response:
"""Handle registration request."""
hass = request.app["hass"]
cloud = hass.data[DOMAIN]
@ -283,7 +307,7 @@ class CloudResendConfirmView(HomeAssistantView):
@_handle_cloud_errors
@RequestDataValidator(vol.Schema({vol.Required("email"): str}))
async def post(self, request, data):
async def post(self, request: web.Request, data: dict[str, Any]) -> web.Response:
"""Handle resending confirm email code request."""
hass = request.app["hass"]
cloud = hass.data[DOMAIN]
@ -302,7 +326,7 @@ class CloudForgotPasswordView(HomeAssistantView):
@_handle_cloud_errors
@RequestDataValidator(vol.Schema({vol.Required("email"): str}))
async def post(self, request, data):
async def post(self, request: web.Request, data: dict[str, Any]) -> web.Response:
"""Handle forgot password request."""
hass = request.app["hass"]
cloud = hass.data[DOMAIN]
@ -330,11 +354,20 @@ async def websocket_cloud_status(
)
def _require_cloud_login(handler):
def _require_cloud_login(
handler: Callable[
[HomeAssistant, websocket_api.ActiveConnection, dict[str, Any]],
None,
]
) -> Callable[[HomeAssistant, websocket_api.ActiveConnection, dict[str, Any]], None,]:
"""Websocket decorator that requires cloud to be logged in."""
@wraps(handler)
def with_cloud_auth(hass, connection, msg):
def with_cloud_auth(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Require to be logged into the cloud."""
cloud = hass.data[DOMAIN]
if not cloud.is_logged_in:
@ -467,7 +500,9 @@ async def websocket_hook_delete(
connection.send_message(websocket_api.result_message(msg["id"]))
async def _account_data(hass: HomeAssistant, cloud: Cloud):
async def _account_data(
hass: HomeAssistant, cloud: Cloud[CloudClient]
) -> dict[str, Any]:
"""Generate the auth data JSON response."""
assert hass.config.api

97
homeassistant/components/cloud/prefs.py
View File

@ -1,14 +1,15 @@
"""Preference management for cloud."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from typing import Any
from homeassistant.auth.const import GROUP_ID_ADMIN
from homeassistant.auth.models import User
from homeassistant.components import webhook
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.storage import Store
from homeassistant.helpers.typing import UNDEFINED
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from homeassistant.util.logging import async_create_catching_coro
from .const import (
@ -63,17 +64,20 @@ class CloudPreferencesStore(Store):
class CloudPreferences:
"""Handle cloud preferences."""
def __init__(self, hass):
_prefs: dict[str, Any]
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize cloud prefs."""
self._hass = hass
self._store = CloudPreferencesStore(
hass, STORAGE_VERSION, STORAGE_KEY, minor_version=STORAGE_VERSION_MINOR
)
self._prefs = None
self._listeners = []
self._listeners: list[
Callable[[CloudPreferences], Coroutine[Any, Any, None]]
] = []
self.last_updated: set[str] = set()
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Finish initializing the preferences."""
if (prefs := await self._store.async_load()) is None:
prefs = self._empty_config("")
@ -89,26 +93,28 @@ class CloudPreferences:
)
@callback
def async_listen_updates(self, listener):
def async_listen_updates(
self, listener: Callable[[CloudPreferences], Coroutine[Any, Any, None]]
) -> None:
"""Listen for updates to the preferences."""
self._listeners.append(listener)
async def async_update(
self,
*,
google_enabled=UNDEFINED,
alexa_enabled=UNDEFINED,
remote_enabled=UNDEFINED,
google_secure_devices_pin=UNDEFINED,
cloudhooks=UNDEFINED,
cloud_user=UNDEFINED,
alexa_report_state=UNDEFINED,
google_report_state=UNDEFINED,
tts_default_voice=UNDEFINED,
remote_domain=UNDEFINED,
alexa_settings_version=UNDEFINED,
google_settings_version=UNDEFINED,
):
google_enabled: bool | UndefinedType = UNDEFINED,
alexa_enabled: bool | UndefinedType = UNDEFINED,
remote_enabled: bool | UndefinedType = UNDEFINED,
google_secure_devices_pin: str | None | UndefinedType = UNDEFINED,
cloudhooks: dict[str, dict[str, str | bool]] | UndefinedType = UNDEFINED,
cloud_user: str | UndefinedType = UNDEFINED,
alexa_report_state: bool | UndefinedType = UNDEFINED,
google_report_state: bool | UndefinedType = UNDEFINED,
tts_default_voice: tuple[str, str] | UndefinedType = UNDEFINED,
remote_domain: str | None | UndefinedType = UNDEFINED,
alexa_settings_version: int | UndefinedType = UNDEFINED,
google_settings_version: int | UndefinedType = UNDEFINED,
) -> None:
"""Update user preferences."""
prefs = {**self._prefs}
@ -131,7 +137,7 @@ class CloudPreferences:
await self._save_prefs(prefs)
async def async_set_username(self, username) -> bool:
async def async_set_username(self, username: str | None) -> bool:
"""Set the username that is logged in."""
# Logging out.
if username is None:
@ -154,7 +160,7 @@ class CloudPreferences:
return True
def as_dict(self):
def as_dict(self) -> dict[str, Any]:
"""Return dictionary version."""
return {
PREF_ALEXA_DEFAULT_EXPOSE: self.alexa_default_expose,
@ -170,7 +176,7 @@ class CloudPreferences:
}
@property
def remote_enabled(self):
def remote_enabled(self) -> bool:
"""Return if remote is enabled on start."""
if not self._prefs.get(PREF_ENABLE_REMOTE, False):
return False
@ -178,17 +184,18 @@ class CloudPreferences:
return True
@property
def remote_domain(self):
def remote_domain(self) -> str | None:
"""Return remote domain."""
return self._prefs.get(PREF_REMOTE_DOMAIN)
@property
def alexa_enabled(self):
def alexa_enabled(self) -> bool:
"""Return if Alexa is enabled."""
return self._prefs[PREF_ENABLE_ALEXA]
alexa_enabled: bool = self._prefs[PREF_ENABLE_ALEXA]
return alexa_enabled
@property
def alexa_report_state(self):
def alexa_report_state(self) -> bool:
"""Return if Alexa report state is enabled."""
return self._prefs.get(PREF_ALEXA_REPORT_STATE, DEFAULT_ALEXA_REPORT_STATE)
@ -201,44 +208,48 @@ class CloudPreferences:
return self._prefs.get(PREF_ALEXA_DEFAULT_EXPOSE)
@property
def alexa_entity_configs(self):
def alexa_entity_configs(self) -> dict[str, Any]:
"""Return Alexa Entity configurations."""
return self._prefs.get(PREF_ALEXA_ENTITY_CONFIGS, {})
@property
def alexa_settings_version(self):
def alexa_settings_version(self) -> int:
"""Return version of Alexa settings."""
return self._prefs[PREF_ALEXA_SETTINGS_VERSION]
alexa_settings_version: int = self._prefs[PREF_ALEXA_SETTINGS_VERSION]
return alexa_settings_version
@property
def google_enabled(self):
def google_enabled(self) -> bool:
"""Return if Google is enabled."""
return self._prefs[PREF_ENABLE_GOOGLE]
google_enabled: bool = self._prefs[PREF_ENABLE_GOOGLE]
return google_enabled
@property
def google_report_state(self):
def google_report_state(self) -> bool:
"""Return if Google report state is enabled."""
return self._prefs.get(PREF_GOOGLE_REPORT_STATE, DEFAULT_GOOGLE_REPORT_STATE)
@property
def google_secure_devices_pin(self):
def google_secure_devices_pin(self) -> str | None:
"""Return if Google is allowed to unlock locks."""
return self._prefs.get(PREF_GOOGLE_SECURE_DEVICES_PIN)
@property
def google_entity_configs(self):
def google_entity_configs(self) -> dict[str, dict[str, Any]]:
"""Return Google Entity configurations."""
return self._prefs.get(PREF_GOOGLE_ENTITY_CONFIGS, {})
@property
def google_settings_version(self):
def google_settings_version(self) -> int:
"""Return version of Google settings."""
return self._prefs[PREF_GOOGLE_SETTINGS_VERSION]
google_settings_version: int = self._prefs[PREF_GOOGLE_SETTINGS_VERSION]
return google_settings_version
@property
def google_local_webhook_id(self):
def google_local_webhook_id(self) -> str:
"""Return Google webhook ID to receive local messages."""
return self._prefs[PREF_GOOGLE_LOCAL_WEBHOOK_ID]
google_local_webhook_id: str = self._prefs[PREF_GOOGLE_LOCAL_WEBHOOK_ID]
return google_local_webhook_id
@property
def google_default_expose(self) -> list[str] | None:
@ -249,12 +260,12 @@ class CloudPreferences:
return self._prefs.get(PREF_GOOGLE_DEFAULT_EXPOSE)
@property
def cloudhooks(self):
def cloudhooks(self) -> dict[str, Any]:
"""Return the published cloud webhooks."""
return self._prefs.get(PREF_CLOUDHOOKS, {})
@property
def tts_default_voice(self):
def tts_default_voice(self) -> tuple[str, str]:
"""Return the default TTS voice."""
return self._prefs.get(PREF_TTS_DEFAULT_VOICE, DEFAULT_TTS_DEFAULT_VOICE)
@ -281,7 +292,7 @@ class CloudPreferences:
# an image was restored without restoring the cloud prefs.
return await self._hass.auth.async_get_user(user_id)
async def _save_prefs(self, prefs):
async def _save_prefs(self, prefs: dict[str, Any]) -> None:
"""Save preferences to disk."""
self.last_updated = {
key for key, value in prefs.items() if value != self._prefs.get(key)
@ -294,7 +305,7 @@ class CloudPreferences:
@callback
@staticmethod
def _empty_config(username):
def _empty_config(username: str) -> dict[str, Any]:
"""Return an empty config."""
return {
PREF_ALEXA_DEFAULT_EXPOSE: DEFAULT_EXPOSED_DOMAINS,

3
homeassistant/components/cloud/repairs.py
View File

@ -12,6 +12,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import issue_registry as ir
from .client import CloudClient
from .const import DOMAIN
from .subscription import async_migrate_paypal_agreement, async_subscription_info
@ -67,7 +68,7 @@ class LegacySubscriptionRepairFlow(RepairsFlow):
async def async_step_change_plan(self, _: None = None) -> FlowResult:
"""Wait for the user to authorize the app installation."""
cloud: Cloud = self.hass.data[DOMAIN]
cloud: Cloud[CloudClient] = self.hass.data[DOMAIN]
async def _async_wait_for_plan_change() -> None:
flow_manager = repairs_flow_manager(self.hass)

13
homeassistant/components/cloud/stt.py
View File

@ -18,15 +18,22 @@ from homeassistant.components.stt import (
SpeechResult,
SpeechResultState,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .client import CloudClient
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_get_engine(hass, config, discovery_info=None):
async def async_get_engine(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> CloudProvider:
"""Set up Cloud speech component."""
cloud: Cloud = hass.data[DOMAIN]
cloud: Cloud[CloudClient] = hass.data[DOMAIN]
cloud_provider = CloudProvider(cloud)
if discovery_info is not None:
@ -37,7 +44,7 @@ async def async_get_engine(hass, config, discovery_info=None):
class CloudProvider(Provider):
"""NabuCasa speech API provider."""
def __init__(self, cloud: Cloud) -> None:
def __init__(self, cloud: Cloud[CloudClient]) -> None:
"""Home Assistant NabuCasa Speech to text."""
self.cloud = cloud

7
homeassistant/components/cloud/subscription.py
View File

@ -9,12 +9,13 @@ from aiohttp.client_exceptions import ClientError
import async_timeout
from hass_nabucasa import Cloud, cloud_api
from .client import CloudClient
from .const import REQUEST_TIMEOUT
_LOGGER = logging.getLogger(__name__)
async def async_subscription_info(cloud: Cloud) -> dict[str, Any] | None:
async def async_subscription_info(cloud: Cloud[CloudClient]) -> dict[str, Any] | None:
"""Fetch the subscription info."""
try:
async with async_timeout.timeout(REQUEST_TIMEOUT):
@ -33,7 +34,9 @@ async def async_subscription_info(cloud: Cloud) -> dict[str, Any] | None:
return None
async def async_migrate_paypal_agreement(cloud: Cloud) -> dict[str, Any] | None:
async def async_migrate_paypal_agreement(
cloud: Cloud[CloudClient],
) -> dict[str, Any] | None:
"""Migrate a paypal agreement from legacy."""
try:
async with async_timeout.timeout(REQUEST_TIMEOUT):

10
homeassistant/components/cloud/system_health.py
View File

@ -1,4 +1,6 @@
"""Provide info to system health."""
from typing import Any
from hass_nabucasa import Cloud
from homeassistant.components import system_health
@ -16,12 +18,12 @@ def async_register(
register.async_register_info(system_health_info, "/config/cloud")
async def system_health_info(hass):
async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
"""Get info for the info page."""
cloud: Cloud = hass.data[DOMAIN]
client: CloudClient = cloud.client
cloud: Cloud[CloudClient] = hass.data[DOMAIN]
client = cloud.client
data = {
data: dict[str, Any] = {
"logged_in": cloud.is_logged_in,
}

39
homeassistant/components/cloud/tts.py
View File

@ -1,6 +1,8 @@
"""Support for the cloud for text to speech service."""
from __future__ import annotations
import logging
from typing import Any
from hass_nabucasa import Cloud
from hass_nabucasa.voice import MAP_VOICE, TTS_VOICES, AudioOutput, VoiceError
@ -12,11 +14,15 @@ from homeassistant.components.tts import (
CONF_LANG,
PLATFORM_SCHEMA,
Provider,
TtsAudioType,
Voice,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .client import CloudClient
from .const import DOMAIN
from .prefs import CloudPreferences
ATTR_GENDER = "gender"
@ -25,7 +31,7 @@ SUPPORT_LANGUAGES = list(TTS_VOICES)
_LOGGER = logging.getLogger(__name__)
def validate_lang(value):
def validate_lang(value: dict[str, Any]) -> dict[str, Any]:
"""Validate chosen gender or language."""
if (lang := value.get(CONF_LANG)) is None:
return value
@ -52,10 +58,16 @@ PLATFORM_SCHEMA = vol.All(
)
async def async_get_engine(hass, config, discovery_info=None):
async def async_get_engine(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> CloudProvider:
"""Set up Cloud speech component."""
cloud: Cloud = hass.data[DOMAIN]
cloud: Cloud[CloudClient] = hass.data[DOMAIN]
language: str | None
gender: str | None
if discovery_info is not None:
language = None
gender = None
@ -72,7 +84,9 @@ async def async_get_engine(hass, config, discovery_info=None):
class CloudProvider(Provider):
"""NabuCasa Cloud speech API provider."""
def __init__(self, cloud: Cloud, language: str, gender: str) -> None:
def __init__(
self, cloud: Cloud[CloudClient], language: str | None, gender: str | None
) -> None:
"""Initialize cloud provider."""
self.cloud = cloud
self.name = "Cloud"
@ -85,22 +99,22 @@ class CloudProvider(Provider):
self._language, self._gender = cloud.client.prefs.tts_default_voice
cloud.client.prefs.async_listen_updates(self._sync_prefs)
async def _sync_prefs(self, prefs):
async def _sync_prefs(self, prefs: CloudPreferences) -> None:
"""Sync preferences."""
self._language, self._gender = prefs.tts_default_voice
@property
def default_language(self):
def default_language(self) -> str | None:
"""Return the default language."""
return self._language
@property
def supported_languages(self):
def supported_languages(self) -> list[str]:
"""Return list of supported languages."""
return SUPPORT_LANGUAGES
@property
def supported_options(self):
def supported_options(self) -> list[str]:
"""Return list of supported options like voice, emotion."""
return [ATTR_GENDER, ATTR_VOICE, ATTR_AUDIO_OUTPUT]
@ -112,17 +126,20 @@ class CloudProvider(Provider):
return [Voice(voice, voice) for voice in voices]
@property
def default_options(self):
def default_options(self) -> dict[str, Any]:
"""Return a dict include default options."""
return {
ATTR_GENDER: self._gender,
ATTR_AUDIO_OUTPUT: AudioOutput.MP3,
}
async def async_get_tts_audio(self, message, language, options=None):
async def async_get_tts_audio(
self, message: str, language: str, options: dict[str, Any] | None = None
) -> TtsAudioType:
"""Load TTS from NabuCasa Cloud."""
# Process TTS
try:
assert options is not None
data = await self.cloud.voice.process_tts(
text=message,
language=language,

26
homeassistant/components/google_assistant/helpers.py
View File

@ -21,7 +21,7 @@ from homeassistant.const import (
CONF_NAME,
STATE_UNAVAILABLE,
)
from homeassistant.core import Context, HomeAssistant, State, callback
from homeassistant.core import CALLBACK_TYPE, Context, HomeAssistant, State, callback
from homeassistant.helpers import (
area_registry as ar,
device_registry as dr,
@ -86,19 +86,19 @@ def _get_registry_entries(
class AbstractConfig(ABC):
"""Hold the configuration for Google Assistant."""
_store: GoogleConfigStore
_unsub_report_state: Callable[[], None] | None = None
def __init__(self, hass):
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize abstract config."""
self.hass = hass
self._store = None
self._google_sync_unsub = {}
self._google_sync_unsub: dict[str, CALLBACK_TYPE] = {}
self._local_sdk_active = False
self._local_last_active: datetime | None = None
self._local_sdk_version_warn = False
self.is_supported_cache: dict[str, tuple[int | None, bool]] = {}
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Perform async initialization of config."""
self._store = GoogleConfigStore(self.hass)
await self._store.async_initialize()
@ -195,7 +195,7 @@ class AbstractConfig(ABC):
await gather(*jobs)
@callback
def async_enable_report_state(self):
def async_enable_report_state(self) -> None:
"""Enable proactive mode."""
# Circular dep
# pylint: disable-next=import-outside-toplevel
@ -205,7 +205,7 @@ class AbstractConfig(ABC):
self._unsub_report_state = async_enable_report_state(self.hass, self)
@callback
def async_disable_report_state(self):
def async_disable_report_state(self) -> None:
"""Disable report state."""
if self._unsub_report_state is not None:
self._unsub_report_state()
@ -220,7 +220,7 @@ class AbstractConfig(ABC):
await self.async_disconnect_agent_user(agent_user_id)
return status
async def async_sync_entities_all(self):
async def async_sync_entities_all(self) -> int:
"""Sync all entities to Google for all registered agents."""
if not self._store.agent_user_ids:
return 204
@ -249,7 +249,7 @@ class AbstractConfig(ABC):
)
@callback
def async_schedule_google_sync_all(self):
def async_schedule_google_sync_all(self) -> None:
"""Schedule a sync for all registered agents."""
for agent_user_id in self._store.agent_user_ids:
self.async_schedule_google_sync(agent_user_id)
@ -279,7 +279,7 @@ class AbstractConfig(ABC):
self._store.pop_agent_user_id(agent_user_id)
@callback
def async_enable_local_sdk(self):
def async_enable_local_sdk(self) -> None:
"""Enable the local SDK."""
setup_successful = True
setup_webhook_ids = []
@ -323,7 +323,7 @@ class AbstractConfig(ABC):
self._local_sdk_active = setup_successful
@callback
def async_disable_local_sdk(self):
def async_disable_local_sdk(self) -> None:
"""Disable the local SDK."""
if not self._local_sdk_active:
return
@ -500,7 +500,7 @@ class GoogleEntity:
self.hass = hass
self.config = config
self.state = state
self._traits = None
self._traits: list[trait._Trait] | None = None
@property
def entity_id(self):
@ -508,7 +508,7 @@ class GoogleEntity:
return self.state.entity_id
@callback
def traits(self):
def traits(self) -> list[trait._Trait]:
"""Return traits for entity."""
if self._traits is not None:
return self._traits

9
homeassistant/components/google_assistant/trait.py
View File

@ -1,6 +1,7 @@
"""Implement the Google Smart Home traits."""
from __future__ import annotations
from abc import ABC, abstractmethod
import logging
from typing import Any, TypeVar
@ -196,9 +197,10 @@ def _next_selected(items: list[str], selected: str | None) -> str | None:
return items[next_item]
class _Trait:
class _Trait(ABC):
"""Represents a Trait inside Google Assistant skill."""
name: str
commands: list[str] = []
@staticmethod
@ -206,6 +208,11 @@ class _Trait:
"""Return if the trait might ask for 2FA."""
return False
@staticmethod
@abstractmethod
def supported(domain, features, device_class, attributes):
"""Test if state is supported."""
def __init__(self, hass, state, config):
"""Initialize a trait for a state."""
self.hass = hass

10
mypy.ini
View File

@ -631,6 +631,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.cloud.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.configurator.*]
check_untyped_defs = true
disallow_incomplete_defs = true

2
tests/components/cloud/test_init.py
View File

@ -135,7 +135,7 @@ async def test_setup_existing_cloud_user(
async def test_on_connect(hass: HomeAssistant, mock_cloud_fixture) -> None:
"""Test cloud on connect triggers."""
cl: Cloud = hass.data["cloud"]
cl: Cloud[cloud.client.CloudClient] = hass.data["cloud"]
assert len(cl.iot._on_connect) == 4